Remove ray.tasks() from API. (#7807)

python/ray/tests/BUILD

@@ -282,15 +282,6 @@ py_test(
     deps = ["//:ray_lib"],
 )
 
-py_test(
-    name = "test_monitors",
-    size = "small",
-    srcs = ["test_monitors.py"],
-    # TODO(ekl) tasks() and objects() are different in direct call mode.
-    tags = ["exclusive", "manual"],
-    deps = ["//:ray_lib"],
-)
-
 py_test(
     name = "test_multiprocessing",
     size = "medium",

python/ray/tests/test_advanced_3.py

@@ -94,15 +94,6 @@ def wait_for_num_actors(num_actors, timeout=10):
     raise RayTestTimeoutException("Timed out while waiting for global state.")
 
 
-def wait_for_num_tasks(num_tasks, timeout=10):
-    start_time = time.time()
-    while time.time() - start_time < timeout:
-        if len(ray.tasks()) >= num_tasks:
-            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:
@@ -123,9 +114,6 @@ def test_global_state_api(shutdown_only):
     with pytest.raises(Exception, match=error_message):
         ray.actors()
 
-    with pytest.raises(Exception, match=error_message):
-        ray.tasks()
-
     with pytest.raises(Exception, match=error_message):
         ray.nodes()
 
@@ -142,22 +130,6 @@ def test_global_state_api(shutdown_only):
 
     job_id = ray.utils.compute_job_id_from_driver(
         ray.WorkerID(ray.worker.global_worker.worker_id))
-    driver_task_id = ray.worker.global_worker.current_task_id.hex()
-
-    # One task is put in the task table which corresponds to this driver.
-    wait_for_num_tasks(1)
-    task_table = ray.tasks()
-    assert len(task_table) == 1
-    assert driver_task_id == list(task_table.keys())[0]
-    task_spec = task_table[driver_task_id]["TaskSpec"]
-    nil_actor_id_hex = ray.ActorID.nil().hex()
-
-    assert task_spec["TaskID"] == driver_task_id
-    assert task_spec["ActorID"] == nil_actor_id_hex
-    assert task_spec["Args"] == []
-    assert task_spec["JobID"] == job_id.hex()
-    assert task_spec["FunctionDescriptor"]["type"] == "EmptyFunctionDescriptor"
-    assert task_spec["ReturnObjectIDs"] == []
 
     client_table = ray.nodes()
     node_ip_address = ray.worker.global_worker.node_ip_address

python/ray/tests/test_monitors.py

@@ -1,114 +0,0 @@
-import multiprocessing
-import os
-import pytest
-import subprocess
-import time
-
-import ray
-
-
-def _test_cleanup_on_driver_exit(num_redis_shards):
-    output = ray.utils.decode(
-        subprocess.check_output(
-            [
-                "ray",
-                "start",
-                "--head",
-                "--num-redis-shards",
-                str(num_redis_shards),
-            ],
-            stderr=subprocess.STDOUT))
-    lines = [m.strip() for m in output.split("\n")]
-    init_cmd = [m for m in lines if m.startswith("ray.init")]
-    assert 1 == len(init_cmd)
-    address = init_cmd[0].split("address=\"")[-1][:-2]
-    max_attempts_before_failing = 100
-    # Wait for monitor.py to start working.
-    time.sleep(2)
-
-    def StateSummary():
-        obj_tbl_len = len(ray.objects())
-        task_tbl_len = len(ray.tasks())
-        return obj_tbl_len, task_tbl_len
-
-    def Driver(success):
-        success.value = True
-        # Start driver.
-        ray.init(address=address)
-        summary_start = StateSummary()
-        if (0, 1) != summary_start:
-            success.value = False
-
-        # Two new objects.
-        ray.get(ray.put(1111))
-        ray.get(ray.put(1111))
-
-        @ray.remote
-        def f():
-            ray.put(1111)  # Yet another object.
-            return 1111  # A returned object as well.
-
-        # 1 new function.
-        attempts = 0
-        while (2, 1) != StateSummary():
-            time.sleep(0.1)
-            attempts += 1
-            if attempts == max_attempts_before_failing:
-                success.value = False
-                break
-
-        ray.get(f.remote())
-        attempts = 0
-        while (4, 2) != StateSummary():
-            time.sleep(0.1)
-            attempts += 1
-            if attempts == max_attempts_before_failing:
-                success.value = False
-                break
-
-        ray.shutdown()
-
-    success = multiprocessing.Value("b", False)
-    driver = multiprocessing.Process(target=Driver, args=(success, ))
-    driver.start()
-    # Wait for client to exit.
-    driver.join()
-
-    # Just make sure Driver() is run and succeeded.
-    assert success.value
-    # Check that objects, tasks, and functions are cleaned up.
-    ray.init(address=address)
-    attempts = 0
-    while (0, 1) != StateSummary():
-        time.sleep(0.1)
-        attempts += 1
-        if attempts == max_attempts_before_failing:
-            break
-    assert (0, 1) == StateSummary()
-
-    ray.shutdown()
-    subprocess.check_output(["ray", "stop"])
-
-
-@pytest.mark.skipif(
-    os.environ.get("RAY_USE_NEW_GCS") == "on",
-    reason="Hanging with the new GCS API.")
-def test_cleanup_on_driver_exit_single_redis_shard():
-    _test_cleanup_on_driver_exit(num_redis_shards=1)
-
-
-@pytest.mark.skipif(
-    os.environ.get("RAY_USE_NEW_GCS") == "on",
-    reason="Hanging with the new GCS API.")
-def test_cleanup_on_driver_exit_many_redis_shards():
-    _test_cleanup_on_driver_exit(num_redis_shards=5)
-    _test_cleanup_on_driver_exit(num_redis_shards=31)
-
-
-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__]))

python/ray/tests/test_multi_node.py

@@ -167,6 +167,39 @@ print("success")
         assert "success" in out
 
 
+def test_cleanup_on_driver_exit(call_ray_start):
+    # This test will create a driver that creates a bunch of objects and then
+    # exits. The entries in the object table should be cleaned up.
+    address = call_ray_start
+
+    ray.init(address=address)
+
+    # Define a driver that creates a bunch of objects and exits.
+    driver_script = """
+import time
+import ray
+ray.init(address="{}")
+object_ids = [ray.put(i) for i in range(1000)]
+start_time = time.time()
+while time.time() - start_time < 30:
+    if len(ray.objects()) == 1000:
+        break
+else:
+    raise Exception("Objects did not appear in object table.")
+print("success")
+""".format(address)
+
+    run_string_as_driver(driver_script)
+
+    # Make sure the objects are removed from the object table.
+    start_time = time.time()
+    while time.time() - start_time < 30:
+        if len(ray.objects()) == 0:
+            break
+    else:
+        raise Exception("Objects were not all removed from object table.")
+
+
 def test_drivers_named_actors(call_ray_start):
     # This test will create some drivers that submit some tasks to the same
     # named actor.