From ba7efafa6765539a5c460d2747daae568e590aec Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Wed, 29 Aug 2018 22:43:47 -0700 Subject: [PATCH] Remove force_start argument from StartWorkerProcess. (#2762) This removes the force_start argument from StartWorkerProcess in the worker pool so that no more than maximum_startup_concurrency are ever started concurrently. In particular, when the raylet starts up, it my start fewer than num_workers workers. --- src/ray/raylet/worker_pool.cc | 7 +++---- src/ray/raylet/worker_pool.h | 12 ++++-------- src/ray/raylet/worker_pool_test.cc | 5 ++--- 3 files changed, 9 insertions(+), 15 deletions(-) diff --git a/src/ray/raylet/worker_pool.cc b/src/ray/raylet/worker_pool.cc index 901eb4833..b31318302 100644 --- a/src/ray/raylet/worker_pool.cc +++ b/src/ray/raylet/worker_pool.cc @@ -62,7 +62,7 @@ WorkerPool::WorkerPool( RAY_CHECK(!state.worker_command.empty()) << "Worker command must not be empty."; // Force-start num_workers worker processes for this language. for (int i = 0; i < num_worker_processes; i++) { - StartWorkerProcess(entry.first, true); + StartWorkerProcess(entry.first); } } } @@ -100,12 +100,11 @@ uint32_t WorkerPool::Size(const Language &language) const { } } -void WorkerPool::StartWorkerProcess(const Language &language, bool force_start) { +void WorkerPool::StartWorkerProcess(const Language &language) { // If we are already starting up too many workers, then return without starting // more. if (static_cast(starting_worker_processes_.size()) >= - maximum_startup_concurrency_ && - !force_start) { + maximum_startup_concurrency_) { // Workers have been started, but not registered. Force start disabled -- returning. RAY_LOG(DEBUG) << starting_worker_processes_.size() << " worker processes pending registration"; diff --git a/src/ray/raylet/worker_pool.h b/src/ray/raylet/worker_pool.h index 46c3a6867..368f0e4f3 100644 --- a/src/ray/raylet/worker_pool.h +++ b/src/ray/raylet/worker_pool.h @@ -32,8 +32,7 @@ class WorkerPool { /// \param num_workers_per_process The number of workers per process. /// \param maximum_startup_concurrency The maximum number of worker processes /// that can be started in parallel (typically this should be set to the number of CPU - /// resources on the machine). Note that this limit can be overridden in - /// StartWorkerProcess by the force_start flag. + /// resources on the machine). /// \param worker_commands The commands used to start the worker process, grouped by /// language. WorkerPool( @@ -48,14 +47,11 @@ class WorkerPool { /// registered with an external server, the process should create and /// register num_workers_per_process_ workers, then add them to the pool. /// Failure to start the worker process is a fatal error. If too many workers - /// are already being started and force_start is false, then this function - /// will return without starting any workers. + /// are already being started, then this function will return without starting + /// any workers. /// /// \param language Which language this worker process should be. - /// \param force_start Controls whether to force starting a worker regardless of any - /// workers that have already been started but not yet registered. This overrides - /// the maximum_startup_concurrency_ value. - void StartWorkerProcess(const Language &language, bool force_start = false); + void StartWorkerProcess(const Language &language); /// Register a new worker. The Worker should be added by the caller to the /// pool after it becomes idle (e.g., requests a work assignment). diff --git a/src/ray/raylet/worker_pool_test.cc b/src/ray/raylet/worker_pool_test.cc index 6e9e9b1bc..6dadeb650 100644 --- a/src/ray/raylet/worker_pool_test.cc +++ b/src/ray/raylet/worker_pool_test.cc @@ -17,9 +17,8 @@ class WorkerPoolMock : public WorkerPool { {{Language::PYTHON, {"dummy_py_worker_command"}}, {Language::JAVA, {"dummy_java_worker_command"}}}) {} - void StartWorkerProcess(pid_t pid, const Language &language = Language::PYTHON, - bool force_start = false) { - if (starting_worker_processes_.size() > 0 && !force_start) { + void StartWorkerProcess(pid_t pid, const Language &language = Language::PYTHON) { + if (starting_worker_processes_.size() > 0) { // Workers have been started, but not registered. Force start disabled -- returning. RAY_LOG(DEBUG) << starting_worker_processes_.size() << " worker processes pending registration";