message: Make max concurrent handlers in run handler configurable.
diff: @@ -146,8 +146,11 @@ class ThreadWorkSource {
  // resource. The non-blocking threads are used only to compensate for
  // threads that may be blocked on some tasks. There is less need to
  // proactively wake up those threads.
  - static int max_rank_to_wakeup = static_cast<int>(ParamFromEnvWithDefault(
  - "TF_RUN_HANDLER_MAX_RANK_TO_WAKE_UP", kMaxConcurrentHandlers));
  + static int max_rank_to_wakeup = static_cast<int>(
  + ParamFromEnvWithDefault("TF_RUN_HANDLER_MAX_RANK_TO_WAKE_UP",
  + static_cast<int32>(ParamFromEnvWithDefault(
  + "TF_RUN_HANDLER_MAX_CONCURRENT_HANDLERS",
  + kMaxConcurrentHandlers))));
  if (max_rank_to_wakeup > 0 &&
  rank_.load(std::memory_order_relaxed) <= max_rank_to_wakeup) {
  Waiter* w = nullptr;
  @@ -414,7 +417,10 @@ class RunHandlerThreadPool {
  private:
  struct ThreadData {
  - ThreadData() : thread_work_sources(kMaxConcurrentHandlers) {}
  + ThreadData()
  + : thread_work_sources(static_cast<int32>(
  + ParamFromEnvWithDefault("TF_RUN_HANDLER_MAX_CONCURRENT_HANDLERS",
  + kMaxConcurrentHandlers))) {}
  mutex mu;
  condition_variable sources_not_empty;
  std::unique_ptr<Thread> thread;
  @@ -604,7 +610,8 @@ class RunHandler::Impl {
  class RunHandlerPool::Impl {
  public:
  explicit Impl(int num_inter_op_threads, int num_intra_op_threads)
  - : max_handlers_(kMaxConcurrentHandlers),
  + : max_handlers_(static_cast<int32>(ParamFromEnvWithDefault(
  + "TF_RUN_HANDLER_MAX_CONCURRENT_HANDLERS", kMaxConcurrentHandlers))),
  run_handler_thread_pool_(new RunHandlerThreadPool(
  num_inter_op_threads, num_intra_op_threads, Env::Default(),
  ThreadOptions(), "tf_run_handler_pool")),
  