[scheduler][6] Integrate ray with syncer.

.buildkite/pipeline.yml

@@ -301,7 +301,6 @@
       --build_tests_only
       --test_tag_filters=release_unit
       release/...
-
 - label: ":python: (Small & Client)"
   conditions: ["RAY_CI_PYTHON_AFFECTED"]
   commands:
@@ -354,6 +353,38 @@
       --test_env=CONDA_DEFAULT_ENV
       python/ray/tests/...
 
+- label: ":construction: :python: (syncer) (Small & Client)"
+  conditions: ["RAY_CI_PYTHON_AFFECTED"]
+  commands:
+    - cleanup() { if [ "${BUILDKITE_PULL_REQUEST}" = "false" ]; then ./ci/travis/upload_build_info.sh; fi }; trap cleanup EXIT
+    - bazel test --config=ci $(./scripts/bazel_export_options)
+      --test_tag_filters=client_tests,small_size_python_tests
+      --test_env=RAY_use_ray_syncer=true
+      -- python/ray/tests/...
+- label: ":construction: :python: (syncer) (Large)"
+  conditions: ["RAY_CI_PYTHON_AFFECTED"]
+  parallelism: 3
+  commands:
+    - cleanup() { if [ "${BUILDKITE_PULL_REQUEST}" = "false" ]; then ./ci/build/upload_build_info.sh; fi }; trap cleanup EXIT
+    - RAY_use_ray_syncer=true . ./ci/ci.sh test_large
+- label: ":construction: :python: (syncer) (Medium A-J)"
+  conditions: ["RAY_CI_PYTHON_AFFECTED"]
+  commands:
+    - cleanup() { if [ "${BUILDKITE_PULL_REQUEST}" = "false" ]; then ./ci/travis/upload_build_info.sh; fi }; trap cleanup EXIT
+    - bazel test --config=ci $(./scripts/bazel_export_options)
+      --test_tag_filters=-kubernetes,medium_size_python_tests_a_to_j
+      --test_env=RAY_use_ray_syncer=true
+      python/ray/tests/...
+- label: ":construction: :python: (syncer) (Medium K-Z)"
+  conditions: ["RAY_CI_PYTHON_AFFECTED"]
+  commands:
+    - cleanup() { if [ "${BUILDKITE_PULL_REQUEST}" = "false" ]; then ./ci/travis/upload_build_info.sh; fi }; trap cleanup EXIT
+    - bazel test --config=ci $(./scripts/bazel_export_options)
+      --test_tag_filters=-kubernetes,medium_size_python_tests_k_to_z
+      --test_env=RAY_use_ray_syncer=true
+      python/ray/tests/...
+
+
 # https://github.com/ray-project/ray/issues/22460
 #- label: ":python: (Privileged test)"
   #conditions: ["RAY_CI_PYTHON_AFFECTED"]

python/ray/includes/ray_config.pxd

@@ -66,3 +66,5 @@ cdef extern from "ray/common/ray_config.h" nogil:
         c_bool gcs_grpc_based_pubsub() const
 
         c_bool start_python_importer_thread() const
+
+        c_bool use_ray_syncer() const

python/ray/includes/ray_config.pxi

@@ -107,3 +107,7 @@ cdef class Config:
     @staticmethod
     def start_python_importer_thread():
         return RayConfig.instance().start_python_importer_thread()
+
+    @staticmethod
+    def use_ray_syncer():
+        return RayConfig.instance().use_ray_syncer()

python/ray/tests/test_metrics_agent.py

@@ -44,7 +44,6 @@
     "ray_internal_num_spilled_tasks",
     # "ray_unintentional_worker_failures_total",
     # "ray_node_failure_total",
-    "ray_outbound_heartbeat_size_kb_sum",
     "ray_operation_count",
     "ray_operation_run_time_ms",
     "ray_operation_queue_time_ms",
@@ -75,6 +74,9 @@
     "ray_gcs_actors_count",
 ]
 
