Provide functionality for local scheduler to start new workers. (#230)

* Provide functionality for local scheduler to start new workers.

* Pass full command for starting new worker in to local scheduler.

* Separate out configuration state of local scheduler.

python/photon/photon_services.py

@@ -10,7 +10,11 @@ import time
 def random_name():
   return str(random.randint(0, 99999999))
 
-def start_local_scheduler(plasma_store_name, plasma_manager_name=None, plasma_address=None, node_ip_address="127.0.0.1", redis_address=None, use_valgrind=False, use_profiler=False, redirect_output=False):
+def start_local_scheduler(plasma_store_name, plasma_manager_name=None,
+                          worker_path=None, plasma_address=None,
+                          node_ip_address="127.0.0.1", redis_address=None,
+                          use_valgrind=False, use_profiler=False,
+                          redirect_output=False):
   """Start a local scheduler process.
 
   Args:
@@ -18,6 +22,8 @@ def start_local_scheduler(plasma_store_name, plasma_manager_name=None, plasma_ad
     plasma_manager_name (str): The name of the plasma manager to connect to.
       This does not need to be provided, but if it is, then the Redis address
       must be provided as well.
+    worker_path (str): The path of the worker script to use when the local
+      scheduler starts up new workers.
     plasma_address (str): The address of the plasma manager to connect to. This
       is only used by the global scheduler to figure out which plasma managers
       are connected to which local schedulers.
@@ -45,6 +51,22 @@ def start_local_scheduler(plasma_store_name, plasma_manager_name=None, plasma_ad
   command = [local_scheduler_executable, "-s", local_scheduler_name, "-p", plasma_store_name, "-h", node_ip_address]
   if plasma_manager_name is not None:
     command += ["-m", plasma_manager_name]
+  if worker_path is not None:
+    assert plasma_store_name is not None
+    assert plasma_manager_name is not None
+    assert redis_address is not None
+    start_worker_command = ("python {} "
+                            "--node-ip-address={} "
+                            "--object-store-name={} "
+                            "--object-store-manager-name={} "
+                            "--local-scheduler-name={} "
+                            "--redis-address={}").format(worker_path,
+                                                         node_ip_address,
+                                                         plasma_store_name,
+                                                         plasma_manager_name,
+                                                         local_scheduler_name,
+                                                         redis_address)
+    command += ["-w", start_worker_command]
   if redis_address is not None:
     command += ["-r", redis_address]
   if plasma_address is not None:

python/ray/services.py

@@ -247,7 +247,9 @@ def start_global_scheduler(redis_address, cleanup=True, redirect_output=False):
   if cleanup:
     all_processes[PROCESS_TYPE_GLOBAL_SCHEDULER].append(p)
 
-def start_local_scheduler(redis_address, node_ip_address, plasma_store_name, plasma_manager_name, plasma_address=None, cleanup=True, redirect_output=False):
+def start_local_scheduler(redis_address, node_ip_address, plasma_store_name,
+                          plasma_manager_name, worker_path, plasma_address=None,
+                          cleanup=True, redirect_output=False):
   """Start a local scheduler process.
 
   Args:
@@ -257,6 +259,8 @@ def start_local_scheduler(redis_address, node_ip_address, plasma_store_name, pla
     plasma_store_name (str): The name of the plasma store socket to connect to.
     plasma_manager_name (str): The name of the plasma manager socket to connect
       to.
+    worker_path (str): The path of the script to use when the local scheduler
+      starts up new workers.
     cleanup (bool): True if using Ray in local mode. If cleanup is true, then
       this process will be killed by serices.cleanup() when the Python process
       that imported services exits.
@@ -266,7 +270,14 @@ def start_local_scheduler(redis_address, node_ip_address, plasma_store_name, pla
   Return:
     The name of the local scheduler socket.
   """
-  local_scheduler_name, p = photon.start_local_scheduler(plasma_store_name, plasma_manager_name, node_ip_address=node_ip_address, redis_address=redis_address, plasma_address=plasma_address, use_profiler=RUN_PHOTON_PROFILER, redirect_output=redirect_output)
+  local_scheduler_name, p = photon.start_local_scheduler(plasma_store_name,
+                                                         plasma_manager_name,
+                                                         worker_path=worker_path,
+                                                         node_ip_address=node_ip_address,
+                                                         redis_address=redis_address,
+                                                         plasma_address=plasma_address,
+                                                         use_profiler=RUN_PHOTON_PROFILER,
+                                                         redirect_output=redirect_output)
   if cleanup:
     all_processes[PROCESS_TYPE_LOCAL_SCHEDULER].append(p)
   return local_scheduler_name
@@ -439,6 +450,7 @@ def start_ray_processes(address_info=None,
                                                  node_ip_address,
                                                  object_store_address.name,
                                                  object_store_address.manager_name,
+                                                 worker_path,
                                                  plasma_address=plasma_address,
                                                  cleanup=cleanup,
                                                  redirect_output=redirect_output)

src/photon/photon.h

@@ -55,8 +55,20 @@ typedef struct {
 /** Internal state of the scheduling algorithm. */
 typedef struct scheduling_algorithm_state scheduling_algorithm_state;
 
+/** A struct storing the configuration state of the local scheduler. This should
+ *  consist of values that don't change over the lifetime of the local
+ *  scheduler. */
+typedef struct {
+  /** The script to use when starting a new worker. */
+  char *start_worker_command;
+  /** Whether there is a global scheduler. */
+  bool global_scheduler_exists;
+} local_scheduler_config;
+
 /** The state of the local scheduler. */
 typedef struct {
+  /** The configuration for the local scheduler. */
+  local_scheduler_config config;
   /** The local scheduler event loop. */
   event_loop *loop;
   /** Association between client socket and worker index. */
@@ -67,8 +79,6 @@ typedef struct {
   UT_array *workers;
   /** The handle to the database. */
   db_handle *db;
-  /** Whether there is a global scheduler. */
-  bool global_scheduler_exists;
   /** The Plasma client. */
   plasma_connection *plasma_conn;
   /** State for the scheduling algorithm. */

src/photon/photon_algorithm.c

@@ -394,13 +394,13 @@ void queue_task_locally(local_scheduler_state *state,
 void give_task_to_global_scheduler(local_scheduler_state *state,
                                    scheduling_algorithm_state *algorithm_state,
                                    task_spec *spec) {
-  if (state->db == NULL || !state->global_scheduler_exists) {
+  if (state->db == NULL || !state->config.global_scheduler_exists) {
     /* A global scheduler is not available, so queue the task locally. */
     queue_task_locally(state, algorithm_state, spec, false);
     return;
   }
   /* Pass on the task to the global scheduler. */
-  DCHECK(state->global_scheduler_exists);
+  DCHECK(state->config.global_scheduler_exists);
   task *task = alloc_task(spec, TASK_STATUS_WAITING, NIL_ID);
   DCHECK(state->db != NULL);
   task_table_add_task(state->db, task, (retry_info *) &photon_retry, NULL,
@@ -447,7 +447,7 @@ void handle_task_scheduled(local_scheduler_state *state,
    * the global scheduler, so we can safely assert that there is a connection
    * to the database. */
   DCHECK(state->db != NULL);
-  DCHECK(state->global_scheduler_exists);
+  DCHECK(state->config.global_scheduler_exists);
   /* Push the task to the appropriate queue. */
   queue_task_locally(state, algorithm_state, spec, true);
   dispatch_tasks(state, algorithm_state);
@@ -499,7 +499,6 @@ void handle_object_available(local_scheduler_state *state,
   if (entry->dependent_tasks != NULL) {
     /* Out of the tasks that were dependent on this object, if they were now
      * ready to run, move them to the dispatch queue. */
-    task_queue_entry *task_entry = NULL;
     for (task_queue_entry **p =
              (task_queue_entry **) utarray_front(entry->dependent_tasks);
          p != NULL;

src/photon/photon_scheduler.c

@@ -34,8 +34,17 @@ local_scheduler_state *init_local_scheduler(
     const char *plasma_store_socket_name,
     const char *plasma_manager_socket_name,
     const char *plasma_manager_address,
-    bool global_scheduler_exists) {
+    bool global_scheduler_exists,
+    const char *start_worker_command) {
   local_scheduler_state *state = malloc(sizeof(local_scheduler_state));
+  /* Set the configuration struct for the local scheduler. */
+  if (start_worker_command != NULL) {
+    state->config.start_worker_command = strdup(start_worker_command);
+  } else {
+    state->config.start_worker_command = NULL;
+  }
+  state->config.global_scheduler_exists = global_scheduler_exists;
+
   state->loop = loop;
   state->worker_index = NULL;
   /* Add scheduler info. */
@@ -73,8 +82,6 @@ local_scheduler_state *init_local_scheduler(
   /* Add the callback that processes the notification to the event loop. */
   event_loop_add_file(loop, plasma_fd, EVENT_LOOP_READ,
                       process_plasma_notification, state);
-  /* Set the flag for whether there is a global scheduler. */
-  state->global_scheduler_exists = global_scheduler_exists;
   /* Add scheduler state. */
   state->algorithm_state = make_scheduling_algorithm_state();
   utarray_new(state->input_buffer, &byte_icd);
@@ -82,6 +89,11 @@ local_scheduler_state *init_local_scheduler(
 };
 
 void free_local_scheduler(local_scheduler_state *state) {
+  if (state->config.start_worker_command != NULL) {
+    free(state->config.start_worker_command);
+    state->config.start_worker_command = NULL;
+  }
+
   if (state->db != NULL) {
     db_disconnect(state->db);
   }
@@ -346,6 +358,14 @@ int heartbeat_handler(event_loop *loop, timer_id id, void *context) {
   return LOCAL_SCHEDULER_HEARTBEAT_TIMEOUT_MILLISECONDS;
 }
 
+void start_new_worker(local_scheduler_state *state) {
+  /* We can't start a worker if we don't have the path to the worker script. */
+  CHECK(state->config.start_worker_command != NULL);
+  /* Launch the process to create the worker. */
+  FILE *p = popen(state->config.start_worker_command, "r");
+  UNUSED(p);
+}
+
 void start_server(const char *node_ip_address,
                   const char *socket_name,
                   const char *redis_addr,
@@ -353,7 +373,8 @@ void start_server(const char *node_ip_address,
                   const char *plasma_store_socket_name,
                   const char *plasma_manager_socket_name,
                   const char *plasma_manager_address,
-                  bool global_scheduler_exists) {
+                  bool global_scheduler_exists,
+                  const char *start_worker_command) {
   /* Ignore SIGPIPE signals. If we don't do this, then when we attempt to write
    * to a client that has already died, the local scheduler could die. */
   signal(SIGPIPE, SIG_IGN);
@@ -362,8 +383,7 @@ void start_server(const char *node_ip_address,
   g_state = init_local_scheduler(
       node_ip_address, loop, redis_addr, redis_port, socket_name,
       plasma_store_socket_name, plasma_manager_socket_name,
-      plasma_manager_address, global_scheduler_exists);
-
+      plasma_manager_address, global_scheduler_exists, start_worker_command);
   /* Register a callback for registering new clients. */
   event_loop_add_file(loop, fd, EVENT_LOOP_READ, new_client_connection,
                       g_state);
@@ -412,9 +432,11 @@ int main(int argc, char *argv[]) {
   char *plasma_manager_address = NULL;
   /* The IP address of the node that this local scheduler is running on. */
   char *node_ip_address = NULL;
+  /* The command to run when starting new workers. */
+  char *start_worker_command = NULL;
   int c;
   bool global_scheduler_exists = true;
-  while ((c = getopt(argc, argv, "s:r:p:m:ga:h:")) != -1) {
+  while ((c = getopt(argc, argv, "s:r:p:m:ga:h:w:")) != -1) {
     switch (c) {
     case 's':
       scheduler_socket_name = optarg;
@@ -437,6 +459,9 @@ int main(int argc, char *argv[]) {
     case 'h':
       node_ip_address = optarg;
       break;
+    case 'w':
+      start_worker_command = optarg;
+      break;
     default:
       LOG_FATAL("unknown option %c", c);
     }
@@ -449,7 +474,7 @@ int main(int argc, char *argv[]) {
         "please specify socket for connecting to Plasma store with -p switch");
   }
   if (!node_ip_address) {
-    LOG_FATAL("please specify the node IP address with -p switch");
+    LOG_FATAL("please specify the node IP address with -h switch");
   }
   if (!redis_addr_port) {
     /* Start the local scheduler without connecting to Redis. In this case, all
@@ -461,7 +486,7 @@ int main(int argc, char *argv[]) {
     }
     start_server(node_ip_address, scheduler_socket_name, NULL, -1,
                  plasma_store_socket_name, NULL, plasma_manager_address,
-                 global_scheduler_exists);
+                 global_scheduler_exists, start_worker_command);
   } else {
     /* Parse the Redis address into an IP address and a port. */
     char redis_addr[16] = {0};
@@ -481,7 +506,7 @@ int main(int argc, char *argv[]) {
     start_server(node_ip_address, scheduler_socket_name, &redis_addr[0],
                  atoi(redis_port), plasma_store_socket_name,
                  plasma_manager_socket_name, plasma_manager_address,
-                 global_scheduler_exists);
+                 global_scheduler_exists, start_worker_command);
   }
 }
 #endif

src/photon/photon_scheduler.h

@@ -75,7 +75,8 @@ local_scheduler_state *init_local_scheduler(
     const char *plasma_manager_socket_name,
     const char *plasma_store_socket_name,
     const char *plasma_manager_address,
-    bool global_scheduler_exists);
+    bool global_scheduler_exists,
+    const char *worker_path);
 
 void free_local_scheduler(local_scheduler_state *state);
 

src/photon/test/photon_tests.c

@@ -67,7 +67,7 @@ photon_mock *init_photon_mock(bool connect_to_redis) {
   mock->photon_state = init_local_scheduler(
       "127.0.0.1", mock->loop, redis_addr, redis_port,
       utstring_body(photon_socket_name), plasma_store_socket_name,
-      utstring_body(plasma_manager_socket_name), NULL, false);
+      utstring_body(plasma_manager_socket_name), NULL, false, NULL);
   /* Connect a Photon client. */
   mock->conn = photon_connect(utstring_body(photon_socket_name));
   new_client_connection(mock->loop, mock->photon_fd,