From 80526f7777f126284631418610f06c345322fd2c Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Tue, 12 Jul 2016 23:54:18 -0700 Subject: [PATCH] add documentation and refactor cluster.py (#238) --- doc/api.rst | 6 + doc/cluster-api.rst | 9 + doc/conf.py | 5 +- doc/index.rst | 14 +- doc/services-api.rst | 5 + doc/using-ray-on-a-cluster.md | 36 +-- lib/python/ray/services.py | 132 +++++---- lib/python/ray/worker.py | 421 +++++++++++++++++++++++----- scripts/cluster.py | 499 +++++++++++++++++----------------- 9 files changed, 719 insertions(+), 408 deletions(-) create mode 100644 doc/cluster-api.rst create mode 100644 doc/services-api.rst diff --git a/doc/api.rst b/doc/api.rst index 152e8e474..b80b6f40b 100644 --- a/doc/api.rst +++ b/doc/api.rst @@ -5,3 +5,9 @@ The Ray API .. autofunction:: ray.put .. autofunction:: ray.get .. autofunction:: ray.remote +.. autofunction:: ray.kill_workers +.. autofunction:: ray.restart_workers_local +.. autofunction:: ray.visualize_computation_graph +.. autofunction:: ray.scheduler_info +.. autofunction:: ray.task_info +.. autofunction:: ray.register_module diff --git a/doc/cluster-api.rst b/doc/cluster-api.rst new file mode 100644 index 000000000..34ffb5c5a --- /dev/null +++ b/doc/cluster-api.rst @@ -0,0 +1,9 @@ +=============== +The Cluster API +=============== + +.. automethod:: cluster.RayCluster.install_ray +.. automethod:: cluster.RayCluster.start_ray +.. automethod:: cluster.RayCluster.stop_ray +.. automethod:: cluster.RayCluster.restart_workers +.. automethod:: cluster.RayCluster.update_ray diff --git a/doc/conf.py b/doc/conf.py index 0adf5dd01..ae34d0a4e 100644 --- a/doc/conf.py +++ b/doc/conf.py @@ -19,7 +19,8 @@ import shlex # If extensions (or modules to document with autodoc) are in another directory, # add these directories to sys.path here. If the directory is relative to the # documentation root, use os.path.abspath to make it absolute, like shown here. -#sys.path.insert(0, os.path.abspath('.')) +sys.path.insert(0, os.path.abspath("../lib/python/")) +sys.path.insert(0, os.path.abspath("../scripts/")) # -- General configuration ------------------------------------------------ @@ -32,6 +33,7 @@ import shlex extensions = [ 'sphinx.ext.autodoc', 'sphinx.ext.pngmath', + 'sphinx.ext.napoleon', ] # Add any paths that contain templates here, relative to this directory. @@ -106,7 +108,6 @@ pygments_style = 'sphinx' # If true, `todo` and `todoList` produce output, else they produce nothing. todo_include_todos = False - # -- Options for HTML output ---------------------------------------------- # The theme to use for HTML and HTML Help pages. See the documentation for diff --git a/doc/index.rst b/doc/index.rst index d65f4c274..c1cdf23d6 100644 --- a/doc/index.rst +++ b/doc/index.rst @@ -6,17 +6,11 @@ Welcome to Ray's documentation! =============================== -Contents: +API: .. toctree:: :maxdepth: 2 - :glob: - * - -Indices and tables -================== - -* :ref:`genindex` -* :ref:`modindex` -* :ref:`search` + api + cluster-api + services-api diff --git a/doc/services-api.rst b/doc/services-api.rst new file mode 100644 index 000000000..f5b8be965 --- /dev/null +++ b/doc/services-api.rst @@ -0,0 +1,5 @@ +================ +The Services API +================ + +.. autofunction:: ray.services.start_ray_local diff --git a/doc/using-ray-on-a-cluster.md b/doc/using-ray-on-a-cluster.md index a5c208bed..14646cb93 100644 --- a/doc/using-ray-on-a-cluster.md +++ b/doc/using-ray-on-a-cluster.md @@ -40,9 +40,9 @@ appropriate values. This assumes that you can connect to each IP address ssh -i @ ``` 4. The previous command should open a Python interpreter. To install Ray on the -cluster, run `install_ray()` in the interpreter. The interpreter should block -until the installation has completed. The standard output from the nodes will -be redirected to your terminal. +cluster, run `cluster.install_ray()` in the interpreter. The interpreter should +block until the installation has completed. The standard output from the nodes +will be redirected to your terminal. 5. To check that the installation succeeded, you can ssh to each node, cd into the directory `ray/test/`, and run the tests (e.g., `python runtest.py`). 6. Create a directory (for example, `mkdir ~/example_ray_code`) containing the @@ -55,9 +55,9 @@ worker `worker.py` code along with the code for any modules imported by ``` 7. Start the cluster (the scheduler, object stores, and workers) with the -command `start_ray("~/example_ray_code")`, where the second argument is the -local path to the worker code that you would like to use. This command will copy -the worker code to each node and will start the cluster. After completing +command `cluster.start_ray("~/example_ray_code")`, where the second argument is +the local path to the worker code that you would like to use. This command will +copy the worker code to each node and will start the cluster. After completing successfully, this command will print out a command that can be run on the head node to attach a shell (the driver) to the cluster. For example, @@ -69,17 +69,19 @@ node to attach a shell (the driver) to the cluster. For example, 8. Note that there are several more commands that can be run from within `cluster.py`. - - `install_ray()` - This pulls the Ray source code on each node, builds all - of the third party libraries, and builds the project itself. - - `start_ray(worker_directory, num_workers_per_node=10)` - This starts a - scheduler process on the head node, and it starts an object store and some - workers on each node. - - `stop_ray()` - This shuts down the cluster (killing all of the processes). - - `restart_workers(worker_directory, num_workers_per_node=10)` - This kills - the current workers and starts new workers using the worker code from the - given file. Currently, this can only run when there are no tasks currently - executing on any of the workers. - - `update_ray()` - This pulls the latest Ray source code and builds it. + - `cluster.install_ray()` - This pulls the Ray source code on each node, + builds all of the third party libraries, and builds the project itself. + - `cluster.start_ray(worker_directory, num_workers_per_node=10)` - This + starts a scheduler process on the head node, and it starts an object store + and some workers on each node. + - `cluster.stop_ray()` - This shuts down the cluster (killing all of the + processes). + - `cluster.restart_workers(worker_directory, num_workers_per_node=10)` - + This kills the current workers and starts new workers using the worker + code from the given file. Currently, this can only run when there are no + tasks currently executing on any of the workers. + - `cluster.update_ray()` - This pulls the latest Ray source code and builds + it. ### Running Ray on a cluster diff --git a/lib/python/ray/services.py b/lib/python/ray/services.py index a0e48beb4..cde00103a 100644 --- a/lib/python/ray/services.py +++ b/lib/python/ray/services.py @@ -43,16 +43,16 @@ def new_objstore_port(): return 20000 + objstore_port_counter def cleanup(): - """ + """When running in local mode, shutdown the Ray processes. + This method is used to shutdown processes that were started with - services.start_ray_local(). It kills all scheduler, object store, and worker - processes that were started by this services module. It disconnects driver - processes but does not kill them. Users should not invoke this manually. It - will automatically run at the end when a Python process that imports - services exits. It is ok to run this twice in a row. Note that we manaully - call services.cleanup() in the tests because we need to start and stop many - clusters in the tests, but in the tests, services is only imported and only - exits once. + services.start_ray_local(). It kills all scheduler, object store, and worker + processes that were started by this services module. It disconnects driver + processes but does not kill them. This will automatically run at the end when + a Python process that imports services exits. It is ok to run this twice in a + row. Note that we manually call services.cleanup() in the tests because we + need to start and stop many clusters in the tests, but in the tests, services + is only imported and only exits once. """ global drivers for driver in drivers: @@ -84,43 +84,47 @@ def cleanup(): atexit.register(cleanup) def start_scheduler(scheduler_address, local): - """ - This method starts a scheduler process. + """This method starts a scheduler process. - :param scheduler_address: The ip address and port to use for the scheduler. - :param local: True if using Ray in local mode. If local is true, then this - process will be killed by serices.cleanup() when the Python process that - imported services exits. + Args: + scheduler_address (str): The ip address and port to use for the scheduler. + local (bool): True if using Ray in local mode. If local is true, then this + process will be killed by serices.cleanup() when the Python process that + imported services exits. """ p = subprocess.Popen(["scheduler", scheduler_address, "--log-file-name", config.get_log_file_path("scheduler.log")], env=_services_env) if local: all_processes.append((p, scheduler_address)) def start_objstore(scheduler_address, objstore_address, local): - """ - This method starts an object store process. + """This method starts an object store process. - :param scheduler_address: The ip address and port of the scheduler to connect to. - :param objstore_address: The ip address and port to use for the object store. - :param local: True if using Ray in local mode. If local is true, then this - process will be killed by serices.cleanup() when the Python process that - imported services exits. + Args: + scheduler_address (str): The ip address and port of the scheduler to connect + to. + objstore_address (str): The ip address and port to use for the object store. + local (bool): True if using Ray in local mode. If local is true, then this + process will be killed by serices.cleanup() when the Python process that + imported services exits. """ p = subprocess.Popen(["objstore", scheduler_address, objstore_address, "--log-file-name", config.get_log_file_path("-".join(["objstore", objstore_address]) + ".log")], env=_services_env) if local: all_processes.append((p, objstore_address)) def start_worker(worker_path, scheduler_address, objstore_address, worker_address, local): - """ - This method starts a worker process. + """This method starts a worker process. - :param worker_path: The path of the source code which the worker process will run. - :param scheduler_address: The ip address and port of the scheduler to connect to. - :param objstore_address: The ip address and port of the object store to connect to. - :param worker_address: The ip address and port to use for the worker. - :param local: True if using Ray in local mode. If local is true, then this - process will be killed by serices.cleanup() when the Python process that - imported services exits. + Args: + worker_path (str): The path of the source code which the worker process will + run. + scheduler_address (str): The ip address and port of the scheduler to connect + to. + objstore_address (str): The ip address and port of the object store to + connect to. + worker_address (str): The ip address and port to use for the worker. + local (bool): True if using Ray in local mode. If local is true, then this + process will be killed by serices.cleanup() when the Python process that + imported services exits. """ p = subprocess.Popen(["python", worker_path, @@ -131,14 +135,18 @@ def start_worker(worker_path, scheduler_address, objstore_address, worker_addres all_processes.append((p, worker_address)) def start_node(scheduler_address, node_ip_address, num_workers, worker_path=None): - """ - Start an object store and associated workers that will be part of a larger cluster. - Assumes the scheduler has already been started. + """Start an object store and associated workers in the cluster setting. - :param scheduler_address: ip address and port of the scheduler (which may run on a different node) - :param node_ip_address: ip address (without port) of the node this function is run on - :param num_workers: the number of workers to be started on this node - :param worker_path: path of the source code that will be run on the worker + This starts an object store and the associated workers when Ray is being used + in the cluster setting. This assumes the scheduler has already been started. + + Args: + scheduler_address (str): ip address and port of the scheduler (which may run + on a different node) + node_ip_address (str): ip address (without port) of the node this function + is run on + num_workers (int): the number of workers to be started on this node + worker_path (str): path of the source code that will be run on the worker """ objstore_address = address(node_ip_address, new_objstore_port()) start_objstore(scheduler_address, objstore_address, local=False) @@ -148,35 +156,41 @@ def start_node(scheduler_address, node_ip_address, num_workers, worker_path=None time.sleep(0.5) def start_workers(scheduler_address, objstore_address, num_workers, worker_path): - """ - Start a new set of workers on this node. This assumes that the scheduler is - already running and that the object store on this node is already running. - The intended use case is that a developer wants to update the code running - on the worker processes so first kills all of the workers and then runs this - method. + """Start a new set of workers on this node. - :param scheduler_address: ip address and port of the scheduler (which may run on a different node) - :param objstore_address: ip address and port of the object store (which runs on the same node) - :param num_workers: the number of workers to be started on this node - :param worker_path: path of the source code that will be run on the worker + Start a new set of workers on this node. This assumes that the scheduler is + already running and that the object store on this node is already running. The + intended use case is that a developer wants to update the code running on the + worker processes so first kills all of the workers and then runs this method. + + Args: + scheduler_address (str): ip address and port of the scheduler (which may run + on a different node) + objstore_address (str): ip address and port of the object store (which runs + on the same node) + num_workers (int): the number of workers to be started on this node + worker_path (str): path of the source code that will be run on the worker """ node_ip_address = objstore_address.split(":")[0] for _ in range(num_workers): start_worker(worker_path, scheduler_address, objstore_address, address(node_ip_address, new_worker_port()), local=False) def start_ray_local(num_workers=0, worker_path=None, driver_mode=ray.SCRIPT_MODE): - """ - This method starts Ray in local mode (as opposed to cluster mode, which is - handled by cluster.py). + """Start Ray in local mode. - :param num_workers: The number of workers to start. - :param worker_path: The path of the source code that will be run by the worker - :param driver_mode: The mode for the driver, this only affects the printing of - error messages. This should be ray.SCRIPT_MODE if the driver is being run in - a script. It should be ray.SHELL_MODE if it is being used interactively in - the shell. It should be ray.PYTHON_MODE to run things in a manner eqivalent - to serial Python code. It should be ray.WORKER_MODE to surpress the printing - of error messages. + This method starts Ray in local mode (as opposed to cluster mode, which is + handled by cluster.py). + + Args: + num_workers (int): The number of workers to start. + worker_path (str): The path of the source code that will be run by the + worker + driver_mode: The mode for the driver, this only affects the printing of + error messages. This should be ray.SCRIPT_MODE if the driver is being run + in a script. It should be ray.SHELL_MODE if it is being used interactively + in the shell. It should be ray.PYTHON_MODE to run things in a manner + equivalent to serial Python code. It should be ray.WORKER_MODE to surpress + the printing of error messages. """ start_services_local(num_objstores=1, num_workers_per_objstore=num_workers, worker_path=worker_path, driver_mode=driver_mode) diff --git a/lib/python/ray/worker.py b/lib/python/ray/worker.py index 3d8c214be..184163b2e 100644 --- a/lib/python/ray/worker.py +++ b/lib/python/ray/worker.py @@ -17,38 +17,146 @@ import ray.graph import services class RayFailedObject(object): - """If a task throws an exception during execution, a RayFailedObject is stored in the object store for each of the tasks outputs.""" + """An object used internally to represent a task that threw an exception. + + If a task throws an exception during execution, a RayFailedObject is stored in + the object store for each of the tasks outputs. When an object is retrieved + from the object store, the Python method that retrieved it should check to see + if the object is a RayFailedObject and if it is then an exception should be + thrown containing the error message. + + Attributes + error_message (str): The error message raised by the task that failed. + """ def __init__(self, error_message=None): + """Initialize a RayFailedObject. + + Args: + error_message (str): The error message raised by the task for which a + RayFailedObject is being created. + """ self.error_message = error_message def deserialize(self, primitives): + """Create a RayFailedObject from a primitive object. + + This initializes a RayFailedObject from a primitive object created by the + serialize method. This method is required in order for Ray to serialize + custom Python classes. + + Note: + This method should not be called by users. + + Args: + primitives (str): The object's error message. + """ self.error_message = primitives def serialize(self): + """Turn a RayFailedObject into a primitive object. + + This method is required in order for Ray to serialize + custom Python classes. + + Note: + The output of this method should only be used by the deserialize method. + This method should not be called by users. + + Args: + primitives (str): The object's error message. + + Returns: + A primitive representation of a RayFailedObject. + """ return self.error_message class RayDealloc(object): + """An object used internally to properly implement reference counting. + + When we call get_object with a particular object reference, we create a + RayDealloc object with the information necessary to properly handle closing + the relevant memory segment when the object is no longer needed by the worker. + The RayDealloc object is stored as a field in the object returned by + get_object so that its destructor is only called when the worker no longer has + any references to the object. + + Attributes + handle (worker capsule): A Python object wrapping a C++ Worker object. + segmentid (int): The id of the segment that contains the object that holds + this RayDealloc object. + """ + def __init__(self, handle, segmentid): + """Initialize a RayDealloc object. + + Args: + handle (worker capsule): A Python object wrapping a C++ Worker object. + segmentid (int): The id of the segment that contains the object that holds + this RayDealloc object. + """ self.handle = handle self.segmentid = segmentid def __del__(self): + """Deallocate the relevant segment to avoid a memory leak.""" ray.lib.unmap_object(self.handle, self.segmentid) class Worker(object): - """The methods in this class are considered unexposed to the user. The functions outside of this class are considered exposed.""" + """A class used to define the control flow of a worker process. + + Note: + The methods in this class are considered unexposed to the user. The + functions outside of this class are considered exposed. + + Attributes: + functions (Dict[str, Callable]): A dictionary mapping the name of a remote + function to the remote function itself. This is the set of remote + functions that can be executed by this worker. + handle (worker capsule): A Python object wrapping a C++ Worker object. + + """ def __init__(self): + """Initialize a Worker object.""" self.functions = {} self.handle = None def set_mode(self, mode): + """Set the mode of the worker. + + The mode ray.SCRIPT_MODE should be used if this Worker is a driver that is + being run as a Python script. It will print information about task failures. + + The mode ray.SHELL_MODE should be used if this Worker is a driver that is + being run interactively in a Python shell. It will print information about + task failures and successes. + + The mode ray.WORKER_MODE should be used if this Worker is not a driver. It + will not print information about tasks. + + The mode ray.PYTHON_MODE should be used if this Worker is a driver and if + you want to run the driver in a manner equivalent to serial Python for + debugging purposes. It will not send remote function calls to the scheduler + and will insead execute them in a blocking fashion. + + args: + mode: One of ray.SCRIPT_MODE, ray.WORKER_MODE, ray.SHELL_MODE, and + ray.PYTHON_MODE. + """ self.mode = mode colorama.init() def put_object(self, objref, value): - """Put `value` in the local object store with objref `objref`. This assumes that the value for `objref` has not yet been placed in the local object store.""" + """Put value in the local object store with object reference objref. + + This assumes that the value for objref has not yet been placed in the + local object store. + + Args: + objref (ray.ObjRef): The object reference of the value to be put. + value (serializable object): The value to put in the object store. + """ if serialization.is_arrow_serializable(value): ray.lib.put_arrow(self.handle, objref, value) else: @@ -56,11 +164,13 @@ class Worker(object): ray.lib.put_object(self.handle, objref, object_capsule, contained_objrefs) def get_object(self, objref): - """ - Return the value from the local object store for objref `objref`. This will - block until the value for `objref` has been written to the local object store. + """Get the value in the local object store associated with objref. - WARNING: get_object can only be called on a canonical objref. + Return the value from the local object store for objref. This will block + until the value for objref has been written to the local object store. + + Args: + objref (ray.ObjRef): The object reference of the value to retrieve. """ if ray.lib.is_arrow(self.handle, objref): result, segmentid = ray.lib.get_arrow(self.handle, objref) @@ -97,27 +207,55 @@ class Worker(object): return result def alias_objrefs(self, alias_objref, target_objref): - """Make `alias_objref` refer to the same object that `target_objref` refers to.""" + """Make two object references refer to the same object.""" ray.lib.alias_objrefs(self.handle, alias_objref, target_objref) def register_function(self, function): - """Notify the scheduler that this worker can execute the function with name `func_name`. Store the function `function` locally.""" + """Register a function with the scheduler. + + Notify the scheduler that this worker can execute the function with name + func_name. After this call, the scheduler can send tasks for executing + the function to this worker. + + Args: + function (Callable): The remote function that this worker can execute. + """ ray.lib.register_function(self.handle, function.func_name, len(function.return_types)) self.functions[function.func_name] = function def submit_task(self, func_name, args): - """Tell the scheduler to schedule the execution of the function with name `func_name` with arguments `args`. Retrieve object references for the outputs of the function from the scheduler and immediately return them.""" + """Submit a remote task to the scheduler. + + Tell the scheduler to schedule the execution of the function with name + func_name with arguments args. Retrieve object references for the outputs of + the function from the scheduler and immediately return them. + + Args: + func_name (str): The name of the function to be executed. + args (List[Any]): The arguments to pass into the function. Arguments can + be object references or they can be values. If they are values, they + must be serializable objecs. + """ task_capsule = serialization.serialize_task(self.handle, func_name, args) objrefs = ray.lib.submit_task(self.handle, task_capsule) if self.mode == ray.SHELL_MODE or self.mode == ray.SCRIPT_MODE: print_task_info(ray.lib.task_info(self.handle), self.mode) return objrefs -# We make `global_worker` a global variable so that there is one worker per worker process. global_worker = Worker() +"""Worker: The global Worker object for this worker process. + +We use a global Worker object to ensure that there is a single worker object +per worker process. +""" -# This is a helper method. It should not be called by users. def print_failed_task(task_status): + """Print information about failed tasks. + + Args: + task_status (Dict): A dictionary containing the name, operationid, and + error message for a failed task. + """ print """ Error: Task failed Function Name: {} @@ -125,8 +263,14 @@ def print_failed_task(task_status): Error Message: \n{} """.format(task_status["function_name"], task_status["operationid"], task_status["error_message"]) -# This is a helper method. It should not be called by users. def print_task_info(task_data, mode): + """Print information about tasks. + + Args: + task_data (Dict): A dictionary containing information about tasks that have + failed, succeeded, or are still running. + mode: The mode of the Worker object. + """ num_tasks_succeeded = task_data["num_succeeded"] num_tasks_in_progress = len(task_data["running_tasks"]) num_tasks_failed = len(task_data["failed_tasks"]) @@ -146,26 +290,26 @@ def print_task_info(task_data, mode): print ", ".join(info_strings) def scheduler_info(worker=global_worker): + """Return information about the state of the scheduler.""" return ray.lib.scheduler_info(worker.handle) def visualize_computation_graph(file_path=None, view=False, worker=global_worker): - """ - Write the computation graph to a pdf file. + """Write the computation graph to a pdf file. Args: - file_path: A .pdf file that the rendered computation graph will be written to + file_path (str): The name of a pdf file that the rendered computation graph + will be written to. If this argument is None, a temporary path will be + used. + view (bool): If true, the result the python graphviz package will try to + open the result in a viewer. - view: If true, the result the python graphviz package will try to open the - result in a viewer + Examples: + In ray/scripts, call "python shell.py" and try the following code. - Example: - In ray/scripts, call "python shell.py" and paste in the following code. - - x = da.zeros([20, 20]) - y = da.zeros([20, 20]) - z = da.dot(x, y) - - ray.visualize_computation_graph("computation_graph.pdf") + >>> x = da.zeros([20, 20]) + >>> y = da.zeros([20, 20]) + >>> z = da.dot(x, y) + >>> ray.visualize_computation_graph(view=True) """ if file_path is None: @@ -186,15 +330,17 @@ def visualize_computation_graph(file_path=None, view=False, worker=global_worker print "Wrote computation graph to file {}.pdf".format(base_path) def task_info(worker=global_worker): - """Tell the scheduler to return task information. Currently includes a list of all failed tasks since the start of the cluster.""" + """Return information about failed tasks.""" return ray.lib.task_info(worker.handle) def register_module(module, worker=global_worker): - """ - This registers each remote function in the module with the scheduler, so tasks - with those functions can be scheduled on this worker. + """Register each remote function in a module with the scheduler. - :param module: The module of functions to register. + This registers each remote function in the module with the scheduler, so tasks + with those functions can be scheduled on this worker. + + args: + module (module): The module of functions to register. """ logging.info("registering functions in module {}.".format(module.__name__)) for name in dir(module): @@ -204,6 +350,17 @@ def register_module(module, worker=global_worker): worker.register_function(val) def connect(scheduler_address, objstore_address, worker_address, is_driver=False, worker=global_worker, mode=ray.WORKER_MODE): + """Connect this worker to the scheduler and an object store. + + Args: + scheduler_address (str): The ip address and port of the scheduler. + objstore_address (str): The ip address and port of the local object store. + worker_address (str): The ip address and port of this worker. The port can + be chosen arbitrarily. + is_driver (bool): True if this worker is a driver and false otherwise. + mode: The mode of the worker. One of ray.SCRIPT_MODE, ray.WORKER_MODE, + ray.SHELL_MODE, and ray.PYTHON_MODE. + """ if hasattr(worker, "handle"): del worker.handle worker.scheduler_address = scheduler_address @@ -216,15 +373,23 @@ def connect(scheduler_address, objstore_address, worker_address, is_driver=False ray.lib.set_log_config(config.get_log_file_path("-".join(["worker", worker_address, "c++"]) + ".log")) def disconnect(worker=global_worker): + """Disconnect this worker from the scheduler and object store.""" if worker.handle is not None: ray.lib.disconnect(worker.handle) def get(objref, worker=global_worker): - """ - Get a remote object from an object store. + """Get a remote object from an object store. - :param objref: Object reference to the object you want to get - :rtype: A Python value + This method blocks until the object corresponding to objref is available in + the local object store. If this object is not in the local object store, it + will be shipped from an object store that has it (once the object has been + created). + + Args: + objref (ray.ObjRef): Object reference to the object to get. + + Returns: + A Python object """ if worker.mode == ray.PYTHON_MODE: return objref # In ray.PYTHON_MODE, ray.get is the identity operation (the input will actually be a value not an objref) @@ -237,11 +402,13 @@ def get(objref, worker=global_worker): return value def put(value, worker=global_worker): - """ - Store an object in the object store. + """Store an object in the object store. - :param value: The Python object to be stored - :rtype: Object reference + Args: + value (serializable object): The Python object to be stored. + + Returns: + The object reference assigned to this value. """ if worker.mode == ray.PYTHON_MODE: return value # In ray.PYTHON_MODE, ray.put is the identity operation @@ -252,35 +419,52 @@ def put(value, worker=global_worker): return objref def kill_workers(worker=global_worker): - """ - This method kills all of the workers in the cluster. It does not kill drivers. + """Kill all of the workers in the cluster. This does not kill drivers. + + Note: + Currently, we only support killing workers if all submitted tasks have been + run. If some workers are still running tasks or if the scheduler still has + tasks in its queue, then this method will not do anything. + + Returns: + True if workers were successfully killed. False otherwise. """ success = ray.lib.kill_workers(worker.handle) if not success: - print "Could not kill all workers. Check that there are no tasks currently running." + print "Could not kill all workers. We currently do not support killing workers when tasks are running." return success def restart_workers_local(num_workers, worker_path, worker=global_worker): - """ - This method kills all of the workers and starts new workers locally on the - same node as the driver. This is intended for use in the case where Ray is - being used on a single node. + """Restart workers locally. - :param num_workers: the number of workers to be started - :param worker_path: path of the source code that will be run on the worker + This method kills all of the workers and starts new workers locally on the + same node as the driver. This is intended for use in the case where Ray is + being used on a single node. + + Args: + num_workers (int): The number of workers to be started. + worker_path (str): The path of the source code that workers will run. + + Returns: + True if workers were successfully restarted. False otherwise. """ if not kill_workers(worker): return False services.start_workers(worker.scheduler_address, worker.objstore_address, num_workers, worker_path) + return True def format_error_message(exception_message): - """ - This method takes an backtrace from an exception and makes it nicer by - removing a few uninformative lines and adding some space to indent the - remaining lines nicely. + """Improve the formatting of an exception thrown by a remote function. - :param exception_message: a string generated by traceback.format_exc() - :rtype: a string + This method takes an backtrace from an exception and makes it nicer by + removing a few uninformative lines and adding some space to indent the + remaining lines nicely. + + Args: + exception_message (str): A message generated by traceback.format_exc(). + + Returns: + A string of the formatted exception message. """ lines = exception_message.split("\n") # Remove lines 1, 2, 3, and 4, which are always the same, they just contain @@ -290,6 +474,21 @@ def format_error_message(exception_message): return "\n".join(lines) def main_loop(worker=global_worker): + """The main loop a worker runs to receive and execute tasks. + + This method is an infinite loop. It waits to receive tasks from the scheduler. + When it receives a task, it first deserializes the task. Then it retrieves the + values for any arguments that were passed in as object references. Then it + passes the arguments to the actual function. Then it stores the outputs of the + function in the local object store. Then it notifies the scheduler that it + completed the task. + + If the process of getting the arguments for execution (which does some type + checking) or the process of executing the task fail, then the main loop will + catch the exception and store RayFailedObject objects containing the relevant + error messages in the object store in place of the actual outputs. These + objects are used to propagate the error messages. + """ if not ray.lib.connected(worker.handle): raise Exception("Worker is attempting to enter main_loop but has not been connected yet.") ray.lib.start_worker_service(worker.handle) @@ -321,15 +520,15 @@ def main_loop(worker=global_worker): process_task(task) def remote(arg_types, return_types, worker=global_worker): - """ - This is a decorator to indicate that a Python function is to be executed remotely. + """This decorator is used to create remote functions. - :param arg_types: List of Python types of the function arguments - :param return_types: List of Python types of the return values + Args: + arg_types (List[type]): List of Python types of the function arguments. + return_types (List[type]): List of Python types of the return values. """ def remote_decorator(func): def func_executor(arguments): - """This is what gets executed remotely on a worker after a remote function is scheduled by the scheduler.""" + """This gets run when the remote function is executed.""" logging.info("Calling function {}".format(func.__name__)) start_time = time.time() result = func(*arguments) @@ -338,7 +537,7 @@ def remote(arg_types, return_types, worker=global_worker): logging.info("Finished executing function {}, it took {} seconds".format(func.__name__, end_time - start_time)) return result def func_call(*args, **kwargs): - """This is what gets run immediately when a worker calls a remote function.""" + """This gets run immediately when a worker calls a remote function.""" args = list(args) args.extend([kwargs[keyword] if kwargs.has_key(keyword) else default for keyword, default in func_call.keyword_defaults[len(args):]]) # fill in the remaining arguments if worker.mode == ray.PYTHON_MODE: @@ -365,10 +564,18 @@ def remote(arg_types, return_types, worker=global_worker): return func_call return remote_decorator -# helper method, this should not be called by the user -# we currently do not support the functionality that we test for in this method, -# but in the future we could def check_signature_supported(function): + """Check if we support the signature of this function. + + We currently do not allow remote functions to have **kwargs. We also do not + support keyword argumens in conjunction with a *args argument. + + Args: + function (Callable): The function to check. + + Raises: + Exception: An exception is raised if the signature is not supported. + """ # check if the user specified kwargs if function.has_kwargs_param: raise "Function {} has a **kwargs argument, which is currently not supported.".format(function.__name__) @@ -377,8 +584,18 @@ def check_signature_supported(function): raise "Function {} has a *args argument as well as a keyword argument, which is currently not supported.".format(function.__name__) -# helper method, this should not be called by the user def check_return_values(function, result): + """Check the types and number of return values. + + Args: + function (Callable): The remote function whose outputs are being checked. + result: The value returned by an invocation of the remote function. The + expected types and number are defined in the remote decorator. + + Raises: + Exception: An exception is raised if the return values have incorrect types + or the function returned the wrong number of return values. + """ # If the @remote decorator declares that the function has no return values, # then all we do is check that there were in fact no return values. if len(function.return_types) == 0: @@ -403,6 +620,17 @@ def check_return_values(function, result): raise Exception("The {}th return value for function {} has type {}, but the @remote decorator expected a return value of type {} or an ObjRef.".format(i, function.__name__, type(result[i]), function.return_types[i])) def typecheck_arg(arg, expected_type, i, function): + """Check that an argument has the expected type. + + Args: + arg: An argument to function. + expected_type (type): The expected type of arg. + i (int): The position of the argument to the function. + function (Callable): The remote function whose argument is being checked. + + Raises: + Exception: An exception is raised if arg does not have the expected type. + """ if issubclass(type(arg), expected_type): # Passed the type-checck # TODO(rkn): This check doesn't really work, e.g., issubclass(type([1, 2, 3]), typing.List[str]) == True @@ -413,8 +641,19 @@ def typecheck_arg(arg, expected_type, i, function): else: raise Exception("Argument {} for function {} has type {} but an argument of type {} was expected.".format(i, function.__name__, type(arg), expected_type)) -# helper method, this should not be called by the user def check_arguments(function, args): + """Check that the arguments to the remote function have the right types. + + This is called by the worker that calls the remote function (not the worker + that executes the remote function). + + Args: + function (Callable): The remote function whose arguments are being checked. + args (List): The arguments to the function + + Raises: + Exception: An exception is raised the args do not all have the right types. + """ # check the number of args if len(args) != len(function.arg_types) and not function.has_vararg_param: raise Exception("Function {} expects {} arguments, but received {}.".format(function.__name__, len(function.arg_types), len(args))) @@ -435,17 +674,33 @@ def check_arguments(function, args): else: typecheck_arg(arg, expected_type, i, function) -# helper method, this should not be called by the user def get_arguments_for_execution(function, args, worker=global_worker): + """Retrieve the arguments for the remote function. + + This retrieves the values for the arguments to the remote function that were + passed in as object references. Argumens that were passed by value are not + changed. This also does some type checking. This is called by the worker that + is executing the remote function. + + Args: + function (Callable): The remote function whose arguments are being + retrieved. + args (List): The arguments to the function. + + Returns: + The retrieved arguments in addition to the arguments that were passed by + value. + + Raises: + Exception: An exception is raised the args do not all have the right types. + """ # TODO(rkn): Eventually, all of the type checking can be put in `check_arguments` above so that the error will happen immediately when calling a remote function. arguments = [] - """ - # check the number of args - if len(args) != len(function.arg_types) and function.arg_types[-1] is not None: - raise Exception("Function {} expects {} arguments, but received {}.".format(function.__name__, len(function.arg_types), len(args))) - elif len(args) < len(function.arg_types) - 1 and function.arg_types[-1] is None: - raise Exception("Function {} expects at least {} arguments, but received {}.".format(function.__name__, len(function.arg_types) - 1, len(args))) - """ + # # check the number of args + # if len(args) != len(function.arg_types) and function.arg_types[-1] is not None: + # raise Exception("Function {} expects {} arguments, but received {}.".format(function.__name__, len(function.arg_types), len(args))) + # elif len(args) < len(function.arg_types) - 1 and function.arg_types[-1] is None: + # raise Exception("Function {} expects at least {} arguments, but received {}.".format(function.__name__, len(function.arg_types) - 1, len(args))) for (i, arg) in enumerate(args): if i <= len(function.arg_types) - 1: @@ -468,8 +723,26 @@ def get_arguments_for_execution(function, args, worker=global_worker): arguments.append(argument) return arguments -# helper method, this should not be called by the user def store_outputs_in_objstore(objrefs, outputs, worker=global_worker): + """Store the outputs of a remote function in the local object store. + + This stores the values that were returned by a remote function in the local + object store. If any of the return values are object references, then these + object references are aliased with the object references that the scheduler + assigned for the return values. This is called by the worker that executes the + remote function. + + Note: + The arguments objrefs and outputs should have the same length. + + Args: + objrefs (List[ray.ObjRef]): The object references that were assigned to the + outputs of the remote function call. + outputs (Tuple): The value returned by the remote function. If the remote + function was supposed to only return one value, then its output was + wrapped in a tuple with one element prior to being passed into this + function. + """ for i in range(len(objrefs)): if isinstance(outputs[i], ray.lib.ObjRef): # An ObjRef is being returned, so we must alias objrefs[i] so that it refers to the same object that outputs[i] refers to diff --git a/scripts/cluster.py b/scripts/cluster.py index 0ff93ece4..51b80c7ec 100644 --- a/scripts/cluster.py +++ b/scripts/cluster.py @@ -1,15 +1,10 @@ -# This script can be used to start Ray on an existing cluster. -# -# How to use it: Create a file "nodes.txt" that contains a list of the IP -# addresses of the nodes in the cluster. Put the head node first. This node will -# host the driver and the scheduler. - import os import subprocess import socket import argparse import threading import IPython +import numpy as np parser = argparse.ArgumentParser(description="Parse information about the cluster.") parser.add_argument("--nodes", type=str, required=True, help="Test file with node IP addresses, one line per address.") @@ -17,33 +12,106 @@ parser.add_argument("--key-file", type=str, required=True, help="Path to the fil parser.add_argument("--username", type=str, required=True, help="User name for logging in.") parser.add_argument("--installation-directory", type=str, required=True, help="The directory in which to install Ray.") -def run_command_over_ssh(node_ip_address, username, key_file, command): - """ - This method is used for connecting to a node with ssh and running a sequence - of commands. +class RayCluster(object): + """A class for setting up, starting, and stopping Ray on a cluster. - :param node_ip_address: the ip address of the node to ssh to - :param username: the username used to ssh to the cluster - :param key_file: the key used to ssh to the cluster - :param command: the command to run over ssh, currently this command is not allowed to have any single quotes + Attributes: + node_ip_addresses (List[str]): A list of the ip addresses of the nodes in + the cluster. The first element is the head node and will host the + scheduler process. + username (str): The username used to ssh to nodes in the cluster. + key_file (str): The path to the key used to ssh to nodes in the cluster. + installation_directory (str): The path on the nodes in the cluster to the + directory in which Ray should be installed. """ - if "'" in command: - raise Exception("Commands run over ssh must not contain the single quote character. This command does: {}".format(command)) - full_command = "ssh -o StrictHostKeyChecking=no -i {} {}@{} '{}'".format(key_file, username, node_ip_address, command) - subprocess.call([full_command], shell=True) - print "Finished running command '{}' on {}@{}.".format(command, username, node_ip_address) -def _install_ray(node_ip_addresses, username, key_file, installation_directory): - """ - This method is used to install Ray on a cluster. For each node in the cluster, - it will ssh to the node and run the build scripts. + def __init__(self, node_ip_addresses, username, key_file, installation_directory): + """Initialize the RayCluster object. - :param node_ip_addresses: ip addresses of the nodes on which to install Ray - :param username: the username used to ssh to the cluster - :param key_file: the key used to ssh to the cluster - :param installation_directory: directory in which Ray is installed, for example "/home/ubuntu/" - """ - def install_ray_over_ssh(node_ip_address, username, key_file, installation_directory): + Args: + node_ip_addresses (List[str]): A list of the ip addresses of the nodes in + the cluster. The first element is the head node and will host the + scheduler process. + username (str): The username used to ssh to nodes in the cluster. + key_file (str): The path to the key used to ssh to nodes in the cluster. + installation_directory (str): The path on the nodes in the cluster to the + directory in which Ray should be installed. + + Raises: + Exception: An exception is raised by check_ip_addresses if one of the ip + addresses is not a valid ip address. + """ + _check_ip_addresses(node_ip_addresses) + self.node_ip_addresses = node_ip_addresses + self.username = username + self.key_file = key_file + self.installation_directory = installation_directory + + def _run_command_over_ssh(self, node_ip_address, command): + """Run a command over ssh. + + Args: + node_ip_address (str): The ip address of the node to ssh to. + command (str): The command to run over ssh, currently this command is not + allowed to have any single quotes. + """ + if "'" in command: + raise Exception("Commands run over ssh must not contain the single quote character. This command does: {}".format(command)) + full_command = "ssh -o StrictHostKeyChecking=no -i {} {}@{} '{}'".format(self.key_file, self.username, node_ip_address, command) + subprocess.call([full_command], shell=True) + print "Finished running command '{}' on {}@{}.".format(command, self.username, node_ip_address) + + def _run_parallel_functions(self, functions, inputs): + """Run functions in parallel. + + This will run each function in functions in a separate thread. This method + blocks until all of the functions have finished. + + Args: + functions (List[Callable]): The functions to execute in parallel. + inputs (List[Tuple]): The inputs to the functions. + """ + threads = [] + for i in range(len(self.node_ip_addresses)): + t = threading.Thread(target=functions[i], args=inputs[i]) + t.start() + threads.append(t) + for t in threads: + t.join() + + def _run_command_over_ssh_on_all_nodes_in_parallel(self, commands): + """Run a command over ssh on all nodes in the cluster in parallel. + + Args: + commands: This is either a single command to run on every node in the + cluster ove ssh, or it is a list of commands of the same length as + node_ip_addresses, in which case the ith command will be run on the ith + element of node_ip_addresses. Currently this command is not allowed to + have any single quotes. + + Raises: + Exception: An exception is raised if commands is not a string or is not a + list with the same length as node_ip_addresses. + """ + if isinstance(commands, str): + # If there is only one command, then run this command on every node in the cluster. + commands = len(self.node_ip_addresses) * [commands] + if len(commands) != len(self.node_ip_addresses): + raise Exception("The number of commands must match the number of nodes.") + functions = [] + inputs = [] + def function(node_ip_address, command): + self._run_command_over_ssh(node_ip_address, command) + inputs = zip(node_ip_addresses, commands) + self._run_parallel_functions(len(self.node_ip_addresses) * [function], inputs) + print "Finished running commands {} on all nodes.".format(inputs) + + def install_ray(self): + """Install Ray on every node in the cluster. + + This method will ssh to each node, clone the Ray repository, install the + dependencies, build the third-party libraries, and build Ray. + """ install_ray_command = """ sudo apt-get update && sudo apt-get -y install git && @@ -54,184 +122,180 @@ def _install_ray(node_ip_addresses, username, key_file, installation_directory): ./install-dependencies.sh; ./setup.sh; ./build.sh - """.format(installation_directory, installation_directory) - run_command_over_ssh(node_ip_address, username, key_file, install_ray_command) - threads = [] - for node_ip_address in node_ip_addresses: - t = threading.Thread(target=install_ray_over_ssh, args=(node_ip_address, username, key_file, installation_directory)) - t.start() - threads.append(t) - for t in threads: - t.join() + """.format(self.installation_directory, self.installation_directory) + self._run_command_over_ssh_on_all_nodes_in_parallel(install_ray_command) -def _start_ray(node_ip_addresses, username, key_file, num_workers_per_node, worker_directory, installation_directory): - """ - This method is used to start Ray on a cluster. It will ssh to the head node, - that is, the first node in the list node_ip_addresses, and it will start - the scheduler. Then it will ssh to each node and start an object store and - some workers. + def start_ray(self, worker_directory, num_workers_per_node=10): + """Start Ray on a cluster. - :param node_ip_addresses: ip addresses of the nodes on which to install Ray - :param username: the username used to ssh to the cluster - :param key_file: the key used to ssh to the cluster - :param worker_directory: local directory containing the worker source code - :param installation_directory: directory in which Ray is installed, for example "/home/ubuntu/" - """ - # First update the worker code on the nodes. - remote_worker_path = _update_worker_code(node_ip_addresses, worker_directory, installation_directory) + This method is used to start Ray on a cluster. It will ssh to the head node, + that is, the first node in the list node_ip_addresses, and it will start the + scheduler. Then it will ssh to each node and start an object store and some + workers. - scripts_directory = os.path.join(installation_directory, "ray/scripts") - # Start the scheduler - # The triple backslashes are used for two rounds of escaping, something like \\\" -> \" -> " - start_scheduler_command = """ - cd "{}"; - source ../setup-env.sh; - python -c "import ray; ray.services.start_scheduler(\\\"{}:10001\\\", local=False)" > start_scheduler.out 2> start_scheduler.err < /dev/null & - """.format(scripts_directory, node_ip_addresses[0]) - run_command_over_ssh(node_ip_addresses[0], username, key_file, start_scheduler_command) + Args: + worker_directory (str): The path to the local directory containing the + worker source code. This directory must contain a file worker.py which + is the code run by the worker processes. + num_workers_per_node (int): The number workers to start on each node. + """ + # First update the worker code on the nodes. + remote_worker_path = self._update_worker_code(worker_directory) - # Start the workers on each node - # The triple backslashes are used for two rounds of escaping, something like \\\" -> \" -> " - for i, node_ip_address in enumerate(node_ip_addresses): - start_workers_command = """ + scripts_directory = os.path.join(self.installation_directory, "ray/scripts") + # Start the scheduler + # The triple backslashes are used for two rounds of escaping, something like \\\" -> \" -> " + start_scheduler_command = """ cd "{}"; source ../setup-env.sh; - python -c "import ray; ray.services.start_node(\\\"{}:10001\\\", \\\"{}\\\", {}, worker_path=\\\"{}\\\")" > start_workers.out 2> start_workers.err < /dev/null & - """.format(scripts_directory, node_ip_addresses[0], node_ip_addresses[i], num_workers_per_node, remote_worker_path) - run_command_over_ssh(node_ip_address, username, key_file, start_workers_command) + python -c "import ray; ray.services.start_scheduler(\\\"{}:10001\\\", local=False)" > start_scheduler.out 2> start_scheduler.err < /dev/null & + """.format(scripts_directory, self.node_ip_addresses[0]) + self._run_command_over_ssh(self.node_ip_addresses[0], start_scheduler_command) - print "cluster started; you can start the shell on the head node with:" - setup_env_path = os.path.join(args.installation_directory, "ray/setup-env.sh") - shell_script_path = os.path.join(args.installation_directory, "ray/scripts/shell.py") - print """ - source "{}"; - python "{}" --scheduler-address={}:10001 --objstore-address={}:20001 --worker-address={}:30001 --attach - """.format(setup_env_path, shell_script_path, node_ip_addresses[0], node_ip_addresses[0], node_ip_addresses[0]) + # Start the workers on each node + # The triple backslashes are used for two rounds of escaping, something like \\\" -> \" -> " + start_workers_commands = [] + for i, node_ip_address in enumerate(self.node_ip_addresses): + start_workers_command = """ + cd "{}"; + source ../setup-env.sh; + python -c "import ray; ray.services.start_node(\\\"{}:10001\\\", \\\"{}\\\", {}, worker_path=\\\"{}\\\")" > start_workers.out 2> start_workers.err < /dev/null & + """.format(scripts_directory, self.node_ip_addresses[0], self.node_ip_addresses[i], num_workers_per_node, remote_worker_path) + start_workers_commands.append(start_workers_command) + self._run_command_over_ssh_on_all_nodes_in_parallel(start_workers_commands) -def _restart_workers(node_ip_addresses, username, key_file, num_workers_per_node, worker_directory, installation_directory): - """ - This method is used for restarting the workers in the cluster, for example, to - use new application code. This is done without shutting down the scheduler - or the object stores so that work is not thrown away. It also does not shut - down any drivers. + print "cluster started; you can start the shell on the head node with:" + setup_env_path = os.path.join(self.installation_directory, "ray/setup-env.sh") + shell_script_path = os.path.join(self.installation_directory, "ray/scripts/shell.py") + print """ + source "{}"; + python "{}" --scheduler-address={}:10001 --objstore-address={}:20001 --worker-address={}:30001 --attach + """.format(setup_env_path, shell_script_path, self.node_ip_addresses[0], self.node_ip_addresses[0], self.node_ip_addresses[0]) - :param node_ip_addresses: ip addresses of the nodes on which to restart the workers - :param username: the username used to ssh to the cluster - :param key_file: the key used to ssh to the cluster - :param worker_directory: local directory containing the worker source code - :param installation_directory: directory in which Ray is installed, for example "/home/ubuntu/" - """ - # First update the worker code on the nodes. - remote_worker_path = _update_worker_code(node_ip_addresses, worker_directory, installation_directory) + def restart_workers(self, worker_directory, num_workers_per_node=10): + """Restart the workers on the cluster. - scripts_directory = os.path.join(installation_directory, "ray/scripts") - head_node_ip_address = node_ip_addresses[0] - scheduler_address = "{}:10001".format(head_node_ip_address) # This needs to be the address of the currently running scheduler, which was presumably created in _start_ray. - objstore_address = "{}:20001".format(head_node_ip_address) # This needs to be the address of the currently running object store, which was presumably created in _start_ray. - shell_address = "{}:30000".format(head_node_ip_address) # This address must be currently unused. In particular, it cannot be the address of any currently running shell. + This method is used for restarting the workers in the cluster, for example, + to use new application code. This is done without shutting down the + scheduler or the object stores so that work is not thrown away. It also does + not shut down any drivers. - # Kill the current workers by attaching a driver to the scheduler and calling ray.kill_workers() - # The triple backslashes are used for two rounds of escaping, something like \\\" -> \" -> " - kill_workers_command = """ - cd "{}"; - source ../setup-env.sh; - python -c "import ray; ray.connect(\\\"{}\\\", \\\"{}\\\", \\\"{}\\\", is_driver=True); ray.kill_workers()" - """.format(scripts_directory, scheduler_address, objstore_address, shell_address) - run_command_over_ssh(head_node_ip_address, username, key_file, kill_workers_command) + Args: + worker_directory (str): The path to the local directory containing the + worker source code. This directory must contain a file worker.py which + is the code run by the worker processes. + num_workers_per_node (int): The number workers to start on each node. + """ + # First update the worker code on the nodes. + remote_worker_path = self._update_worker_code(worker_directory) - # Start new workers on each node - # The triple backslashes are used for two rounds of escaping, something like \\\" -> \" -> " - for i, node_ip_address in enumerate(node_ip_addresses): - start_workers_command = """ + scripts_directory = os.path.join(self.installation_directory, "ray/scripts") + head_node_ip_address = self.node_ip_addresses[0] + scheduler_address = "{}:10001".format(head_node_ip_address) # This needs to be the address of the currently running scheduler, which was presumably created in _start_ray. + objstore_address = "{}:20001".format(head_node_ip_address) # This needs to be the address of the currently running object store, which was presumably created in _start_ray. + shell_address = "{}:{}".format(head_node_ip_address, np.random.randint(30000, 40000)) # This address must be currently unused. In particular, it cannot be the address of any currently running shell. + + # Kill the current workers by attaching a driver to the scheduler and calling ray.kill_workers() + # The triple backslashes are used for two rounds of escaping, something like \\\" -> \" -> " + kill_workers_command = """ cd "{}"; source ../setup-env.sh; - python -c "import ray; ray.services.start_workers(\\\"{}:10001\\\", \\\"{}:20001\\\", {}, worker_path=\\\"{}\\\")" > start_workers.out 2> start_workers.err < /dev/null & - """.format(scripts_directory, node_ip_addresses[0], node_ip_addresses[i], num_workers_per_node, remote_worker_path) - run_command_over_ssh(node_ip_address, username, key_file, start_workers_command) + python -c "import ray; ray.connect(\\\"{}\\\", \\\"{}\\\", \\\"{}\\\", is_driver=True); ray.kill_workers()" + """.format(scripts_directory, scheduler_address, objstore_address, shell_address) + self._run_command_over_ssh(head_node_ip_address, kill_workers_command) -def _stop_ray(node_ip_addresses, username, key_file): - """ - This method is used for stopping a Ray cluster. It will ssh to each node and + # Start new workers on each node + # The triple backslashes are used for two rounds of escaping, something like \\\" -> \" -> " + start_workers_commands = [] + for i, node_ip_address in enumerate(self.node_ip_addresses): + start_workers_command = """ + cd "{}"; + source ../setup-env.sh; + python -c "import ray; ray.services.start_workers(\\\"{}:10001\\\", \\\"{}:20001\\\", {}, worker_path=\\\"{}\\\")" > start_workers.out 2> start_workers.err < /dev/null & + """.format(scripts_directory, self.node_ip_addresses[0], self.node_ip_addresses[i], num_workers_per_node, remote_worker_path) + start_workers_commands.append(start_workers_command) + self._run_command_over_ssh_on_all_nodes_in_parallel(start_workers_commands) + + def stop_ray(self): + """Kill all of the processes in the Ray cluster. + + This method is used for stopping a Ray cluster. It will ssh to each node and kill every schedule, object store, and Python process. + """ + kill_cluster_command = "killall scheduler objstore python > /dev/null 2> /dev/null" + self._run_command_over_ssh_on_all_nodes_in_parallel(kill_cluster_command) - :param node_ip_addresses: ip addresses of the nodes on which to restart the workers - :param username: the username used to ssh to the cluster - :param key_file: the key used to ssh to the cluster - """ - kill_cluster_command = "killall scheduler objstore python > /dev/null 2> /dev/null" - for node_ip_address in node_ip_addresses: - run_command_over_ssh(node_ip_address, username, key_file, kill_cluster_command) + def update_ray(self): + """Pull the latest Ray source code and rebuild Ray. -def _update_ray(node_ip_addresses, username, key_file, installation_directory): - """ - This method is used for updating the Ray source code on a Ray cluster. It + This method is used for updating the Ray source code on a Ray cluster. It will ssh to each node, will pull the latest source code from the Ray repository, and will rerun the build script (though currently it will not rebuild the third party libraries). + """ + ray_directory = os.path.join(self.installation_directory, "ray") + update_cluster_command = """ + cd "{}" && + git fetch && + git reset --hard "@{{upstream}}" -- && + (make -C "./build" clean || rm -rf "./build") && + ./build.sh + """.format(ray_directory) + self._run_command_over_ssh_on_all_nodes_in_parallel(update_cluster_command) - :param node_ip_addresses: ip addresses of the nodes on which to restart the workers - :param username: the username used to ssh to the cluster - :param key_file: the key used to ssh to the cluster - :param installation_directory: directory in which Ray is installed, for example "/home/ubuntu/" - """ - ray_directory = os.path.join(installation_directory, "ray") - update_cluster_command = """ - cd "{}" && - git fetch && - git reset --hard "@{{upstream}}" -- && - (make -C "./build" clean || rm -rf "./build") && - ./build.sh - """.format(ray_directory) - for node_ip_address in node_ip_addresses: - run_command_over_ssh(node_ip_address, username, key_file, update_cluster_command) + def _update_worker_code(self, worker_directory): + """Update the worker code on each node in the cluster. -def _update_worker_code(node_ip_addresses, worker_directory, installation_directory): - """ - This method is used to sync update the worker source code on each node in the - cluster. The worker_directory will be copied under installation_directory. - For example, we call _update_worker_code(node_ip_addresses, "~/a/b/c", - "/d/e/f"), then the contents of ~/a/b/c on the local machine will be copied - to /d/e/f/ray_worker_files/c on each node in the cluster. + This method is used to update the worker source code on each node in the + cluster. The local worker_directory will be copied under ray_worker_files in + the installation_directory. For example, if installation_directory is + "/d/e/f" and we call _update_worker_code("~/a/b/c"), then the contents of + "~/a/b/c" on the local machine will be copied to "/d/e/f/ray_worker_files/c" + on each node in the cluster. - :param node_ip_addresses: ip addresses of the nodes on which to restart the - workers - :param worker_directory: The path on the local machine to the directory that - contains the worker code. This directory must contain a file worker.py. - :param installation_directory: Directory in which ray is installed, for - example "/home/ubuntu/". + Args: + worker_directory (str): The path on the local machine to the directory + that contains the worker code. This directory must contain a file + worker.py. - :rtype: A string with the path to the source code of the worker on the remote - nodes. - """ - worker_directory = os.path.expanduser(worker_directory) - if not os.path.isdir(worker_directory): - raise Exception("Directory {} does not exist.".format(worker_directory)) - if not os.path.exists(os.path.join(worker_directory, "worker.py")): - raise Exception("Directory {} does not contain a file named worker.py.".format(worker_directory)) - # If worker_directory is "/a/b/c", then local_directory_name is "c". - local_directory_name = os.path.split(os.path.realpath(worker_directory))[1] - remote_directory = os.path.join(installation_directory, "ray_worker_files", local_directory_name) - for node_ip_address in node_ip_addresses: + Returns: + A string with the path to the source code of the worker on the remote + nodes. + """ + worker_directory = os.path.expanduser(worker_directory) + if not os.path.isdir(worker_directory): + raise Exception("Directory {} does not exist.".format(worker_directory)) + if not os.path.exists(os.path.join(worker_directory, "worker.py")): + raise Exception("Directory {} does not contain a file named worker.py.".format(worker_directory)) + # If worker_directory is "/a/b/c", then local_directory_name is "c". + local_directory_name = os.path.split(os.path.realpath(worker_directory))[1] + remote_directory = os.path.join(self.installation_directory, "ray_worker_files", local_directory_name) # Remove and recreate the directory on the node. recreate_directory_command = """ rm -r "{}"; mkdir -p "{}" """.format(remote_directory, remote_directory) - run_command_over_ssh(node_ip_address, username, key_file, recreate_directory_command) + self._run_command_over_ssh_on_all_nodes_in_parallel(recreate_directory_command) # Copy the files from the local machine to the node. - copy_command = """ - scp -r -i {} {}/* {}@{}:{}/ - """.format(key_file, worker_directory, username, node_ip_address, remote_directory) - subprocess.call([copy_command], shell=True) - remote_worker_path = os.path.join(remote_directory, "worker.py") - return remote_worker_path + def copy_function(node_ip_address): + copy_command = """ + scp -r -i {} {}/* {}@{}:{}/ + """.format(self.key_file, worker_directory, self.username, node_ip_address, remote_directory) + subprocess.call([copy_command], shell=True) + inputs = [(node_ip_address,) for node_ip_address in node_ip_addresses] + self._run_parallel_functions(len(self.node_ip_addresses) * [copy_function], inputs) + # Return the path to worker.py on the remote nodes. + remote_worker_path = os.path.join(remote_directory, "worker.py") + return remote_worker_path -def is_valid_ip(ip_address): - """ - This method returns true if an address is a valid IPv4 address and returns - false otherwise. +def _is_valid_ip(ip_address): + """Check if ip_addess is a valid IPv4 address. - :param ip_address: the ip address to check + Args: + ip_address (str): The ip address to check. + + Returns: + True if the address is a valid IPv4 address and False otherwise. """ try: socket.inet_aton(ip_address) @@ -239,20 +303,22 @@ def is_valid_ip(ip_address): except socket.error: return False -def check_ip_addresses(node_ip_addresses): - """ - This method checks if all of the addresses in a list are valid IPv4 address. - If not, it returns false and prints an error message for each invalid - address. +def _check_ip_addresses(node_ip_addresses): + """Check if a list of ip addresses are all valid IPv4 addresses. - :param node_ip_addresses: the list of ip addresses to check + This method checks if all of the addresses in a list are valid IPv4 address. + It prints an error message for each invalid address. + + Args: + node_ip_addresses (List[str]): The list of ip addresses to check. + + Raises: + Exception: An exception is raisd if one of the addresses is not a valid IPv4 + address. """ - addresses_valid = True - for index, node_ip_address in enumerate(node_ip_addresses): - if not is_valid_ip(node_ip_address): - print "ERROR: node_ip_addresses[{}] is '{}', which is not a valid IP address.".format(index, node_ip_address) - addresses_valid = False - return addresses_valid + for i, node_ip_address in enumerate(node_ip_addresses): + if not _is_valid_ip(node_ip_address): + raise Exception("node_ip_addresses[{}] is '{}', which is not a valid IP address.".format(i, node_ip_address)) if __name__ == "__main__": args = parser.parse_args() @@ -260,64 +326,5 @@ if __name__ == "__main__": key_file = args.key_file installation_directory = args.installation_directory node_ip_addresses = map(lambda s: str(s.strip()), open(args.nodes).readlines()) - - def install_ray(node_ip_addresses=node_ip_addresses): - """ - This method is used to install Ray on a cluster. For each node in the cluster, - it will ssh to the node and run the build scripts. - - :param node_ip_addresses: ip addresses of the nodes on which to install Ray - """ - if check_ip_addresses(node_ip_addresses): - _install_ray(node_ip_addresses, username, key_file, installation_directory) - - def start_ray(worker_directory, num_workers_per_node=10, node_ip_addresses=node_ip_addresses): - """ - This method is used to start Ray on a cluster. It will ssh to the head node, - that is, the first node in the list node_ip_addresses, and it will start - the scheduler. Then it will ssh to each node and start an object store and - some workers. - - :param worker_directory: path of the source code to have the workers run - :param node_ip_addresses: ip addresses of the nodes on which to install Ray - """ - if check_ip_addresses(node_ip_addresses): - _start_ray(node_ip_addresses, username, key_file, num_workers_per_node, worker_directory, installation_directory) - - def restart_workers(worker_directory, num_workers_per_node=10, node_ip_addresses=node_ip_addresses): - """ - This method is used for restarting the workers in the cluster, for example, to - use new application code. This is done without shutting down the scheduler - or the object stores so that work is not thrown away. It also does not - shut down any drivers. - - :param node_ip_addresses: ip addresses of the nodes on which to restart the workers - :param worker_directory: path of the source code to have the workers run - :param installation_directory: directory in which Ray is installed, for example "/home/ubuntu/" - """ - if check_ip_addresses(node_ip_addresses): - _restart_workers(node_ip_addresses, username, key_file, num_workers_per_node, worker_directory, installation_directory) - - def stop_ray(node_ip_addresses=node_ip_addresses): - """ - This method is used for stopping a Ray cluster. It will ssh to each node and - kill every schedule, object store, and Python process. - - :param node_ip_addresses: ip addresses of the nodes on which to restart the workers - """ - if check_ip_addresses(node_ip_addresses): - _stop_ray(node_ip_addresses, username, key_file) - - def update_ray(node_ip_addresses=node_ip_addresses): - """ - This method is used for updating the Ray source code on a Ray cluster. It - will ssh to each node, will pull the latest source code from the Ray - repository, and will rerun the build script (though currently it will not - rebuild the third party libraries). - - :param node_ip_addresses: ip addresses of the nodes on which to restart the workers - """ - if check_ip_addresses(node_ip_addresses): - _update_ray(node_ip_addresses, username, key_file, installation_directory) - + cluster = RayCluster(node_ip_addresses, username, key_file, installation_directory) IPython.embed()