Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Core] fix gRPC handlers' unlimited active calls configuration #25626

Merged
merged 3 commits into from
Jun 10, 2022
Merged

[Core] fix gRPC handlers' unlimited active calls configuration #25626

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
18e164b
fix
mwtian Jun 9, 2022
3251f2d
fix unlimited
mwtian Jun 9, 2022
764f3a9
increase buffer size
mwtian Jun 10, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions src/ray/rpc/grpc_server.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,10 +107,11 @@ void GrpcServer::Run() {
// Create calls for all the server call factories.
for (auto &entry : server_call_factories_) {
for (int i = 0; i < num_threads_; i++) {
// Create a buffer of 100 calls for each RPC handler.
// TODO(edoakes): a small buffer should be fine and seems to have better
// performance, but we don't currently handle backpressure on the client.
int buffer_size = 100;
// When there is no max active RPC limit, a call will be added to the completetion
// queue before RPC processing starts. In this case, the buffer size only
// determines the number of tags in the completion queue, instead of the number of
// inflight RPCs being processed.
int buffer_size = 128;
if (entry->GetMaxActiveRPCs() != -1) {
buffer_size = entry->GetMaxActiveRPCs();
}
Expand Down
18 changes: 7 additions & 11 deletions src/ray/rpc/server_call.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,6 +162,12 @@ class ServerCallImpl : public ServerCall {
start_time_ = absl::GetCurrentTimeNanos();
ray::stats::STATS_grpc_server_req_handling.Record(1.0, call_name_);
if (!io_service_.stopped()) {
if (factory_.GetMaxActiveRPCs() == -1) {
// Create a new `ServerCall` as completion queue tag before handling the request
// when no back pressure limit is set, so that new requests can continue to be
// pulled from the completion queue before this request is done.
factory_.CreateCall();
}
io_service_.post([this] { HandleRequestImpl(); }, call_name_);
} else {
// Handle service for rpc call has stopped, we must handle the call here
Expand All @@ -173,16 +179,6 @@ 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_;
if (factory.GetMaxActiveRPCs() == -1) {
// Create a new `ServerCall` to accept the next incoming request.
// We create this before handling the request only when no back pressure limit is
// set. So that the it can be populated by the completion queue in the background if
// a new request comes in.
factory.CreateCall();
}
(service_handler_.*handle_request_function_)(
request_,
reply_,
Expand Down Expand Up @@ -377,7 +373,7 @@ class ServerCallFactoryImpl : public ServerCallFactory {

/// Maximum request number to handle at the same time.
/// -1 means no limit.
uint64_t max_active_rpcs_;
int64_t max_active_rpcs_;
};

} // namespace rpc
Expand Down