diff --git a/src/ray/gcs/gcs_server/gcs_server.cc b/src/ray/gcs/gcs_server/gcs_server.cc index b17e5d336..6fd1f3b0d 100644 --- a/src/ray/gcs/gcs_server/gcs_server.cc +++ b/src/ray/gcs/gcs_server/gcs_server.cc @@ -238,8 +238,8 @@ std::unique_ptr GcsServer::InitErrorInfoHandler() { } std::unique_ptr GcsServer::InitWorkerInfoHandler() { - return std::unique_ptr( - new rpc::DefaultWorkerInfoHandler(*redis_gcs_client_, gcs_pub_sub_)); + return std::unique_ptr(new rpc::DefaultWorkerInfoHandler( + *redis_gcs_client_, gcs_table_storage_, gcs_pub_sub_)); } } // namespace gcs diff --git a/src/ray/gcs/gcs_server/worker_info_handler_impl.cc b/src/ray/gcs/gcs_server/worker_info_handler_impl.cc index 87495bc4a..5bd7745fb 100644 --- a/src/ray/gcs/gcs_server/worker_info_handler_impl.cc +++ b/src/ray/gcs/gcs_server/worker_info_handler_impl.cc @@ -38,8 +38,8 @@ void DefaultWorkerInfoHandler::HandleReportWorkerFailure( GCS_RPC_SEND_REPLY(send_reply_callback, reply, status); }; - Status status = - gcs_client_.Workers().AsyncReportWorkerFailure(worker_failure_data, on_done); + Status status = gcs_table_storage_->WorkerFailureTable().Put( + worker_id, *worker_failure_data, on_done); if (!status.ok()) { on_done(status); } diff --git a/src/ray/gcs/gcs_server/worker_info_handler_impl.h b/src/ray/gcs/gcs_server/worker_info_handler_impl.h index 1cf33b217..b8d844111 100644 --- a/src/ray/gcs/gcs_server/worker_info_handler_impl.h +++ b/src/ray/gcs/gcs_server/worker_info_handler_impl.h @@ -15,6 +15,7 @@ #ifndef RAY_GCS_WORKER_INFO_HANDLER_IMPL_H #define RAY_GCS_WORKER_INFO_HANDLER_IMPL_H +#include "gcs_table_storage.h" #include "ray/gcs/pubsub/gcs_pub_sub.h" #include "ray/gcs/redis_gcs_client.h" #include "ray/rpc/gcs_server/gcs_rpc_server.h" @@ -25,9 +26,13 @@ namespace rpc { /// This implementation class of `WorkerInfoHandler`. class DefaultWorkerInfoHandler : public rpc::WorkerInfoHandler { public: - explicit DefaultWorkerInfoHandler(gcs::RedisGcsClient &gcs_client, - std::shared_ptr &gcs_pub_sub) - : gcs_client_(gcs_client), gcs_pub_sub_(gcs_pub_sub) {} + explicit DefaultWorkerInfoHandler( + gcs::RedisGcsClient &gcs_client, + std::shared_ptr gcs_table_storage, + std::shared_ptr &gcs_pub_sub) + : gcs_client_(gcs_client), + gcs_table_storage_(gcs_table_storage), + gcs_pub_sub_(gcs_pub_sub) {} void HandleReportWorkerFailure(const ReportWorkerFailureRequest &request, ReportWorkerFailureReply *reply, @@ -39,6 +44,7 @@ class DefaultWorkerInfoHandler : public rpc::WorkerInfoHandler { private: gcs::RedisGcsClient &gcs_client_; + std::shared_ptr gcs_table_storage_; std::shared_ptr gcs_pub_sub_; };