diff --git a/src/ray/rpc/grpc_server.cc b/src/ray/rpc/grpc_server.cc index be7fe90ea..bac98a583 100644 --- a/src/ray/rpc/grpc_server.cc +++ b/src/ray/rpc/grpc_server.cc @@ -64,9 +64,7 @@ void GrpcServer::PollEventsFromCompletionQueue() { switch (server_call->GetState()) { case ServerCallState::PENDING: // We've received a new incoming request. Now this call object is used to - // track this request. So we need to create another call to handle next - // incoming request. - server_call->GetFactory().CreateCall(); + // track this request. server_call->SetState(ServerCallState::PROCESSING); server_call->HandleRequest(); break; diff --git a/src/ray/rpc/grpc_server.h b/src/ray/rpc/grpc_server.h index 8f4c975df..727301329 100644 --- a/src/ray/rpc/grpc_server.h +++ b/src/ray/rpc/grpc_server.h @@ -87,7 +87,7 @@ class GrpcServer { /// The `grpc::Service` objects which should be registered to `ServerBuilder`. std::vector> services_; /// The `ServerCallFactory` objects, and the maximum number of concurrent requests that - /// gRPC server can accept. + /// this gRPC server can handle. std::vector, int>> server_call_factories_and_concurrencies_; /// The `ServerCompletionQueue` object used for polling events. @@ -121,13 +121,11 @@ class GrpcService { /// Subclasses should implement this method to initialize the `ServerCallFactory` /// instances, as well as specify maximum number of concurrent requests that gRPC - /// server can "accept" (not "handle"). Each factory will be used to create - /// `accept_concurrency` `ServerCall` objects, each of which will be used to accept and - /// handle an incoming request. + /// server can handle. /// /// \param[in] cq The grpc completion queue. /// \param[out] server_call_factories_and_concurrencies The `ServerCallFactory` objects, - /// and the maximum number of concurrent requests that gRPC server can accept. + /// and the maximum number of concurrent requests that this gRPC server can handle. virtual void InitServerCallFactories( const std::unique_ptr &cq, std::vector, int>> diff --git a/src/ray/rpc/server_call.h b/src/ray/rpc/server_call.h index 750ecc8f6..87fab5019 100644 --- a/src/ray/rpc/server_call.h +++ b/src/ray/rpc/server_call.h @@ -63,9 +63,6 @@ class ServerCall { /// `GrpcServer` when the request is received. virtual void HandleRequest() = 0; - /// Get the factory that created this `ServerCall`. - virtual const ServerCallFactory &GetFactory() const = 0; - /// Invoked when sending reply successes. virtual void OnReplySent() = 0; @@ -141,6 +138,9 @@ class ServerCallImpl : public ServerCall { void HandleRequestImpl() { state_ = ServerCallState::PROCESSING; + // NOTE(hchen): This `factory` local variable is needed. Because `SendReply` runs in + // a different thread, and will cause `this` to be deleted. + const auto &factory = factory_; (service_handler_.*handle_request_function_)( request_, &reply_, [this](Status status, std::function success, @@ -155,10 +155,11 @@ class ServerCallImpl : public ServerCall { // this server call might be deleted SendReply(status); }); + // We've finished handling this request, + // create a new `ServerCall` to accept the next incoming request. + factory.CreateCall(); } - const ServerCallFactory &GetFactory() const override { return factory_; } - void OnReplySent() override { if (send_reply_success_callback_ && !io_service_.stopped()) { auto callback = std::move(send_reply_success_callback_);