code: def test_spill_worker_failure(ray_start_regular):
      def run_workload():
          @ray.remote
          def f():
              return np.zeros(50 * 1024 * 1024, dtype=np.uint8)
  
          ids = []
          for _ in range(5):
              x = f.remote()
              ids.append(x)
          for id in ids:
              ray.get(id)
          del ids
  
      run_workload()
  
      def get_spill_worker():
          import psutil
  
          for proc in psutil.process_iter():
              try:
                  name = (
                      ray._private.ray_constants.WORKER_PROCESS_TYPE_SPILL_WORKER_IDLE
                  )
                  if name in proc.name():
                      return proc
                  # for macOS
                  if proc.cmdline() and name in proc.cmdline()[0]:
                      return proc
                  # for Windows
                  if (
                      proc.cmdline()
                      and "--worker-type=SPILL_WORKER" in proc.cmdline()
                  ):
                      return proc
              except psutil.AccessDenied:
                  pass
              except psutil.NoSuchProcess:
                  pass
  
      # Spilling occurred. Get the PID of the spill worker.
      spill_worker_proc = get_spill_worker()
      assert spill_worker_proc
  
      # Kill the spill worker
      spill_worker_proc.kill()
      spill_worker_proc.wait()
  
      # Now we trigger spilling again
      run_workload()
  
      # A new spill worker should be created
      spill_worker_proc = get_spill_worker()
      assert spill_worker_proc
  
imports: import copy
  import json
  import platform
  import random
  import sys
  from datetime import datetime, timedelta
  from unittest.mock import patch
  import numpy
  import pytest
  import ray
  from ray._private.external_storage import create_url_with_offset, parse_url_with_offset, _get_unique_spill_filename, FileSystemStorage, ExternalStorageSmartOpenImpl
  from ray._private.internal_api import memory_summary
  from ray._private.test_utils import wait_for_condition
  from ray._raylet import GcsClientOptions
  from ray.tests.conftest import buffer_object_spilling_config, file_system_object_spilling_config, mock_distributed_fs_object_spilling_config
label_desc: Use image node_id 16 for calling a global function with example usage: test_spill_worker_failure(ray_start_regular) and returns: np, proc, proc, proc