From 0d713e3eba91c1102525c171c44df86c9924ca39 Mon Sep 17 00:00:00 2001
From: Lingxuan Zuo <skyzlxuan@gmail.com>
Date: Fri, 10 Apr 2020 11:56:59 +0800
Subject: [PATCH] [Streaming] Try to trigger mock transfer tests ci (#7885)

* try to trigger mock transfer tests ci

* execute transfer tests

* show all logs when bazel test streaming

* temporary repeated ci runs

* Revert "temporary repeated ci runs"

This reverts commit dc77d2f9f79b5fa7b490221a8e9089e6349e067d.
---
 .travis.yml                               |  2 +-
 streaming/src/event_service.cc            |  4 ++--
 streaming/src/test/mock_transfer_tests.cc | 19 +++++++++++--------
 3 files changed, 14 insertions(+), 11 deletions(-)

diff --git a/.travis.yml b/.travis.yml
index 1f3b19ae6..e17d52ee1 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -43,7 +43,7 @@ matrix:
         - ./ci/suppress_output ./ci/travis/install-ray.sh
       script:
         # Streaming cpp test.
-        - if [ $RAY_CI_STREAMING_CPP_AFFECTED == "1" ]; then ./ci/suppress_output bazel test //streaming:all && bash streaming/src/test/run_streaming_queue_test.sh; fi
+        - if [ $RAY_CI_STREAMING_CPP_AFFECTED == "1" ]; then bazel test --test_output=all //streaming:all && bash streaming/src/test/run_streaming_queue_test.sh; fi
         - if [ $RAY_CI_STREAMING_PYTHON_AFFECTED == "1" ]; then python -m pytest -v --durations=5 --timeout=300 streaming/python/tests/; fi
         - if [ $RAY_CI_STREAMING_JAVA_AFFECTED == "1" ]; then ./streaming/java/test.sh; fi
 
diff --git a/streaming/src/event_service.cc b/streaming/src/event_service.cc
index 04377871c..2e788c9fe 100644
--- a/streaming/src/event_service.cc
+++ b/streaming/src/event_service.cc
@@ -182,7 +182,7 @@ void EventService::RemoveDestroyedChannelEvent(const std::vector<ObjectID> &remo
   STREAMING_LOG(INFO) << "Remove Destroyed channel event, removed_ids size "
                       << removed_ids.size() << ", total event size " << total_event_nums;
   size_t removed_related_num = 0;
-  event_queue_->Freeze();
+  event_queue_->Unfreeze();
   for (size_t i = 0; i < total_event_nums; ++i) {
     Event event;
     if (!event_queue_->Get(event) || !event.channel_info) {
@@ -196,7 +196,7 @@ void EventService::RemoveDestroyedChannelEvent(const std::vector<ObjectID> &remo
     }
     event_queue_->Pop();
   }
-  event_queue_->Unfreeze();
+  event_queue_->Freeze();
   STREAMING_LOG(INFO) << "Total event num => " << total_event_nums
                       << ", removed related num => " << removed_related_num;
 }
diff --git a/streaming/src/test/mock_transfer_tests.cc b/streaming/src/test/mock_transfer_tests.cc
index f5cfcecd6..012c9edd5 100644
--- a/streaming/src/test/mock_transfer_tests.cc
+++ b/streaming/src/test/mock_transfer_tests.cc
@@ -37,10 +37,12 @@ TEST(StreamingMockTransfer, mock_produce_consume) {
 class StreamingTransferTest : public ::testing::Test {
  public:
   StreamingTransferTest() {
-    runtime_context = std::make_shared<RuntimeContext>();
-    runtime_context->MarkMockTest();
-    writer = std::make_shared<DataWriter>(runtime_context);
-    reader = std::make_shared<DataReader>(runtime_context);
+    writer_runtime_context = std::make_shared<RuntimeContext>();
+    reader_runtime_context = std::make_shared<RuntimeContext>();
+    writer_runtime_context->MarkMockTest();
+    reader_runtime_context->MarkMockTest();
+    writer = std::make_shared<DataWriter>(writer_runtime_context);
+    reader = std::make_shared<DataReader>(reader_runtime_context);
   }
   virtual ~StreamingTransferTest() = default;
   void InitTransfer(int channel_num = 1) {
@@ -64,7 +66,8 @@ class StreamingTransferTest : public ::testing::Test {
   std::shared_ptr<DataWriter> writer;
   std::shared_ptr<DataReader> reader;
   std::vector<ObjectID> queue_vec;
-  std::shared_ptr<RuntimeContext> runtime_context;
+  std::shared_ptr<RuntimeContext> writer_runtime_context;
+  std::shared_ptr<RuntimeContext> reader_runtime_context;
 };
 
 TEST_F(StreamingTransferTest, exchange_single_channel_test) {
@@ -149,8 +152,8 @@ TEST_F(StreamingTransferTest, flow_control_test) {
   std::unordered_map<ObjectID, ConsumerChannelInfo> *reader_offset_info = nullptr;
   writer->GetOffsetInfo(writer_offset_info);
   reader->GetOffsetInfo(reader_offset_info);
-  uint32_t writer_step = runtime_context->GetConfig().GetWriterConsumedStep();
-  uint32_t reader_step = runtime_context->GetConfig().GetReaderConsumedStep();
+  uint32_t writer_step = writer_runtime_context->GetConfig().GetWriterConsumedStep();
+  uint32_t reader_step = reader_runtime_context->GetConfig().GetReaderConsumedStep();
   uint64_t &writer_current_seq_id = (*writer_offset_info)[queue_vec[0]].current_seq_id;
   uint64_t &writer_current_message_id =
       (*writer_offset_info)[queue_vec[0]].current_message_id;
@@ -167,7 +170,7 @@ TEST_F(StreamingTransferTest, flow_control_test) {
   std::list<StreamingMessagePtr> read_message_list;
   while (read_message_list.size() < num) {
     std::shared_ptr<DataBundle> msg;
-    reader->GetBundle(5000, msg);
+    reader->GetBundle(1000, msg);
     StreamingMessageBundlePtr bundle_ptr = StreamingMessageBundle::FromBytes(msg->data);
     auto &message_list = bundle_ptr->GetMessageList();
     std::copy(message_list.begin(), message_list.end(),