diff --git a/python/photon/photon_services.py b/python/photon/photon_services.py index 5452fbdb4..9e330497a 100644 --- a/python/photon/photon_services.py +++ b/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: diff --git a/python/ray/services.py b/python/ray/services.py index d7edde005..c7a7d216d 100644 --- a/python/ray/services.py +++ b/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) diff --git a/src/photon/photon.h b/src/photon/photon.h index 9208a1cd8..b1b2b2699 100644 --- a/src/photon/photon.h +++ b/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. */ diff --git a/src/photon/photon_algorithm.c b/src/photon/photon_algorithm.c index 7dba09b77..73d9ba42f 100644 --- a/src/photon/photon_algorithm.c +++ b/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; diff --git a/src/photon/photon_scheduler.c b/src/photon/photon_scheduler.c index 9f0185151..89fc4dd0c 100644 --- a/src/photon/photon_scheduler.c +++ b/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 diff --git a/src/photon/photon_scheduler.h b/src/photon/photon_scheduler.h index 5e52d8f1f..aebcee48a 100644 --- a/src/photon/photon_scheduler.h +++ b/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); diff --git a/src/photon/test/photon_tests.c b/src/photon/test/photon_tests.c index 2e5788628..edea72140 100644 --- a/src/photon/test/photon_tests.c +++ b/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,