[CI] Format Python code with Black (ray-project#21975)

See ray-project#21316 and ray-project#21311 for the motivation behind these changes.

.buildkite/copy_files.py

@@ -39,14 +39,16 @@ def perform_auth():
     resp = requests.get(
         "https://vop4ss7n22.execute-api.us-west-2.amazonaws.com/endpoint/",
         auth=auth,
-        params={"job_id": os.environ["BUILDKITE_JOB_ID"]})
+        params={"job_id": os.environ["BUILDKITE_JOB_ID"]},
+    )
     return resp
 
 
 def handle_docker_login(resp):
     pwd = resp.json()["docker_password"]
     subprocess.call(
-        ["docker", "login", "--username", "raytravisbot", "--password", pwd])
+        ["docker", "login", "--username", "raytravisbot", "--password", pwd]
+    )
 
 
 def gather_paths(dir_path) -> List[str]:
@@ -86,7 +88,7 @@ def upload_paths(paths, resp, destination):
             "branch_wheels": f"{branch}/{sha}/{fn}",
             "jars": f"jars/latest/{current_os}/{fn}",
             "branch_jars": f"jars/{branch}/{sha}/{current_os}/{fn}",
-            "logs": f"bazel_events/{branch}/{sha}/{bk_job_id}/{fn}"
+            "logs": f"bazel_events/{branch}/{sha}/{bk_job_id}/{fn}",
         }[destination]
         of["file"] = open(path, "rb")
         r = requests.post(c["url"], files=of)
@@ -95,14 +97,19 @@ def upload_paths(paths, resp, destination):
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(
-        description="Helper script to upload files to S3 bucket")
+        description="Helper script to upload files to S3 bucket"
+    )
     parser.add_argument("--path", type=str, required=False)
     parser.add_argument("--destination", type=str)
     args = parser.parse_args()
 
     assert args.destination in {
-        "branch_jars", "branch_wheels", "jars", "logs", "wheels",
-        "docker_login"
+        "branch_jars",
+        "branch_wheels",
+        "jars",
+        "logs",
+        "wheels",
+        "docker_login",
     }
     assert "BUILDKITE_JOB_ID" in os.environ
     assert "BUILDKITE_COMMIT" in os.environ

benchmarks/distributed/test_many_actors.py

@@ -51,8 +51,10 @@ def no_resource_leaks():
 test_utils.wait_for_condition(no_resource_leaks)
 
 rate = MAX_ACTORS_IN_CLUSTER / (end_time - start_time)
-print(f"Success! Started {MAX_ACTORS_IN_CLUSTER} actors in "
-      f"{end_time - start_time}s. ({rate} actors/s)")
+print(
+    f"Success! Started {MAX_ACTORS_IN_CLUSTER} actors in "
+    f"{end_time - start_time}s. ({rate} actors/s)"
+)
 
 if "TEST_OUTPUT_JSON" in os.environ:
     out_file = open(os.environ["TEST_OUTPUT_JSON"], "w")
@@ -62,6 +64,6 @@ def no_resource_leaks():
         "time": end_time - start_time,
         "success": "1",
         "_peak_memory": round(used_gb, 2),
-        "_peak_process_memory": usage
+        "_peak_process_memory": usage,
     }
     json.dump(results, out_file)

benchmarks/distributed/test_many_pgs.py

@@ -77,8 +77,10 @@ def no_resource_leaks():
 test_utils.wait_for_condition(no_resource_leaks)
 
 rate = MAX_PLACEMENT_GROUPS / (end_time - start_time)
-print(f"Success! Started {MAX_PLACEMENT_GROUPS} pgs in "
-      f"{end_time - start_time}s. ({rate} pgs/s)")
+print(
+    f"Success! Started {MAX_PLACEMENT_GROUPS} pgs in "
+    f"{end_time - start_time}s. ({rate} pgs/s)"
+)
 
 if "TEST_OUTPUT_JSON" in os.environ:
     out_file = open(os.environ["TEST_OUTPUT_JSON"], "w")
