set num server call thread for core worker to two

Signed-off-by: yicheng <[email protected]>

Author: yicheng

src/ray/common/ray_config_def.h

@@ -898,11 +898,19 @@ RAY_CONFIG(int64_t, health_check_timeout_ms, 10000)
 /// The threshold to consider a node dead.
 RAY_CONFIG(int64_t, health_check_failure_threshold, 5)
 
-/// The pool size for grpc server call.
+/// The pool size for grpc server call for system components (raylet, GCS, etc.). All
+/// processes that run a CoreWorker (workers, drivers, io workers) have a dedicated config.
 RAY_CONFIG(int64_t,
            num_server_call_thread,
            std::max((int64_t)1, (int64_t)(std::thread::hardware_concurrency() / 4U)))
 
+/// The pool size for grpc server call inside CoreWorker processes (workers, drivers, io
+/// workers). https://github.com/ray-project/ray/issues/58351 shows the
+/// reply path is light enough that 2 threads is sufficient.
+RAY_CONFIG(int64_t,
+           core_worker_num_server_call_thread,
+           std::min<int64_t>(2, (int64_t)std::thread::hardware_concurrency()))
+
 /// Use madvise to prevent worker/raylet coredumps from including
 /// the mapped plasma pages.
 RAY_CONFIG(bool, worker_core_dump_exclude_plasma_store, true)

src/ray/core_worker/core_worker_process.cc

@@ -15,6 +15,7 @@
 #include "ray/core_worker/core_worker_process.h"
 
 #include <chrono>
+#include <cstdlib>
 #include <memory>
 #include <string>
 #include <thread>
@@ -34,6 +35,7 @@
 #include "ray/pubsub/subscriber.h"
 #include "ray/raylet_ipc_client/raylet_ipc_client.h"
 #include "ray/raylet_rpc_client/raylet_client.h"
+#include "ray/rpc/server_call.h"
 #include "ray/stats/stats.h"
 #include "ray/stats/tag_defs.h"
 #include "ray/util/env.h"
@@ -775,6 +777,7 @@ CoreWorkerProcessImpl::CoreWorkerProcessImpl(const CoreWorkerOptions &options)
   // NOTE(kfstorm): any initialization depending on RayConfig must happen after this
   // line.
   InitializeSystemConfig();
+  rpc::SetServerCallThreadPoolMode(rpc::ServerCallThreadPoolMode::CORE_WORKER);
 
   // Assume stats module will be initialized exactly once in once process.
   // So it must be called in CoreWorkerProcess constructor and will be reused

src/ray/rpc/server_call.cc

@@ -14,6 +14,7 @@
 
 #include "ray/rpc/server_call.h"
 
+#include <atomic>
 #include <memory>
 
 #include "ray/common/ray_config.h"
@@ -22,9 +23,17 @@ namespace ray {
 namespace rpc {
 namespace {
 
+std::atomic<ServerCallThreadPoolMode> &ThreadPoolMode() {
+  static std::atomic<ServerCallThreadPoolMode> mode(ServerCallThreadPoolMode::SYSTEM_COMPONENT);
+  return mode;
+}
+
 std::unique_ptr<boost::asio::thread_pool> &_GetServerCallExecutor() {
   static auto thread_pool = std::make_unique<boost::asio::thread_pool>(
-      ::RayConfig::instance().num_server_call_thread());
+      ThreadPoolMode().load(std::memory_order_acquire) ==
+              ServerCallThreadPoolMode::CORE_WORKER
+          ? ::RayConfig::instance().core_worker_num_server_call_thread()
+          : ::RayConfig::instance().num_server_call_thread());
   return thread_pool;
 }
 
@@ -36,7 +45,14 @@ void DrainServerCallExecutor() { GetServerCallExecutor().join(); }
 
 void ResetServerCallExecutor() {
   _GetServerCallExecutor() = std::make_unique<boost::asio::thread_pool>(
-      ::RayConfig::instance().num_server_call_thread());
+      ThreadPoolMode().load(std::memory_order_acquire) ==
+              ServerCallThreadPoolMode::CORE_WORKER
+          ? ::RayConfig::instance().core_worker_num_server_call_thread()
+          : ::RayConfig::instance().num_server_call_thread());
+}
+
+void SetServerCallThreadPoolMode(ServerCallThreadPoolMode mode) {
+  ThreadPoolMode().store(mode, std::memory_order_release);
 }
 
 }  // namespace rpc

src/ray/rpc/server_call.h

@@ -50,6 +50,8 @@ enum class ClusterIdAuthType {
 /// This pool is shared across gRPC servers.
 boost::asio::thread_pool &GetServerCallExecutor();
 
+enum class ServerCallThreadPoolMode { SYSTEM_COMPONENT = 0, CORE_WORKER = 1 };
+
 /// Drain the executor.
 void DrainServerCallExecutor();
 
@@ -59,6 +61,10 @@ void DrainServerCallExecutor();
 /// because they are global.
 void ResetServerCallExecutor();
 
+/// Set the executor mode for the current process. This should be called before any RPC
+/// replies are processed so the thread pool is created with the correct size.
+void SetServerCallThreadPoolMode(ServerCallThreadPoolMode mode);
+
 /// Represents state of a `ServerCall`.
 enum class ServerCallState {
   /// The call is created and waiting for an incoming request.