diff --git a/python/ray/_raylet.pyx b/python/ray/_raylet.pyx index ca719f517..fdbff32d8 100644 --- a/python/ray/_raylet.pyx +++ b/python/ray/_raylet.pyx @@ -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 task_type == 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. diff --git a/python/ray/experimental/signal.py b/python/ray/experimental/signal.py deleted file mode 100644 index 477d7821e..000000000 --- a/python/ray/experimental/signal.py +++ /dev/null @@ -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") diff --git a/python/ray/test_utils.py b/python/ray/test_utils.py index 65be5fd98..9fb3c53bd 100644 --- a/python/ray/test_utils.py +++ b/python/ray/test_utils.py @@ -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 diff --git a/python/ray/tests/BUILD b/python/ray/tests/BUILD index 96a844c00..84696a919 100644 --- a/python/ray/tests/BUILD +++ b/python/ray/tests/BUILD @@ -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", diff --git a/python/ray/tests/test_component_failures.py b/python/ray/tests/test_component_failures.py index 416802035..90c921f2f 100644 --- a/python/ray/tests/test_component_failures.py +++ b/python/ray/tests/test_component_failures.py @@ -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. diff --git a/python/ray/tests/test_failure.py b/python/ray/tests/test_failure.py index ba266b60b..e169e1b21 100644 --- a/python/ray/tests/test_failure.py +++ b/python/ray/tests/test_failure.py @@ -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. diff --git a/python/ray/tests/test_signal.py b/python/ray/tests/test_signal.py deleted file mode 100644 index af41c5126..000000000 --- a/python/ray/tests/test_signal.py +++ /dev/null @@ -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__]))