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.