@@ -88,6 +90,6 @@ def no_resource_leaks():
         "time": end_time - start_time,
         "success": "1",
         "_peak_memory": round(used_gb, 2),
-        "_peak_process_memory": usage
+        "_peak_process_memory": usage,
     }
     json.dump(results, out_file)

benchmarks/distributed/test_many_tasks.py

@@ -16,9 +16,7 @@ def test_max_running_tasks(num_tasks):
     def task():
         time.sleep(sleep_time)
 
-    refs = [
-        task.remote() for _ in tqdm.trange(num_tasks, desc="Launching tasks")
-    ]
+    refs = [task.remote() for _ in tqdm.trange(num_tasks, desc="Launching tasks")]
 
     max_cpus = ray.cluster_resources()["CPU"]
     min_cpus_available = max_cpus
@@ -48,8 +46,7 @@ def no_resource_leaks():
 
 
 @click.command()
-@click.option(
-    "--num-tasks", required=True, type=int, help="Number of tasks to launch.")
+@click.option("--num-tasks", required=True, type=int, help="Number of tasks to launch.")
 def test(num_tasks):
     ray.init(address="auto")
 
@@ -66,8 +63,10 @@ def test(num_tasks):
     test_utils.wait_for_condition(no_resource_leaks)
 
     rate = num_tasks / (end_time - start_time - sleep_time)
-    print(f"Success! Started {num_tasks} tasks in {end_time - start_time}s. "
-          f"({rate} tasks/s)")
+    print(
+        f"Success! Started {num_tasks} tasks in {end_time - start_time}s. "
+        f"({rate} tasks/s)"
+    )
 
     if "TEST_OUTPUT_JSON" in os.environ:
         out_file = open(os.environ["TEST_OUTPUT_JSON"], "w")
@@ -77,7 +76,7 @@ def test(num_tasks):
             "time": end_time - start_time,
             "success": "1",
             "_peak_memory": round(used_gb, 2),
-            "_peak_process_memory": usage
+            "_peak_process_memory": usage,
         }
         json.dump(results, out_file)
 

benchmarks/distributed/test_scheduling.py

@@ -25,10 +25,12 @@ def do_job(self):
 
 
 def start_tasks(num_task, num_cpu_per_task, task_duration):
-    ray.get([
-        simple_task.options(num_cpus=num_cpu_per_task).remote(task_duration)
-        for _ in range(num_task)
-    ])
+    ray.get(
+        [
+            simple_task.options(num_cpus=num_cpu_per_task).remote(task_duration)
+            for _ in range(num_task)
+        ]
+    )
 
 
 def measure(f):
@@ -40,47 +42,49 @@ def measure(f):
 
 def start_actor(num_actors, num_actors_per_nodes, job):
     resources = {"node": floor(1.0 / num_actors_per_nodes)}
-    submission_cost, actors = measure(lambda: [
-        SimpleActor.options(resources=resources, num_cpus=0).remote(job)
-        for _ in range(num_actors)])
-    ready_cost, _ = measure(
-        lambda: ray.get([actor.ready.remote() for actor in actors]))
+    submission_cost, actors = measure(
+        lambda: [
+            SimpleActor.options(resources=resources, num_cpus=0).remote(job)
+            for _ in range(num_actors)
+        ]
+    )
+    ready_cost, _ = measure(lambda: ray.get([actor.ready.remote() for actor in actors]))
     actor_job_cost, _ = measure(
-        lambda: ray.get([actor.do_job.remote() for actor in actors]))
+        lambda: ray.get([actor.do_job.remote() for actor in actors])
+    )
     return (submission_cost, ready_cost, actor_job_cost)
 
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(prog="Test Scheduling")
     # Task workloads
     parser.add_argument(
-        "--total-num-task",
-        type=int,
-        help="Total number of tasks.",
-        required=False)
+        "--total-num-task", type=int, help="Total number of tasks.", required=False
+    )
     parser.add_argument(
         "--num-cpu-per-task",
         type=int,
         help="Resources needed for tasks.",
-        required=False)
+        required=False,
+    )
     parser.add_argument(
         "--task-duration-s",
         type=int,
         help="How long does each task execute.",
         required=False,
-        default=1)
+        default=1,
+    )
 
     # Actor workloads
     parser.add_argument(
-        "--total-num-actors",
-        type=int,
-        help="Total number of actors.",
-        required=True)
+        "--total-num-actors", type=int, help="Total number of actors.", required=True
+    )
     parser.add_argument(
         "--num-actors-per-nodes",
         type=int,
         help="How many actors to allocate for each nodes.",
-        required=True)
+        required=True,
+    )
 
     ray.init(address="auto")
 
