From f37b6874523cdab4dc69118ff44b1bb8469a9217 Mon Sep 17 00:00:00 2001
From: Alex Wu <itswu.alex@gmail.com>
Date: Fri, 18 Sep 2020 09:31:07 -0700
Subject: [PATCH] [release test] Spillback test (#10788)

* .

* marked things as hidden

* removed remaining redis args

* removed huge pages and include_java

* adjust hidden fields

* lint

* readded include_java

* Delete temp

* lint

* .

* test_cli is flakey but ok

* .

* .

* .

* .

* .

* .

* .

* .

* .

* .

* print

* lint

* lint
---
 .../stress_tests/test_many_tasks.py           | 44 +++++++++++++++++++
 1 file changed, 44 insertions(+)

diff --git a/ci/regression_test/stress_tests/test_many_tasks.py b/ci/regression_test/stress_tests/test_many_tasks.py
index f486b2d18..1abebfb51 100644
--- a/ci/regression_test/stress_tests/test_many_tasks.py
+++ b/ci/regression_test/stress_tests/test_many_tasks.py
@@ -1,5 +1,6 @@
 #!/usr/bin/env python
 
+from collections import defaultdict
 import numpy as np
 import logging
 import time
@@ -110,6 +111,46 @@ for N in [1000, 100000]:
 stage_3_time = time.time() - start_time
 logger.info("Finished stage 3 in %s seconds.", stage_3_time)
 
+# This tests https://github.com/ray-project/ray/issues/10150. The only way to
+# integration test this is via performance. The goal is to fill up the cluster
+# so that all tasks can be run, but spillback is required. Since the driver
+# submits all these tasks it should easily be able to schedule each task in
+# O(1) iterative spillback queries. If spillback behavior is incorrect, each
+# task will require O(N) queries. Since we limit the number of inflight
+# requests, we will run into head of line blocking and we should be able to
+# measure this timing.
+num_tasks = int(ray.cluster_resources()["GPU"])
+logger.info(f"Scheduling many tasks for spillback.")
+
+
+@ray.remote(num_gpus=1)
+def func(t):
+    if t % 100 == 0:
+        logger.info(f"[spillback test] {t}/{num_tasks}")
+    start = time.perf_counter()
+    time.sleep(1)
+    end = time.perf_counter()
+    return start, end, ray.worker.global_worker.node.unique_id
+
+
+results = ray.get([func.remote(i) for i in range(num_tasks)])
+
+host_to_start_times = defaultdict(list)
+for start, end, host in results:
+    host_to_start_times[host].append(start)
+
+spreads = []
+for host in host_to_start_times:
+    last = max(host_to_start_times[host])
+    first = min(host_to_start_times[host])
+    spread = last - first
+    spreads.append(spread)
+    logger.info(f"Spread: {last - first}\tLast: {last}\tFirst: {first}")
+
+# avg_spread ~ 115 with Ray 1.0 scheduler. ~695 with (buggy) 0.8.7 scheduler.
+avg_spread = sum(spreads) / len(spreads)
+logger.info(f"Avg spread: {sum(spreads)/len(spreads)}")
+
 print("Stage 0 results:")
 print("\tTotal time: {}".format(stage_0_time))
 
@@ -131,6 +172,9 @@ print("Stage 3 results:")
 print("\tActor creation time: {}".format(stage_3_creation_time))
 print("\tTotal time: {}".format(stage_3_time))
 
+print("Stage 4 results:")
+print(f"\tScheduling spread: {avg_spread}.")
+
 # TODO(rkn): The test below is commented out because it currently does not
 # pass.
 # # Submit a bunch of actor tasks with all-to-all communication.