wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-20 12:40:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

GCS server worker info handler use storage instead of redis accessor (#8543)

Browse Source
This commit is contained in:
fangfengbin
2020-05-23 23:17:36 +08:00
committed by GitHub
parent 351839bf69
commit 2ab1b773d4
3 changed files with 13 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/gcs/gcs_server/gcs_server.cc
+2 -2
View File
@@ -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
src/ray/gcs/gcs_server/worker_info_handler_impl.cc
+2 -2
View File
@@ -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);
}
src/ray/gcs/gcs_server/worker_info_handler_impl.h
+9 -3
View File
@@ -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_;
};