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

Remove experimental.signal API (#7477)

Browse Source 
* Remove experimental.signal API

* fix test
This commit is contained in:
Edward Oakes
2020-03-09 16:03:36 -07:00
committed by GitHub
parent 023d4c02a9
commit 0c254295b0
7 changed files with 13 additions and 634 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/_raylet.pyx
-12
View File
@@ -80,7 +80,6 @@ from ray.includes.ray_config cimport RayConfig
import ray
from ray.async_compat import (sync_to_async,
AsyncGetResponse, AsyncMonitorState)
import ray.experimental.signal as ray_signal
import ray.memory_monitor as memory_monitor
import ray.ray_constants as ray_constants
from ray import profiling
@@ -475,17 +474,6 @@ cdef execute_task(
str(failure_object),
job_id=worker.current_job_id)
# Send signal with the error.
ray_signal.send(ray_signal.ErrorSignal(str(failure_object)))
# Don't need to reset `current_job_id` if the worker is an
# actor. Because the following tasks should all have the
# same driver id.
if <int>task_type == <int>TASK_TYPE_NORMAL_TASK:
# Reset signal counters so that the next task can get
# all past signals.
ray_signal.reset()
if execution_info.max_calls != 0:
# Reset the state of the worker for the next task to execute.
# Increase the task execution counter.
python/ray/experimental/signal.py
-198
View File
@@ -1,198 +0,0 @@
import logging
from collections import defaultdict
import ray
import ray.cloudpickle as cloudpickle
# This string should be identical to the name of the signal sent upon
# detecting that an actor died.
# This constant is also used in NodeManager::PublishActorStateTransition()
# in node_manager.cc
ACTOR_DIED_STR = "ACTOR_DIED_SIGNAL"
logger = logging.getLogger(__name__)
class Signal:
"""Base class for Ray signals."""
pass
class ErrorSignal(Signal):
"""Signal raised if an exception happens in a task or actor method."""
def __init__(self, error):
self.error = error
class ActorDiedSignal(Signal):
"""Signal raised if an exception happens in a task or actor method."""
def __init__(self):
pass
def _get_task_id(source):
"""Return the task id associated to the generic source of the signal.
Args:
source: source of the signal, it can be either an object id returned
by a task, a task id, or an actor handle.
Returns:
- If source is an object id, return id of task which creted object.
- If source is an actor handle, return id of actor's task creator.
- If source is a task id, return same task id.
"""
if type(source) is ray.actor.ActorHandle:
return source._actor_id
else:
if type(source) is ray.TaskID:
return source
else:
return ray._raylet.compute_task_id(source)
def send(signal):
"""Send signal.
The signal has a unique identifier that is computed from (1) the id
of the actor or task sending this signal (i.e., the actor or task calling
this function), and (2) an index that is incremented every time this
source sends a signal. This index starts from 1.
Args:
signal: Signal to be sent.
"""
if ray.worker.global_worker.actor_id.is_nil():
source_key = ray.worker.global_worker.current_task_id.hex()
else:
source_key = ray.worker.global_worker.actor_id.hex()
encoded_signal = ray.utils.binary_to_hex(cloudpickle.dumps(signal))
ray.worker.global_worker.redis_client.execute_command(
"XADD " + source_key + " * signal " + encoded_signal)
def receive(sources, timeout=None):
"""Get all outstanding signals from sources.
A source can be either (1) an object ID returned by the task (we want
to receive signals from), or (2) an actor handle.
When invoked by the same entity E (where E can be an actor, task or
driver), for each source S in sources, this function returns all signals
generated by S since the last receive() was invoked by E on S. If this is
the first call on S, this function returns all past signals generated by S
so far. Note that different actors, tasks or drivers that call receive()
on the same source S will get independent copies of the signals generated
by S.
Args:
sources: List of sources from which the caller waits for signals.
A source is either an object ID returned by a task (in this case
the object ID is used to identify that task), or an actor handle.
If the user passes the IDs of multiple objects returned by the
same task, this function returns a copy of the signals generated
by that task for each object ID.
timeout: Maximum time (in seconds) this function waits to get a signal
from a source in sources. If None, the timeout is infinite.
Returns:
A list of pairs (S, sig), where S is a source in the sources argument,
and sig is a signal generated by S since the last time receive()
was called on S. Thus, for each S in sources, the return list can
contain zero or multiple entries.
"""
# If None, initialize the timeout to a huge value (i.e., over 30,000 years
# in this case) to "approximate" infinity.
if timeout is None:
timeout = 10**12
if timeout < 0:
raise ValueError("The 'timeout' argument cannot be less than 0.")
if not hasattr(ray.worker.global_worker, "signal_counters"):
ray.worker.global_worker.signal_counters = defaultdict(lambda: b"0")
signal_counters = ray.worker.global_worker.signal_counters
# Map the ID of each source task to the source itself.
task_id_to_sources = defaultdict(lambda: [])
for s in sources:
task_id_to_sources[_get_task_id(s).hex()].append(s)
if timeout < 1e-3:
logger.warning("Timeout too small. Using 1ms minimum")
timeout = 1e-3
timeout_ms = int(1000 * timeout)
# Construct the redis query.
query = "XREAD BLOCK "
# redis expects ms.
query += str(timeout_ms)
query += " STREAMS "
query += " ".join(task_id_to_sources)
query += " "
query += " ".join([
ray.utils.decode(signal_counters[ray.utils.hex_to_binary(task_id)])
for task_id in task_id_to_sources
])
answers = ray.worker.global_worker.redis_client.execute_command(query)
if not answers:
return []
results = []
# Decoding is a little bit involved. Iterate through all the answers:
for i, answer in enumerate(answers):
# Make sure the answer corresponds to a source, s, in sources.
task_id = ray.utils.decode(answer[0])
task_source_list = task_id_to_sources[task_id]
# The list of results for source s is stored in answer[1]
for r in answer[1]:
for s in task_source_list:
if r[1][1].decode("ascii") == ACTOR_DIED_STR:
results.append((s, ActorDiedSignal()))
else:
# Now it gets tricky: r[0] is the redis internal sequence
# id
signal_counters[ray.utils.hex_to_binary(task_id)] = r[0]
# r[1] contains a list with elements (key, value), in our
# case we only have one key "signal" and the value is the
# signal.
signal = cloudpickle.loads(
ray.utils.hex_to_binary(r[1][1]))
results.append((s, signal))
return results
def forget(sources):
"""Ignore all previous signals associated with each source S in sources.
The index of the next expected signal from S is set to the index of
the last signal that S sent plus 1. This means that the next receive()
on S will only get the signals generated after this function was invoked.
Args:
sources: list of sources whose past signals are forgotten.
"""
# Just read all signals sent by all sources so far.
# This will results in ignoring these signals.
receive(sources, timeout=0)
def reset():
"""
Reset the worker state associated with any signals that this worker
has received so far.
If the worker calls receive() on a source next, it will get all the
signals generated by that source starting with index = 1.
"""
if hasattr(ray.worker.global_worker, "signal_counters"):
ray.worker.global_worker.signal_counters = defaultdict(lambda: b"0")
python/ray/test_utils.py
-11
View File
@@ -226,17 +226,6 @@ class SignalActor:
await self.ready_event.wait()
class RemoteSignal:
def __init__(self):
self.signal_actor = SignalActor.remote()
def send(self):
ray.get(self.signal_actor.send.remote())
def wait(self):
ray.get(self.signal_actor.wait.remote())
@ray.remote
def _put(obj):
return obj
python/ray/tests/BUILD
-8
View File
@@ -328,14 +328,6 @@ py_test(
deps = ["//:ray_lib"],
)
py_test(
name = "test_signal",
size = "medium",
srcs = ["test_signal.py"],
tags = ["exclusive"],
deps = ["//:ray_lib"],
)
py_test(
name = "test_tempfile",
size = "small",
python/ray/tests/test_component_failures.py
+10 -8
View File
@@ -1,12 +1,12 @@
import os
import signal
from signal import SIGKILL
import sys
import time
import pytest
import ray
from ray.test_utils import run_string_as_driver_nonblocking
from ray.test_utils import run_string_as_driver_nonblocking, SignalActor
# This test checks that when a worker dies in the middle of a get, the plasma
@@ -16,16 +16,18 @@ from ray.test_utils import run_string_as_driver_nonblocking
reason="Not working with new GCS API.")
def test_dying_worker_get(ray_start_2_cpus):
@ray.remote
def sleep_forever():
ray.experimental.signal.send("ready")
def sleep_forever(signal):
ray.get(signal.send.remote())
time.sleep(10**6)
@ray.remote
def get_worker_pid():
return os.getpid()
x_id = sleep_forever.remote()
ray.experimental.signal.receive([x_id]) # Block until it is scheduled.
signal = SignalActor.remote()
x_id = sleep_forever.remote(signal)
ray.get(signal.wait.remote())
# Get the PID of the other worker.
worker_pid = ray.get(get_worker_pid.remote())
@@ -42,7 +44,7 @@ def test_dying_worker_get(ray_start_2_cpus):
assert len(ready_ids) == 0
# Kill the worker.
os.kill(worker_pid, signal.SIGKILL)
os.kill(worker_pid, SIGKILL)
time.sleep(0.1)
# Make sure the sleep task hasn't finished.
@@ -129,7 +131,7 @@ def test_dying_worker_wait(ray_start_2_cpus):
time.sleep(0.1)
# Kill the worker.
os.kill(worker_pid, signal.SIGKILL)
os.kill(worker_pid, SIGKILL)
time.sleep(0.1)
# Create the object.
python/ray/tests/test_failure.py
+3 -2
View File
@@ -329,11 +329,12 @@ def test_incorrect_method_calls(ray_start_regular):
def test_worker_raising_exception(ray_start_regular):
@ray.remote
@ray.remote(max_calls=2)
def f():
# This is the only reasonable variable we can set here that makes the
# execute_task function fail after the task got executed.
ray.experimental.signal.reset = None
worker = ray.worker.global_worker
worker.function_actor_manager.increase_task_counter = None
# Running this task should cause the worker to raise an exception after
# the task has successfully completed.
python/ray/tests/test_signal.py
-395
View File
@@ -1,395 +0,0 @@
import pytest
import time
import ray
import ray.experimental.signal as signal
class UserSignal(signal.Signal):
def __init__(self, value):
self.value = value
def receive_all_signals(sources, timeout):
# Get all signals from sources, until there is no signal for a time
# period of timeout.
results = []
while True:
r = signal.receive(sources, timeout=timeout)
if len(r) == 0:
return results
else:
results.extend(r)
def test_task_to_driver(ray_start_regular):
# Send a signal from a task to the driver.
@ray.remote
def task_send_signal(value):
signal.send(UserSignal(value))
return
signal_value = "simple signal"
object_id = task_send_signal.remote(signal_value)
result_list = signal.receive([object_id], timeout=10)
print(result_list[0][1])
assert len(result_list) == 1
def test_send_signal_from_actor_to_driver(ray_start_regular):
# Send several signals from an actor, and receive them in the driver.
@ray.remote
class ActorSendSignal:
def __init__(self):
pass
def send_signal(self, value):
signal.send(UserSignal(value))
a = ActorSendSignal.remote()
signal_value = "simple signal"
count = 6
for i in range(count):
ray.get(a.send_signal.remote(signal_value + str(i)))
result_list = receive_all_signals([a], timeout=5)
assert len(result_list) == count
for i in range(count):
assert signal_value + str(i) == result_list[i][1].value
def test_send_signals_from_actor_to_driver(ray_start_regular):
# Send "count" signal at intervals from an actor and get
# these signals in the driver.
@ray.remote
class ActorSendSignals:
def __init__(self):
pass
def send_signals(self, value, count):
for i in range(count):
signal.send(UserSignal(value + str(i)))
a = ActorSendSignals.remote()
signal_value = "simple signal"
count = 20
a.send_signals.remote(signal_value, count)
received_count = 0
while True:
result_list = signal.receive([a], timeout=5)
received_count += len(result_list)
if (received_count == count):
break
assert True
def test_task_crash(ray_start_regular):
# Get an error when ray.get() is called on the return of a failed task.
@ray.remote
def crashing_function():
raise Exception("exception message")
object_id = crashing_function.remote()
try:
ray.get(object_id)
except Exception as e:
assert type(e) == ray.exceptions.RayTaskError
finally:
result_list = signal.receive([object_id], timeout=5)
assert len(result_list) == 1
assert type(result_list[0][1]) == signal.ErrorSignal
def test_task_crash_without_get(ray_start_regular):
# Get an error when task failed.
@ray.remote
def crashing_function():
raise Exception("exception message")
object_id = crashing_function.remote()
result_list = signal.receive([object_id], timeout=5)
assert len(result_list) == 1
assert type(result_list[0][1]) == signal.ErrorSignal
def test_actor_crash(ray_start_regular):
# Get an error when ray.get() is called on a return parameter
# of a method that failed.
@ray.remote
class Actor:
def __init__(self):
pass
def crash(self):
raise Exception("exception message")
a = Actor.remote()
try:
ray.get(a.crash.remote())
except Exception as e:
assert type(e) == ray.exceptions.RayTaskError
finally:
result_list = signal.receive([a], timeout=5)
assert len(result_list) == 1
assert type(result_list[0][1]) == signal.ErrorSignal
def test_actor_crash_init(ray_start_regular):
# Get an error when an actor's __init__ failed.
@ray.remote
class ActorCrashInit:
def __init__(self):
raise Exception("exception message")
def m(self):
return 1
# Do not catch the exception in the __init__.
a = ActorCrashInit.remote()
result_list = signal.receive([a], timeout=5)
assert len(result_list) == 1
assert type(result_list[0][1]) == signal.ErrorSignal
def test_actor_crash_init2(ray_start_regular):
# Get errors when (1) __init__ fails, and (2) subsequently when
# ray.get() is called on the return parameter of another method
# of the actor.
@ray.remote
class ActorCrashInit:
def __init__(self):
raise Exception("exception message")
def method(self):
return 1
a = ActorCrashInit.remote()
try:
ray.get(a.method.remote())
except Exception as e:
assert type(e) == ray.exceptions.RayTaskError
finally:
result_list = receive_all_signals([a], timeout=5)
assert len(result_list) == 2
assert type(result_list[0][1]) == signal.ErrorSignal
def test_actor_crash_init3(ray_start_regular):
# Get errors when (1) __init__ fails, and (2) subsequently when
# another method of the actor is invoked.
@ray.remote
class ActorCrashInit:
def __init__(self):
raise Exception("exception message")
def method(self):
return 1
a = ActorCrashInit.remote()
a.method.remote()
# Wait for a.method.remote() to finish and generate an error.
time.sleep(10)
result_list = signal.receive([a], timeout=5)
assert len(result_list) == 2
assert type(result_list[0][1]) == signal.ErrorSignal
def test_send_signals_from_actor_to_actor(ray_start_regular):
# Send "count" signal at intervals of 100ms from two actors and get
# these signals in another actor.
@ray.remote
class ActorSendSignals:
def __init__(self):
pass
def send_signals(self, value, count):
for i in range(count):
signal.send(UserSignal(value + str(i)))
@ray.remote
class ActorGetSignalsAll:
def __init__(self):
self.received_signals = []
def register_handle(self, handle):
self.this_actor = handle
def get_signals(self, source_ids, count):
new_signals = receive_all_signals(source_ids, timeout=5)
for s in new_signals:
self.received_signals.append(s)
if len(self.received_signals) < count:
self.this_actor.get_signals.remote(source_ids, count)
else:
return
def get_count(self):
return len(self.received_signals)
a1 = ActorSendSignals.remote()
a2 = ActorSendSignals.remote()
signal_value = "simple signal"
count = 20
ray.get(a1.send_signals.remote(signal_value, count))
ray.get(a2.send_signals.remote(signal_value, count))
b = ActorGetSignalsAll.remote()
ray.get(b.register_handle.remote(b))
b.get_signals.remote([a1, a2], count)
received_count = ray.get(b.get_count.remote())
assert received_count == 2 * count
def test_forget(ray_start_regular):
# Send "count" signals on behalf of an actor, then ignore all these
# signals, and then send anther "count" signals on behalf of the same
# actor. Then show that the driver only gets the last "count" signals.
@ray.remote
class ActorSendSignals:
def __init__(self):
pass
def send_signals(self, value, count):
for i in range(count):
signal.send(UserSignal(value + str(i)))
a = ActorSendSignals.remote()
signal_value = "simple signal"
count = 5
ray.get(a.send_signals.remote(signal_value, count))
signal.forget([a])
ray.get(a.send_signals.remote(signal_value, count))
result_list = receive_all_signals([a], timeout=5)
assert len(result_list) == count
def test_signal_on_node_failure(two_node_cluster):
"""Test actor checkpointing on a remote node."""
class ActorSignal:
def __init__(self):
pass
def node_id(self):
return ray.worker.global_worker.node.unique_id
# Place the actor on the remote node.
cluster, remote_node = two_node_cluster
actor_cls = ray.remote(max_reconstructions=0)(ActorSignal)
actor = actor_cls.remote()
# Try until we put an actor on a different node.
while (ray.get(actor.node_id.remote()) != remote_node.unique_id):
actor = actor_cls.remote()
# Kill actor process.
cluster.remove_node(remote_node)
# Wait on signal from the actor on the failed node.
result_list = signal.receive([actor], timeout=10)
assert len(result_list) == 1
assert type(result_list[0][1]) == signal.ActorDiedSignal
def test_send_signal_from_two_tasks_to_driver(ray_start_regular):
# Define a remote function that sends a user-defined signal.
@ray.remote
def send_signal(value):
signal.send(UserSignal(value))
a = send_signal.remote(0)
b = send_signal.remote(0)
ray.get([a, b])
result_list = ray.experimental.signal.receive([a])
assert len(result_list) == 1
# Call again receive on "a" with no new signal.
result_list = ray.experimental.signal.receive([a, b])
assert len(result_list) == 1
def test_receiving_on_two_returns(ray_start_regular):
@ray.remote(num_return_vals=2)
def send_signal(value):
signal.send(UserSignal(value))
return 1, 2
x, y = send_signal.remote(0)
ray.get([x, y])
results = ray.experimental.signal.receive([x, y])
assert ((x == results[0][0] and y == results[1][0])
or (x == results[1][0] and y == results[0][0]))
def test_serial_tasks_reading_same_signal(shutdown_only):
ray.init(num_cpus=2)
@ray.remote
def send_signal(value):
signal.send(UserSignal(value))
a = send_signal.remote(0)
@ray.remote
def f(sources):
return ray.experimental.signal.receive(sources, timeout=1)
result_list = ray.get(f.remote([a]))
assert len(result_list) == 1
result_list = ray.get(f.remote([a]))
assert len(result_list) == 1
result_list = ray.get(f.remote([a]))
assert len(result_list) == 1
def test_non_integral_receive_timeout(ray_start_regular):
@ray.remote
def send_signal(value):
signal.send(UserSignal(value))
a = send_signal.remote(0)
# make sure send_signal had a chance to execute
ray.get(a)
result_list = ray.experimental.signal.receive([a], timeout=0.1)
assert len(result_list) == 1
def test_small_receive_timeout(ray_start_regular):
""" Test that receive handles timeout smaller than the 1ms min
"""
# 0.1 ms
small_timeout = 1e-4
@ray.remote
def send_signal(value):
signal.send(UserSignal(value))
a = send_signal.remote(0)
# make sure send_signal had a chance to execute
ray.get(a)
result_list = ray.experimental.signal.receive([a], timeout=small_timeout)
assert len(result_list) == 1
if __name__ == "__main__":
import sys
sys.exit(pytest.main(["-v", __file__]))