From 0fbee4da0c9bdc57a40d424d7dde2213991e1744 Mon Sep 17 00:00:00 2001 From: Tao Wang Date: Mon, 26 Oct 2020 12:06:28 +0800 Subject: [PATCH] [GCS] Remove unused ReportBatchHeartbeat/SubscribeHeartbeat (#11567) * Remove unused message ReportBatchHeartbeat * add up --- src/ray/gcs/accessor.h | 18 ------------- .../gcs/gcs_client/service_based_accessor.cc | 18 ------------- .../gcs/gcs_client/service_based_accessor.h | 8 ------ .../gcs_server/test/gcs_server_test_util.h | 15 ----------- src/ray/gcs/redis_accessor.cc | 25 ------------------- src/ray/gcs/redis_accessor.h | 12 --------- src/ray/gcs/subscription_executor.cc | 1 - src/ray/protobuf/gcs_service.proto | 8 ------ 8 files changed, 105 deletions(-) diff --git a/src/ray/gcs/accessor.h b/src/ray/gcs/accessor.h index 72453c092..96436e6bd 100644 --- a/src/ray/gcs/accessor.h +++ b/src/ray/gcs/accessor.h @@ -574,24 +574,6 @@ class NodeInfoAccessor { /// Resend heartbeat when GCS restarts from a failure. virtual void AsyncReReportHeartbeat() = 0; - /// Subscribe to the heartbeat of each node from GCS. - /// - /// \param subscribe Callback that will be called each time when heartbeat is updated. - /// \param done Callback that will be called when subscription is complete. - /// \return Status - virtual Status AsyncSubscribeHeartbeat( - const SubscribeCallback &subscribe, - const StatusCallback &done) = 0; - - /// Report state of all nodes to GCS asynchronously. - /// - /// \param data_ptr The heartbeats that will be reported to GCS. - /// \param callback Callback that will be called after report finishes. - /// \return Status - virtual Status AsyncReportBatchHeartbeat( - const std::shared_ptr &data_ptr, - const StatusCallback &callback) = 0; - /// Subscribe batched state of all nodes from GCS. /// /// \param subscribe Callback that will be called each time when batch heartbeat is diff --git a/src/ray/gcs/gcs_client/service_based_accessor.cc b/src/ray/gcs/gcs_client/service_based_accessor.cc index 35cce0bcd..649642864 100644 --- a/src/ray/gcs/gcs_client/service_based_accessor.cc +++ b/src/ray/gcs/gcs_client/service_based_accessor.cc @@ -706,24 +706,6 @@ void ServiceBasedNodeInfoAccessor::AsyncReReportHeartbeat() { } } -Status ServiceBasedNodeInfoAccessor::AsyncSubscribeHeartbeat( - const SubscribeCallback &subscribe, - const StatusCallback &done) { - const std::string error_msg = - "Unsupported method of AsyncSubscribeHeartbeat in ServiceBasedNodeInfoAccessor."; - RAY_LOG(FATAL) << error_msg; - return Status::Invalid(error_msg); -} - -Status ServiceBasedNodeInfoAccessor::AsyncReportBatchHeartbeat( - const std::shared_ptr &data_ptr, - const StatusCallback &callback) { - const std::string error_msg = - "Unsupported method of AsyncReportBatchHeartbeat in ServiceBasedNodeInfoAccessor."; - RAY_LOG(FATAL) << error_msg; - return Status::Invalid(error_msg); -} - Status ServiceBasedNodeInfoAccessor::AsyncSubscribeBatchHeartbeat( const ItemCallback &subscribe, const StatusCallback &done) { diff --git a/src/ray/gcs/gcs_client/service_based_accessor.h b/src/ray/gcs/gcs_client/service_based_accessor.h index 56e740479..c1ce2a453 100644 --- a/src/ray/gcs/gcs_client/service_based_accessor.h +++ b/src/ray/gcs/gcs_client/service_based_accessor.h @@ -193,14 +193,6 @@ class ServiceBasedNodeInfoAccessor : public NodeInfoAccessor { void AsyncReReportHeartbeat() override; - Status AsyncSubscribeHeartbeat( - const SubscribeCallback &subscribe, - const StatusCallback &done) override; - - Status AsyncReportBatchHeartbeat( - const std::shared_ptr &data_ptr, - const StatusCallback &callback) override; - Status AsyncSubscribeBatchHeartbeat( const ItemCallback &subscribe, const StatusCallback &done) override; diff --git a/src/ray/gcs/gcs_server/test/gcs_server_test_util.h b/src/ray/gcs/gcs_server/test/gcs_server_test_util.h index 6ef863baf..e004a17ca 100644 --- a/src/ray/gcs/gcs_server/test/gcs_server_test_util.h +++ b/src/ray/gcs/gcs_server/test/gcs_server_test_util.h @@ -380,21 +380,6 @@ struct GcsServerMocker { return Status::NotImplemented(""); } - Status AsyncSubscribeHeartbeat( - const gcs::SubscribeCallback &subscribe, - const gcs::StatusCallback &done) override { - return Status::NotImplemented(""); - } - - Status AsyncReportBatchHeartbeat( - const std::shared_ptr &data_ptr, - const gcs::StatusCallback &callback) override { - if (callback) { - callback(Status::OK()); - } - return Status::OK(); - } - Status AsyncSubscribeBatchHeartbeat( const gcs::ItemCallback &subscribe, const gcs::StatusCallback &done) override { diff --git a/src/ray/gcs/redis_accessor.cc b/src/ray/gcs/redis_accessor.cc index f37be4d6a..9231ee508 100644 --- a/src/ray/gcs/redis_accessor.cc +++ b/src/ray/gcs/redis_accessor.cc @@ -495,7 +495,6 @@ Status RedisObjectInfoAccessor::AsyncUnsubscribeToLocations(const ObjectID &obje RedisNodeInfoAccessor::RedisNodeInfoAccessor(RedisGcsClient *client_impl) : client_impl_(client_impl), resource_sub_executor_(client_impl_->resource_table()), - heartbeat_sub_executor_(client_impl->heartbeat_table()), heartbeat_batch_sub_executor_(client_impl->heartbeat_batch_table()) {} Status RedisNodeInfoAccessor::RegisterSelf(const GcsNodeInfo &local_node_info) { @@ -600,30 +599,6 @@ Status RedisNodeInfoAccessor::AsyncReportHeartbeat( void RedisNodeInfoAccessor::AsyncReReportHeartbeat() {} -Status RedisNodeInfoAccessor::AsyncSubscribeHeartbeat( - const SubscribeCallback &subscribe, - const StatusCallback &done) { - RAY_CHECK(subscribe != nullptr); - auto on_subscribe = [subscribe](const NodeID &node_id, const HeartbeatTableData &data) { - subscribe(node_id, data); - }; - - return heartbeat_sub_executor_.AsyncSubscribeAll(NodeID::Nil(), on_subscribe, done); -} - -Status RedisNodeInfoAccessor::AsyncReportBatchHeartbeat( - const std::shared_ptr &data_ptr, - const StatusCallback &callback) { - HeartbeatBatchTable::WriteCallback on_done = nullptr; - if (callback != nullptr) { - on_done = [callback](RedisGcsClient *client, const NodeID &node_id, - const HeartbeatBatchTableData &data) { callback(Status::OK()); }; - } - - HeartbeatBatchTable &hb_batch_table = client_impl_->heartbeat_batch_table(); - return hb_batch_table.Add(JobID::Nil(), NodeID::Nil(), data_ptr, on_done); -} - Status RedisNodeInfoAccessor::AsyncSubscribeBatchHeartbeat( const ItemCallback &subscribe, const StatusCallback &done) { RAY_CHECK(subscribe != nullptr); diff --git a/src/ray/gcs/redis_accessor.h b/src/ray/gcs/redis_accessor.h index e1e0da9ba..d7b3d49e5 100644 --- a/src/ray/gcs/redis_accessor.h +++ b/src/ray/gcs/redis_accessor.h @@ -369,14 +369,6 @@ class RedisNodeInfoAccessor : public NodeInfoAccessor { void AsyncReReportHeartbeat() override; - Status AsyncSubscribeHeartbeat( - const SubscribeCallback &subscribe, - const StatusCallback &done) override; - - Status AsyncReportBatchHeartbeat( - const std::shared_ptr &data_ptr, - const StatusCallback &callback) override; - Status AsyncSubscribeBatchHeartbeat( const ItemCallback &subscribe, const StatusCallback &done) override; @@ -401,10 +393,6 @@ class RedisNodeInfoAccessor : public NodeInfoAccessor { DynamicResourceSubscriptionExecutor; DynamicResourceSubscriptionExecutor resource_sub_executor_; - typedef SubscriptionExecutor - HeartbeatSubscriptionExecutor; - HeartbeatSubscriptionExecutor heartbeat_sub_executor_; - typedef SubscriptionExecutor HeartbeatBatchSubscriptionExecutor; HeartbeatBatchSubscriptionExecutor heartbeat_batch_sub_executor_; diff --git a/src/ray/gcs/subscription_executor.cc b/src/ray/gcs/subscription_executor.cc index 26014169f..fa4fb00aa 100644 --- a/src/ray/gcs/subscription_executor.cc +++ b/src/ray/gcs/subscription_executor.cc @@ -207,7 +207,6 @@ template class SubscriptionExecutor, TaskLeaseTable>; template class SubscriptionExecutor; -template class SubscriptionExecutor; template class SubscriptionExecutor; template class SubscriptionExecutor; diff --git a/src/ray/protobuf/gcs_service.proto b/src/ray/protobuf/gcs_service.proto index d5a40c2db..6e6ca8f2a 100644 --- a/src/ray/protobuf/gcs_service.proto +++ b/src/ray/protobuf/gcs_service.proto @@ -198,14 +198,6 @@ message ReportHeartbeatReply { GcsStatus status = 1; } -message ReportBatchHeartbeatRequest { - HeartbeatBatchTableData heartbeat_batch = 1; -} - -message ReportBatchHeartbeatReply { - GcsStatus status = 1; -} - message GetResourcesRequest { bytes node_id = 1; }