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

Add RayParams to refactor the parameters used by ray python. (#3558)

Browse Source
This commit is contained in:
Yuhong Guo
2018-12-29 22:04:27 +08:00
committed by Hao Chen
parent eb1e5fa2cf
commit c9b8ecca51
12 changed files with 543 additions and 697 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/parameter.py
+192
View File
@@ -0,0 +1,192 @@
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import logging
import ray.ray_constants as ray_constants
class RayParams(object):
"""A class used to store the parameters used by Ray.
Attributes:
address_info (dict): A dictionary with address information for
processes in a partially-started Ray cluster. If
start_ray_local=True, any processes not in this dictionary will be
started. If provided, an updated address_info dictionary will be
returned to include processes that are newly started.
start_ray_local (bool): If True then this will start any processes not
already in address_info, including Redis, a global scheduler, local
scheduler(s), object store(s), and worker(s). It will also kill
these processes when Python exits. If False, this will attach to an
existing Ray cluster.
redis_address (str): The address of the Redis server to connect to. If
this address is not provided, then this command will start Redis, a
global scheduler, a local scheduler, a plasma store, a plasma
manager, and some workers. It will also kill these processes when
Python exits.
redis_port (int): The port that the primary Redis shard should listen
to. If None, then a random port will be chosen. If the key
"redis_address" is in address_info, then this argument will be
ignored.
redis_shard_ports: A list of the ports to use for the non-primary Redis
shards.
num_cpus (int): Number of cpus the user wishes all local schedulers to
be configured with.
num_gpus (int): Number of gpus the user wishes all local schedulers to
be configured with.
num_local_schedulers (int): The number of local schedulers to start.
This is only provided if start_ray_local is True.
resources: A dictionary mapping the name of a resource to the quantity
of that resource available.
object_store_memory: The amount of memory (in bytes) to start the
object store with.
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. This only applies to the
sharded redis tables (task and object tables).
object_manager_ports (list): A list of the ports to use for the object
managers. There should be one per object manager being started on
this node (typically just one).
node_manager_ports (list): A list of the ports to use for the node
managers. There should be one per node manager being started on
this node (typically just one).
collect_profiling_data: Whether to collect profiling data from workers.
node_ip_address (str): The IP address of the node that we are on.
object_id_seed (int): Used to seed the deterministic generation of
object IDs. The same value can be used across multiple runs of the
same job in order to generate the object IDs in a consistent
manner. However, the same ID should not be used for different jobs.
local_mode (bool): True if the code should be executed serially
without Ray. This is useful for debugging.
redirect_worker_output: True if the stdout and stderr of worker
processes should be redirected to files.
redirect_output (bool): True if stdout and stderr for non-worker
processes should be redirected to files and false otherwise.
num_redis_shards: The number of Redis shards to start in addition to
the primary Redis shard.
redis_max_clients: If provided, attempt to configure Redis with this
maxclients number.
redis_password (str): Prevents external clients without the password
from connecting to Redis if provided.
plasma_directory: A directory where the Plasma memory mapped files will
be created.
worker_path (str): The path of the source code that will be run by the
worker.
huge_pages: Boolean flag indicating whether to start the Object
Store with hugetlbfs support. Requires plasma_directory.
include_webui: Boolean flag indicating whether to start the web
UI, which is a Jupyter notebook.
plasma_store_socket_name (str): If provided, it will specify the socket
name used by the plasma store.
raylet_socket_name (str): If provided, it will specify the socket path
used by the raylet process.
temp_dir (str): If provided, it will specify the root temporary
directory for the Ray process.
include_log_monitor (bool): If True, then start a log monitor to
monitor the log files for all processes on this node and push their
contents to Redis.
autoscaling_config: path to autoscaling config file.
_internal_config (str): JSON configuration for overriding
RayConfig defaults. For testing purposes ONLY.
"""
def __init__(self,
address_info=None,
start_ray_local=False,
redis_address=None,
num_cpus=None,
num_gpus=None,
num_local_schedulers=None,
resources=None,
object_store_memory=None,
redis_max_memory=None,
redis_port=None,
redis_shard_ports=None,
object_manager_ports=None,
node_manager_ports=None,
collect_profiling_data=True,
node_ip_address=None,
object_id_seed=None,
num_workers=None,
local_mode=False,
driver_mode=None,
redirect_worker_output=False,
redirect_output=True,
num_redis_shards=None,
redis_max_clients=None,
redis_password=None,
plasma_directory=None,
worker_path=None,
huge_pages=False,
include_webui=None,
logging_level=logging.INFO,
logging_format=ray_constants.LOGGER_FORMAT,
plasma_store_socket_name=None,
raylet_socket_name=None,
temp_dir=None,
include_log_monitor=None,
autoscaling_config=None,
_internal_config=None):
self.address_info = address_info
self.start_ray_local = start_ray_local
self.object_id_seed = object_id_seed
self.redis_address = redis_address
self.num_cpus = num_cpus
self.num_gpus = num_gpus
self.num_local_schedulers = num_local_schedulers
self.resources = resources
self.object_store_memory = object_store_memory
self.redis_max_memory = redis_max_memory
self.redis_port = redis_port
self.redis_shard_ports = redis_shard_ports
self.object_manager_ports = object_manager_ports
self.node_manager_ports = node_manager_ports
self.collect_profiling_data = collect_profiling_data
self.node_ip_address = node_ip_address
self.num_workers = num_workers
self.local_mode = local_mode
self.driver_mode = driver_mode
self.redirect_worker_output = redirect_worker_output
self.redirect_output = redirect_output
self.num_redis_shards = num_redis_shards
self.redis_max_clients = redis_max_clients
self.redis_password = redis_password
self.plasma_directory = plasma_directory
self.worker_path = worker_path
self.huge_pages = huge_pages
self.include_webui = include_webui
self.plasma_store_socket_name = plasma_store_socket_name
self.raylet_socket_name = raylet_socket_name
self.temp_dir = temp_dir
self.include_log_monitor = include_log_monitor
self.autoscaling_config = autoscaling_config
self._internal_config = _internal_config
def update(self, **kwargs):
"""Update the settings according to the keyword arguments.
Args:
kwargs: The keyword arguments to set corresponding fields.
"""
for arg in kwargs:
if (hasattr(self, arg)):
setattr(self, arg, kwargs[arg])
else:
raise ValueError("Invalid RayParams parameter in"
" update: %s" % arg)
def update_if_absent(self, **kwargs):
"""Update the settings when the target fields are None.
Args:
kwargs: The keyword arguments to set corresponding fields.
"""
for arg in kwargs:
if (hasattr(self, arg)):
if getattr(self, arg) is None:
setattr(self, arg, kwargs[arg])
else:
raise ValueError("Invalid RayParams parameter in"
" update_if_absent: %s" % arg)
python/ray/scripts/scripts.py
+32 -44
View File
@@ -15,6 +15,8 @@ from ray.autoscaler.commands import (attach_cluster, exec_cluster,
import ray.ray_constants as ray_constants
import ray.utils
from ray.parameter import RayParams
logger = logging.getLogger(__name__)
@@ -243,6 +245,22 @@ def start(node_ip_address, redis_address, redis_port, num_redis_shards,
resources["CPU"] = num_cpus
if num_gpus is not None:
resources["GPU"] = num_gpus
ray_params = RayParams(
node_ip_address=node_ip_address,
object_manager_ports=[object_manager_port],
node_manager_ports=[node_manager_port],
num_workers=num_workers,
object_store_memory=object_store_memory,
redis_password=redis_password,
redirect_worker_output=not no_redirect_worker_output,
redirect_output=not no_redirect_output,
resources=resources,
plasma_directory=plasma_directory,
huge_pages=huge_pages,
plasma_store_socket_name=plasma_store_socket_name,
raylet_socket_name=raylet_socket_name,
temp_dir=temp_dir,
_internal_config=internal_config)
if head:
# Start Ray on the head node.
@@ -266,36 +284,21 @@ def start(node_ip_address, redis_address, redis_port, num_redis_shards,
"provided.")
# Get the node IP address if one is not provided.
if node_ip_address is None:
node_ip_address = services.get_node_ip_address()
ray_params.update_if_absent(
node_ip_address=services.get_node_ip_address())
logger.info("Using IP address {} for this node."
.format(node_ip_address))
address_info = services.start_ray_head(
object_manager_ports=[object_manager_port],
node_manager_ports=[node_manager_port],
node_ip_address=node_ip_address,
.format(ray_params.node_ip_address))
ray_params.update_if_absent(
redis_port=redis_port,
redis_shard_ports=redis_shard_ports,
object_store_memory=object_store_memory,
redis_max_memory=redis_max_memory,
collect_profiling_data=collect_profiling_data,
num_workers=num_workers,
cleanup=False,
redirect_worker_output=not no_redirect_worker_output,
redirect_output=not no_redirect_output,
resources=resources,
num_redis_shards=num_redis_shards,
redis_max_clients=redis_max_clients,
redis_password=redis_password,
include_webui=(not no_ui),
plasma_directory=plasma_directory,
huge_pages=huge_pages,
autoscaling_config=autoscaling_config,
plasma_store_socket_name=plasma_store_socket_name,
raylet_socket_name=raylet_socket_name,
temp_dir=temp_dir,
_internal_config=internal_config)
autoscaling_config=autoscaling_config)
address_info = services.start_ray_head(ray_params, cleanup=False)
logger.info(address_info)
logger.info(
"\nStarted Ray on this node. You can add additional nodes to "
@@ -350,32 +353,17 @@ def start(node_ip_address, redis_address, redis_port, num_redis_shards,
services.check_version_info(redis_client)
# Get the node IP address if one is not provided.
if node_ip_address is None:
node_ip_address = services.get_node_ip_address(redis_address)
ray_params.update_if_absent(
node_ip_address=services.get_node_ip_address(redis_address))
logger.info("Using IP address {} for this node."
.format(node_ip_address))
.format(ray_params.node_ip_address))
# Check that there aren't already Redis clients with the same IP
# address connected with this Redis instance. This raises an exception
# if the Redis server already has clients on this node.
check_no_existing_redis_clients(node_ip_address, redis_client)
address_info = services.start_ray_node(
node_ip_address=node_ip_address,
redis_address=redis_address,
object_manager_ports=[object_manager_port],
node_manager_ports=[node_manager_port],
num_workers=num_workers,
object_store_memory=object_store_memory,
redis_password=redis_password,
cleanup=False,
redirect_worker_output=not no_redirect_worker_output,
redirect_output=not no_redirect_output,
resources=resources,
plasma_directory=plasma_directory,
huge_pages=huge_pages,
plasma_store_socket_name=plasma_store_socket_name,
raylet_socket_name=raylet_socket_name,
temp_dir=temp_dir,
_internal_config=internal_config)
check_no_existing_redis_clients(ray_params.node_ip_address,
redis_client)
ray_params.redis_address = redis_address
address_info = services.start_ray_node(ray_params, cleanup=False)
logger.info(address_info)
logger.info("\nStarted Ray on this node. If you wish to terminate the "
"processes that have been started, run\n\n"
python/ray/services.py
+163 -423
View File
@@ -867,41 +867,31 @@ def check_and_update_resources(resources):
return resources
def start_raylet(redis_address,
node_ip_address,
def start_raylet(ray_params,
index,
raylet_name,
plasma_store_name,
worker_path,
resources=None,
object_manager_port=None,
node_manager_port=None,
num_workers=0,
use_valgrind=False,
use_profiler=False,
stdout_file=None,
stderr_file=None,
cleanup=True,
config=None,
redis_password=None,
collect_profiling_data=True):
config=None):
"""Start a raylet, which is a combined local scheduler and object manager.
Args:
redis_address (str): The address of the Redis instance.
node_ip_address (str): The IP address of the node that this local
scheduler is running on.
plasma_store_name (str): The name of the plasma store socket to connect
to.
ray_params (ray.params.RayParams): The RayParams instance. The
following parameters could be checked: redis_address,
node_ip_address, worker_path, resources, object_manager_ports,
node_manager_ports, redis_password
index (int): Usually, this index is 0. When index > 0, it means
starting multiple raylet locally. The index will be used in
resources, object_manager_ports, node_manager_ports.
raylet_name (str): The name of the raylet socket to create.
worker_path (str): The path of the script to use when the local
scheduler starts up new workers.
resources: The resources that this raylet has.
object_manager_port (int): The port to use for the object manager. If
this is not provided, we will use 0 and the object manager will
choose its own port.
node_manager_port (int): The port to use for the node manager. If
this is not provided, we will use 0 and the node manager will
choose its own port.
plasma_store_name (str): The name of the plasma store socket to connect
to.
num_workers (int): The number of workers to start.
use_valgrind (bool): True if the raylet should be started inside
of valgrind. If this is True, use_profiler must be False.
use_profiler (bool): True if the raylet should be started inside
@@ -915,8 +905,6 @@ def start_raylet(redis_address,
Python process that imported services exits.
config (dict|None): Optional Raylet configuration that will
override defaults in RayConfig.
redis_password (str): The password of the redis server.
collect_profiling_data: Whether to collect profiling data from workers.
Returns:
The raylet socket name.
@@ -927,7 +915,7 @@ def start_raylet(redis_address,
if use_valgrind and use_profiler:
raise Exception("Cannot use valgrind and profiler at the same time.")
static_resources = check_and_update_resources(resources)
static_resources = check_and_update_resources(ray_params.resources[index])
# Limit the number of workers that can be started in parallel by the
# raylet. However, make sure it is at least 1.
@@ -938,7 +926,7 @@ def start_raylet(redis_address,
resource_argument = ",".join(
["{},{}".format(*kv) for kv in static_resources.items()])
gcs_ip_address, gcs_port = redis_address.split(":")
gcs_ip_address, gcs_port = ray_params.redis_address.split(":")
# Create the command that the Raylet will use to start workers.
start_worker_command = ("{} {} "
@@ -948,29 +936,31 @@ def start_raylet(redis_address,
"--redis-address={} "
"--collect-profiling-data={} "
"--temp-dir={}".format(
sys.executable, worker_path, node_ip_address,
plasma_store_name, raylet_name, redis_address,
"1" if collect_profiling_data else "0",
sys.executable, ray_params.worker_path,
ray_params.node_ip_address, plasma_store_name,
raylet_name, ray_params.redis_address, "1"
if ray_params.collect_profiling_data else "0",
get_temp_root()))
if redis_password:
start_worker_command += " --redis-password {}".format(redis_password)
if ray_params.redis_password:
start_worker_command += " --redis-password {}".format(
ray_params.redis_password)
# If the object manager port is None, then use 0 to cause the object
# manager to choose its own port.
if object_manager_port is None:
object_manager_port = 0
if ray_params.object_manager_ports[index] is None:
ray_params.object_manager_ports[index] = 0
# If the node manager port is None, then use 0 to cause the node manager
# to choose its own port.
if node_manager_port is None:
node_manager_port = 0
if ray_params.node_manager_ports[index] is None:
ray_params.node_manager_ports[index] = 0
command = [
RAYLET_EXECUTABLE,
raylet_name,
plasma_store_name,
str(object_manager_port),
str(node_manager_port),
node_ip_address,
str(ray_params.object_manager_ports[index]),
str(ray_params.node_manager_ports[index]),
ray_params.node_ip_address,
gcs_ip_address,
gcs_port,
str(num_workers),
@@ -979,7 +969,7 @@ def start_raylet(redis_address,
config_str,
start_worker_command,
"", # Worker command for Java, not needed for Python.
redis_password or "",
ray_params.redis_password or "",
get_temp_root(),
]
@@ -1009,9 +999,9 @@ def start_raylet(redis_address,
if cleanup:
all_processes[PROCESS_TYPE_RAYLET].append(pid)
record_log_files_in_redis(
redis_address,
node_ip_address, [stdout_file, stderr_file],
password=redis_password)
ray_params.redis_address,
ray_params.node_ip_address, [stdout_file, stderr_file],
password=ray_params.redis_password)
return raylet_name
@@ -1276,502 +1266,252 @@ def start_raylet_monitor(redis_address,
all_processes[PROCESS_TYPE_MONITOR].append(p)
def start_ray_processes(address_info=None,
object_manager_ports=None,
node_manager_ports=None,
node_ip_address="127.0.0.1",
redis_port=None,
redis_shard_ports=None,
num_workers=None,
num_local_schedulers=1,
object_store_memory=None,
redis_max_memory=None,
collect_profiling_data=True,
num_redis_shards=1,
redis_max_clients=None,
redis_password=None,
worker_path=None,
cleanup=True,
redirect_worker_output=False,
redirect_output=False,
include_log_monitor=False,
include_webui=False,
start_workers_from_local_scheduler=True,
resources=None,
plasma_directory=None,
huge_pages=False,
autoscaling_config=None,
plasma_store_socket_name=None,
raylet_socket_name=None,
temp_dir=None,
_internal_config=None):
def start_ray_processes(ray_params, cleanup=True):
"""Helper method to start Ray processes.
Args:
address_info (dict): A dictionary with address information for
processes that have already been started. If provided, address_info
will be modified to include processes that are newly started.
object_manager_ports (list): A list of the ports to use for the object
managers. There should be one per object manager being started on
this node (typically just one).
node_manager_ports (list): A list of the ports to use for the node
managers. There should be one per node manager being started on
this node (typically just one).
node_ip_address (str): The IP address of this node.
redis_port (int): The port that the primary Redis shard should listen
to. If None, then a random port will be chosen. If the key
"redis_address" is in address_info, then this argument will be
ignored.
redis_shard_ports: A list of the ports to use for the non-primary Redis
shards.
num_workers (int): The number of workers to start.
num_local_schedulers (int): The total number of local schedulers
required. This is also the total number of object stores required.
This method will start new instances of local schedulers and object
stores until there are num_local_schedulers existing instances of
each, including ones already registered with the given
address_info.
object_store_memory: The amount of memory (in bytes) to start the
object store with.
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. This only applies to the
sharded redis tables (task and object tables).
collect_profiling_data: Whether to collect profiling data. Note that
profiling data cannot be LRU evicted, so if you set
redis_max_memory then profiling will also be disabled to prevent
it from consuming all available redis memory.
num_redis_shards: The number of Redis shards to start in addition to
the primary Redis shard.
redis_max_clients: If provided, attempt to configure Redis with this
maxclients number.
redis_password (str): Prevents external clients without the password
from connecting to Redis if provided.
worker_path (str): The path of the source code that will be run by the
worker.
ray_params (ray.params.RayParams): The RayParams instance. The
following parameters will be set to default values if it's None:
node_ip_address("127.0.0.1"), num_local_schedulers(1),
include_webui(False), worker_path(path of default_worker.py),
include_log_monitor(False)
cleanup (bool): If cleanup is true, then the processes started here
will be killed by services.cleanup() when the Python process that
called this method exits.
redirect_worker_output: True if the stdout and stderr of worker
processes should be redirected to files.
redirect_output (bool): True if stdout and stderr for non-worker
processes should be redirected to files and false otherwise.
include_log_monitor (bool): If True, then start a log monitor to
monitor the log files for all processes on this node and push their
contents to Redis.
include_webui (bool): If True, then attempt to start the web UI. Note
that this is only possible with Python 3.
start_workers_from_local_scheduler (bool): If this flag is True, then
start the initial workers from the local scheduler. Else, start
them from Python.
resources: A dictionary mapping resource name to the quantity of that
resource.
plasma_directory: A directory where the Plasma memory mapped files will
be created.
huge_pages: Boolean flag indicating whether to start the Object
Store with hugetlbfs support. Requires plasma_directory.
autoscaling_config: path to autoscaling config file.
plasma_store_socket_name (str): If provided, it will specify the socket
name used by the plasma store.
raylet_socket_name (str): If provided, it will specify the socket path
used by the raylet process.
temp_dir (str): If provided, it will specify the root temporary
directory for the Ray process.
_internal_config (str): JSON configuration for overriding
RayConfig defaults. For testing purposes ONLY.
Returns:
A dictionary of the address information for the processes that were
started.
"""
set_temp_root(temp_dir)
set_temp_root(ray_params.temp_dir)
logger.info("Process STDOUT and STDERR is being redirected to {}.".format(
get_logs_dir_path()))
config = json.loads(_internal_config) if _internal_config else None
config = json.loads(
ray_params._internal_config) if ray_params._internal_config else None
if resources is None:
resources = {}
if not isinstance(resources, list):
resources = num_local_schedulers * [resources]
ray_params.update_if_absent(
include_log_monitor=False,
resources={},
num_local_schedulers=1,
include_webui=False,
node_ip_address="127.0.0.1")
if not isinstance(ray_params.resources, list):
ray_params.resources = ray_params.num_local_schedulers * [
ray_params.resources
]
if num_workers is not None:
if ray_params.num_workers is not None:
raise Exception("The 'num_workers' argument is deprecated. Please use "
"'num_cpus' instead.")
else:
workers_per_local_scheduler = []
for resource_dict in resources:
for resource_dict in ray_params.resources:
cpus = resource_dict.get("CPU")
workers_per_local_scheduler.append(cpus if cpus is not None else
multiprocessing.cpu_count())
if address_info is None:
address_info = {}
address_info["node_ip_address"] = node_ip_address
if worker_path is None:
worker_path = os.path.join(
ray_params.update_if_absent(
address_info={},
worker_path=os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"workers/default_worker.py")
"workers/default_worker.py"))
ray_params.address_info["node_ip_address"] = ray_params.node_ip_address
# Start Redis if there isn't already an instance running. TODO(rkn): We are
# suppressing the output of Redis because on Linux it prints a bunch of
# warning messages when it starts up. Instead of suppressing the output, we
# should address the warnings.
redis_address = address_info.get("redis_address")
redis_shards = address_info.get("redis_shards", [])
if redis_address is None:
redis_address, redis_shards = start_redis(
node_ip_address,
port=redis_port,
redis_shard_ports=redis_shard_ports,
num_redis_shards=num_redis_shards,
redis_max_clients=redis_max_clients,
ray_params.redis_address = ray_params.address_info.get("redis_address")
ray_params.redis_shards = ray_params.address_info.get("redis_shards", [])
if ray_params.redis_address is None:
ray_params.redis_address, ray_params.redis_shards = start_redis(
ray_params.node_ip_address,
port=ray_params.redis_port,
redis_shard_ports=ray_params.redis_shard_ports,
num_redis_shards=ray_params.num_redis_shards,
redis_max_clients=ray_params.redis_max_clients,
redirect_output=True,
redirect_worker_output=redirect_worker_output,
redirect_worker_output=ray_params.redirect_worker_output,
cleanup=cleanup,
password=redis_password,
redis_max_memory=redis_max_memory)
address_info["redis_address"] = redis_address
password=ray_params.redis_password,
redis_max_memory=ray_params.redis_max_memory)
ray_params.address_info["redis_address"] = ray_params.redis_address
time.sleep(0.1)
# Start monitoring the processes.
monitor_stdout_file, monitor_stderr_file = new_monitor_log_file(
redirect_output)
ray_params.redirect_output)
start_monitor(
redis_address,
node_ip_address,
ray_params.redis_address,
ray_params.node_ip_address,
stdout_file=monitor_stdout_file,
stderr_file=monitor_stderr_file,
cleanup=cleanup,
autoscaling_config=autoscaling_config,
redis_password=redis_password)
autoscaling_config=ray_params.autoscaling_config,
redis_password=ray_params.redis_password)
start_raylet_monitor(
redis_address,
ray_params.redis_address,
stdout_file=monitor_stdout_file,
stderr_file=monitor_stderr_file,
cleanup=cleanup,
redis_password=redis_password,
redis_password=ray_params.redis_password,
config=config)
if redis_shards == []:
if ray_params.redis_shards == []:
# Get redis shards from primary redis instance.
redis_ip_address, redis_port = redis_address.split(":")
redis_ip_address, redis_port = ray_params.redis_address.split(":")
redis_client = redis.StrictRedis(
host=redis_ip_address, port=redis_port, password=redis_password)
host=redis_ip_address,
port=redis_port,
password=ray_params.redis_password)
redis_shards = redis_client.lrange("RedisShards", start=0, end=-1)
redis_shards = [ray.utils.decode(shard) for shard in redis_shards]
address_info["redis_shards"] = redis_shards
ray_params.redis_shards = [
ray.utils.decode(shard) for shard in redis_shards
]
ray_params.address_info["redis_shards"] = ray_params.redis_shards
# Start the log monitor, if necessary.
if include_log_monitor:
if ray_params.include_log_monitor:
log_monitor_stdout_file, log_monitor_stderr_file = (
new_log_monitor_log_file())
start_log_monitor(
redis_address,
node_ip_address,
ray_params.redis_address,
ray_params.node_ip_address,
stdout_file=log_monitor_stdout_file,
stderr_file=log_monitor_stderr_file,
cleanup=cleanup,
redis_password=redis_password)
redis_password=ray_params.redis_password)
# Initialize with existing services.
if "object_store_addresses" not in address_info:
address_info["object_store_addresses"] = []
object_store_addresses = address_info["object_store_addresses"]
if "raylet_socket_names" not in address_info:
address_info["raylet_socket_names"] = []
raylet_socket_names = address_info["raylet_socket_names"]
if "object_store_addresses" not in ray_params.address_info:
ray_params.address_info["object_store_addresses"] = []
object_store_addresses = ray_params.address_info["object_store_addresses"]
if "raylet_socket_names" not in ray_params.address_info:
ray_params.address_info["raylet_socket_names"] = []
raylet_socket_names = ray_params.address_info["raylet_socket_names"]
# Get the ports to use for the object managers if any are provided.
if not isinstance(object_manager_ports, list):
assert object_manager_ports is None or num_local_schedulers == 1
object_manager_ports = num_local_schedulers * [object_manager_ports]
assert len(object_manager_ports) == num_local_schedulers
if not isinstance(node_manager_ports, list):
assert node_manager_ports is None or num_local_schedulers == 1
node_manager_ports = num_local_schedulers * [node_manager_ports]
assert len(node_manager_ports) == num_local_schedulers
if not isinstance(ray_params.object_manager_ports, list):
assert (ray_params.object_manager_ports is None
or ray_params.num_local_schedulers == 1)
ray_params.object_manager_ports = (ray_params.num_local_schedulers *
[ray_params.object_manager_ports])
assert len(
ray_params.object_manager_ports) == ray_params.num_local_schedulers
if not isinstance(ray_params.node_manager_ports, list):
assert (ray_params.node_manager_ports is None
or ray_params.num_local_schedulers == 1)
ray_params.node_manager_ports = (
ray_params.num_local_schedulers * [ray_params.node_manager_ports])
assert len(
ray_params.node_manager_ports) == ray_params.num_local_schedulers
# Start any object stores that do not yet exist.
for i in range(num_local_schedulers - len(object_store_addresses)):
for i in range(ray_params.num_local_schedulers -
len(object_store_addresses)):
# Start Plasma.
plasma_store_stdout_file, plasma_store_stderr_file = (
new_plasma_store_log_file(i, redirect_output))
new_plasma_store_log_file(i, ray_params.redirect_output))
object_store_address = start_plasma_store(
node_ip_address,
redis_address,
ray_params.node_ip_address,
ray_params.redis_address,
store_stdout_file=plasma_store_stdout_file,
store_stderr_file=plasma_store_stderr_file,
object_store_memory=object_store_memory,
object_store_memory=ray_params.object_store_memory,
cleanup=cleanup,
plasma_directory=plasma_directory,
huge_pages=huge_pages,
plasma_store_socket_name=plasma_store_socket_name,
redis_password=redis_password)
plasma_directory=ray_params.plasma_directory,
huge_pages=ray_params.huge_pages,
plasma_store_socket_name=ray_params.plasma_store_socket_name,
redis_password=ray_params.redis_password)
object_store_addresses.append(object_store_address)
time.sleep(0.1)
# Start any raylets that do not exist yet.
for i in range(len(raylet_socket_names), num_local_schedulers):
for raylet_index in range(
len(raylet_socket_names), ray_params.num_local_schedulers):
raylet_stdout_file, raylet_stderr_file = new_raylet_log_file(
i, redirect_output=redirect_worker_output)
address_info["raylet_socket_names"].append(
raylet_index, redirect_output=ray_params.redirect_worker_output)
ray_params.address_info["raylet_socket_names"].append(
start_raylet(
redis_address,
node_ip_address,
raylet_socket_name or get_raylet_socket_name(),
object_store_addresses[i],
worker_path,
object_manager_port=object_manager_ports[i],
node_manager_port=node_manager_ports[i],
resources=resources[i],
num_workers=workers_per_local_scheduler[i],
ray_params,
raylet_index,
ray_params.raylet_socket_name or get_raylet_socket_name(),
object_store_addresses[raylet_index],
num_workers=workers_per_local_scheduler[raylet_index],
stdout_file=raylet_stdout_file,
stderr_file=raylet_stderr_file,
cleanup=cleanup,
redis_password=redis_password,
collect_profiling_data=collect_profiling_data,
config=config))
# Try to start the web UI.
if include_webui:
if ray_params.include_webui:
ui_stdout_file, ui_stderr_file = new_webui_log_file()
address_info["webui_url"] = start_ui(
redis_address,
ray_params.address_info["webui_url"] = start_ui(
ray_params.redis_address,
stdout_file=ui_stdout_file,
stderr_file=ui_stderr_file,
cleanup=cleanup)
else:
address_info["webui_url"] = ""
ray_params.address_info["webui_url"] = ""
# Return the addresses of the relevant processes.
return address_info
return ray_params.address_info
def start_ray_node(node_ip_address,
redis_address,
object_manager_ports=None,
node_manager_ports=None,
num_workers=None,
num_local_schedulers=1,
object_store_memory=None,
redis_password=None,
worker_path=None,
cleanup=True,
redirect_worker_output=False,
redirect_output=False,
resources=None,
plasma_directory=None,
huge_pages=False,
plasma_store_socket_name=None,
raylet_socket_name=None,
temp_dir=None,
_internal_config=None):
def start_ray_node(ray_params, cleanup=True):
"""Start the Ray processes for a single node.
This assumes that the Ray processes on some master node have already been
started.
Args:
node_ip_address (str): The IP address of this node.
redis_address (str): The address of the Redis server.
object_manager_ports (list): A list of the ports to use for the object
managers. There should be one per object manager being started on
this node (typically just one).
node_manager_ports (list): A list of the ports to use for the node
managers. There should be one per node manager being started on
this node (typically just one).
num_workers (int): The number of workers to start.
num_local_schedulers (int): The number of local schedulers to start.
This is also the number of plasma stores and raylets to start.
object_store_memory (int): The maximum amount of memory (in bytes) to
let the plasma store use.
redis_password (str): Prevents external clients without the password
from connecting to Redis if provided.
worker_path (str): The path of the source code that will be run by the
worker.
ray_params (ray.params.RayParams): The RayParams instance. The
following parameters could be checked: node_ip_address,
redis_address, object_manager_ports, node_manager_ports,
num_workers, num_local_schedulers, object_store_memory,
redis_password, worker_path, cleanup, redirect_worker_output,
redirect_output, resources, plasma_directory, huge_pages,
plasma_store_socket_name, raylet_socket_name, temp_dir,
_internal_config
cleanup (bool): If cleanup is true, then the processes started here
will be killed by services.cleanup() when the Python process that
called this method exits.
redirect_worker_output: True if the stdout and stderr of worker
processes should be redirected to files.
redirect_output (bool): True if stdout and stderr for non-worker
processes should be redirected to files and false otherwise.
resources: A dictionary mapping resource name to the available quantity
of that resource.
plasma_directory: A directory where the Plasma memory mapped files will
be created.
huge_pages: Boolean flag indicating whether to start the Object
Store with hugetlbfs support. Requires plasma_directory.
plasma_store_socket_name (str): If provided, it will specify the socket
name used by the plasma store.
raylet_socket_name (str): If provided, it will specify the socket path
used by the raylet process.
temp_dir (str): If provided, it will specify the root temporary
directory for the Ray process.
_internal_config (str): JSON configuration for overriding
RayConfig defaults. For testing purposes ONLY.
Returns:
A dictionary of the address information for the processes that were
started.
"""
address_info = {
"redis_address": redis_address,
ray_params.address_info = {
"redis_address": ray_params.redis_address,
}
return start_ray_processes(
address_info=address_info,
object_manager_ports=object_manager_ports,
node_manager_ports=node_manager_ports,
node_ip_address=node_ip_address,
num_workers=num_workers,
num_local_schedulers=num_local_schedulers,
object_store_memory=object_store_memory,
redis_password=redis_password,
worker_path=worker_path,
include_log_monitor=True,
cleanup=cleanup,
redirect_worker_output=redirect_worker_output,
redirect_output=redirect_output,
resources=resources,
plasma_directory=plasma_directory,
huge_pages=huge_pages,
plasma_store_socket_name=plasma_store_socket_name,
raylet_socket_name=raylet_socket_name,
temp_dir=temp_dir,
_internal_config=_internal_config)
ray_params.update(include_log_monitor=True)
return start_ray_processes(ray_params, cleanup=cleanup)
def start_ray_head(address_info=None,
object_manager_ports=None,
node_manager_ports=None,
node_ip_address="127.0.0.1",
redis_port=None,
redis_shard_ports=None,
num_workers=None,
num_local_schedulers=1,
object_store_memory=None,
redis_max_memory=None,
collect_profiling_data=True,
worker_path=None,
cleanup=True,
redirect_worker_output=False,
redirect_output=False,
start_workers_from_local_scheduler=True,
resources=None,
num_redis_shards=None,
redis_max_clients=None,
redis_password=None,
include_webui=True,
plasma_directory=None,
huge_pages=False,
autoscaling_config=None,
plasma_store_socket_name=None,
raylet_socket_name=None,
temp_dir=None,
_internal_config=None):
def start_ray_head(ray_params, cleanup=True):
"""Start Ray in local mode.
Args:
address_info (dict): A dictionary with address information for
processes that have already been started. If provided, address_info
will be modified to include processes that are newly started.
object_manager_ports (list): A list of the ports to use for the object
managers. There should be one per object manager being started on
this node (typically just one).
node_manager_ports (list): A list of the ports to use for the node
managers. There should be one per node manager being started on
this node (typically just one).
node_ip_address (str): The IP address of this node.
redis_port (int): The port that the primary Redis shard should listen
to. If None, then a random port will be chosen. If the key
"redis_address" is in address_info, then this argument will be
ignored.
redis_shard_ports: A list of the ports to use for the non-primary Redis
shards.
num_workers (int): The number of workers to start.
num_local_schedulers (int): The total number of local schedulers
required. This is also the total number of object stores required.
This method will start new instances of local schedulers and object
stores until there are at least num_local_schedulers existing
instances of each, including ones already registered with the given
address_info.
object_store_memory: The amount of memory (in bytes) to start the
object store with.
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. This only applies to the
sharded redis tables (task and object tables).
collect_profiling_data: Whether to collect profiling data from workers.
worker_path (str): The path of the source code that will be run by the
worker.
ray_params (ray.params.RayParams): The RayParams instance. The
following parameters could be checked: address_info,
object_manager_ports, node_manager_ports, node_ip_address,
redis_port, redis_shard_ports, num_workers, num_local_schedulers,
object_store_memory, redis_max_memory, collect_profiling_data,
worker_path, cleanup, redirect_worker_output, redirect_output,
start_workers_from_local_scheduler, resources, num_redis_shards,
redis_max_clients, redis_password, include_webui, huge_pages,
plasma_directory, autoscaling_config, plasma_store_socket_name,
raylet_socket_name, temp_dir, _internal_config
cleanup (bool): If cleanup is true, then the processes started here
will be killed by services.cleanup() when the Python process that
called this method exits.
redirect_worker_output: True if the stdout and stderr of worker
processes should be redirected to files.
redirect_output (bool): True if stdout and stderr for non-worker
processes should be redirected to files and false otherwise.
start_workers_from_local_scheduler (bool): If this flag is True, then
start the initial workers from the local scheduler. Else, start
them from Python.
resources: A dictionary mapping resource name to the available quantity
of that resource.
num_redis_shards: The number of Redis shards to start in addition to
the primary Redis shard.
redis_max_clients: If provided, attempt to configure Redis with this
maxclients number.
redis_password (str): Prevents external clients without the password
from connecting to Redis if provided.
include_webui: True if the UI should be started and false otherwise.
plasma_directory: A directory where the Plasma memory mapped files will
be created.
huge_pages: Boolean flag indicating whether to start the Object
Store with hugetlbfs support. Requires plasma_directory.
autoscaling_config: path to autoscaling config file.
plasma_store_socket_name (str): If provided, it will specify the socket
name used by the plasma store.
raylet_socket_name (str): If provided, it will specify the socket path
used by the raylet process.
temp_dir (str): If provided, it will specify the root temporary
directory for the Ray process.
_internal_config (str): JSON configuration for overriding
RayConfig defaults. For testing purposes ONLY.
Returns:
A dictionary of the address information for the processes that were
started.
"""
num_redis_shards = 1 if num_redis_shards is None else num_redis_shards
return start_ray_processes(
address_info=address_info,
object_manager_ports=object_manager_ports,
node_manager_ports=node_manager_ports,
node_ip_address=node_ip_address,
redis_port=redis_port,
redis_shard_ports=redis_shard_ports,
num_workers=num_workers,
num_local_schedulers=num_local_schedulers,
object_store_memory=object_store_memory,
redis_max_memory=redis_max_memory,
collect_profiling_data=collect_profiling_data,
worker_path=worker_path,
cleanup=cleanup,
redirect_worker_output=redirect_worker_output,
redirect_output=redirect_output,
include_log_monitor=True,
include_webui=include_webui,
start_workers_from_local_scheduler=start_workers_from_local_scheduler,
resources=resources,
num_redis_shards=num_redis_shards,
redis_max_clients=redis_max_clients,
redis_password=redis_password,
plasma_directory=plasma_directory,
huge_pages=huge_pages,
autoscaling_config=autoscaling_config,
plasma_store_socket_name=plasma_store_socket_name,
raylet_socket_name=raylet_socket_name,
temp_dir=temp_dir,
_internal_config=_internal_config)
ray_params.update_if_absent(num_redis_shards=1, include_webui=True)
ray_params.update(include_log_monitor=True)
return start_ray_processes(ray_params, cleanup=cleanup)
python/ray/test/cluster_utils.py
+7 -8
View File
@@ -7,6 +7,7 @@ import logging
import time
import ray
from ray.parameter import RayParams
import ray.services as services
logger = logging.getLogger(__name__)
@@ -73,19 +74,18 @@ class Cluster(object):
Node object of the added Ray node.
"""
node_kwargs = {
"cleanup": True,
"resources": {
"CPU": 1
},
"object_store_memory": 100 * (2**20) # 100 MB
}
node_kwargs.update(override_kwargs)
ray_params = RayParams(
node_ip_address=services.get_node_ip_address(), **node_kwargs)
if self.head_node is None:
address_info = services.start_ray_head(
node_ip_address=services.get_node_ip_address(),
include_webui=False,
**node_kwargs)
ray_params.update(include_webui=False)
address_info = services.start_ray_head(ray_params, cleanup=True)
self.redis_address = address_info["redis_address"]
# TODO(rliaw): Find a more stable way than modifying global state.
process_dict_copy = services.all_processes.copy()
@@ -94,9 +94,8 @@ class Cluster(object):
node = Node(address_info, process_dict_copy)
self.head_node = node
else:
address_info = services.start_ray_node(
services.get_node_ip_address(), self.redis_address,
**node_kwargs)
ray_params.update(redis_address=self.redis_address)
address_info = services.start_ray_node(ray_params, cleanup=True)
# TODO(rliaw): Find a more stable way than modifying global state.
process_dict_copy = services.all_processes.copy()
for key in services.all_processes:
python/ray/worker.py
+85 -180
View File
@@ -37,6 +37,7 @@ import ray.ray_constants as ray_constants
from ray import import_thread
from ray import profiling
from ray.function_manager import (FunctionActorManager, FunctionDescriptor)
from ray.parameter import RayParams
from ray.utils import (
check_oversized_pickle,
is_cython,
@@ -1269,33 +1270,7 @@ def _normalize_resource_arguments(num_cpus, num_gpus, resources,
return new_resources
def _init(address_info=None,
start_ray_local=False,
object_id_seed=None,
num_workers=None,
num_local_schedulers=None,
object_store_memory=None,
redis_max_memory=None,
collect_profiling_data=True,
local_mode=False,
driver_mode=None,
redirect_worker_output=False,
redirect_output=True,
start_workers_from_local_scheduler=True,
num_cpus=None,
num_gpus=None,
resources=None,
num_redis_shards=None,
redis_max_clients=None,
redis_password=None,
plasma_directory=None,
huge_pages=False,
include_webui=True,
driver_id=None,
plasma_store_socket_name=None,
raylet_socket_name=None,
temp_dir=None,
_internal_config=None):
def _init(ray_params, driver_id=None):
"""Helper method to connect to an existing Ray cluster or start a new one.
This method handles two cases. Either a Ray cluster already exists and we
@@ -1303,67 +1278,17 @@ def _init(address_info=None,
with a Ray cluster and attach to the newly started cluster.
Args:
address_info (dict): A dictionary with address information for
processes in a partially-started Ray cluster. If
start_ray_local=True, any processes not in this dictionary will be
started. If provided, an updated address_info dictionary will be
returned to include processes that are newly started.
start_ray_local (bool): If True then this will start any processes not
already in address_info, including Redis, a global scheduler, local
scheduler(s), object store(s), and worker(s). It will also kill
these processes when Python exits. If False, this will attach to an
existing Ray cluster.
object_id_seed (int): Used to seed the deterministic generation of
object IDs. The same value can be used across multiple runs of the
same job in order to generate the object IDs in a consistent
manner. However, the same ID should not be used for different jobs.
num_local_schedulers (int): The number of local schedulers to start.
This is only provided if start_ray_local is True.
object_store_memory: The maximum amount of memory (in bytes) to
allow the object store to use.
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. This only applies to the
sharded redis tables (task and object tables).
collect_profiling_data: Whether to collect profiling data from workers.
local_mode (bool): True if the code should be executed serially
without Ray. This is useful for debugging.
redirect_worker_output: True if the stdout and stderr of worker
processes should be redirected to files.
redirect_output (bool): True if stdout and stderr for non-worker
processes should be redirected to files and false otherwise.
start_workers_from_local_scheduler (bool): If this flag is True, then
start the initial workers from the local scheduler. Else, start
them from Python. The latter case is for debugging purposes only.
num_cpus (int): Number of cpus the user wishes all local schedulers to
be configured with.
num_gpus (int): Number of gpus the user wishes all local schedulers to
be configured with. If unspecified, Ray will attempt to autodetect
the number of GPUs available on the node (note that autodetection
currently only works for Nvidia GPUs).
resources: A dictionary mapping resource names to the quantity of that
resource available.
num_redis_shards: The number of Redis shards to start in addition to
the primary Redis shard.
redis_max_clients: If provided, attempt to configure Redis with this
maxclients number.
redis_password (str): Prevents external clients without the password
from connecting to Redis if provided.
plasma_directory: A directory where the Plasma memory mapped files will
be created.
huge_pages: Boolean flag indicating whether to start the Object
Store with hugetlbfs support. Requires plasma_directory.
include_webui: Boolean flag indicating whether to start the web
UI, which is a Jupyter notebook.
ray_params (ray.params.RayParams): The RayParams instance. The
following parameters could be checked: address_info,
start_ray_local, object_id_seed, num_workers,
num_local_schedulers, object_store_memory, redis_max_memory,
collect_profiling_data, local_mode, redirect_worker_output,
driver_mode, redirect_output, start_workers_from_local_scheduler,
num_cpus, num_gpus, resources, num_redis_shards,
redis_max_clients, redis_password, plasma_directory, huge_pages,
include_webui, driver_id, plasma_store_socket_name, temp_dir,
raylet_socket_name, _internal_config
driver_id: The ID of driver.
plasma_store_socket_name (str): If provided, it will specify the socket
name used by the plasma store.
raylet_socket_name (str): If provided, it will specify the socket path
used by the raylet process.
temp_dir (str): If provided, it will specify the root temporary
directory for the Ray process.
_internal_config (str): JSON configuration for overriding
RayConfig defaults. For testing purposes ONLY.
Returns:
Address information about the started processes.
@@ -1372,157 +1297,137 @@ def _init(address_info=None,
Exception: An exception is raised if an inappropriate combination of
arguments is passed in.
"""
if driver_mode is not None:
if ray_params.driver_mode is not None:
raise Exception("The 'driver_mode' argument has been deprecated. "
"To run Ray in local mode, pass in local_mode=True.")
if local_mode:
driver_mode = LOCAL_MODE
if ray_params.local_mode:
ray_params.driver_mode = LOCAL_MODE
else:
driver_mode = SCRIPT_MODE
ray_params.driver_mode = SCRIPT_MODE
if redis_max_memory and collect_profiling_data:
if ray_params.redis_max_memory and ray_params.collect_profiling_data:
logger.warning(
"Profiling data cannot be LRU evicted, so it is disabled "
"when redis_max_memory is set.")
collect_profiling_data = False
ray_params.collect_profiling_data = False
# Get addresses of existing services.
if address_info is None:
address_info = {}
if ray_params.address_info is None:
ray_params.address_info = {}
else:
assert isinstance(address_info, dict)
node_ip_address = address_info.get("node_ip_address")
redis_address = address_info.get("redis_address")
assert isinstance(ray_params.address_info, dict)
ray_params.node_ip_address = ray_params.address_info.get("node_ip_address")
ray_params.redis_address = ray_params.address_info.get("redis_address")
# Start any services that do not yet exist.
if driver_mode == LOCAL_MODE:
if ray_params.driver_mode == LOCAL_MODE:
# If starting Ray in LOCAL_MODE, don't start any other processes.
pass
elif start_ray_local:
elif ray_params.start_ray_local:
# In this case, we launch a scheduler, a new object store, and some
# workers, and we connect to them. We do not launch any processes that
# are already registered in address_info.
if node_ip_address is None:
node_ip_address = ray.services.get_node_ip_address()
ray_params.update_if_absent(
node_ip_address=ray.services.get_node_ip_address())
# Use 1 local scheduler if num_local_schedulers is not provided. If
# existing local schedulers are provided, use that count as
# num_local_schedulers.
if num_local_schedulers is None:
num_local_schedulers = 1
ray_params.update_if_absent(num_local_schedulers=1)
# Use 1 additional redis shard if num_redis_shards is not provided.
num_redis_shards = 1 if num_redis_shards is None else num_redis_shards
ray_params.update_if_absent(num_redis_shards=1)
# Stick the CPU and GPU resources into the resource dictionary.
resources = _normalize_resource_arguments(
num_cpus, num_gpus, resources, num_local_schedulers)
ray_params.resources = _normalize_resource_arguments(
ray_params.num_cpus, ray_params.num_gpus, ray_params.resources,
ray_params.num_local_schedulers)
# Start the scheduler, object store, and some workers. These will be
# killed by the call to shutdown(), which happens when the Python
# script exits.
address_info = services.start_ray_head(
address_info=address_info,
node_ip_address=node_ip_address,
num_workers=num_workers,
num_local_schedulers=num_local_schedulers,
object_store_memory=object_store_memory,
redis_max_memory=redis_max_memory,
collect_profiling_data=collect_profiling_data,
redirect_worker_output=redirect_worker_output,
redirect_output=redirect_output,
start_workers_from_local_scheduler=(
start_workers_from_local_scheduler),
resources=resources,
num_redis_shards=num_redis_shards,
redis_max_clients=redis_max_clients,
redis_password=redis_password,
plasma_directory=plasma_directory,
huge_pages=huge_pages,
include_webui=include_webui,
plasma_store_socket_name=plasma_store_socket_name,
raylet_socket_name=raylet_socket_name,
temp_dir=temp_dir,
_internal_config=_internal_config)
ray_params.address_info = services.start_ray_head(ray_params)
else:
if redis_address is None:
if ray_params.redis_address is None:
raise Exception("When connecting to an existing cluster, "
"redis_address must be provided.")
if num_workers is not None:
if ray_params.num_workers is not None:
raise Exception("When connecting to an existing cluster, "
"num_workers must not be provided.")
if num_local_schedulers is not None:
if ray_params.num_local_schedulers is not None:
raise Exception("When connecting to an existing cluster, "
"num_local_schedulers must not be provided.")
if num_cpus is not None or num_gpus is not None:
if ray_params.num_cpus is not None or ray_params.num_gpus is not None:
raise Exception("When connecting to an existing cluster, num_cpus "
"and num_gpus must not be provided.")
if resources is not None:
if ray_params.resources is not None:
raise Exception("When connecting to an existing cluster, "
"resources must not be provided.")
if num_redis_shards is not None:
if ray_params.num_redis_shards is not None:
raise Exception("When connecting to an existing cluster, "
"num_redis_shards must not be provided.")
if redis_max_clients is not None:
if ray_params.redis_max_clients is not None:
raise Exception("When connecting to an existing cluster, "
"redis_max_clients must not be provided.")
if object_store_memory is not None:
if ray_params.object_store_memory is not None:
raise Exception("When connecting to an existing cluster, "
"object_store_memory must not be provided.")
if redis_max_memory is not None:
if ray_params.redis_max_memory is not None:
raise Exception("When connecting to an existing cluster, "
"redis_max_memory must not be provided.")
if plasma_directory is not None:
if ray_params.plasma_directory is not None:
raise Exception("When connecting to an existing cluster, "
"plasma_directory must not be provided.")
if huge_pages:
if ray_params.huge_pages:
raise Exception("When connecting to an existing cluster, "
"huge_pages must not be provided.")
if temp_dir is not None:
if ray_params.temp_dir is not None:
raise Exception("When connecting to an existing cluster, "
"temp_dir must not be provided.")
if plasma_store_socket_name is not None:
if ray_params.plasma_store_socket_name is not None:
raise Exception("When connecting to an existing cluster, "
"plasma_store_socket_name must not be provided.")
if raylet_socket_name is not None:
if ray_params.raylet_socket_name is not None:
raise Exception("When connecting to an existing cluster, "
"raylet_socket_name must not be provided.")
if _internal_config is not None:
if ray_params._internal_config is not None:
raise Exception("When connecting to an existing cluster, "
"_internal_config must not be provided.")
# Get the node IP address if one is not provided.
if node_ip_address is None:
node_ip_address = services.get_node_ip_address(redis_address)
ray_params.update_if_absent(
node_ip_address=services.get_node_ip_address(
ray_params.redis_address))
# Get the address info of the processes to connect to from Redis.
address_info = get_address_info_from_redis(
redis_address, node_ip_address, redis_password=redis_password)
ray_params.address_info = get_address_info_from_redis(
ray_params.redis_address,
ray_params.node_ip_address,
redis_password=ray_params.redis_password)
# Connect this driver to Redis, the object store, and the local scheduler.
# Choose the first object store and local scheduler if there are multiple.
# The corresponding call to disconnect will happen in the call to
# shutdown() when the Python script exits.
if driver_mode == LOCAL_MODE:
if ray_params.driver_mode == LOCAL_MODE:
driver_address_info = {}
else:
driver_address_info = {
"node_ip_address": node_ip_address,
"redis_address": address_info["redis_address"],
"store_socket_name": address_info["object_store_addresses"][0],
"webui_url": address_info["webui_url"],
"node_ip_address": ray_params.node_ip_address,
"redis_address": ray_params.address_info["redis_address"],
"store_socket_name": ray_params.address_info[
"object_store_addresses"][0],
"webui_url": ray_params.address_info["webui_url"],
}
driver_address_info["raylet_socket_name"] = (
address_info["raylet_socket_names"][0])
ray_params.address_info["raylet_socket_names"][0])
# We only pass `temp_dir` to a worker (WORKER_MODE).
# It can't be a worker here.
connect(
ray_params,
driver_address_info,
object_id_seed=object_id_seed,
mode=driver_mode,
mode=ray_params.driver_mode,
worker=global_worker,
driver_id=driver_id,
redis_password=redis_password,
collect_profiling_data=collect_profiling_data)
return address_info
driver_id=driver_id)
return ray_params.address_info
def init(redis_address=None,
@@ -1674,7 +1579,7 @@ def init(redis_address=None,
redis_address = services.address_to_ip(redis_address)
info = {"node_ip_address": node_ip_address, "redis_address": redis_address}
ret = _init(
ray_params = RayParams(
address_info=info,
start_ray_local=(redis_address is None),
num_workers=num_workers,
@@ -1695,11 +1600,12 @@ def init(redis_address=None,
object_store_memory=object_store_memory,
redis_max_memory=redis_max_memory,
collect_profiling_data=collect_profiling_data,
driver_id=driver_id,
plasma_store_socket_name=plasma_store_socket_name,
raylet_socket_name=raylet_socket_name,
temp_dir=temp_dir,
_internal_config=_internal_config)
_internal_config=_internal_config,
)
ret = _init(ray_params, driver_id=driver_id)
for hook in _post_init_hooks:
hook()
return ret
@@ -1900,26 +1806,23 @@ def print_error_messages(worker):
pass
def connect(info,
object_id_seed=None,
def connect(ray_params,
info,
mode=WORKER_MODE,
worker=global_worker,
driver_id=None,
redis_password=None,
collect_profiling_data=True):
driver_id=None):
"""Connect this worker to the local scheduler, to Plasma, and to Redis.
Args:
ray_params (ray.params.RayParams): The RayParams instance. The
following parameters could be checked: object_id_seed,
redis_password, collect_profiling_data
info (dict): A dictionary with address of the Redis server and the
sockets of the plasma store and raylet.
object_id_seed: A seed to use to make the generation of object IDs
deterministic.
mode: The mode of the worker. One of SCRIPT_MODE, WORKER_MODE, and
LOCAL_MODE.
worker: The ray.Worker instance.
driver_id: The ID of driver. If it's None, then we will generate one.
redis_password (str): Prevents external clients without the password
from connecting to Redis if provided.
collect_profiling_data: Whether to collect profiling data from workers.
"""
# Do some basic checking to make sure we didn't call ray.init twice.
error_message = "Perhaps you called ray.init twice by accident?"
@@ -1929,7 +1832,7 @@ def connect(info,
# Enable nice stack traces on SIGSEGV etc.
faulthandler.enable(all_threads=False)
if collect_profiling_data:
if ray_params.collect_profiling_data:
worker.profiler = profiling.Profiler(worker)
else:
worker.profiler = profiling.NoopProfiler()
@@ -1977,7 +1880,7 @@ def connect(info,
redis.StrictRedis(
host=redis_ip_address,
port=int(redis_port),
password=redis_password))
password=ray_params.redis_password))
# For driver's check that the version information matches the version
# information that the Ray cluster was started with.
@@ -2015,11 +1918,13 @@ def connect(info,
services.record_log_files_in_redis(
info["redis_address"],
info["node_ip_address"], [log_stdout_file, log_stderr_file],
password=redis_password)
password=ray_params.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)
redis_ip_address,
int(redis_port),
redis_password=ray_params.redis_password)
# Register the worker with Redis.
if mode == SCRIPT_MODE:
@@ -2068,8 +1973,8 @@ def connect(info,
# the user's random number generator). Otherwise, set the current task
# ID randomly to avoid object ID collisions.
numpy_state = np.random.get_state()
if object_id_seed is not None:
np.random.seed(object_id_seed)
if ray_params.object_id_seed is not None:
np.random.seed(ray_params.object_id_seed)
else:
# Try to use true randomness.
np.random.seed(None)
python/ray/workers/default_worker.py
+9 -10
View File
@@ -8,6 +8,7 @@ import traceback
import ray
import ray.actor
from ray.parameter import RayParams
import ray.ray_constants as ray_constants
import ray.tempfile_services as tempfile_services
@@ -35,11 +36,6 @@ parser.add_argument(
required=True,
type=str,
help="the object store's name")
parser.add_argument(
"--object-store-manager-name",
required=False,
type=str,
help="the object store manager's name")
parser.add_argument(
"--raylet-name", required=False, type=str, help="the raylet's name")
parser.add_argument(
@@ -75,7 +71,6 @@ if __name__ == "__main__":
"redis_address": args.redis_address,
"redis_password": args.redis_password,
"store_socket_name": args.object_store_name,
"manager_socket_name": args.object_store_manager_name,
"raylet_socket_name": args.raylet_name,
}
@@ -86,11 +81,15 @@ if __name__ == "__main__":
# Override the temporary directory.
tempfile_services.set_temp_root(args.temp_dir)
ray.worker.connect(
info,
mode=ray.WORKER_MODE,
ray_params = RayParams(
node_ip_address=args.node_ip_address,
redis_address=args.redis_address,
redis_password=args.redis_password,
collect_profiling_data=args.collect_profiling_data)
plasma_store_socket_name=args.object_store_name,
raylet_socket_name=args.raylet_name,
temp_dir=args.temp_dir)
ray.worker.connect(ray_params, info, mode=ray.WORKER_MODE)
error_explanation = """
This error is unexpected and should not have happened. Somehow a worker