code: def test_backpressure_pause_signal(shutdown_only):
      """Verify the signal can be caught while the main thread is blocked
      by a backpressure.
      """
  
      @ray.remote(
          _generator_backpressure_num_objects=1,
          max_retries=0,
      )
      def backpressure_task():
          for i in range(5):
              print("yield ", i)
              yield i
  
      gen = backpressure_task.remote()
      print(ray.get(next(gen)))
  
      def wait_for_task():
          t = list_tasks(filters=[("name", "=", "backpressure_task")])[
              0
          ]
          assert t.state == "RUNNING"
          return True
  
      wait_for_condition(wait_for_task)
  
      t = list_tasks(filters=[("name", "=", "backpressure_task")])[0]
      os.kill(t.worker_pid, signal.SIGTERM)
      with pytest.raises(ray.exceptions.WorkerCrashedError):
          ray.get(gen._generator_ref)
  
file_path: ray/python/ray/tests/test_streaming_generator_backpressure.py
imports: import asyncio
  import pytest
  import numpy
  import sys
  import time
  import signal
  import os
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.util.state import list_tasks
label_desc: Use image node_id 6 for calling a global function with example usage: test_backpressure_pause_signal(shutdown_only) and returns: True