Add --worker-port-list option to ray start (#11481)

src/ray/raylet/main.cc

@@ -36,6 +36,8 @@ DEFINE_int32(min_worker_port, 0,
              "The lowest port that workers' gRPC servers will bind on.");
 DEFINE_int32(max_worker_port, 0,
              "The highest port that workers' gRPC servers will bind on.");
+DEFINE_string(worker_port_list, "",
+              "An explicit list of ports that workers' gRPC servers will bind on.");
 DEFINE_int32(num_initial_workers, 0, "Number of initial workers.");
 DEFINE_int32(num_initial_python_workers_for_first_job, 0,
              "Number of initial Python workers for the first job.");
@@ -75,6 +77,7 @@ int main(int argc, char *argv[]) {
   const int redis_port = static_cast<int>(FLAGS_redis_port);
   const int min_worker_port = static_cast<int>(FLAGS_min_worker_port);
   const int max_worker_port = static_cast<int>(FLAGS_max_worker_port);
+  const std::string worker_port_list = FLAGS_worker_port_list;
   const int num_initial_workers = static_cast<int>(FLAGS_num_initial_workers);
   const int num_initial_python_workers_for_first_job =
       static_cast<int>(FLAGS_num_initial_python_workers_for_first_job);
@@ -150,6 +153,15 @@ int main(int argc, char *argv[]) {
 
         RayConfig::instance().initialize(raylet_config);
 
+        // Parse the worker port list.
+        std::istringstream worker_port_list_string(worker_port_list);
+        std::string worker_port;
+        std::vector<int> worker_ports;
+
+        while (std::getline(worker_port_list_string, worker_port, ',')) {
+          worker_ports.push_back(std::stoi(worker_port));
+        }
+
         // Parse the resource list.
         std::istringstream resource_string(static_resource_list);
         std::string resource_name;
@@ -157,8 +169,6 @@ int main(int argc, char *argv[]) {
 
         while (std::getline(resource_string, resource_name, ',')) {
           RAY_CHECK(std::getline(resource_string, resource_quantity, ','));
-          // TODO(rkn): The line below could throw an exception. What should we do
-          // about this?
           static_resource_conf[resource_name] = std::stod(resource_quantity);
         }
         auto num_cpus_it = static_resource_conf.find("CPU");
@@ -180,6 +190,7 @@ int main(int argc, char *argv[]) {
         node_manager_config.maximum_startup_concurrency = maximum_startup_concurrency;
         node_manager_config.min_worker_port = min_worker_port;
         node_manager_config.max_worker_port = max_worker_port;
+        node_manager_config.worker_ports = worker_ports;
 
         if (!python_worker_command.empty()) {
           node_manager_config.worker_commands.emplace(

src/ray/raylet/node_manager.cc

@@ -142,8 +142,8 @@ NodeManager::NodeManager(boost::asio::io_service &io_service, const NodeID &self
           io_service, config.num_initial_workers, config.num_workers_soft_limit,
           config.num_initial_python_workers_for_first_job,
           config.maximum_startup_concurrency, config.min_worker_port,
-          config.max_worker_port, gcs_client_, config.worker_commands,
-          config.raylet_config,
+          config.max_worker_port, config.worker_ports, gcs_client_,
+          config.worker_commands, config.raylet_config,
           /*starting_worker_timeout_callback=*/
           [this]() { this->DispatchTasks(this->local_queues_.GetReadyTasksByClass()); }),
       scheduling_policy_(local_queues_),

src/ray/raylet/node_manager.h

@@ -65,6 +65,9 @@ struct NodeManagerConfig {
   /// The highest port number that workers started will bind on.
   /// If this is not set to 0, min_worker_port must also not be set to 0.
   int max_worker_port;
+  /// An explicit list of open ports that workers started will bind
+  /// on. This takes precedence over min_worker_port and max_worker_port.
+  std::vector<int> worker_ports;
   /// The initial number of workers to create.
   int num_initial_workers;
   /// The soft limit of the number of workers.

src/ray/raylet/worker_pool.cc

@@ -58,7 +58,8 @@ WorkerPool::WorkerPool(boost::asio::io_service &io_service, int num_workers,
                        int num_workers_soft_limit,
                        int num_initial_python_workers_for_first_job,
                        int maximum_startup_concurrency, int min_worker_port,
-                       int max_worker_port, std::shared_ptr<gcs::GcsClient> gcs_client,
+                       int max_worker_port, const std::vector<int> &worker_ports,
+                       std::shared_ptr<gcs::GcsClient> gcs_client,
                        const WorkerCommandMap &worker_commands,
                        const std::unordered_map<std::string, std::string> &raylet_config,
                        std::function<void()> starting_worker_timeout_callback)
@@ -114,7 +115,12 @@ WorkerPool::WorkerPool(boost::asio::io_service &io_service, int num_workers,
     RAY_CHECK(!state.worker_command.empty()) << "Worker command must not be empty.";
   }
   // Initialize free ports list with all ports in the specified range.
-  if (min_worker_port != 0) {
+  if (!worker_ports.empty()) {
+    free_ports_ = std::unique_ptr<std::queue<int>>(new std::queue<int>());
+    for (int port : worker_ports) {
+      free_ports_->push(port);
+    }
+  } else if (min_worker_port != 0) {
     if (max_worker_port == 0) {
       max_worker_port = 65535;  // Maximum valid port number.
     }

src/ray/raylet/worker_pool.h

@@ -83,6 +83,8 @@ class WorkerPool : public WorkerPoolInterface {
   /// If this is set to 0, workers will bind on random ports.
   /// \param max_worker_port The highest port number that workers started will bind on.
   /// If this is not set to 0, min_worker_port must also not be set to 0.
+  /// \param worker_ports An explicit list of open ports that workers started will bind
+  /// on. This takes precedence over min_worker_port and max_worker_port.
   /// \param worker_commands The commands used to start the worker process, grouped by
   /// language.
   /// \param raylet_config The raylet config list of this node.
@@ -91,6 +93,7 @@ class WorkerPool : public WorkerPoolInterface {
   WorkerPool(boost::asio::io_service &io_service, int num_workers,
              int num_workers_soft_limit, int num_initial_python_workers_for_first_job,
              int maximum_startup_concurrency, int min_worker_port, int max_worker_port,
+             const std::vector<int> &worker_ports,
              std::shared_ptr<gcs::GcsClient> gcs_client,
              const WorkerCommandMap &worker_commands,
              const std::unordered_map<std::string, std::string> &raylet_config,

src/ray/raylet/worker_pool_test.cc

@@ -34,7 +34,7 @@ class WorkerPoolMock : public WorkerPool {
  public:
   explicit WorkerPoolMock(boost::asio::io_service &io_service,
                           const WorkerCommandMap &worker_commands)
-      : WorkerPool(io_service, 0, 0, 0, MAXIMUM_STARTUP_CONCURRENCY, 0, 0, nullptr,
+      : WorkerPool(io_service, 0, 0, 0, MAXIMUM_STARTUP_CONCURRENCY, 0, 0, {}, nullptr,
                    worker_commands, {}, []() {}),
         last_worker_process_() {
     states_by_lang_[ray::Language::JAVA].num_workers_per_process =