Optimize heartbeat manager initialization (#12911)

src/ray/gcs/gcs_server/gcs_heartbeat_manager.cc

@@ -34,6 +34,14 @@ GcsHeartbeatManager::GcsHeartbeatManager(
   }));
 }
 
+void GcsHeartbeatManager::Initialize(const GcsInitData &gcs_init_data) {
+  for (const auto &item : gcs_init_data.Nodes()) {
+    if (item.second.state() == rpc::GcsNodeInfo::ALIVE) {
+      heartbeats_.emplace(item.first, num_heartbeats_timeout_);
+    }
+  }
+}
+
 void GcsHeartbeatManager::Start() {
   io_service_.post([this] {
     if (!is_started_) {

src/ray/gcs/gcs_server/gcs_heartbeat_manager.h

@@ -43,6 +43,12 @@ class GcsHeartbeatManager : public rpc::HeartbeatInfoHandler {
                              rpc::ReportHeartbeatReply *reply,
                              rpc::SendReplyCallback send_reply_callback) override;
 
+  /// Initialize with the gcs tables data synchronously.
+  /// This should be called when GCS server restarts after a failure.
+  ///
+  /// \param gcs_init_data.
+  void Initialize(const GcsInitData &gcs_init_data);
+
   /// Start node failure detect loop.
   void Start();
 

src/ray/gcs/gcs_server/gcs_server.cc

@@ -151,9 +151,8 @@ void GcsServer::InitGcsHeartbeatManager(const GcsInitData &gcs_init_data) {
         main_service_.post(
             [this, node_id] { return gcs_node_manager_->OnNodeFailure(node_id); });
       });
-  for (const auto &node : gcs_init_data.Nodes()) {
-    gcs_heartbeat_manager_->AddNode(node.first);
-  }
+  // Initialize by gcs tables data.
+  gcs_heartbeat_manager_->Initialize(gcs_init_data);
   // Register service.
   heartbeat_info_service_.reset(new rpc::HeartbeatInfoGrpcService(
       heartbeat_manager_io_service_, *gcs_heartbeat_manager_));