code: def test_multi_drivers(shutdown_only):
      info = ray.init(num_cpus=10)
  
      driver_code = """
  import os
  import sys
  import ray
  
  
  ray.init(address="{}")
  
  @ray.remote
  class Actor:
      def get_pid(self):
          return os.getpid()
  
  @ray.remote
  def get_pid():
      return os.getpid()
  
  pid_objs = []
  # Submit some normal tasks and get the PIDs of workers which execute the tasks.
  pid_objs = pid_objs + [get_pid.remote() for _ in range(2)]
  # Create some actors and get the PIDs of actors.
  actors = [Actor.remote() for _ in range(2)]
  pid_objs = pid_objs + [actor.get_pid.remote() for actor in actors]
  
  pids = set([ray.get(obj) for obj in pid_objs])
  # Write pids to stdout
  print("PID:" + str.join(",", [str(_) for _ in pids]))
  
  ray.shutdown()
      """.format(
          info["address"]
      )
  
      driver_count = 3
      processes = [
          run_string_as_driver_nonblocking(driver_code)
          for _ in range(driver_count)
      ]
      outputs = []
      for p in processes:
          out = p.stdout.read().decode("ascii")
          err = p.stderr.read().decode("ascii")
          p.wait()
          # out, err = p.communicate()
          # out = ray._private.utils.decode(out)
          # err = ray._private.utils.decode(err)
          if p.returncode != 0:
              print(
                  "Driver with PID {} returned error code {}".format(
                      p.pid, p.returncode
                  )
              )
              print("STDOUT:\n{}".format(out))
              print("STDERR:\n{}".format(err))
          outputs.append((p, out))
  
      all_worker_pids = set()
      for p, out in outputs:
          assert p.returncode == 0
          for line in out.splitlines():
              if line.startswith("PID:"):
                  worker_pids = [
                      int(_) for _ in line.split(":")[1].split(",")
                  ]
                  assert len(worker_pids) > 0
                  for worker_pid in worker_pids:
                      assert worker_pid not in all_worker_pids, (
                          "Worker process with PID {} is shared"
                          + " by multiple drivers."
                      ).format(worker_pid)
                      all_worker_pids.add(worker_pid)
  
file_path: ray/python/ray/tests/test_multi_tenancy.py
imports: import os
  import sys
  import time
  import pytest
  import numpy
  import ray
  from ray.core.generated import common_pb2
  from ray.core.generated import node_manager_pb2, node_manager_pb2_grpc
  from ray._private.test_utils import wait_for_condition, run_string_as_driver, run_string_as_driver_nonblocking
  from ray._private.utils import init_grpc_channel
label_desc: Use image node_id 3 for calling a global function with example usage: test_multi_drivers(shutdown_only) without return types