diff --git a/test/actor_test.py b/test/actor_test.py index 17c413f43..e51401b1b 100644 --- a/test/actor_test.py +++ b/test/actor_test.py @@ -1253,6 +1253,57 @@ class ActorsWithGPUs(unittest.TestCase): assert remaining_ids == [x_id] +@unittest.skipIf( + os.environ.get("RAY_USE_XRAY") != "1", "This test only works with xray.") +class ActorExceptionFailures(unittest.TestCase): + def tearDown(self): + ray.shutdown() + + def testExceptionRaisedWhenActorNodeDies(self): + ray.worker._init( + start_ray_local=True, num_local_schedulers=2, num_cpus=1) + + @ray.remote + class Counter(object): + def __init__(self): + self.x = 0 + + def local_plasma(self): + return ray.worker.global_worker.plasma_client.store_socket_name + + def inc(self): + self.x += 1 + return self.x + + local_plasma = ray.worker.global_worker.plasma_client.store_socket_name + + # Create an actor that is not on the local scheduler. + actor = Counter.remote() + while ray.get(actor.local_plasma.remote()) == local_plasma: + actor = Counter.remote() + + # Kill the second plasma store to get rid of the cached objects and + # trigger the corresponding local scheduler to exit. + process = ray.services.all_processes[ + ray.services.PROCESS_TYPE_PLASMA_STORE][1] + process.kill() + + # Submit some new actor tasks. + x_ids = [actor.inc.remote() for _ in range(100)] + + # Make sure that getting the result raises an exception. + for _ in range(10): + for x_id in x_ids: + with pytest.raises(ray.worker.RayGetError): + # There is some small chance that ray.get will actually + # succeed (if the object is transferred before the raylet + # dies). + ray.get(x_id) + + # Make sure the process has exited. + process.wait() + + @unittest.skipIf( os.environ.get("RAY_USE_XRAY") == "1", "This test does not work with xray yet.")