wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-13 17:45:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

Various Python code cleanups. (#3837)

Browse Source
This commit is contained in:
Si-Yuan
2019-02-04 02:16:24 +08:00
committed by Robert Nishihara
parent a1bcd2a4f5
commit 9295ab8f60
12 changed files with 177 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/actor.py
+3 -16
View File
@@ -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
python/ray/experimental/state.py
+1 -1
View File
@@ -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))
python/ray/function_manager.py
+8 -14
View File
@@ -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
python/ray/import_thread.py
+1 -4
View File
@@ -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))
python/ray/log_monitor.py
-2
View File
@@ -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
python/ray/monitor.py
-2
View File
@@ -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):
python/ray/plasma/.gitkeep
View File
python/ray/profiling.py
+2 -2
View File
@@ -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):
python/ray/services.py
+127 -100
View File
@@ -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()
]
python/ray/tempfile_services.py
+2 -7
View File
@@ -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):
python/ray/utils.py
+3 -15
View File
@@ -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":
python/ray/worker.py
+30 -98
View File
@@ -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,