[core][scalability] Change ray syncer from unary call to streaming call

BUILD.bazel

@@ -2619,6 +2619,7 @@ cc_test(
     copts = COPTS,
     tags = ["team:core"],
     deps = [
+        ":grpc_common_lib",
         ":ray_common",
         ":ray_mock",
         "@com_google_googletest//:gtest",

release/nightly_tests/many_nodes_tests/actor_test.py

@@ -16,11 +16,6 @@ def foo(self):
     return actors
 
 
-def test_actor_ready(actors):
-    remaining = [actor.foo.remote() for actor in actors]
-    ray.get(remaining)
-
-
 def parse_script_args():
     parser = argparse.ArgumentParser()
     parser.add_argument("--cpus-per-actor", type=float, default=0.2)
@@ -43,7 +38,15 @@ def main():
         sleep(10)
         return
     actor_ready_start = perf_counter()
-    test_actor_ready(actors)
+    total_actors = len(actors)
+    objs = [actor.foo.remote() for actor in actors]
+
+    while len(objs) != 0:
+        objs_ready, objs = ray.wait(objs, timeout=10)
+        print(
+            f"Status: {total_actors - len(objs)}/{total_actors}, "
+            f"{perf_counter() - actor_ready_start}"
+        )
     actor_ready_end = perf_counter()
     actor_ready_time = actor_ready_end - actor_ready_start
 

src/mock/ray/common/ray_syncer/ray_syncer.h

@@ -43,10 +43,24 @@ class MockReceiverInterface : public ReceiverInterface {
 namespace ray {
 namespace syncer {
 
-class MockNodeSyncConnection : public NodeSyncConnection {
+class MockRaySyncerBidiReactor : public RaySyncerBidiReactor {
  public:
-  using NodeSyncConnection::NodeSyncConnection;
-  MOCK_METHOD(void, DoSend, (), (override));
+  using RaySyncerBidiReactor::RaySyncerBidiReactor;
+
+  MOCK_METHOD(void, Disconnect, (), (override));
+
+  MOCK_METHOD(bool,
+              PushToSendingQueue,
+              (std::shared_ptr<const RaySyncMessage>),
+              (override));
+};
+
+template <typename T>
+class MockRaySyncerBidiReactorBase : public RaySyncerBidiReactorBase<T> {
+ public:
+  using RaySyncerBidiReactorBase<T>::RaySyncerBidiReactorBase;
+
+  MOCK_METHOD(void, Disconnect, (), (override));
 };
 
 }  // namespace syncer

src/ray/common/id.cc

@@ -337,4 +337,7 @@ ID_OSTREAM_OPERATOR(ActorID);
 ID_OSTREAM_OPERATOR(TaskID);
 ID_OSTREAM_OPERATOR(ObjectID);
 ID_OSTREAM_OPERATOR(PlacementGroupID);
+
+const NodeID kGCSNodeID = NodeID::FromBinary(std::string(kUniqueIDSize, 0));
+
 }  // namespace ray

src/ray/common/id.h

@@ -561,3 +561,7 @@ DEFINE_UNIQUE_ID(PlacementGroupID);
 
 #undef DEFINE_UNIQUE_ID
 }  // namespace std
+
+namespace ray {
+extern const NodeID kGCSNodeID;
+}

src/ray/common/ray_config_def.h

@@ -672,6 +672,10 @@ RAY_CONFIG(int64_t, grpc_keepalive_time_ms, 10000)
 /// grpc keepalive timeout
 RAY_CONFIG(int64_t, grpc_keepalive_timeout_ms, 20000)
 
+/// grpc streaming buffer size
+/// Set it to 512kb
+RAY_CONFIG(int64_t, grpc_stream_buffer_size, 512 * 1024);
+
 /// Whether to use log reporter in event framework
 RAY_CONFIG(bool, event_log_reporter_enabled, false)