From 72335dbe464e58938d172bccf676e44dd5b5b102 Mon Sep 17 00:00:00 2001 From: Zhijun Fu <37800433+zhijunfu@users.noreply.github.com> Date: Tue, 7 Jan 2020 22:03:42 +0800 Subject: [PATCH] [rpc] refactor RPC server code (#6661) * refactor RPC client * remove unused code * format * fix * resolve comments * format * update * refactor rpc server * update * format * fix * fix * Update src/ray/rpc/worker/core_worker_server.h Co-Authored-By: Edward Oakes * resolve comments * format * update * update * add a comment * fix Co-authored-by: Edward Oakes --- src/ray/core_worker/core_worker.h | 32 +++--- src/ray/object_manager/object_manager.cc | 16 ++- src/ray/object_manager/object_manager.h | 14 +-- src/ray/raylet/node_manager.cc | 18 ++-- src/ray/raylet/node_manager.h | 14 +-- src/ray/rpc/gcs_server/gcs_rpc_server.h | 98 +++++++------------ src/ray/rpc/grpc_server.h | 15 +++ .../rpc/node_manager/node_manager_server.h | 70 +++---------- .../object_manager/object_manager_server.h | 47 +++------ src/ray/rpc/worker/core_worker_server.cc | 28 ------ src/ray/rpc/worker/core_worker_server.h | 47 ++++++++- 11 files changed, 168 insertions(+), 231 deletions(-) delete mode 100644 src/ray/rpc/worker/core_worker_server.cc diff --git a/src/ray/core_worker/core_worker.h b/src/ray/core_worker/core_worker.h index f78e13a80..ca4dc1174 100644 --- a/src/ray/core_worker/core_worker.h +++ b/src/ray/core_worker/core_worker.h @@ -26,23 +26,19 @@ /// The set of gRPC handlers and their associated level of concurrency. If you want to /// add a new call to the worker gRPC server, do the following: /// 1) Add the rpc to the CoreWorkerService in core_worker.proto, e.g., "ExampleCall" -/// 2) Add a new handler to the macro below: "RAY_CORE_WORKER_RPC_HANDLER(ExampleCall, 1)" -/// 3) Add a method to the CoreWorker class below: "CoreWorker::HandleExampleCall" -#define RAY_CORE_WORKER_RPC_HANDLERS \ - RAY_CORE_WORKER_RPC_HANDLER(AssignTask, 5) \ - RAY_CORE_WORKER_RPC_HANDLER(PushTask, 9999) \ - RAY_CORE_WORKER_RPC_HANDLER(DirectActorCallArgWaitComplete, 100) \ - RAY_CORE_WORKER_RPC_HANDLER(GetObjectStatus, 9999) \ - RAY_CORE_WORKER_RPC_HANDLER(WaitForObjectEviction, 9999) \ - RAY_CORE_WORKER_RPC_HANDLER(KillActor, 9999) \ - RAY_CORE_WORKER_RPC_HANDLER(GetCoreWorkerStats, 100) +/// 2) Add a new macro to RAY_CORE_WORKER_DECLARE_RPC_HANDLERS +/// in core_worker_server.h, +// e.g. "DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(ExampleCall)" +/// 3) Add a new macro to RAY_CORE_WORKER_RPC_HANDLERS in core_worker_server.h, e.g. +/// "RPC_SERVICE_HANDLER(CoreWorkerService, ExampleCall, 1)" +/// 4) Add a method to the CoreWorker class below: "CoreWorker::HandleExampleCall" namespace ray { /// The root class that contains all the core and language-independent functionalities /// of the worker. This class is supposed to be used to implement app-language (Java, /// Python, etc) workers. -class CoreWorker { +class CoreWorker : public rpc::CoreWorkerServiceHandler { // Callback that must be implemented and provided by the language-specific worker // frontend to execute tasks and return their results. using TaskExecutionCallback = std::function check_signals = nullptr, bool ref_counting_enabled = false); - ~CoreWorker(); + virtual ~CoreWorker(); void Disconnect(); @@ -400,22 +396,22 @@ class CoreWorker { /// Implements gRPC server handler. void HandleAssignTask(const rpc::AssignTaskRequest &request, rpc::AssignTaskReply *reply, - rpc::SendReplyCallback send_reply_callback); + rpc::SendReplyCallback send_reply_callback) override; /// Implements gRPC server handler. void HandlePushTask(const rpc::PushTaskRequest &request, rpc::PushTaskReply *reply, - rpc::SendReplyCallback send_reply_callback); + rpc::SendReplyCallback send_reply_callback) override; /// Implements gRPC server handler. void HandleDirectActorCallArgWaitComplete( const rpc::DirectActorCallArgWaitCompleteRequest &request, rpc::DirectActorCallArgWaitCompleteReply *reply, - rpc::SendReplyCallback send_reply_callback); + rpc::SendReplyCallback send_reply_callback) override; /// Implements gRPC server handler. void HandleGetObjectStatus(const rpc::GetObjectStatusRequest &request, rpc::GetObjectStatusReply *reply, - rpc::SendReplyCallback send_reply_callback); + rpc::SendReplyCallback send_reply_callback) override; /// Implements gRPC server handler. void HandleWaitForObjectEviction(const rpc::WaitForObjectEvictionRequest &request, @@ -424,12 +420,12 @@ class CoreWorker { /// Implements gRPC server handler. void HandleKillActor(const rpc::KillActorRequest &request, rpc::KillActorReply *reply, - rpc::SendReplyCallback send_reply_callback); + rpc::SendReplyCallback send_reply_callback) override; /// Get statistics from core worker. void HandleGetCoreWorkerStats(const rpc::GetCoreWorkerStatsRequest &request, rpc::GetCoreWorkerStatsReply *reply, - rpc::SendReplyCallback send_reply_callback); + rpc::SendReplyCallback send_reply_callback) override; /// /// Public methods related to async actor call. This should only be used when diff --git a/src/ray/object_manager/object_manager.cc b/src/ray/object_manager/object_manager.cc index 1bfedaa7f..3a5c7276a 100644 --- a/src/ray/object_manager/object_manager.cc +++ b/src/ray/object_manager/object_manager.cc @@ -665,9 +665,8 @@ void ObjectManager::WaitComplete(const UniqueID &wait_id) { } /// Implementation of ObjectManagerServiceHandler -void ObjectManager::HandlePushRequest(const rpc::PushRequest &request, - rpc::PushReply *reply, - rpc::SendReplyCallback send_reply_callback) { +void ObjectManager::HandlePush(const rpc::PushRequest &request, rpc::PushReply *reply, + rpc::SendReplyCallback send_reply_callback) { ObjectID object_id = ObjectID::FromBinary(request.object_id()); ClientID client_id = ClientID::FromBinary(request.client_id()); @@ -712,9 +711,8 @@ ray::Status ObjectManager::ReceiveObjectChunk(const ClientID &client_id, return status; } -void ObjectManager::HandlePullRequest(const rpc::PullRequest &request, - rpc::PullReply *reply, - rpc::SendReplyCallback send_reply_callback) { +void ObjectManager::HandlePull(const rpc::PullRequest &request, rpc::PullReply *reply, + rpc::SendReplyCallback send_reply_callback) { ObjectID object_id = ObjectID::FromBinary(request.object_id()); ClientID client_id = ClientID::FromBinary(request.client_id()); RAY_LOG(DEBUG) << "Received pull request from client " << client_id << " for object [" @@ -735,9 +733,9 @@ void ObjectManager::HandlePullRequest(const rpc::PullRequest &request, send_reply_callback(Status::OK(), nullptr, nullptr); } -void ObjectManager::HandleFreeObjectsRequest(const rpc::FreeObjectsRequest &request, - rpc::FreeObjectsReply *reply, - rpc::SendReplyCallback send_reply_callback) { +void ObjectManager::HandleFreeObjects(const rpc::FreeObjectsRequest &request, + rpc::FreeObjectsReply *reply, + rpc::SendReplyCallback send_reply_callback) { std::vector object_ids; for (const auto &e : request.object_ids()) { object_ids.emplace_back(ObjectID::FromBinary(e)); diff --git a/src/ray/object_manager/object_manager.h b/src/ray/object_manager/object_manager.h index 917768fdf..8a1f1ca5d 100644 --- a/src/ray/object_manager/object_manager.h +++ b/src/ray/object_manager/object_manager.h @@ -81,25 +81,25 @@ class ObjectManager : public ObjectManagerInterface, /// \param request Push request including the object chunk data /// \param reply Reply to the sender /// \param send_reply_callback Callback of the request - void HandlePushRequest(const rpc::PushRequest &request, rpc::PushReply *reply, - rpc::SendReplyCallback send_reply_callback) override; + void HandlePush(const rpc::PushRequest &request, rpc::PushReply *reply, + rpc::SendReplyCallback send_reply_callback) override; /// Handle pull request from remote object manager /// /// \param request Pull request /// \param reply Reply /// \param send_reply_callback Callback of request - void HandlePullRequest(const rpc::PullRequest &request, rpc::PullReply *reply, - rpc::SendReplyCallback send_reply_callback) override; + void HandlePull(const rpc::PullRequest &request, rpc::PullReply *reply, + rpc::SendReplyCallback send_reply_callback) override; /// Handle free objects request /// /// \param request Free objects request /// \param reply Reply /// \param send_reply_callback - void HandleFreeObjectsRequest(const rpc::FreeObjectsRequest &request, - rpc::FreeObjectsReply *reply, - rpc::SendReplyCallback send_reply_callback) override; + void HandleFreeObjects(const rpc::FreeObjectsRequest &request, + rpc::FreeObjectsReply *reply, + rpc::SendReplyCallback send_reply_callback) override; /// Send object to remote object manager /// diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index f418a9b28..25d21825a 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -1496,7 +1496,7 @@ void NodeManager::WaitForTaskArgsRequests(std::pair &work) { } }; -void NodeManager::HandleWorkerLeaseRequest(const rpc::RequestWorkerLeaseRequest &request, +void NodeManager::HandleRequestWorkerLease(const rpc::RequestWorkerLeaseRequest &request, rpc::RequestWorkerLeaseReply *reply, rpc::SendReplyCallback send_reply_callback) { rpc::Task task_message; @@ -2968,9 +2968,9 @@ std::string compact_tag_string(const opencensus::stats::ViewDescriptor &view, return result.str(); } -void NodeManager::HandlePinObjectIDsRequest(const rpc::PinObjectIDsRequest &request, - rpc::PinObjectIDsReply *reply, - rpc::SendReplyCallback send_reply_callback) { +void NodeManager::HandlePinObjectIDs(const rpc::PinObjectIDsRequest &request, + rpc::PinObjectIDsReply *reply, + rpc::SendReplyCallback send_reply_callback) { if (!object_pinning_enabled_) { send_reply_callback(Status::OK(), nullptr, nullptr); return; @@ -3044,9 +3044,9 @@ void NodeManager::HandlePinObjectIDsRequest(const rpc::PinObjectIDsRequest &requ send_reply_callback(Status::OK(), nullptr, nullptr); } -void NodeManager::HandleNodeStatsRequest(const rpc::GetNodeStatsRequest &request, - rpc::GetNodeStatsReply *reply, - rpc::SendReplyCallback send_reply_callback) { +void NodeManager::HandleGetNodeStats(const rpc::GetNodeStatsRequest &request, + rpc::GetNodeStatsReply *reply, + rpc::SendReplyCallback send_reply_callback) { for (const auto &driver : worker_pool_.GetAllDrivers()) { auto worker_stats = reply->add_workers_stats(); worker_stats->set_pid(driver->Pid()); @@ -3094,11 +3094,11 @@ void NodeManager::HandleNodeStatsRequest(const rpc::GetNodeStatsRequest &request } } } - // As a result of the HandleNodeStatsRequest, we are collecting information from all + // As a result of the HandleGetNodeStats, we are collecting information from all // workers on this node. This is done by calling GetCoreWorkerStats on each worker. In // order to send up-to-date information back, we wait until all workers have replied, // and return the information from HandleNodesStatsRequest. The caller of - // HandleNodeStatsRequest should set a timeout so that the rpc finishes even if not all + // HandleGetNodeStats should set a timeout so that the rpc finishes even if not all // workers have replied. auto all_workers = worker_pool_.GetAllWorkers(); for (const auto &worker : all_workers) { diff --git a/src/ray/raylet/node_manager.h b/src/ray/raylet/node_manager.h index ddfda825f..5809f50d0 100644 --- a/src/ray/raylet/node_manager.h +++ b/src/ray/raylet/node_manager.h @@ -534,7 +534,7 @@ class NodeManager : public rpc::NodeManagerServiceHandler { bool success); /// Handle a `WorkerLease` request. - void HandleWorkerLeaseRequest(const rpc::RequestWorkerLeaseRequest &request, + void HandleRequestWorkerLease(const rpc::RequestWorkerLeaseRequest &request, rpc::RequestWorkerLeaseReply *reply, rpc::SendReplyCallback send_reply_callback) override; @@ -549,14 +549,14 @@ class NodeManager : public rpc::NodeManagerServiceHandler { rpc::SendReplyCallback send_reply_callback) override; /// Handle a `PinObjectIDs` request. - void HandlePinObjectIDsRequest(const rpc::PinObjectIDsRequest &request, - rpc::PinObjectIDsReply *reply, - rpc::SendReplyCallback send_reply_callback) override; + void HandlePinObjectIDs(const rpc::PinObjectIDsRequest &request, + rpc::PinObjectIDsReply *reply, + rpc::SendReplyCallback send_reply_callback) override; /// Handle a `NodeStats` request. - void HandleNodeStatsRequest(const rpc::GetNodeStatsRequest &request, - rpc::GetNodeStatsReply *reply, - rpc::SendReplyCallback send_reply_callback) override; + void HandleGetNodeStats(const rpc::GetNodeStatsRequest &request, + rpc::GetNodeStatsReply *reply, + rpc::SendReplyCallback send_reply_callback) override; /// Push an error to the driver if this node is full of actors and so we are /// unable to schedule new tasks or actors at all. diff --git a/src/ray/rpc/gcs_server/gcs_rpc_server.h b/src/ray/rpc/gcs_server/gcs_rpc_server.h index a3f28055d..c4bc63b0c 100644 --- a/src/ray/rpc/gcs_server/gcs_rpc_server.h +++ b/src/ray/rpc/gcs_server/gcs_rpc_server.h @@ -9,54 +9,24 @@ namespace ray { namespace rpc { -#define JOB_INFO_SERVICE_RPC_HANDLER(HANDLER, CONCURRENCY) \ - std::unique_ptr HANDLER##_call_factory( \ - new ServerCallFactoryImpl( \ - service_, &JobInfoGcsService::AsyncService::Request##HANDLER, \ - service_handler_, &JobInfoHandler::Handle##HANDLER, cq, main_service_)); \ - server_call_factories_and_concurrencies->emplace_back( \ - std::move(HANDLER##_call_factory), CONCURRENCY); +#define JOB_INFO_SERVICE_RPC_HANDLER(HANDLER, CONCURRENCY) \ + RPC_SERVICE_HANDLER(JobInfoGcsService, HANDLER, CONCURRENCY) -#define ACTOR_INFO_SERVICE_RPC_HANDLER(HANDLER, CONCURRENCY) \ - std::unique_ptr HANDLER##_call_factory( \ - new ServerCallFactoryImpl( \ - service_, &ActorInfoGcsService::AsyncService::Request##HANDLER, \ - service_handler_, &ActorInfoHandler::Handle##HANDLER, cq, main_service_)); \ - server_call_factories_and_concurrencies->emplace_back( \ - std::move(HANDLER##_call_factory), CONCURRENCY); +#define ACTOR_INFO_SERVICE_RPC_HANDLER(HANDLER, CONCURRENCY) \ + RPC_SERVICE_HANDLER(ActorInfoGcsService, HANDLER, CONCURRENCY) -#define NODE_INFO_SERVICE_RPC_HANDLER(HANDLER, CONCURRENCY) \ - std::unique_ptr HANDLER##_call_factory( \ - new ServerCallFactoryImpl( \ - service_, &NodeInfoGcsService::AsyncService::Request##HANDLER, \ - service_handler_, &NodeInfoHandler::Handle##HANDLER, cq, main_service_)); \ - server_call_factories_and_concurrencies->emplace_back( \ - std::move(HANDLER##_call_factory), CONCURRENCY); +#define NODE_INFO_SERVICE_RPC_HANDLER(HANDLER, CONCURRENCY) \ + RPC_SERVICE_HANDLER(NodeInfoGcsService, HANDLER, CONCURRENCY) -#define OBJECT_INFO_SERVICE_RPC_HANDLER(HANDLER, CONCURRENCY) \ - std::unique_ptr HANDLER##_call_factory( \ - new ServerCallFactoryImpl( \ - service_, &ObjectInfoGcsService::AsyncService::Request##HANDLER, \ - service_handler_, &ObjectInfoHandler::Handle##HANDLER, cq, main_service_)); \ - server_call_factories_and_concurrencies->emplace_back( \ - std::move(HANDLER##_call_factory), CONCURRENCY); +#define OBJECT_INFO_SERVICE_RPC_HANDLER(HANDLER, CONCURRENCY) \ + RPC_SERVICE_HANDLER(ObjectInfoGcsService, HANDLER, CONCURRENCY) -#define TASK_INFO_SERVICE_RPC_HANDLER(HANDLER, CONCURRENCY) \ - std::unique_ptr HANDLER##_call_factory( \ - new ServerCallFactoryImpl( \ - service_, &TaskInfoGcsService::AsyncService::Request##HANDLER, \ - service_handler_, &TaskInfoHandler::Handle##HANDLER, cq, main_service_)); \ - server_call_factories_and_concurrencies->emplace_back( \ - std::move(HANDLER##_call_factory), CONCURRENCY); +#define TASK_INFO_SERVICE_RPC_HANDLER(HANDLER, CONCURRENCY) \ + RPC_SERVICE_HANDLER(TaskInfoGcsService, HANDLER, CONCURRENCY) -class JobInfoHandler { +class JobInfoGcsServiceHandler { public: - virtual ~JobInfoHandler() = default; + virtual ~JobInfoGcsServiceHandler() = default; virtual void HandleAddJob(const AddJobRequest &request, AddJobReply *reply, SendReplyCallback send_reply_callback) = 0; @@ -73,7 +43,7 @@ class JobInfoGrpcService : public GrpcService { /// /// \param[in] handler The service handler that actually handle the requests. explicit JobInfoGrpcService(boost::asio::io_service &io_service, - JobInfoHandler &handler) + JobInfoGcsServiceHandler &handler) : GrpcService(io_service), service_handler_(handler){}; protected: @@ -91,12 +61,12 @@ class JobInfoGrpcService : public GrpcService { /// The grpc async service object. JobInfoGcsService::AsyncService service_; /// The service handler that actually handle the requests. - JobInfoHandler &service_handler_; + JobInfoGcsServiceHandler &service_handler_; }; -class ActorInfoHandler { +class ActorInfoGcsServiceHandler { public: - virtual ~ActorInfoHandler() = default; + virtual ~ActorInfoGcsServiceHandler() = default; virtual void HandleGetActorInfo(const GetActorInfoRequest &request, GetActorInfoReply *reply, @@ -130,7 +100,7 @@ class ActorInfoGrpcService : public GrpcService { /// /// \param[in] handler The service handler that actually handle the requests. explicit ActorInfoGrpcService(boost::asio::io_service &io_service, - ActorInfoHandler &handler) + ActorInfoGcsServiceHandler &handler) : GrpcService(io_service), service_handler_(handler){}; protected: @@ -152,12 +122,12 @@ class ActorInfoGrpcService : public GrpcService { /// The grpc async service object. ActorInfoGcsService::AsyncService service_; /// The service handler that actually handle the requests. - ActorInfoHandler &service_handler_; + ActorInfoGcsServiceHandler &service_handler_; }; -class NodeInfoHandler { +class NodeInfoGcsServiceHandler { public: - virtual ~NodeInfoHandler() = default; + virtual ~NodeInfoGcsServiceHandler() = default; virtual void HandleRegisterNode(const RegisterNodeRequest &request, RegisterNodeReply *reply, @@ -199,7 +169,7 @@ class NodeInfoGrpcService : public GrpcService { /// /// \param[in] handler The service handler that actually handle the requests. explicit NodeInfoGrpcService(boost::asio::io_service &io_service, - NodeInfoHandler &handler) + NodeInfoGcsServiceHandler &handler) : GrpcService(io_service), service_handler_(handler){}; protected: @@ -223,12 +193,12 @@ class NodeInfoGrpcService : public GrpcService { /// The grpc async service object. NodeInfoGcsService::AsyncService service_; /// The service handler that actually handle the requests. - NodeInfoHandler &service_handler_; + NodeInfoGcsServiceHandler &service_handler_; }; -class ObjectInfoHandler { +class ObjectInfoGcsServiceHandler { public: - virtual ~ObjectInfoHandler() = default; + virtual ~ObjectInfoGcsServiceHandler() = default; virtual void HandleGetObjectLocations(const GetObjectLocationsRequest &request, GetObjectLocationsReply *reply, @@ -243,14 +213,14 @@ class ObjectInfoHandler { SendReplyCallback send_reply_callback) = 0; }; -/// The `GrpcService` for `ObjectInfoHandler`. +/// The `GrpcService` for `ObjectInfoGcsServiceHandler`. class ObjectInfoGrpcService : public GrpcService { public: /// Constructor. /// /// \param[in] handler The service handler that actually handle the requests. explicit ObjectInfoGrpcService(boost::asio::io_service &io_service, - ObjectInfoHandler &handler) + ObjectInfoGcsServiceHandler &handler) : GrpcService(io_service), service_handler_(handler){}; protected: @@ -269,12 +239,12 @@ class ObjectInfoGrpcService : public GrpcService { /// The grpc async service object. ObjectInfoGcsService::AsyncService service_; /// The service handler that actually handle the requests. - ObjectInfoHandler &service_handler_; + ObjectInfoGcsServiceHandler &service_handler_; }; -class TaskInfoHandler { +class TaskInfoGcsServiceHandler { public: - virtual ~TaskInfoHandler() = default; + virtual ~TaskInfoGcsServiceHandler() = default; virtual void HandleAddTask(const AddTaskRequest &request, AddTaskReply *reply, SendReplyCallback send_reply_callback) = 0; @@ -294,7 +264,7 @@ class TaskInfoGrpcService : public GrpcService { /// /// \param[in] handler The service handler that actually handle the requests. explicit TaskInfoGrpcService(boost::asio::io_service &io_service, - TaskInfoHandler &handler) + TaskInfoGcsServiceHandler &handler) : GrpcService(io_service), service_handler_(handler){}; protected: @@ -313,9 +283,15 @@ class TaskInfoGrpcService : public GrpcService { /// The grpc async service object. TaskInfoGcsService::AsyncService service_; /// The service handler that actually handle the requests. - TaskInfoHandler &service_handler_; + TaskInfoGcsServiceHandler &service_handler_; }; +using JobInfoHandler = JobInfoGcsServiceHandler; +using ActorInfoHandler = ActorInfoGcsServiceHandler; +using NodeInfoHandler = NodeInfoGcsServiceHandler; +using ObjectInfoHandler = ObjectInfoGcsServiceHandler; +using TaskInfoHandler = TaskInfoGcsServiceHandler; + } // namespace rpc } // namespace ray diff --git a/src/ray/rpc/grpc_server.h b/src/ray/rpc/grpc_server.h index 9e5fb60b8..5f5e8f876 100644 --- a/src/ray/rpc/grpc_server.h +++ b/src/ray/rpc/grpc_server.h @@ -13,6 +13,21 @@ namespace ray { namespace rpc { +#define RPC_SERVICE_HANDLER(SERVICE, HANDLER, CONCURRENCY) \ + std::unique_ptr HANDLER##_call_factory( \ + new ServerCallFactoryImpl( \ + service_, &SERVICE::AsyncService::Request##HANDLER, service_handler_, \ + &SERVICE##Handler::Handle##HANDLER, cq, main_service_)); \ + server_call_factories_and_concurrencies->emplace_back( \ + std::move(HANDLER##_call_factory), CONCURRENCY); + +// Define a void RPC client method. +#define DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(METHOD) \ + virtual void Handle##METHOD(const rpc::METHOD##Request &request, \ + rpc::METHOD##Reply *reply, \ + rpc::SendReplyCallback send_reply_callback) = 0; + class GrpcService; /// Class that represents an gRPC server. diff --git a/src/ray/rpc/node_manager/node_manager_server.h b/src/ray/rpc/node_manager/node_manager_server.h index 9c292884d..bcb2328de 100644 --- a/src/ray/rpc/node_manager/node_manager_server.h +++ b/src/ray/rpc/node_manager/node_manager_server.h @@ -9,6 +9,14 @@ namespace ray { namespace rpc { +/// NOTE: See src/ray/core_worker/core_worker.h on how to add a new grpc handler. +#define RAY_NODE_MANAGER_RPC_HANDLERS \ + RPC_SERVICE_HANDLER(NodeManagerService, RequestWorkerLease, 100) \ + RPC_SERVICE_HANDLER(NodeManagerService, ReturnWorker, 100) \ + RPC_SERVICE_HANDLER(NodeManagerService, ForwardTask, 100) \ + RPC_SERVICE_HANDLER(NodeManagerService, PinObjectIDs, 100) \ + RPC_SERVICE_HANDLER(NodeManagerService, GetNodeStats, 1) + /// Interface of the `NodeManagerService`, see `src/ray/protobuf/node_manager.proto`. class NodeManagerServiceHandler { public: @@ -23,7 +31,7 @@ class NodeManagerServiceHandler { /// \param[out] reply The reply message. /// \param[in] send_reply_callback The callback to be called when the request is done. - virtual void HandleWorkerLeaseRequest(const RequestWorkerLeaseRequest &request, + virtual void HandleRequestWorkerLease(const RequestWorkerLeaseRequest &request, RequestWorkerLeaseReply *reply, SendReplyCallback send_reply_callback) = 0; @@ -35,13 +43,13 @@ class NodeManagerServiceHandler { ForwardTaskReply *reply, SendReplyCallback send_reply_callback) = 0; - virtual void HandlePinObjectIDsRequest(const PinObjectIDsRequest &request, - PinObjectIDsReply *reply, - SendReplyCallback send_reply_callback) = 0; + virtual void HandlePinObjectIDs(const PinObjectIDsRequest &request, + PinObjectIDsReply *reply, + SendReplyCallback send_reply_callback) = 0; - virtual void HandleNodeStatsRequest(const GetNodeStatsRequest &request, - GetNodeStatsReply *reply, - SendReplyCallback send_reply_callback) = 0; + virtual void HandleGetNodeStats(const GetNodeStatsRequest &request, + GetNodeStatsReply *reply, + SendReplyCallback send_reply_callback) = 0; }; /// The `GrpcService` for `NodeManagerService`. @@ -62,53 +70,7 @@ class NodeManagerGrpcService : public GrpcService { const std::unique_ptr &cq, std::vector, int>> *server_call_factories_and_concurrencies) override { - // Initialize the factory for requests. - std::unique_ptr request_worker_lease_call_factory( - new ServerCallFactoryImpl( - service_, &NodeManagerService::AsyncService::RequestRequestWorkerLease, - service_handler_, &NodeManagerServiceHandler::HandleWorkerLeaseRequest, cq, - main_service_)); - - std::unique_ptr release_worker_call_factory( - new ServerCallFactoryImpl( - service_, &NodeManagerService::AsyncService::RequestReturnWorker, - service_handler_, &NodeManagerServiceHandler::HandleReturnWorker, cq, - main_service_)); - - std::unique_ptr forward_task_call_factory( - new ServerCallFactoryImpl( - service_, &NodeManagerService::AsyncService::RequestForwardTask, - service_handler_, &NodeManagerServiceHandler::HandleForwardTask, cq, - main_service_)); - - std::unique_ptr pin_object_ids_call_factory( - new ServerCallFactoryImpl( - service_, &NodeManagerService::AsyncService::RequestPinObjectIDs, - service_handler_, &NodeManagerServiceHandler::HandlePinObjectIDsRequest, cq, - main_service_)); - - std::unique_ptr node_stats_call_factory( - new ServerCallFactoryImpl( - service_, &NodeManagerService::AsyncService::RequestGetNodeStats, - service_handler_, &NodeManagerServiceHandler::HandleNodeStatsRequest, cq, - main_service_)); - - // Set accept concurrency. - server_call_factories_and_concurrencies->emplace_back( - std::move(request_worker_lease_call_factory), 100); - server_call_factories_and_concurrencies->emplace_back( - std::move(release_worker_call_factory), 100); - server_call_factories_and_concurrencies->emplace_back( - std::move(forward_task_call_factory), 100); - server_call_factories_and_concurrencies->emplace_back( - std::move(pin_object_ids_call_factory), 100); - server_call_factories_and_concurrencies->emplace_back( - std::move(node_stats_call_factory), 1); + RAY_NODE_MANAGER_RPC_HANDLERS } private: diff --git a/src/ray/rpc/object_manager/object_manager_server.h b/src/ray/rpc/object_manager/object_manager_server.h index 496d22591..c8c62fc25 100644 --- a/src/ray/rpc/object_manager/object_manager_server.h +++ b/src/ray/rpc/object_manager/object_manager_server.h @@ -10,6 +10,11 @@ namespace ray { namespace rpc { +#define RAY_OBJECT_MANAGER_RPC_HANDLERS \ + RPC_SERVICE_HANDLER(ObjectManagerService, Push, 5) \ + RPC_SERVICE_HANDLER(ObjectManagerService, Pull, 5) \ + RPC_SERVICE_HANDLER(ObjectManagerService, FreeObjects, 2) + /// Implementations of the `ObjectManagerGrpcService`, check interface in /// `src/ray/protobuf/object_manager.proto`. class ObjectManagerServiceHandler { @@ -21,15 +26,15 @@ class ObjectManagerServiceHandler { /// \param[in] request The request message. /// \param[out] reply The reply message. /// \param[in] send_reply_callback The callback to be called when the request is done. - virtual void HandlePushRequest(const PushRequest &request, PushReply *reply, - SendReplyCallback send_reply_callback) = 0; + virtual void HandlePush(const PushRequest &request, PushReply *reply, + SendReplyCallback send_reply_callback) = 0; /// Handle a `Pull` request - virtual void HandlePullRequest(const PullRequest &request, PullReply *reply, - SendReplyCallback send_reply_callback) = 0; + virtual void HandlePull(const PullRequest &request, PullReply *reply, + SendReplyCallback send_reply_callback) = 0; /// Handle a `FreeObjects` request - virtual void HandleFreeObjectsRequest(const FreeObjectsRequest &request, - FreeObjectsReply *reply, - SendReplyCallback send_reply_callback) = 0; + virtual void HandleFreeObjects(const FreeObjectsRequest &request, + FreeObjectsReply *reply, + SendReplyCallback send_reply_callback) = 0; }; /// The `GrpcService` for `ObjectManagerGrpcService`. @@ -50,33 +55,7 @@ class ObjectManagerGrpcService : public GrpcService { const std::unique_ptr &cq, std::vector, int>> *server_call_factories_and_concurrencies) override { - // Initialize the factory for `Push` requests. - std::unique_ptr push_call_factory( - new ServerCallFactoryImpl( - service_, &ObjectManagerService::AsyncService::RequestPush, service_handler_, - &ObjectManagerServiceHandler::HandlePushRequest, cq, main_service_)); - server_call_factories_and_concurrencies->emplace_back(std::move(push_call_factory), - 5); - - // Initialize the factory for `Pull` requests. - std::unique_ptr pull_call_factory( - new ServerCallFactoryImpl( - service_, &ObjectManagerService::AsyncService::RequestPull, service_handler_, - &ObjectManagerServiceHandler::HandlePullRequest, cq, main_service_)); - server_call_factories_and_concurrencies->emplace_back(std::move(pull_call_factory), - 5); - - // Initialize the factory for `FreeObjects` requests. - std::unique_ptr free_objects_call_factory( - new ServerCallFactoryImpl( - service_, &ObjectManagerService::AsyncService::RequestFreeObjects, - service_handler_, &ObjectManagerServiceHandler::HandleFreeObjectsRequest, cq, - main_service_)); - server_call_factories_and_concurrencies->emplace_back( - std::move(free_objects_call_factory), 2); + RAY_OBJECT_MANAGER_RPC_HANDLERS } private: diff --git a/src/ray/rpc/worker/core_worker_server.cc b/src/ray/rpc/worker/core_worker_server.cc deleted file mode 100644 index 165c9031a..000000000 --- a/src/ray/rpc/worker/core_worker_server.cc +++ /dev/null @@ -1,28 +0,0 @@ -#include "ray/rpc/worker/core_worker_server.h" -#include "ray/core_worker/core_worker.h" - -namespace ray { -namespace rpc { - -#define RAY_CORE_WORKER_RPC_HANDLER(HANDLER, CONCURRENCY) \ - std::unique_ptr HANDLER##_call_factory( \ - new ServerCallFactoryImpl( \ - service_, &CoreWorkerService::AsyncService::Request##HANDLER, core_worker_, \ - &CoreWorker::Handle##HANDLER, cq, main_service_)); \ - server_call_factories_and_concurrencies->emplace_back( \ - std::move(HANDLER##_call_factory), CONCURRENCY); - -CoreWorkerGrpcService::CoreWorkerGrpcService(boost::asio::io_service &main_service, - CoreWorker &core_worker) - : GrpcService(main_service), core_worker_(core_worker){}; - -void CoreWorkerGrpcService::InitServerCallFactories( - const std::unique_ptr &cq, - std::vector, int>> - *server_call_factories_and_concurrencies) { - RAY_CORE_WORKER_RPC_HANDLERS -} - -} // namespace rpc -} // namespace ray diff --git a/src/ray/rpc/worker/core_worker_server.h b/src/ray/rpc/worker/core_worker_server.h index 9e58929c5..10bd3441f 100644 --- a/src/ray/rpc/worker/core_worker_server.h +++ b/src/ray/rpc/worker/core_worker_server.h @@ -13,6 +13,41 @@ class CoreWorker; namespace rpc { +/// NOTE: See src/ray/core_worker/core_worker.h on how to add a new grpc handler. +#define RAY_CORE_WORKER_RPC_HANDLERS \ + RPC_SERVICE_HANDLER(CoreWorkerService, AssignTask, 5) \ + RPC_SERVICE_HANDLER(CoreWorkerService, PushTask, 9999) \ + RPC_SERVICE_HANDLER(CoreWorkerService, DirectActorCallArgWaitComplete, 100) \ + RPC_SERVICE_HANDLER(CoreWorkerService, GetObjectStatus, 9999) \ + RPC_SERVICE_HANDLER(CoreWorkerService, WaitForObjectEviction, 9999) \ + RPC_SERVICE_HANDLER(CoreWorkerService, KillActor, 9999) \ + RPC_SERVICE_HANDLER(CoreWorkerService, GetCoreWorkerStats, 100) + +#define RAY_CORE_WORKER_DECLARE_RPC_HANDLERS \ + DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(AssignTask) \ + DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(PushTask) \ + DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(DirectActorCallArgWaitComplete) \ + DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(GetObjectStatus) \ + DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(WaitForObjectEviction) \ + DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(KillActor) \ + DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(GetCoreWorkerStats) + +/// Interface of the `CoreWorkerServiceHandler`, see `src/ray/protobuf/core_worker.proto`. +class CoreWorkerServiceHandler { + public: + /// Handlers. For all of the following handlers, the implementations can + /// handle the request asynchronously. When handling is done, the + /// `send_reply_callback` should be called. See + /// src/ray/rpc/node_manager/node_manager_client.h and + /// src/ray/protobuf/node_manager.proto for a description of the + /// functionality of each handler. + /// + /// \param[in] request The request message. + /// \param[out] reply The reply message. + /// \param[in] send_reply_callback The callback to be called when the request is done. + RAY_CORE_WORKER_DECLARE_RPC_HANDLERS +}; + /// The `GrpcServer` for `CoreWorkerService`. class CoreWorkerGrpcService : public GrpcService { public: @@ -20,7 +55,9 @@ class CoreWorkerGrpcService : public GrpcService { /// /// \param[in] main_service See super class. /// \param[in] handler The service handler that actually handle the requests. - CoreWorkerGrpcService(boost::asio::io_service &main_service, CoreWorker &core_worker); + CoreWorkerGrpcService(boost::asio::io_service &main_service, + CoreWorkerServiceHandler &service_handler) + : GrpcService(main_service), service_handler_(service_handler) {} protected: grpc::Service &GetGrpcService() override { return service_; } @@ -28,14 +65,16 @@ class CoreWorkerGrpcService : public GrpcService { void InitServerCallFactories( const std::unique_ptr &cq, std::vector, int>> - *server_call_factories_and_concurrencies) override; + *server_call_factories_and_concurrencies) override { + RAY_CORE_WORKER_RPC_HANDLERS + } private: /// The grpc async service object. CoreWorkerService::AsyncService service_; - /// The core worker that actually handles the requests. - CoreWorker &core_worker_; + /// The service handler that actually handles the requests. + CoreWorkerServiceHandler &service_handler_; }; } // namespace rpc