diff --git a/python/ray/actor.py b/python/ray/actor.py index 6bb279573..8d4e55e88 100644 --- a/python/ray/actor.py +++ b/python/ray/actor.py @@ -102,7 +102,6 @@ def save_and_log_checkpoint(worker, actor): Args: worker: The worker to use to log errors. actor: The actor to checkpoint. - checkpoint_index: The number of tasks that have executed so far. """ try: actor.__ray_checkpoint__() @@ -113,11 +112,7 @@ def save_and_log_checkpoint(worker, actor): worker, ray_constants.CHECKPOINT_PUSH_ERROR, traceback_str, - driver_id=worker.task_driver_id, - data={ - "actor_class": actor.__class__.__name__, - "function_name": actor.__ray_checkpoint__.__name__ - }) + driver_id=worker.task_driver_id) def restore_and_log_checkpoint(worker, actor): @@ -137,11 +132,7 @@ def restore_and_log_checkpoint(worker, actor): worker, ray_constants.CHECKPOINT_PUSH_ERROR, traceback_str, - driver_id=worker.task_driver_id, - data={ - "actor_class": actor.__class__.__name__, - "function_name": actor.__ray_checkpoint_restore__.__name__ - }) + driver_id=worker.task_driver_id) return checkpoint_resumed @@ -550,11 +541,7 @@ class ActorHandle(object): method_name: The name of the actor method to execute. args: A list of arguments for the actor method. kwargs: A dictionary of keyword arguments for the actor method. - dependency: The object ID that this method is dependent on. - Defaults to None, for no dependencies. Most tasks should - pass in the dummy object returned by the preceding task. - Some tasks, such as checkpoint and terminate methods, have - no dependencies. + num_return_vals (int): The number of return values for the method. Returns: object_ids: A list of object IDs returned by the remote actor diff --git a/python/ray/experimental/state.py b/python/ray/experimental/state.py index 2e43af8ce..156d0e52d 100644 --- a/python/ray/experimental/state.py +++ b/python/ray/experimental/state.py @@ -152,7 +152,7 @@ class GlobalState(object): time.sleep(1) continue num_redis_shards = int(num_redis_shards) - if (num_redis_shards < 1): + if num_redis_shards < 1: raise Exception("Expected at least one Redis shard, found " "{}.".format(num_redis_shards)) diff --git a/python/ray/function_manager.py b/python/ray/function_manager.py index aa081a4d9..06ec3e49e 100644 --- a/python/ray/function_manager.py +++ b/python/ray/function_manager.py @@ -403,12 +403,10 @@ class FunctionActorManager(object): push_error_to_driver( self._worker, ray_constants.REGISTER_REMOTE_FUNCTION_PUSH_ERROR, - traceback_str, - driver_id=driver_id, - data={ - "function_id": function_id.binary(), - "function_name": function_name - }) + "Failed to unpickle the remote function '{}' with function ID " + "{}. Traceback:\n{}".format(function_name, function_id.hex(), + traceback_str), + driver_id=driver_id) else: # The below line is necessary. Because in the driver process, # if the function is defined in the file where the python script @@ -506,7 +504,6 @@ class FunctionActorManager(object): Args: key: The key to store the actor class info at. actor_class_info: Information about the actor class. - worker: The worker to use to connect to Redis. """ # We set the driver ID here because it may not have been available when # the actor class was defined. @@ -577,7 +574,6 @@ class FunctionActorManager(object): Args: actor_class_key: The key in Redis to use to fetch the actor. - worker: The worker to use. """ actor_id = self._worker.actor_id (driver_id_str, class_name, module, pickled_class, checkpoint_interval, @@ -642,11 +638,10 @@ class FunctionActorManager(object): traceback.format_exc()) # Log the error message. push_error_to_driver( - self._worker, - ray_constants.REGISTER_ACTOR_PUSH_ERROR, - traceback_str, - driver_id, - data={"actor_id": actor_id.binary()}) + self._worker, ray_constants.REGISTER_ACTOR_PUSH_ERROR, + "Failed to unpickle actor class '{}' for actor ID {}. " + "Traceback:\n{}".format(class_name, actor_id.hex(), + traceback_str), driver_id) # TODO(rkn): In the future, it might make sense to have the worker # exit here. However, currently that would lead to hanging if # someone calls ray.get on a method invoked on the actor. @@ -680,7 +675,6 @@ class FunctionActorManager(object): necessary checkpointing operations. Args: - worker (Worker): The worker that is executing the actor. method_name (str): The name of the actor method. method (instancemethod): The actor method to wrap. This should be a method defined on the actor class and should therefore take an diff --git a/python/ray/import_thread.py b/python/ray/import_thread.py index c848517ed..8aa1149db 100644 --- a/python/ray/import_thread.py +++ b/python/ray/import_thread.py @@ -125,11 +125,8 @@ class ImportThread(object): # the traceback and notify the scheduler of the failure. traceback_str = traceback.format_exc() # Log the error message. - name = function.__name__ if ("function" in locals() and hasattr( - function, "__name__")) else "" utils.push_error_to_driver( self.worker, ray_constants.FUNCTION_TO_RUN_PUSH_ERROR, traceback_str, - driver_id=ray.DriverID(driver_id), - data={"name": name}) + driver_id=ray.DriverID(driver_id)) diff --git a/python/ray/log_monitor.py b/python/ray/log_monitor.py index 84ed2ce65..ff2910e88 100644 --- a/python/ray/log_monitor.py +++ b/python/ray/log_monitor.py @@ -27,8 +27,6 @@ class LogMonitor(object): process is running on. This will be used to determine which log files to track. redis_client: A client used to communicate with the Redis server. - log_filenames: A list of the names of the log files that this monitor - process is monitoring. log_files: A dictionary mapping the name of a log file to a list of strings representing its contents. log_file_handles: A dictionary mapping the name of a log file to a file diff --git a/python/ray/monitor.py b/python/ray/monitor.py index 238d0dc96..260ef1d87 100644 --- a/python/ray/monitor.py +++ b/python/ray/monitor.py @@ -239,7 +239,6 @@ class Monitor(object): data = message["data"] # Determine the appropriate message handler. - message_handler = None if channel == ray.gcs_utils.XRAY_HEARTBEAT_BATCH_CHANNEL: # Similar functionality as local scheduler info channel message_handler = self.xray_heartbeat_batch_handler @@ -250,7 +249,6 @@ class Monitor(object): raise Exception("This code should be unreachable.") # Call the handler. - assert (message_handler is not None) message_handler(channel, data) def update_local_scheduler_map(self): diff --git a/python/ray/plasma/.gitkeep b/python/ray/plasma/.gitkeep deleted file mode 100644 index e69de29bb..000000000 diff --git a/python/ray/profiling.py b/python/ray/profiling.py index ba0daed19..6f4818379 100644 --- a/python/ray/profiling.py +++ b/python/ray/profiling.py @@ -111,7 +111,7 @@ class Profiler(object): timeline until after the task has completed. For very long-running tasks, we may want profiling information to appear more quickly. In such cases, this function can be called. Note that as an - aalternative, we could start thread in the background on workers that + alternative, we could start a thread in the background on workers that calls this automatically. """ with self.lock: @@ -137,7 +137,7 @@ class RayLogSpanRaylet(object): Attributes: event_type (str): The type of the event being logged. - contents: Additional information to log. + extra_data: Additional information to log. """ def __init__(self, profiler, event_type, extra_data=None): diff --git a/python/ray/services.py b/python/ray/services.py index e7f801c09..adec81aa0 100644 --- a/python/ray/services.py +++ b/python/ray/services.py @@ -2,6 +2,7 @@ from __future__ import absolute_import from __future__ import division from __future__ import print_function +import binascii import collections import json import logging @@ -47,6 +48,11 @@ CREDIS_MEMBER_MODULE = os.path.join( os.path.abspath(os.path.dirname(__file__)), "core/src/credis/build/src/libmember.so") +# Location of the plasma object store executable. +PLASMA_STORE_EXECUTABLE = os.path.join( + os.path.abspath(os.path.dirname(__file__)), + "core/src/plasma/plasma_store_server") + # Location of the raylet executables. RAYLET_MONITOR_EXECUTABLE = os.path.join( os.path.abspath(os.path.dirname(__file__)), @@ -87,7 +93,7 @@ def new_port(): return random.randint(10000, 65535) -def remaining_processes_alive(exclude=None): +def remaining_processes_alive(): """See if the remaining processes are alive or not. Note that this ignores processes that have been explicitly killed, @@ -493,47 +499,51 @@ def start_redis(node_ip_address, if use_credis is None: use_credis = ("RAY_USE_NEW_GCS" in os.environ) - if use_credis and password is not None: - # TODO(pschafhalter) remove this once credis supports - # authenticating Redis ports - raise Exception("Setting the `redis_password` argument is not " - "supported in credis. To run Ray with " - "password-protected Redis ports, ensure that " - "the environment variable `RAY_USE_NEW_GCS=off`.") - if not use_credis: - assigned_port, p = _start_redis_instance( - node_ip_address=node_ip_address, - port=port, - redis_max_clients=redis_max_clients, - stdout_file=redis_stdout_file, - stderr_file=redis_stderr_file, - password=password, - # Below we use None to indicate no limit on the memory of the - # primary Redis shard. - redis_max_memory=None) - processes.append(p) + if use_credis: + if password is not None: + # TODO(pschafhalter) remove this once credis supports + # authenticating Redis ports + raise Exception("Setting the `redis_password` argument is not " + "supported in credis. To run Ray with " + "password-protected Redis ports, ensure that " + "the environment variable `RAY_USE_NEW_GCS=off`.") + assert num_redis_shards == 1, ( + "For now, RAY_USE_NEW_GCS supports 1 shard, and credis " + "supports 1-node chain for that shard only.") + + if use_credis: + redis_executable = CREDIS_EXECUTABLE + # TODO(suquark): We need credis here because some symbols need to be + # imported from credis dynamically through dlopen when Ray is built + # with RAY_USE_NEW_GCS=on. We should remove them later for the primary + # shard. + # See src/ray/gcs/redis_module/ray_redis_module.cc + redis_modules = [CREDIS_MASTER_MODULE, REDIS_MODULE] else: - assigned_port, p = _start_redis_instance( - node_ip_address=node_ip_address, - port=port, - redis_max_clients=redis_max_clients, - stdout_file=redis_stdout_file, - stderr_file=redis_stderr_file, - executable=CREDIS_EXECUTABLE, - # It is important to load the credis module BEFORE the ray module, - # as the latter contains an extern declaration that the former - # supplies. - modules=[CREDIS_MASTER_MODULE, REDIS_MODULE], - password=password, - # Below we use None to indicate no limit on the memory of the - # primary Redis shard. - redis_max_memory=None) - processes.append(p) - if port is not None: - assert assigned_port == port - port = assigned_port + redis_executable = REDIS_EXECUTABLE + redis_modules = [REDIS_MODULE] + + # Start the primary Redis shard. + port, p = _start_redis_instance( + redis_executable, + modules=redis_modules, + port=port, + password=password, + redis_max_clients=redis_max_clients, + # Below we use None to indicate no limit on the memory of the + # primary Redis shard. + redis_max_memory=None, + stdout_file=redis_stdout_file, + stderr_file=redis_stderr_file) + processes.append(p) redis_address = address(node_ip_address, port) + # Record the log files in Redis. + record_log_files_in_redis( + redis_address, + node_ip_address, [redis_stdout_file, redis_stderr_file], + password=password) + # Register the number of Redis shards in the primary shard, so that clients # know how many redis shards to expect under RedisShards. primary_redis_client = redis.StrictRedis( @@ -563,69 +573,95 @@ def start_redis(node_ip_address, for i in range(num_redis_shards): redis_stdout_file, redis_stderr_file = new_redis_log_file( redirect_output, shard_number=i) - if not use_credis: - redis_shard_port, p = _start_redis_instance( - node_ip_address=node_ip_address, - port=redis_shard_ports[i], - redis_max_clients=redis_max_clients, - stdout_file=redis_stdout_file, - stderr_file=redis_stderr_file, - password=password, - redis_max_memory=redis_max_memory) - processes.append(p) - else: - assert num_redis_shards == 1, \ - "For now, RAY_USE_NEW_GCS supports 1 shard, and credis "\ - "supports 1-node chain for that shard only." - redis_shard_port, p = _start_redis_instance( - node_ip_address=node_ip_address, - port=redis_shard_ports[i], - redis_max_clients=redis_max_clients, - stdout_file=redis_stdout_file, - stderr_file=redis_stderr_file, - password=password, - executable=CREDIS_EXECUTABLE, - # It is important to load the credis module BEFORE the ray - # module, as the latter contains an extern declaration that the - # former supplies. - modules=[CREDIS_MEMBER_MODULE, REDIS_MODULE], - redis_max_memory=redis_max_memory) - processes.append(p) - if redis_shard_ports[i] is not None: - assert redis_shard_port == redis_shard_ports[i] + if use_credis: + redis_executable = CREDIS_EXECUTABLE + # It is important to load the credis module BEFORE the ray module, + # as the latter contains an extern declaration that the former + # supplies. + redis_modules = [CREDIS_MEMBER_MODULE, REDIS_MODULE] + else: + redis_executable = REDIS_EXECUTABLE + redis_modules = [REDIS_MODULE] + + redis_shard_port, p = _start_redis_instance( + redis_executable, + modules=redis_modules, + port=redis_shard_ports[i], + password=password, + redis_max_clients=redis_max_clients, + redis_max_memory=redis_max_memory, + stdout_file=redis_stdout_file, + stderr_file=redis_stderr_file) + processes.append(p) + shard_address = address(node_ip_address, redis_shard_port) redis_shards.append(shard_address) # Store redis shard information in the primary redis shard. primary_redis_client.rpush("RedisShards", shard_address) + record_log_files_in_redis( + redis_address, + node_ip_address, [redis_stdout_file, redis_stderr_file], + password=password) + if use_credis: + # Configure the chain state. The way it is intended to work is + # the following: + # + # PRIMARY_SHARD + # + # SHARD_1 (master replica) -> SHARD_1 (member replica) + # -> SHARD_1 (member replica) + # + # SHARD_2 (master replica) -> SHARD_2 (member replica) + # -> SHARD_2 (member replica) + # ... + # + # + # If we have credis members in future, their modules should be: + # [CREDIS_MEMBER_MODULE, REDIS_MODULE], and they will be initialized by + # execute_command("MEMBER.CONNECT_TO_MASTER", node_ip_address, port) + # + # Currently we have num_redis_shards == 1, so only one chain will be + # created, and the chain only contains master. + + # TODO(suquark): Currently, this is not correct because we are + # using the master replica as the primary shard. This should be + # fixed later. I had tried to fix it but failed because of heartbeat + # issues. + primary_client = redis.StrictRedis( + host=node_ip_address, port=port, password=password) shard_client = redis.StrictRedis( host=node_ip_address, port=redis_shard_port, password=password) - # Configure the chain state. - primary_redis_client.execute_command("MASTER.ADD", node_ip_address, - redis_shard_port) + primary_client.execute_command("MASTER.ADD", node_ip_address, + redis_shard_port) shard_client.execute_command("MEMBER.CONNECT_TO_MASTER", node_ip_address, port) return redis_address, redis_shards, processes -def _start_redis_instance(node_ip_address="127.0.0.1", +def _start_redis_instance(executable, + modules, port=None, redis_max_clients=None, num_retries=20, stdout_file=None, stderr_file=None, password=None, - executable=REDIS_EXECUTABLE, - modules=None, redis_max_memory=None): """Start a single Redis server. + Notes: + If "port" is not None, then we will only use this port and try + only once. Otherwise, random ports will be used and the maximum + retries count is "num_retries". + Args: - node_ip_address (str): The IP address of the current node. This is only - used for recording the log filenames in Redis. + executable (str): Full path of the redis-server executable. + modules (list of str): A list of pathnames, pointing to the redis + module(s) that will be loaded in this redis server. port (int): If provided, start a Redis server with this port. redis_max_clients: If this is provided, Ray will attempt to configure Redis with this maxclients number. @@ -637,10 +673,6 @@ def _start_redis_instance(node_ip_address="127.0.0.1", no redirection should happen, then this should be None. password (str): Prevents external clients without the password from connecting to Redis if provided. - executable (str): Full path tho the redis-server executable. - modules (list of str): A list of pathnames, pointing to the redis - module(s) that will be loaded in this redis server. If None, load - the default Ray redis module. redis_max_memory: The max amount of memory (in bytes) to allow redis to use, or None for no limit. Once the limit is exceeded, redis will start LRU eviction of entries. @@ -654,13 +686,12 @@ def _start_redis_instance(node_ip_address="127.0.0.1", Exception: An exception is raised if Redis could not be started. """ assert os.path.isfile(executable) - if modules is None: - modules = [REDIS_MODULE] for module in modules: assert os.path.isfile(module) counter = 0 if port is not None: # If a port is specified, then try only once to connect. + # This ensures that we will use the given port. num_retries = 1 else: port = new_port() @@ -749,11 +780,6 @@ def _start_redis_instance(node_ip_address="127.0.0.1", " ".join(cur_config_list)) # Put a time stamp in Redis to indicate when it was started. redis_client.set("redis_start_time", time.time()) - # Record the log files in Redis. - record_log_files_in_redis( - address(node_ip_address, port), - node_ip_address, [stdout_file, stderr_file], - password=password) return port, process_info @@ -821,10 +847,12 @@ def start_ui(redis_address, stdout_file=None, stderr_file=None): break except socket.error: port += 1 + + notebook_name = get_ipython_notebook_path() + new_notebook_directory = os.path.dirname(notebook_name) # We generate the token used for authentication ourselves to avoid # querying the jupyter server. - new_notebook_directory, webui_url, token = ( - get_ipython_notebook_path(port)) + token = ray.utils.decode(binascii.hexlify(os.urandom(24))) # The --ip=0.0.0.0 flag is intended to enable connecting to a notebook # running within a docker container (from the outside). command = [ @@ -849,6 +877,8 @@ def start_ui(redis_address, stdout_file=None, stderr_file=None): logger.warning("Failed to start the UI, you may need to run " "'pip install jupyter'.") else: + webui_url = ("http://localhost:{}/notebooks/{}?token={}".format( + port, os.path.basename(notebook_name), token)) print("\n" + "=" * 70) print("View the web UI at {}".format(webui_url)) print("=" * 70 + "\n") @@ -1170,12 +1200,8 @@ def _start_plasma_store(plasma_store_memory, if not isinstance(plasma_store_memory, int): raise Exception("plasma_store_memory should be an integer.") - plasma_store_executable = os.path.join( - os.path.abspath(os.path.dirname(__file__)), - "core/src/plasma/plasma_store_server") - plasma_store_name = socket_name command = [ - plasma_store_executable, "-s", plasma_store_name, "-m", + PLASMA_STORE_EXECUTABLE, "-s", socket_name, "-m", str(plasma_store_memory) ] if plasma_directory is not None: @@ -1189,7 +1215,7 @@ def _start_plasma_store(plasma_store_memory, use_valgrind_profiler=use_profiler, stdout_file=stdout_file, stderr_file=stderr_file) - return plasma_store_name, process_info + return process_info def start_plasma_store(node_ip_address, @@ -1236,7 +1262,7 @@ def start_plasma_store(node_ip_address, logger.info("Starting the Plasma object store with {} GB memory " "using {}.".format(object_store_memory_str, plasma_directory)) # Start the Plasma store. - plasma_store_name, process_info = _start_plasma_store( + process_info = _start_plasma_store( object_store_memory, use_profiler=RUN_PLASMA_STORE_PROFILER, stdout_file=stdout_file, @@ -1255,7 +1281,7 @@ def start_plasma_store(node_ip_address, def start_worker(node_ip_address, object_store_name, - local_scheduler_name, + raylet_name, redis_address, worker_path, stdout_file=None, @@ -1265,8 +1291,8 @@ def start_worker(node_ip_address, Args: node_ip_address (str): The IP address of the node that this worker is running on. - object_store_name (str): The name of the object store. - local_scheduler_name (str): The name of the local scheduler. + object_store_name (str): The socket name of the object store. + raylet_name (str): The socket name of the raylet server. redis_address (str): The address that the Redis server is listening on. worker_path (str): The path of the source code which the worker process will run. @@ -1282,6 +1308,7 @@ def start_worker(node_ip_address, sys.executable, "-u", worker_path, "--node-ip-address=" + node_ip_address, "--object-store-name=" + object_store_name, + "--raylet-name=" + raylet_name, "--redis-address=" + str(redis_address), "--temp-dir=" + get_temp_root() ] diff --git a/python/ray/tempfile_services.py b/python/ray/tempfile_services.py index c8da4e58b..156843b79 100644 --- a/python/ray/tempfile_services.py +++ b/python/ray/tempfile_services.py @@ -1,4 +1,3 @@ -import binascii import collections import datetime import errno @@ -130,7 +129,7 @@ def get_object_store_socket_name(): return make_inc_temp(prefix="plasma_store", directory_name=sockets_dir) -def get_ipython_notebook_path(port): +def get_ipython_notebook_path(): """Get a new ipython notebook path""" notebook_filepath = os.path.join( @@ -140,11 +139,7 @@ def get_ipython_notebook_path(port): notebook_name = make_inc_temp( suffix=".ipynb", prefix="ray_ui", directory_name=get_temp_root()) shutil.copy(notebook_filepath, notebook_name) - new_notebook_directory = os.path.dirname(notebook_name) - token = ray.utils.decode(binascii.hexlify(os.urandom(24))) - webui_url = ("http://localhost:{}/notebooks/{}?token={}".format( - port, os.path.basename(notebook_name), token)) - return new_notebook_directory, webui_url, token + return notebook_name def new_log_files(name, redirect_output): diff --git a/python/ray/utils.py b/python/ray/utils.py index b814fc86f..64de24187 100644 --- a/python/ray/utils.py +++ b/python/ray/utils.py @@ -49,11 +49,7 @@ def format_error_message(exception_message, task_exception=False): return "\n".join(lines) -def push_error_to_driver(worker, - error_type, - message, - driver_id=None, - data=None): +def push_error_to_driver(worker, error_type, message, driver_id=None): """Push an error message to the driver to be printed in the background. Args: @@ -63,12 +59,9 @@ def push_error_to_driver(worker, on the driver. driver_id: The ID of the driver to push the error message to. If this is None, then the message will be pushed to all drivers. - data: This should be a dictionary mapping strings to strings. It - will be serialized with json and stored in Redis. """ if driver_id is None: driver_id = ray.DriverID.nil() - data = {} if data is None else data worker.raylet_client.push_error(driver_id, error_type, message, time.time()) @@ -76,8 +69,7 @@ def push_error_to_driver(worker, def push_error_to_driver_through_redis(redis_client, error_type, message, - driver_id=None, - data=None): + driver_id=None): """Push an error message to the driver to be printed in the background. Normally the push_error_to_driver function should be used. However, in some @@ -92,12 +84,9 @@ def push_error_to_driver_through_redis(redis_client, on the driver. driver_id: The ID of the driver to push the error message to. If this is None, then the message will be pushed to all drivers. - data: This should be a dictionary mapping strings to strings. It - will be serialized with json and stored in Redis. """ if driver_id is None: driver_id = ray.DriverID.nil() - data = {} if data is None else data # Do everything in Python and through the Python Redis client instead # of through the raylet. error_data = ray.gcs_utils.construct_error_message(driver_id, error_type, @@ -132,7 +121,7 @@ def is_function_or_method(obj): Returns: True if the object is an function or method. """ - return (inspect.isfunction(obj) or inspect.ismethod(obj) or is_cython(obj)) + return inspect.isfunction(obj) or inspect.ismethod(obj) or is_cython(obj) def is_class_method(f): @@ -367,7 +356,6 @@ def get_system_memory(): except ImportError: pass - memory_in_bytes = None if psutil_memory_in_bytes is not None: memory_in_bytes = psutil_memory_in_bytes elif sys.platform == "linux" or sys.platform == "linux2": diff --git a/python/ray/worker.py b/python/ray/worker.py index c50f50c07..72198bcaf 100644 --- a/python/ray/worker.py +++ b/python/ray/worker.py @@ -83,7 +83,6 @@ class RayTaskError(Exception): Attributes: function_name (str): The name of the function that failed and produced the RayTaskError. - exception (Exception): The exception object thrown by the failed task. traceback_str (str): The traceback from the exception. """ @@ -522,8 +521,7 @@ class Worker(object): num_return_vals=None, resources=None, placement_resources=None, - driver_id=None, - language=ray.gcs_utils.Language.PYTHON): + driver_id=None): """Submit a remote task to the scheduler. Tell the scheduler to schedule the execution of the function with @@ -658,7 +656,7 @@ class Worker(object): takes only one argument, a worker info dict. If it returns anything, its return values will not be used. run_on_other_drivers: The boolean that indicates whether we want to - run this funtion on other drivers. One case is we may need to + run this function on other drivers. One case is we may need to share objects across drivers. """ # If ray.init has not been called yet, then cache the function and @@ -864,7 +862,6 @@ class Worker(object): def _handle_process_task_failure(self, function_descriptor, return_object_ids, error, backtrace): function_name = function_descriptor.function_name - function_id = function_descriptor.function_id failure_object = RayTaskError(function_name, backtrace) failure_objects = [ failure_object for _ in range(len(return_object_ids)) @@ -875,13 +872,7 @@ class Worker(object): self, ray_constants.TASK_PUSH_ERROR, str(failure_object), - driver_id=self.task_driver_id, - data={ - "function_id": function_id.binary(), - "function_name": function_name, - "module_name": function_descriptor.module_name, - "class_name": function_descriptor.class_name - }) + driver_id=self.task_driver_id) # Mark the actor init as failed if not self.actor_id.is_nil() and function_name == "__init__": self.mark_actor_init_failed(error) @@ -1089,19 +1080,6 @@ def print_failed_task(task_status): task_status["error_message"])) -def error_applies_to_driver(error_key, worker=global_worker): - """Return True if the error is for this driver and false otherwise.""" - # TODO(rkn): Should probably check that this is only called on a driver. - # Check that the error key is formatted as in push_error_to_driver. - assert len(error_key) == (len(ERROR_KEY_PREFIX) + ray_constants.ID_SIZE + 1 - + ray_constants.ID_SIZE), error_key - # If the driver ID in the error message is a sequence of all zeros, then - # the message is intended for all drivers. - driver_id = DriverID(error_key[len(ERROR_KEY_PREFIX):( - len(ERROR_KEY_PREFIX) + ray_constants.ID_SIZE)]) - return (driver_id == worker.task_driver_id or driver_id == DriverID.nil()) - - def error_info(worker=global_worker): """Return information about failed tasks.""" worker.check_connected() @@ -1513,9 +1491,8 @@ def init(redis_address=None, "redis_address": address_info["redis_address"], "store_socket_name": address_info["object_store_address"], "webui_url": address_info["webui_url"], + "raylet_socket_name": address_info["raylet_socket_name"], } - driver_address_info["raylet_socket_name"] = ( - address_info["raylet_socket_name"]) # We only pass `temp_dir` to a worker (WORKER_MODE). # It can't be a worker here. @@ -1680,49 +1657,6 @@ def is_initialized(): return ray.worker.global_worker.connected -def print_error_messages(worker): - """Print error messages in the background on the driver. - - This runs in a separate thread on the driver and prints error messages in - the background. - """ - # TODO(rkn): All error messages should have a "component" field indicating - # which process the error came from (e.g., a worker or a plasma store). - # Currently all error messages come from workers. - - worker.error_message_pubsub_client = worker.redis_client.pubsub() - # Exports that are published after the call to - # error_message_pubsub_client.subscribe and before the call to - # error_message_pubsub_client.listen will still be processed in the loop. - worker.error_message_pubsub_client.subscribe("__keyspace@0__:ErrorKeys") - num_errors_received = 0 - - # Get the exports that occurred before the call to subscribe. - with worker.lock: - error_keys = worker.redis_client.lrange("ErrorKeys", 0, -1) - for error_key in error_keys: - if error_applies_to_driver(error_key, worker=worker): - error_message = ray.utils.decode( - worker.redis_client.hget(error_key, "message")) - logger.error(error_message) - num_errors_received += 1 - - try: - for msg in worker.error_message_pubsub_client.listen(): - with worker.lock: - for error_key in worker.redis_client.lrange( - "ErrorKeys", num_errors_received, -1): - if error_applies_to_driver(error_key, worker=worker): - error_message = ray.utils.decode( - worker.redis_client.hget(error_key, "message")) - logger.error(error_message) - num_errors_received += 1 - except redis.ConnectionError: - # When Redis terminates the listen call will throw a ConnectionError, - # which we catch here. - pass - - def connect(info, redis_password=None, object_id_seed=None, @@ -1817,9 +1751,32 @@ def connect(info, worker.lock = threading.Lock() - # Check the RedirectOutput key in Redis and based on its value redirect - # worker output and error to their own files. - if mode == WORKER_MODE: + # Create an object for interfacing with the global state. + global_state._initialize_global_state( + redis_ip_address, int(redis_port), redis_password=redis_password) + + # Register the worker with Redis. + if mode == SCRIPT_MODE: + # The concept of a driver is the same as the concept of a "job". + # Register the driver/job with Redis here. + import __main__ as main + driver_info = { + "node_ip_address": worker.node_ip_address, + "driver_id": worker.worker_id, + "start_time": time.time(), + "plasma_store_socket": info["store_socket_name"], + "raylet_socket": info.get("raylet_socket_name"), + "name": (main.__file__ + if hasattr(main, "__file__") else "INTERACTIVE MODE") + } + worker.redis_client.hmset(b"Drivers:" + worker.worker_id, driver_info) + if (not worker.redis_client.exists("webui") + and info["webui_url"] is not None): + worker.redis_client.hmset("webui", {"url": info["webui_url"]}) + is_worker = False + elif mode == WORKER_MODE: + # Check the RedirectOutput key in Redis and based on its value redirect + # worker output and error to their own files. # This key is set in services.py when Redis is started. redirect_worker_output_val = worker.redis_client.get("RedirectOutput") if (redirect_worker_output_val is not None @@ -1838,31 +1795,6 @@ def connect(info, info["redis_address"], info["node_ip_address"], [log_stdout_file, log_stderr_file], password=redis_password) - - # Create an object for interfacing with the global state. - global_state._initialize_global_state( - redis_ip_address, int(redis_port), redis_password=redis_password) - - # Register the worker with Redis. - if mode == SCRIPT_MODE: - # The concept of a driver is the same as the concept of a "job". - # Register the driver/job with Redis here. - import __main__ as main - driver_info = { - "node_ip_address": worker.node_ip_address, - "driver_id": worker.worker_id, - "start_time": time.time(), - "plasma_store_socket": info["store_socket_name"], - "raylet_socket": info.get("raylet_socket_name") - } - driver_info["name"] = (main.__file__ if hasattr(main, "__file__") else - "INTERACTIVE MODE") - worker.redis_client.hmset(b"Drivers:" + worker.worker_id, driver_info) - if (not worker.redis_client.exists("webui") - and info["webui_url"] is not None): - worker.redis_client.hmset("webui", {"url": info["webui_url"]}) - is_worker = False - elif mode == WORKER_MODE: # Register the worker with Redis. worker_dict = { "node_ip_address": worker.node_ip_address,