From 2ab1b773d422858e4420d6acfe003b0a56c8d097 Mon Sep 17 00:00:00 2001
From: fangfengbin <869218239a@zju.edu.cn>
Date: Sat, 23 May 2020 23:17:36 +0800
Subject: [PATCH] GCS server worker info handler use storage instead of redis
 accessor (#8543)

---
 src/ray/gcs/gcs_server/gcs_server.cc               |  4 ++--
 src/ray/gcs/gcs_server/worker_info_handler_impl.cc |  4 ++--
 src/ray/gcs/gcs_server/worker_info_handler_impl.h  | 12 +++++++++---
 3 files changed, 13 insertions(+), 7 deletions(-)

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<rpc::ErrorInfoHandler> GcsServer::InitErrorInfoHandler() {
 }
 
 std::unique_ptr<rpc::WorkerInfoHandler> GcsServer::InitWorkerInfoHandler() {
-  return std::unique_ptr<rpc::DefaultWorkerInfoHandler>(
-      new rpc::DefaultWorkerInfoHandler(*redis_gcs_client_, gcs_pub_sub_));
+  return std::unique_ptr<rpc::DefaultWorkerInfoHandler>(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::GcsPubSub> &gcs_pub_sub)
-      : gcs_client_(gcs_client), gcs_pub_sub_(gcs_pub_sub) {}
+  explicit DefaultWorkerInfoHandler(
+      gcs::RedisGcsClient &gcs_client,
+      std::shared_ptr<gcs::GcsTableStorage> gcs_table_storage,
+      std::shared_ptr<gcs::GcsPubSub> &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::GcsTableStorage> gcs_table_storage_;
   std::shared_ptr<gcs::GcsPubSub> gcs_pub_sub_;
 };