[GCS] Add integration test for actor info cleaning (#8900)

python/ray/test_utils.py

@@ -155,6 +155,15 @@ def relevant_errors(error_type):
     return [error for error in flat_errors() if error["type"] == error_type]
 
 
+def wait_for_num_actors(num_actors, timeout=10):
+    start_time = time.time()
+    while time.time() - start_time < timeout:
+        if len(ray.actors()) >= num_actors:
+            return
+        time.sleep(0.1)
+    raise RayTestTimeoutException("Timed out while waiting for global state.")
+
+
 def wait_for_errors(error_type, num_errors, timeout=20):
     start_time = time.time()
     while time.time() - start_time < timeout:

python/ray/tests/BUILD

@@ -331,6 +331,14 @@ py_test(
     deps = ["//:ray_lib"],
 )
 
+py_test(
+    name = "test_job",
+    size = "small",
+    srcs = SRCS + ["test_job.py"],
+    tags = ["exclusive"],
+    deps = ["//:ray_lib"],
+)
+
 py_test(
     name = "test_joblib",
     size = "medium",

python/ray/tests/test_advanced_3.py

@@ -20,7 +20,7 @@ import ray.cluster_utils
 import ray.test_utils
 import setproctitle
 
-from ray.test_utils import RayTestTimeoutException
+from ray.test_utils import RayTestTimeoutException, wait_for_num_actors
 
 logger = logging.getLogger(__name__)
 
@@ -85,15 +85,6 @@ def test_load_balancing_with_dependencies(ray_start_cluster):
     attempt_to_load_balance(f, [x], 100, num_nodes, 25)
 
 
-def wait_for_num_actors(num_actors, timeout=10):
-    start_time = time.time()
-    while time.time() - start_time < timeout:
-        if len(ray.actors()) >= num_actors:
-            return
-        time.sleep(0.1)
-    raise RayTestTimeoutException("Timed out while waiting for global state.")
-
-
 def wait_for_num_objects(num_objects, timeout=10):
     start_time = time.time()
     while time.time() - start_time < timeout:

python/ray/tests/test_job.py

@@ -0,0 +1,96 @@
+import os
+
+import ray
+from ray.test_utils import (
+    run_string_as_driver_nonblocking,
+    wait_for_condition,
+    wait_for_num_actors,
+)
+
+
+def test_job_gc(call_ray_start):
+    address = call_ray_start
+
+    ray.init(address=address)
+    driver = """
+import ray
+
+ray.init(address="{}")
+
+@ray.remote
+class Actor:
+    def __init__(self):
+        pass
+
+_ = Actor.remote()
+""".format(address)
+
+    p = run_string_as_driver_nonblocking(driver)
+    # Wait for actor to be created
+    wait_for_num_actors(1)
+
+    actor_table = ray.actors()
+    assert len(actor_table) == 1
+
+    job_table = ray.jobs()
+    assert len(job_table) == 2
+
+    # Kill the driver process.
+    p.kill()
+    p.wait()
+
+    def actor_finish():
+        actor_table = ray.actors()
+        if (len(actor_table) == 0):
+            return True
+        else:
+            return False
+
+    assert wait_for_condition(actor_finish)
+
+
+def test_job_gc_with_detached_actor(call_ray_start):
+    address = call_ray_start
+
+    ray.init(address=address)
+    driver = """
+import ray
+
+ray.init(address="{}")
+
+@ray.remote
+class Actor:
+    def __init__(self):
+        pass
+
+    def value(self):
+        return 1
+
+_ = Actor.options(name="DetachedActor").remote()
+""".format(address)
+
+    p = run_string_as_driver_nonblocking(driver)
+    # Wait for actor to be created
+    wait_for_num_actors(1)
+
+    actor_table = ray.actors()
+    assert len(actor_table) == 1
+
+    job_table = ray.jobs()
+    assert len(job_table) == 2
+
+    # Kill the driver process.
+    p.kill()
+    p.wait()
+
+    detached_actor = ray.get_actor("DetachedActor")
+    assert ray.get(detached_actor.value.remote()) == 1
+
+
+if __name__ == "__main__":
+    import pytest
+    import sys
+    # Make subprocess happy in bazel.
+    os.environ["LC_ALL"] = "en_US.UTF-8"
+    os.environ["LANG"] = "en_US.UTF-8"
+    sys.exit(pytest.main(["-v", __file__]))