@@ -92,13 +96,14 @@ def start_actor(num_actors, num_actors_per_nodes, job):
     job = None
     if args.total_num_task is not None:
         if args.num_cpu_per_task is None:
-            args.num_cpu_per_task = floor(
-                1.0 * total_cpus / args.total_num_task)
+            args.num_cpu_per_task = floor(1.0 * total_cpus / args.total_num_task)
         job = lambda: start_tasks(  # noqa: E731
-            args.total_num_task, args.num_cpu_per_task, args.task_duration_s)
+            args.total_num_task, args.num_cpu_per_task, args.task_duration_s
+        )
 
     submission_cost, ready_cost, actor_job_cost = start_actor(
-        args.total_num_actors, args.num_actors_per_nodes, job)
+        args.total_num_actors, args.num_actors_per_nodes, job
+    )
 
     output = os.environ.get("TEST_OUTPUT_JSON")
 
@@ -118,6 +123,7 @@ def start_actor(num_actors, num_actors_per_nodes, job):
 
     if output is not None:
         from pathlib import Path
+
         p = Path(output)
         p.write_text(json.dumps(result))
 

benchmarks/distributed/wait_cluster.py

@@ -12,8 +12,7 @@ def num_alive_nodes():
 
 
 @click.command()
-@click.option(
-    "--num-nodes", required=True, type=int, help="The target number of nodes")
+@click.option("--num-nodes", required=True, type=int, help="The target number of nodes")
 def wait_cluster(num_nodes: int):
     ray.init(address="auto")
     while num_alive_nodes() != num_nodes:

benchmarks/object_store/test_object_store.py

@@ -9,7 +9,7 @@
 from tqdm import tqdm
 
 NUM_NODES = 50
-OBJECT_SIZE = 2**30
+OBJECT_SIZE = 2 ** 30
 
 
 def num_alive_nodes():
@@ -60,6 +60,6 @@ def sum(self, arr):
         "broadcast_time": end - start,
         "object_size": OBJECT_SIZE,
         "num_nodes": NUM_NODES,
-        "success": "1"
+        "success": "1",
     }
     json.dump(results, out_file)

benchmarks/single_node/test_single_node.py

@@ -13,7 +13,7 @@
 MAX_RETURNS = 3000
 MAX_RAY_GET_ARGS = 10000
 MAX_QUEUED_TASKS = 1_000_000
-MAX_RAY_GET_SIZE = 100 * 2**30
+MAX_RAY_GET_SIZE = 100 * 2 ** 30
 
 
 def assert_no_leaks():
@@ -189,8 +189,7 @@ def test_large_object():
 print(f"Many returns time: {returns_time} ({MAX_RETURNS} returns)")
 print(f"Ray.get time: {get_time} ({MAX_RAY_GET_ARGS} args)")
 print(f"Queued task time: {queued_time} ({MAX_QUEUED_TASKS} tasks)")
-print(f"Ray.get large object time: {large_object_time} "
-      f"({MAX_RAY_GET_SIZE} bytes)")
+print(f"Ray.get large object time: {large_object_time} " f"({MAX_RAY_GET_SIZE} bytes)")
 
 if "TEST_OUTPUT_JSON" in os.environ:
     out_file = open(os.environ["TEST_OUTPUT_JSON"], "w")
@@ -205,6 +204,6 @@ def test_large_object():
         "num_queued": MAX_QUEUED_TASKS,
         "large_object_time": large_object_time,
         "large_object_size": MAX_RAY_GET_SIZE,
-        "success": "1"
+        "success": "1",
     }
     json.dump(results, out_file)