From f395e480316e31d47883c746f06a1b45e1bc7711 Mon Sep 17 00:00:00 2001
From: Tao Wang <dooku.wt@antfin.com>
Date: Wed, 8 Jul 2020 10:25:08 +0800
Subject: [PATCH] [GCS] Add integration test for actor info cleaning (#8900)

---
 ci/travis/ci.sh                     |  1 +
 python/ray/test_utils.py            |  9 +++
 python/ray/tests/BUILD              |  8 +++
 python/ray/tests/test_advanced_3.py | 11 +---
 python/ray/tests/test_job.py        | 96 +++++++++++++++++++++++++++++
 5 files changed, 115 insertions(+), 10 deletions(-)
 create mode 100644 python/ray/tests/test_job.py

diff --git a/ci/travis/ci.sh b/ci/travis/ci.sh
index ec430eb44..18d312e96 100755
--- a/ci/travis/ci.sh
+++ b/ci/travis/ci.sh
@@ -132,6 +132,7 @@ test_python() {
       -python/ray/tests:test_cython
       -python/ray/tests:test_failure
       -python/ray/tests:test_global_gc
+      -python/ray/tests:test_job
       -python/ray/tests:test_memstat
       -python/ray/tests:test_metrics
       -python/ray/tests:test_multi_node
diff --git a/python/ray/test_utils.py b/python/ray/test_utils.py
index c9158ed36..fd59fad7d 100644
--- a/python/ray/test_utils.py
+++ b/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:
diff --git a/python/ray/tests/BUILD b/python/ray/tests/BUILD
index d765dc900..6dc95fa4c 100644
--- a/python/ray/tests/BUILD
+++ b/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",
diff --git a/python/ray/tests/test_advanced_3.py b/python/ray/tests/test_advanced_3.py
index 131ce14e6..7d7095d8e 100644
--- a/python/ray/tests/test_advanced_3.py
+++ b/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:
diff --git a/python/ray/tests/test_job.py b/python/ray/tests/test_job.py
new file mode 100644
index 000000000..b38e21a59
--- /dev/null
+++ b/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__]))