code: def test_worker_failed(ray_start_workers_separate_multinode):
      (
          num_nodes,
          num_initial_workers,
      ) = ray_start_workers_separate_multinode
  
      block_worker = Semaphore.remote(0)
      block_driver = Semaphore.remote(0)
      ray.get(
          [block_worker.locked.remote(), block_driver.locked.remote()]
      )
  
      # Acquire a custom resource that isn't released on `ray.get` to make sure
      # this task gets spread across all the nodes.
      @ray.remote(num_cpus=1, resources={"custom": 1})
      def get_pids():
          ray.get(block_driver.release.remote())
          ray.get(block_worker.acquire.remote())
          return os.getpid()
  
      total_num_workers = num_nodes * num_initial_workers
      pid_refs = [get_pids.remote() for _ in range(total_num_workers)]
      ray.get(
          [
              block_driver.acquire.remote()
              for _ in range(total_num_workers)
          ]
      )
      ray.get(
          [
              block_worker.release.remote()
              for _ in range(total_num_workers)
          ]
      )
  
      pids = set(ray.get(pid_refs))
  
      @ray.remote
      def f(x):
          time.sleep(0.5)
          return x
  
      # Submit more tasks than there are workers so that all workers and
      # cores are utilized.
      object_refs = [
          f.remote(i) for i in range(num_initial_workers * num_nodes)
      ]
      object_refs += [
          f.remote(object_ref) for object_ref in object_refs
      ]
      # Allow the tasks some time to begin executing.
      time.sleep(0.1)
      # Kill the workers as the tasks execute.
      for pid in pids:
          try:
              os.kill(pid, SIGKILL)
          except OSError:
              # The process may have already exited due to worker capping.
              pass
          time.sleep(0.1)
      # Make sure that we either get the object or we get an appropriate
      # exception.
      for object_ref in object_refs:
          try:
              ray.get(object_ref)
          except (
              ray.exceptions.RayTaskError,
              ray.exceptions.WorkerCrashedError,
          ):
              pass
  
file_path: ray/python/ray/tests/test_multinode_failures.py
imports: import os
  import signal
  import sys
  import time
  import pytest
  import ray
  import ray._private.ray_constants
  from ray._private.test_utils import Semaphore, get_other_nodes
  from ray.cluster_utils import Cluster, cluster_not_supported
label_desc: Use image node_id 2 for calling a global function with example usage: test_worker_failed(ray_start_workers_separate_multinode) and returns: os, x