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

@@ -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<int>(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";

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).

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";