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

Factor out code for starting new processes and test plasma store in valgrind. (#3824)

Browse Source 
* Factor out starting Ray processes.

* Detect flags through environment variables.

* Return ProcessInfo from start_ray_process.

* Print valgrind errors at exit.

* Test valgrind in travis.

* Some valgrind fixes.

* Undo raylet monitor change.

* Only test plasma store in valgrind.
This commit is contained in:
Robert Nishihara
2019-01-22 14:59:11 -08:00
committed by Philipp Moritz
parent f0e6523323
commit 0b1608a546
11 changed files with 352 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/services.py
+249 -58
View File
@@ -2,6 +2,7 @@ from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import collections
import json
import logging
import multiprocessing
@@ -16,8 +17,8 @@ import redis
import pyarrow
# Ray modules
import ray
import ray.ray_constants as ray_constants
import ray.plasma
from ray.tempfile_services import (get_ipython_notebook_path, get_temp_root,
new_redis_log_file)
@@ -58,6 +59,11 @@ RAYLET_EXECUTABLE = os.path.join(
# using logging.basicConfig in its entry/init points.
logger = logging.getLogger(__name__)
ProcessInfo = collections.namedtuple("ProcessInfo", [
"process", "stdout_file", "stderr_file", "use_valgrind", "use_gdb",
"use_valgrind_profiler", "use_perftools_profiler", "use_tmux"
])
def address(ip_address, port):
return ip_address + ":" + str(port)
@@ -198,6 +204,118 @@ def create_redis_client(redis_address, password=None):
host=redis_ip_address, port=int(redis_port), password=password)
def start_ray_process(command,
process_type,
env_updates=None,
cwd=None,
use_valgrind=False,
use_gdb=False,
use_valgrind_profiler=False,
use_perftools_profiler=False,
use_tmux=False,
stdout_file=None,
stderr_file=None):
"""Start one of the Ray processes.
TODO(rkn): We need to figure out how these commands interact. For example,
it may only make sense to start a process in gdb if we also start it in
tmux. Similarly, certain combinations probably don't make sense, like
simultaneously running the process in valgrind and the profiler.
Args:
command (List[str]): The command to use to start the Ray process.
process_type (str): The type of the process that is being started
(e.g., "raylet").
env_updates (dict): A dictionary of additional environment variables to
run the command with (in addition to the caller's environment
variables).
cwd (str): The directory to run the process in.
use_valgrind (bool): True if we should start the process in valgrind.
use_gdb (bool): True if we should start the process in gdb.
use_valgrind_profiler (bool): True if we should start the process in
the valgrind profiler.
use_perftools_profiler (bool): True if we should profile the process
using perftools.
use_tmux (bool): True if we should start the process in tmux.
stdout_file: A file handle opened for writing to redirect stdout to. If
no redirection should happen, then this should be None.
stderr_file: A file handle opened for writing to redirect stderr to. If
no redirection should happen, then this should be None.
Returns:
Inormation about the process that was started including a handle to the
process that was started.
"""
# Detect which flags are set through environment variables.
valgrind_env_var = "RAY_{}_VALGRIND".format(process_type.upper())
if os.environ.get(valgrind_env_var) == "1":
logger.info("Detected environment variable '%s'.", valgrind_env_var)
use_valgrind = True
valgrind_profiler_env_var = "RAY_{}_VALGRIND_PROFILER".format(
process_type.upper())
if os.environ.get(valgrind_profiler_env_var) == "1":
logger.info("Detected environment variable '%s'.",
valgrind_profiler_env_var)
use_valgrind_profiler = True
perftools_profiler_env_var = "RAY_{}_PERFTOOLS_PROFILER".format(
process_type.upper())
if os.environ.get(perftools_profiler_env_var) == "1":
logger.info("Detected environment variable '%s'.",
perftools_profiler_env_var)
use_perftools_profiler = True
tmux_env_var = "RAY_{}_TMUX".format(process_type.upper())
if os.environ.get(tmux_env_var) == "1":
logger.info("Detected environment variable '%s'.", tmux_env_var)
use_tmux = True
if use_gdb:
raise NotImplementedError
if use_tmux:
raise NotImplementedError
if sum([use_valgrind, use_valgrind_profiler, use_perftools_profiler]) > 1:
raise ValueError(
"At most one of the 'use_valgrind', 'use_valgrind_profiler', and "
"'use_perftools_profiler' flags can be used at a time.")
if env_updates is None:
env_updates = {}
if not isinstance(env_updates, dict):
raise ValueError("The 'env_updates' argument must be a dictionary.")
modified_env = os.environ.copy()
modified_env.update(env_updates)
if use_valgrind:
command = [
"valgrind", "--track-origins=yes", "--leak-check=full",
"--show-leak-kinds=all", "--leak-check-heuristics=stdstring",
"--error-exitcode=1"
] + command
if use_valgrind_profiler:
command = ["valgrind", "--tool=callgrind"] + command
if use_perftools_profiler:
modified_env["LD_PRELOAD"] = os.environ["PERFTOOLS_PATH"]
modified_env["CPUPROFILE"] = os.environ["PERFTOOLS_LOGFILE"]
process = subprocess.Popen(
command,
env=modified_env,
cwd=cwd,
stdout=stdout_file,
stderr=stderr_file)
return ProcessInfo(
process=process,
stdout_file=stdout_file.name if stdout_file is not None else None,
stderr_file=stderr_file.name if stderr_file is not None else None,
use_valgrind=use_valgrind,
use_gdb=use_gdb,
use_valgrind_profiler=use_valgrind_profiler,
use_perftools_profiler=use_perftools_profiler,
use_tmux=use_tmux)
def wait_for_redis_to_start(redis_ip_address,
redis_port,
password=None,
@@ -528,9 +646,9 @@ def _start_redis_instance(node_ip_address="127.0.0.1",
will start LRU eviction of entries.
Returns:
A tuple of the port used by Redis and a handle to the process that was
started. If a port is passed in, then the returned port value is
the same.
A tuple of the port used by Redis and ProcessInfo for the process that
was started. If a port is passed in, then the returned port value
is the same.
Raises:
Exception: An exception is raised if Redis could not be started.
@@ -561,12 +679,15 @@ def _start_redis_instance(node_ip_address="127.0.0.1",
command += ["--requirepass", password]
command += (
["--port", str(port), "--loglevel", "warning"] + load_module_args)
p = subprocess.Popen(command, stdout=stdout_file, stderr=stderr_file)
process_info = start_ray_process(
command,
ray_constants.PROCESS_TYPE_REDIS_SERVER,
stdout_file=stdout_file,
stderr_file=stderr_file)
time.sleep(0.1)
# Check if Redis successfully started (or at least if it the executable
# did not exit within 0.1 seconds).
if p.poll() is None:
if process_info.process.poll() is None:
break
port = new_port()
counter += 1
@@ -633,7 +754,7 @@ def _start_redis_instance(node_ip_address="127.0.0.1",
address(node_ip_address, port),
node_ip_address, [stdout_file, stderr_file],
password=password)
return port, p
return port, process_info
def start_log_monitor(redis_address,
@@ -654,7 +775,7 @@ def start_log_monitor(redis_address,
redis_password (str): The password of the redis server.
Returns:
The process that was started.
ProcessInfo for the process that was started.
"""
log_monitor_filepath = os.path.join(
os.path.dirname(os.path.abspath(__file__)), "log_monitor.py")
@@ -664,12 +785,16 @@ def start_log_monitor(redis_address,
]
if redis_password:
command += ["--redis-password", redis_password]
p = subprocess.Popen(command, stdout=stdout_file, stderr=stderr_file)
process_info = start_ray_process(
command,
ray_constants.PROCESS_TYPE_LOG_MONITOR,
stdout_file=stdout_file,
stderr_file=stderr_file)
record_log_files_in_redis(
redis_address,
node_ip_address, [stdout_file, stderr_file],
password=redis_password)
return p
return process_info
def start_ui(redis_address, stdout_file=None, stderr_file=None):
@@ -683,7 +808,8 @@ def start_ui(redis_address, stdout_file=None, stderr_file=None):
no redirection should happen, then this should be None.
Returns:
A tuple of the web UI url and the process that was started.
A tuple of the web UI url and ProcessInfo for the process that was
started.
"""
port = 8888
@@ -695,8 +821,6 @@ def start_ui(redis_address, stdout_file=None, stderr_file=None):
break
except socket.error:
port += 1
new_env = os.environ.copy()
new_env["REDIS_ADDRESS"] = redis_address
# We generate the token used for authentication ourselves to avoid
# querying the jupyter server.
new_notebook_directory, webui_url, token = (
@@ -714,12 +838,13 @@ def start_ui(redis_address, stdout_file=None, stderr_file=None):
command.append("--allow-root")
try:
ui_process = subprocess.Popen(
process_info = start_ray_process(
command,
env=new_env,
ray_constants.PROCESS_TYPE_WEB_UI,
env_updates={"REDIS_ADDRESS": redis_address},
cwd=new_notebook_directory,
stdout=stdout_file,
stderr=stderr_file)
stdout_file=stdout_file,
stderr_file=stderr_file)
except Exception:
logger.warning("Failed to start the UI, you may need to run "
"'pip install jupyter'.")
@@ -727,7 +852,7 @@ def start_ui(redis_address, stdout_file=None, stderr_file=None):
logger.info("\n" + "=" * 70)
logger.info("View the web UI at {}".format(webui_url))
logger.info("=" * 70 + "\n")
return webui_url, ui_process
return webui_url, process_info
return None, None
@@ -836,7 +961,7 @@ def start_raylet(redis_address,
override defaults in RayConfig.
Returns:
The process that was started.
ProcessInfo for the process that was started.
"""
config = config or {}
config_str = ",".join(["{},{}".format(*kv) for kv in config.items()])
@@ -901,36 +1026,21 @@ def start_raylet(redis_address,
redis_password or "",
get_temp_root(),
]
if use_valgrind:
p = subprocess.Popen(
[
"valgrind", "--track-origins=yes", "--leak-check=full",
"--show-leak-kinds=all", "--leak-check-heuristics=stdstring",
"--error-exitcode=1"
] + command,
stdout=stdout_file,
stderr=stderr_file)
elif use_profiler:
p = subprocess.Popen(
["valgrind", "--tool=callgrind"] + command,
stdout=stdout_file,
stderr=stderr_file)
elif "RAYLET_PERFTOOLS_PATH" in os.environ:
modified_env = os.environ.copy()
modified_env["LD_PRELOAD"] = os.environ["RAYLET_PERFTOOLS_PATH"]
modified_env["CPUPROFILE"] = os.environ["RAYLET_PERFTOOLS_LOGFILE"]
p = subprocess.Popen(
command, stdout=stdout_file, stderr=stderr_file, env=modified_env)
else:
p = subprocess.Popen(command, stdout=stdout_file, stderr=stderr_file)
process_info = start_ray_process(
command,
ray_constants.PROCESS_TYPE_RAYLET,
use_valgrind=use_valgrind,
use_gdb=False,
use_valgrind_profiler=use_profiler,
use_perftools_profiler=("RAYLET_PERFTOOLS_PATH" in os.environ),
stdout_file=stdout_file,
stderr_file=stderr_file)
record_log_files_in_redis(
redis_address,
node_ip_address, [stdout_file, stderr_file],
password=redis_password)
return p
return process_info
def determine_plasma_store_config(object_store_memory=None,
@@ -1013,6 +1123,75 @@ def determine_plasma_store_config(object_store_memory=None,
return object_store_memory, plasma_directory
def _start_plasma_store(plasma_store_memory,
use_valgrind=False,
use_profiler=False,
stdout_file=None,
stderr_file=None,
plasma_directory=None,
huge_pages=False,
socket_name=None):
"""Start a plasma store process.
Args:
plasma_store_memory (int): The amount of memory in bytes to start the
plasma store with.
use_valgrind (bool): True if the plasma store should be started inside
of valgrind. If this is True, use_profiler must be False.
use_profiler (bool): True if the plasma store should be started inside
a profiler. If this is True, use_valgrind must be False.
stdout_file: A file handle opened for writing to redirect stdout to. If
no redirection should happen, then this should be None.
stderr_file: A file handle opened for writing to redirect stderr to. If
no redirection should happen, then this should be None.
plasma_directory: A directory where the Plasma memory mapped files will
be created.
huge_pages: a boolean flag indicating whether to start the
Object Store with hugetlbfs support. Requires plasma_directory.
socket_name (str): If provided, it will specify the socket
name used by the plasma store.
Return:
A tuple of the name of the plasma store socket and ProcessInfo for the
plasma store process.
"""
if use_valgrind and use_profiler:
raise Exception("Cannot use valgrind and profiler at the same time.")
if huge_pages and not (sys.platform == "linux"
or sys.platform == "linux2"):
raise Exception("The huge_pages argument is only supported on "
"Linux.")
if huge_pages and plasma_directory is None:
raise Exception("If huge_pages is True, then the "
"plasma_directory argument must be provided.")
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",
str(plasma_store_memory)
]
if plasma_directory is not None:
command += ["-d", plasma_directory]
if huge_pages:
command += ["-h"]
process_info = start_ray_process(
command,
ray_constants.PROCESS_TYPE_PLASMA_STORE,
use_valgrind=use_valgrind,
use_valgrind_profiler=use_profiler,
stdout_file=stdout_file,
stderr_file=stderr_file)
return plasma_store_name, process_info
def start_plasma_store(node_ip_address,
redis_address,
stdout_file=None,
@@ -1041,7 +1220,7 @@ def start_plasma_store(node_ip_address,
redis_password (str): The password of the redis server.
Returns:
The process that was started.
ProcessInfo for the process that was started.
"""
object_store_memory, plasma_directory = determine_plasma_store_config(
object_store_memory, plasma_directory, huge_pages)
@@ -1057,8 +1236,8 @@ 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, p = ray.plasma.start_plasma_store(
plasma_store_memory=object_store_memory,
plasma_store_name, process_info = _start_plasma_store(
object_store_memory,
use_profiler=RUN_PLASMA_STORE_PROFILER,
stdout_file=stdout_file,
stderr_file=stderr_file,
@@ -1071,7 +1250,7 @@ def start_plasma_store(node_ip_address,
node_ip_address, [stdout_file, stderr_file],
password=redis_password)
return p
return process_info
def start_worker(node_ip_address,
@@ -1097,7 +1276,7 @@ def start_worker(node_ip_address,
no redirection should happen, then this should be None.
Returns:
The process that was started.
ProcessInfo for the process that was started.
"""
command = [
sys.executable, "-u", worker_path,
@@ -1106,10 +1285,14 @@ def start_worker(node_ip_address,
"--redis-address=" + str(redis_address),
"--temp-dir=" + get_temp_root()
]
p = subprocess.Popen(command, stdout=stdout_file, stderr=stderr_file)
process_info = start_ray_process(
command,
ray_constants.PROCESS_TYPE_WORKER,
stdout_file=stdout_file,
stderr_file=stderr_file)
record_log_files_in_redis(redis_address, node_ip_address,
[stdout_file, stderr_file])
return p
return process_info
def start_monitor(redis_address,
@@ -1132,7 +1315,7 @@ def start_monitor(redis_address,
redis_password (str): The password of the redis server.
Returns:
The process that was started.
ProcessInfo for the process that was started.
"""
monitor_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)), "monitor.py")
@@ -1144,12 +1327,16 @@ def start_monitor(redis_address,
command.append("--autoscaling-config=" + str(autoscaling_config))
if redis_password:
command.append("--redis-password=" + redis_password)
p = subprocess.Popen(command, stdout=stdout_file, stderr=stderr_file)
process_info = start_ray_process(
command,
ray_constants.PROCESS_TYPE_MONITOR,
stdout_file=stdout_file,
stderr_file=stderr_file)
record_log_files_in_redis(
redis_address,
node_ip_address, [stdout_file, stderr_file],
password=redis_password)
return p
return process_info
def start_raylet_monitor(redis_address,
@@ -1170,7 +1357,7 @@ def start_raylet_monitor(redis_address,
override defaults in RayConfig.
Returns:
The process that was started.
ProcessInfo for the process that was started.
"""
gcs_ip_address, gcs_port = redis_address.split(":")
redis_password = redis_password or ""
@@ -1179,5 +1366,9 @@ def start_raylet_monitor(redis_address,
command = [RAYLET_MONITOR_EXECUTABLE, gcs_ip_address, gcs_port, config_str]
if redis_password:
command += [redis_password]
p = subprocess.Popen(command, stdout=stdout_file, stderr=stderr_file)
return p
process_info = start_ray_process(
command,
ray_constants.PROCESS_TYPE_RAYLET_MONITOR,
stdout_file=stdout_file,
stderr_file=stderr_file)
return process_info