code: def test_dying_worker_get(ray_start_2_cpus):
      @ray.remote
      def sleep_forever(signal):
          ray.get(signal.send.remote())
          time.sleep(10**6)
  
      @ray.remote
      def get_worker_pid():
          return os.getpid()
  
      signal = SignalActor.remote()
  
      x_id = sleep_forever.remote(signal)
      ray.get(signal.wait.remote())
      # Get the PID of the other worker.
      worker_pid = ray.get(get_worker_pid.remote())
  
      @ray.remote
      def f(id_in_a_list):
          ray.get(id_in_a_list[0])
  
      # Have the worker wait in a get call.
      result_id = f.remote([x_id])
      time.sleep(1)
  
      # Make sure the task hasn't finished.
      ready_ids, _ = ray.wait([result_id], timeout=0)
      assert len(ready_ids) == 0
  
      # Kill the worker.
      os.kill(worker_pid, SIGKILL)
      time.sleep(0.1)
  
      # Make sure the sleep task hasn't finished.
      ready_ids, _ = ray.wait([x_id], timeout=0)
      assert len(ready_ids) == 0
      # Seal the object so the store attempts to notify the worker that the
      # get has been fulfilled.
      obj = np.ones(200 * 1024, dtype=np.uint8)
      ray._private.worker.global_worker.put_object(obj, x_id)
      time.sleep(0.1)
  
      # Make sure that nothing has died.
      assert ray._private.services.remaining_processes_alive()
  
file_path: ray/python/ray/tests/test_component_failures.py
imports: import os
  import signal
  import sys
  import time
  import numpy
  import pytest
  import ray
  from ray._private.test_utils import SignalActor, run_string_as_driver_nonblocking
label_desc: Use image node_id 1 for calling a global function with example usage: test_dying_worker_get(ray_start_2_cpus) and returns: os