Fix a bug from many drivers. (#22248)

After this PR (https://github.com/ray-project/ray/pull/22156), for some reasons the driver script has some string that cannot be encoded with ascii. It seems like using utf-8 solves the problem.

python/ray/_private/test_utils.py

@@ -207,7 +207,7 @@ def kill_process_by_name(name, SIGKILL=False):
                 p.terminate()
 
 
-def run_string_as_driver(driver_script: str, env: Dict = None):
+def run_string_as_driver(driver_script: str, env: Dict = None, encode: str = "ascii"):
     """Run a driver as a separate process.
 
     Args:
@@ -235,13 +235,13 @@ def run_string_as_driver(driver_script: str, env: Dict = None):
         env=env,
     )
     with proc:
-        output = proc.communicate(driver_script.encode("ascii"))[0]
+        output = proc.communicate(driver_script.encode(encoding=encode))[0]
         if proc.returncode:
-            print(ray._private.utils.decode(output))
+            print(ray._private.utils.decode(output, encode_type=encode))
             raise subprocess.CalledProcessError(
                 proc.returncode, proc.args, output, proc.stderr
             )
-        out = ray._private.utils.decode(output)
+        out = ray._private.utils.decode(output, encode_type=encode)
     return out
 
 

python/ray/_private/utils.py

@@ -181,7 +181,7 @@ def random_string():
     return random_id
 
 
-def decode(byte_str, allow_none=False):
+def decode(byte_str: str, allow_none: bool = False, encode_type: str = "ascii"):
     """Make this unicode in Python 3, otherwise leave it as bytes.
 
     Args:
@@ -199,7 +199,7 @@ def decode(byte_str, allow_none=False):
     if not isinstance(byte_str, bytes):
         raise ValueError(f"The argument {byte_str} must be a bytes object.")
     if sys.version_info >= (3, 0):
-        return byte_str.decode("ascii")
+        return byte_str.decode(encode_type)
     else:
         return byte_str
 

python/ray/_raylet.pyx

@@ -501,7 +501,6 @@ cdef execute_task(
     worker = ray.worker.global_worker
     manager = worker.function_actor_manager
     actor = None
-
     cdef:
         dict execution_infos = manager.execution_infos
         CoreWorker core_worker = worker.core_worker

python/ray/worker.py

@@ -1163,7 +1163,6 @@ def shutdown(_exiting_interpreter: bool = False):
         # This is a duration to sleep before shutting down everything in order
         # to make sure that log messages finish printing.
         time.sleep(0.5)
-
     disconnect(_exiting_interpreter)
 
     # disconnect internal kv

release/long_running_tests/workloads/many_drivers.py

@@ -74,8 +74,11 @@ class Actor(object):
 for _ in range(5):
     for i in range(num_nodes):
         assert (ray.get(
-            f._remote(args=[], kwargs={{}}, resources={{str(i): 1}})) == 1)
-        actor = Actor._remote(args=[], kwargs={{}}, resources={{str(i): 1}})
+            f._remote(args=[],
+            kwargs={{}},
+            resources={{str(i): 1}})) == 1)
+        actor = Actor._remote(
+            args=[], kwargs={{}}, resources={{str(i): 1}})
         assert ray.get(actor.method.remote()) == 1
 
 # Tests datasets doesn't leak workers.
@@ -89,7 +92,7 @@ print("success")
 
 @ray.remote
 def run_driver():
-    output = run_string_as_driver(driver_script)
+    output = run_string_as_driver(driver_script, encode="utf-8")
     assert "success" in output