Add service based gcs client (#6686)

streaming/src/test/queue_tests_base.h

@@ -14,9 +14,10 @@ static void flushall_redis(void) {
 class StreamingQueueTestBase : public ::testing::TestWithParam<uint64_t> {
  public:
   StreamingQueueTestBase(int num_nodes, std::string raylet_exe, std::string store_exe,
-                         int port, std::string actor_exe)
+                         int port, std::string actor_exe, std::string gcs_server_exe)
       : gcs_options_("127.0.0.1", 6379, ""),
         raylet_executable_(raylet_exe),
+        gcs_server_executable_(gcs_server_exe),
         store_executable_(store_exe),
         actor_executable_(actor_exe),
         node_manager_port_(port) {
@@ -34,6 +35,9 @@ class StreamingQueueTestBase : public ::testing::TestWithParam<uint64_t> {
       store_socket = StartStore();
     }
 
+    // start gcs server
+    gcs_server_pid_ = StartGcsServer("127.0.0.1");
+
     // start raylet on each node. Assign each node with different resources so that
     // a task can be scheduled to the desired node.
     for (int i = 0; i < num_nodes; i++) {
@@ -52,6 +56,8 @@ class StreamingQueueTestBase : public ::testing::TestWithParam<uint64_t> {
     for (const auto &store_socket : raylet_store_socket_names_) {
       StopStore(store_socket);
     }
+
+    StopGcsServer(gcs_server_pid_);
   }
 
   JobID NextJobId() const {
@@ -80,6 +86,30 @@ class StreamingQueueTestBase : public ::testing::TestWithParam<uint64_t> {
     ASSERT_EQ(system(("rm -rf " + store_socket_name + ".pid").c_str()), 0);
   }
 
+  std::string StartGcsServer(std::string redis_address) {
+    std::string gcs_server_socket_name = "/tmp/gcs_server" + ObjectID::FromRandom().Hex();
+    std::string ray_start_cmd = gcs_server_executable_;
+    ray_start_cmd.append(" --redis_address=" + redis_address)
+        .append(" --redis_port=6379")
+        .append(" --config_list=initial_reconstruction_timeout_milliseconds,2000")
+        .append(" & echo $! > " + gcs_server_socket_name + ".pid");
+
+    RAY_LOG(INFO) << "Start gcs server command: " << ray_start_cmd;
+    RAY_CHECK(system(ray_start_cmd.c_str()) == 0);
+    usleep(200 * 1000);
+    RAY_LOG(INFO) << "Finished start gcs server.";
+    return gcs_server_socket_name;
+  }
+
+  void StopGcsServer(std::string gcs_server_socket_name) {
+    std::string gcs_server_pid = gcs_server_socket_name + ".pid";
+    std::string kill_9 = "kill -9 `cat " + gcs_server_pid + "`";
+    RAY_LOG(DEBUG) << kill_9;
+    ASSERT_TRUE(system(kill_9.c_str()) == 0);
+    ASSERT_TRUE(system(("rm -rf " + gcs_server_socket_name).c_str()) == 0);
+    ASSERT_TRUE(system(("rm -rf " + gcs_server_socket_name + ".pid").c_str()) == 0);
+  }
+
   std::string StartRaylet(std::string store_socket_name, std::string node_ip_address,
                           int port, std::string redis_address, std::string resource) {
     std::string raylet_socket_name = "/tmp/raylet" + RandomObjectID().Hex();
@@ -304,9 +334,11 @@ class StreamingQueueTestBase : public ::testing::TestWithParam<uint64_t> {
   std::vector<std::string> raylet_store_socket_names_;
   gcs::GcsClientOptions gcs_options_;
   std::string raylet_executable_;
+  std::string gcs_server_executable_;
   std::string store_executable_;
   std::string actor_executable_;
   int node_manager_port_;
+  std::string gcs_server_pid_;
 };
 
 }  // namespace streaming

streaming/src/test/run_streaming_queue_test.sh

@@ -35,7 +35,7 @@ if [ -z "$RAY_ROOT" ] ; then
   exit 1
 fi
 
-bazel build "//:core_worker_test" "//:mock_worker"  "//:raylet" "//:libray_redis_module.so" "@plasma//:plasma_store_server"
+bazel build "//:core_worker_test" "//:mock_worker"  "//:raylet" "//:gcs_server" "//:libray_redis_module.so" "@plasma//:plasma_store_server"
 bazel build //streaming:streaming_test_worker
 bazel build //streaming:streaming_queue_tests
 
@@ -50,6 +50,7 @@ LOAD_MODULE_ARGS="--loadmodule ${REDIS_MODULE}"
 STORE_EXEC="./bazel-bin/external/plasma/plasma_store_server"
 RAYLET_EXEC="./bazel-bin/raylet"
 STREAMING_TEST_WORKER_EXEC="./bazel-bin/streaming/streaming_test_worker"
+GCS_SERVER_EXEC="./bazel-bin/gcs_server"
 
 # Allow cleanup commands to fail.
 bazel run //:redis-cli -- -p 6379 shutdown || true
@@ -61,7 +62,7 @@ sleep 2s
 bazel run //:redis-server -- --loglevel warning ${LOAD_MODULE_ARGS} --port 6380 &
 sleep 2s
 # Run tests.
-./bazel-bin/streaming/streaming_queue_tests $STORE_EXEC $RAYLET_EXEC $RAYLET_PORT $STREAMING_TEST_WORKER_EXEC
+./bazel-bin/streaming/streaming_queue_tests $STORE_EXEC $RAYLET_EXEC $RAYLET_PORT $STREAMING_TEST_WORKER_EXEC $GCS_SERVER_EXEC
 sleep 1s
 bazel run //:redis-cli -- -p 6379 shutdown
 bazel run //:redis-cli -- -p 6380 shutdown

streaming/src/test/streaming_queue_tests.cc

@@ -18,6 +18,7 @@ namespace streaming {
 
 static std::string store_executable;
 static std::string raylet_executable;
+static std::string gcs_server_executable;
 static std::string actor_executable;
 static int node_manager_port;
 
@@ -25,14 +26,14 @@ class StreamingWriterTest : public StreamingQueueTestBase {
  public:
   StreamingWriterTest()
       : StreamingQueueTestBase(1, raylet_executable, store_executable, node_manager_port,
-                               actor_executable) {}
+                               actor_executable, gcs_server_executable) {}
 };
 
 class StreamingExactlySameTest : public StreamingQueueTestBase {
  public:
   StreamingExactlySameTest()
       : StreamingQueueTestBase(1, raylet_executable, store_executable, node_manager_port,
-                               actor_executable) {}
+                               actor_executable, gcs_server_executable) {}
 };
 
 TEST_P(StreamingWriterTest, streaming_writer_exactly_once_test) {
@@ -56,10 +57,11 @@ INSTANTIATE_TEST_CASE_P(StreamingTest, StreamingExactlySameTest,
 int main(int argc, char **argv) {
   // set_streaming_log_config("streaming_writer_test", StreamingLogLevel::INFO, 0);
   ::testing::InitGoogleTest(&argc, argv);
-  RAY_CHECK(argc == 5);
+  RAY_CHECK(argc == 6);
   ray::streaming::store_executable = std::string(argv[1]);
   ray::streaming::raylet_executable = std::string(argv[2]);
   ray::streaming::node_manager_port = std::stoi(std::string(argv[3]));
   ray::streaming::actor_executable = std::string(argv[4]);
+  ray::streaming::gcs_server_executable = std::string(argv[5]);
   return RUN_ALL_TESTS();
 }