+if not ray._raylet.Config.use_ray_syncer():
+    _METRICS.append("ray_outbound_heartbeat_size_kb_sum")
+
 # This list of metrics should be kept in sync with
 # ray/python/ray/autoscaler/_private/prom_metrics.py
 _AUTOSCALER_METRICS = [

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

@@ -19,7 +19,7 @@ class MockReporterInterface : public ReporterInterface {
  public:
   MOCK_METHOD(std::optional<RaySyncMessage>,
               CreateSyncMessage,
-              (int64_t current_version, RayComponentId component_id),
+              (int64_t current_version, MessageType message_type),
               (const, override));
 };
 

src/mock/ray/raylet_client/raylet_client.h

@@ -132,6 +132,7 @@ class MockRayletClientInterface : public RayletClientInterface {
               WaitForDirectActorCallArgs,
               (const std::vector<rpc::ObjectReference> &references, int64_t tag),
               (override));
+  MOCK_METHOD(std::shared_ptr<grpc::Channel>, GetChannel, (), (const));
   MOCK_METHOD(void,
               ReportWorkerBacklog,
               (const WorkerID &worker_id,

src/ray/common/ray_config_def.h

@@ -64,6 +64,10 @@ RAY_CONFIG(uint64_t, num_heartbeats_warning, 5)
 
 /// The duration between reporting resources sent by the raylets.
 RAY_CONFIG(uint64_t, raylet_report_resources_period_milliseconds, 100)
+
+/// The duration between raylet check memory pressure and send gc request
+RAY_CONFIG(uint64_t, raylet_check_gc_period_milliseconds, 100)
+
 /// For a raylet, if the last resource report was sent more than this many
 /// report periods ago, then a warning will be logged that the report
 /// handler is drifting.
@@ -335,6 +339,9 @@ RAY_CONFIG(int32_t, gcs_rpc_server_reconnect_timeout_s, 60)
 /// Minimum interval between reconnecting gcs rpc server when gcs server restarts.
 RAY_CONFIG(int32_t, minimum_gcs_reconnect_interval_milliseconds, 5000)
 
+/// Feature flag to use the ray syncer for resource synchronization
+RAY_CONFIG(bool, use_ray_syncer, false)
+
 /// The interval at which the gcs client will check if the address of gcs service has
 /// changed. When the address changed, we will resubscribe again.
 RAY_CONFIG(uint64_t, gcs_service_address_check_interval_milliseconds, 1000)

src/ray/common/ray_syncer/ray_syncer-inl.h

@@ -25,7 +25,7 @@ class NodeState {
 
   /// Set the local component.
   ///
-  /// \param cid The component id.
+  /// \param message_type The type of the message for this component.
   /// \param reporter The reporter is defined to be the local module which wants to
   /// broadcast its internal status to the whole clsuter. When it's null, it means there
   /// is no reporter in the local node for this component. This is the place there
@@ -36,16 +36,16 @@ class NodeState {
   /// received messages are consumed.
   ///
   /// \return true if set successfully.
-  bool SetComponent(RayComponentId cid,
+  bool SetComponent(MessageType message_type,
                     const ReporterInterface *reporter,
                     ReceiverInterface *receiver);
 
   /// Get the snapshot of a component for a newer version.
   ///
-  /// \param cid The component id to take the snapshot.
+  /// \param message_type The component to take the snapshot.
   ///
   /// \return If a snapshot is taken, return the message, otherwise std::nullopt.
-  std::optional<RaySyncMessage> CreateSyncMessage(RayComponentId cid);
+  std::optional<RaySyncMessage> CreateSyncMessage(MessageType message_type);
 
   /// Consume a message. Receiver will consume this message if it doesn't have
   /// this message.
@@ -127,7 +127,7 @@ class NodeSyncConnection {
   std::function<void(std::shared_ptr<RaySyncMessage>)> message_processor_;
 
   /// Buffering all the updates. Sending will be done in an async way.
-  absl::flat_hash_map<std::pair<std::string, RayComponentId>,
+  absl::flat_hash_map<std::pair<std::string, MessageType>,
                       std::shared_ptr<const RaySyncMessage>>
       sending_buffer_;
 

src/ray/common/ray_syncer/ray_syncer.cc

@@ -22,38 +22,38 @@ namespace syncer {
 
 NodeState::NodeState() { sync_message_versions_taken_.fill(-1); }
 
-bool NodeState::SetComponent(RayComponentId cid,
+bool NodeState::SetComponent(MessageType message_type,
                              const ReporterInterface *reporter,
                              ReceiverInterface *receiver) {
-  if (cid < static_cast<RayComponentId>(kComponentArraySize) &&
-      reporters_[cid] == nullptr && receivers_[cid] == nullptr) {
-    reporters_[cid] = reporter;
-    receivers_[cid] = receiver;
+  if (message_type < static_cast<MessageType>(kComponentArraySize) &&
+      reporters_[message_type] == nullptr && receivers_[message_type] == nullptr) {
+    reporters_[message_type] = reporter;
+    receivers_[message_type] = receiver;
     return true;
   } else {
-    RAY_LOG(FATAL) << "Fail to set components, component_id:" << cid
+    RAY_LOG(FATAL) << "Fail to set components, message_type:" << message_type
                    << ", reporter:" << reporter << ", receiver:" << receiver;
     return false;
   }
 }
 
-std::optional<RaySyncMessage> NodeState::CreateSyncMessage(RayComponentId cid) {
-  if (reporters_[cid] == nullptr) {
+std::optional<RaySyncMessage> NodeState::CreateSyncMessage(MessageType message_type) {
+  if (reporters_[message_type] == nullptr) {
     return std::nullopt;
   }
-  auto message =
-      reporters_[cid]->CreateSyncMessage(sync_message_versions_taken_[cid], cid);
+  auto message = reporters_[message_type]->CreateSyncMessage(
+      sync_message_versions_taken_[message_type], message_type);
   if (message != std::nullopt) {
-    sync_message_versions_taken_[cid] = message->version();
-    RAY_LOG(DEBUG) << "Sync message taken: cid:" << cid
+    sync_message_versions_taken_[message_type] = message->version();
+    RAY_LOG(DEBUG) << "Sync message taken: message_type:" << message_type
                    << ", version:" << message->version()
                    << ", node:" << NodeID::FromBinary(message->node_id());
   }
   return message;
 }
 
 bool NodeState::ConsumeSyncMessage(std::shared_ptr<const RaySyncMessage> message) {
-  auto &current = cluster_view_[message->node_id()][message->component_id()];
+  auto &current = cluster_view_[message->node_id()][message->message_type()];
 
   RAY_LOG(DEBUG) << "ConsumeSyncMessage: " << (current ? current->version() : -1)
                  << " message_version: " << message->version()
@@ -64,7 +64,7 @@ bool NodeState::ConsumeSyncMessage(std::shared_ptr<const RaySyncMessage> message
   }
 
   current = message;
-  auto receiver = receivers_[message->component_id()];
+  auto receiver = receivers_[message->message_type()];
   if (receiver != nullptr) {
     receiver->ConsumeSyncMessage(message);
   }
@@ -83,11 +83,11 @@ void NodeSyncConnection::ReceiveUpdate(RaySyncMessages messages) {
   for (auto &message : *messages.mutable_sync_messages()) {
     auto &node_versions = GetNodeComponentVersions(message.node_id());
     RAY_LOG(DEBUG) << "Receive update: "
-                   << " component_id=" << message.component_id()
+                   << " message_type=" << message.message_type()
                    << ", message_version=" << message.version()
-                   << ", local_message_version=" << node_versions[message.component_id()];
-    if (node_versions[message.component_id()] < message.version()) {
-      node_versions[message.component_id()] = message.version();
+                   << ", local_message_version=" << node_versions[message.message_type()];
+    if (node_versions[message.message_type()] < message.version()) {
+      node_versions[message.message_type()] = message.version();
       message_processor_(std::make_shared<RaySyncMessage>(std::move(message)));
     }
   }
@@ -104,9 +104,9 @@ bool NodeSyncConnection::PushToSendingQueue(
   }
 
   auto &node_versions = GetNodeComponentVersions(message->node_id());
-  if (node_versions[message->component_id()] < message->version()) {
-    node_versions[message->component_id()] = message->version();
-    sending_buffer_[std::make_pair(message->node_id(), message->component_id())] =
+  if (node_versions[message->message_type()] < message->version()) {
+    node_versions[message->message_type()] = message->version();
+    sending_buffer_[std::make_pair(message->node_id(), message->message_type())] =
         message;
     return true;
   }
@@ -323,30 +323,30 @@ void RaySyncer::Disconnect(const std::string &node_id) {
       "RaySyncerDisconnect");
 }
 
-bool RaySyncer::Register(RayComponentId component_id,
+bool RaySyncer::Register(MessageType message_type,
                          const ReporterInterface *reporter,
                          ReceiverInterface *receiver,
                          int64_t pull_from_reporter_interval_ms) {
-  if (!node_state_->SetComponent(component_id, reporter, receiver)) {
+  if (!node_state_->SetComponent(message_type, reporter, receiver)) {
     return false;
   }
 
   // Set job to pull from reporter periodically
   if (reporter != nullptr && pull_from_reporter_interval_ms > 0) {
     timer_.RunFnPeriodically(
-        [this, component_id]() { OnDemandBroadcasting(component_id); },
+        [this, message_type]() { OnDemandBroadcasting(message_type); },
         pull_from_reporter_interval_ms);
   }
 
   RAY_LOG(DEBUG) << "Registered components: "
-                 << "component_id:" << component_id << ", reporter:" << reporter
+                 << "message_type:" << message_type << ", reporter:" << reporter
                  << ", receiver:" << receiver
                  << ", pull_from_reporter_interval_ms:" << pull_from_reporter_interval_ms;
   return true;
 }
 
-bool RaySyncer::OnDemandBroadcasting(RayComponentId component_id) {
-  auto msg = node_state_->CreateSyncMessage(component_id);
+bool RaySyncer::OnDemandBroadcasting(MessageType message_type) {
+  auto msg = node_state_->CreateSyncMessage(message_type);
   if (msg) {
     RAY_CHECK(msg->node_id() == GetLocalNodeID());
     BroadcastMessage(std::make_shared<RaySyncMessage>(std::move(*msg)));

src/ray/common/ray_syncer/ray_syncer.h

@@ -29,14 +29,14 @@ namespace syncer {
 
 using ray::rpc::syncer::DummyRequest;
 using ray::rpc::syncer::DummyResponse;
-using ray::rpc::syncer::RayComponentId;
+using ray::rpc::syncer::MessageType;
 using ray::rpc::syncer::RaySyncMessage;
 using ray::rpc::syncer::RaySyncMessages;
 using ray::rpc::syncer::StartSyncRequest;
 using ray::rpc::syncer::StartSyncResponse;
 
 static constexpr size_t kComponentArraySize =
-    static_cast<size_t>(ray::rpc::syncer::RayComponentId_ARRAYSIZE);
+    static_cast<size_t>(ray::rpc::syncer::MessageType_ARRAYSIZE);
 
 /// The interface for a reporter. Reporter is defined to be a local module which would
 /// like to let the other nodes know its state. For example, local cluster resource
@@ -49,13 +49,13 @@ struct ReporterInterface {
   ///
   /// \param version_after Request message with version after `version_after`. If the
   /// reporter doesn't have the qualified one, just return std::nullopt
-  /// \param component_id The component id asked for.
+  /// \param message_type The message type asked for.
   ///
   /// \return std::nullopt if the reporter doesn't have such component or the current
   /// snapshot of the component is not newer the asked one. Otherwise, return the
   /// actual message.
   virtual std::optional<RaySyncMessage> CreateSyncMessage(
-      int64_t version_after, RayComponentId component_id) const = 0;
+      int64_t version_after, MessageType message_type) const = 0;
   virtual ~ReporterInterface() {}
 };
 
@@ -114,14 +114,14 @@ class RaySyncer {
   /// it'll have a global view of the cluster.
   ///
   ///
-  /// \param component_id The component to sync.
+  /// \param message_type The component to sync.
   /// \param reporter The local component to be broadcasted.
   /// \param receiver The consumer of the sync message sent by the other nodes in the
   /// cluster.
   /// \param pull_from_reporter_interval_ms The frequence to pull a message. 0 means
   /// never pull a message in syncer.
   /// from reporter and push it to sending queue.
-  bool Register(RayComponentId component_id,
+  bool Register(MessageType message_type,
                 const ReporterInterface *reporter,
                 ReceiverInterface *receiver,
                 int64_t pull_from_reporter_interval_ms = 100);
@@ -132,10 +132,10 @@ class RaySyncer {
   /// Request trigger a broadcasting for a specific component immediately instead of
   /// waiting for ray syncer to poll the message.
   ///
-  /// \param component_id The component to check.
+  /// \param message_type The component to check.
   /// \return true if a message is generated. If the component doesn't have a new
   /// version of message, false will be returned.
-  bool OnDemandBroadcasting(RayComponentId component_id);
+  bool OnDemandBroadcasting(MessageType message_type);
 
  private:
   /// Get the io_context used by RaySyncer.