Rebased onto run_string_as_driver_stdout_stderr

Signed-off-by: Alexey Kudinkin <ak@anyscale.com>

python/ray/dashboard/datacenter.py

@@ -102,9 +102,7 @@ async def organize(cls, thread_pool_executor):
         DataSource.core_worker_stats.reset(core_worker_stats)
 
     @classmethod
-    def _extract_workers_for_node(
-        cls, node_physical_stats, node_stats
-    ):
+    def _extract_workers_for_node(cls, node_physical_stats, node_stats):
         workers = []
         # Merge coreWorkerStats (node stats) to workers (node physical stats)
         pid_to_worker_stats = {}

python/ray/tests/test_output.py

@@ -102,19 +102,17 @@ def test_spill_logs():
 for _ in range(10):
     x.append(ray.put(bytes(100 * 1024 * 1024)))
 
-# NOTE: This ray.get is necessary to make sure object is persisted
-#       (and spilled) int he OS
-ray.get(x)
 """
-
-    proc = run_string_as_driver_nonblocking(script, env={"RAY_verbose_spill_logs": "1"})
-    out_str = proc.stdout.read().decode("ascii") + proc.stderr.read().decode("ascii")
-    print(out_str)
+    stdout_str, stderr_str = run_string_as_driver_stdout_stderr(
+        script, env={"RAY_verbose_spill_logs": "1"}
+    )
+    out_str = stdout_str + stderr_str
     assert "Spilled " in out_str
 
-    proc = run_string_as_driver_nonblocking(script, env={"RAY_verbose_spill_logs": "0"})
-    out_str = proc.stdout.read().decode("ascii") + proc.stderr.read().decode("ascii")
-    print(out_str)
+    stdout_str, stderr_str = run_string_as_driver_stdout_stderr(
+        script, env={"RAY_verbose_spill_logs": "0"}
+    )
+    out_str = stdout_str + stderr_str
     assert "Spilled " not in out_str