code: def test_core_worker_resubscription(
      tmp_path, ray_start_regular_with_external_redis
  ):
      # This test is to ensure core worker will resubscribe to GCS after GCS
      # restarts.
      from filelock import FileLock
  
      lock_file = str(tmp_path / "lock")
      lock = FileLock(lock_file)
      lock.acquire()
  
      @ray.remote
      class Actor:
          def __init__(self):
              lock = FileLock(lock_file)
              lock.acquire()
  
          def ready(self):
              return
  
      a = Actor.remote()
      r = a.ready.remote()
      # Actor is not ready before GCS is down.
      ray._private.worker._global_node.kill_gcs_server()
  
      lock.release()
      # Actor is ready after GCS starts
      ray._private.worker._global_node.start_gcs_server()
      # Test the resubscribe works: if not, it'll timeout because worker
      # will think the actor is not ready.
      ray.get(r, timeout=5)
  
imports: import sys
  import os
  import threading
  from time import sleep
  import pytest
  import ray
  from ray._private.utils import get_or_create_event_loop
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  import ray._private.gcs_utils
  from ray._private import ray_constants
  from ray._private.test_utils import convert_actor_state, enable_external_redis, generate_system_config_map, wait_for_condition, wait_for_pid_to_exit, run_string_as_driver
  from ray.job_submission import JobSubmissionClient, JobStatus
  from ray._raylet import GcsClient
  import psutil
label_desc: Use image node_id 9 for calling a global function with example usage: test_core_worker_resubscription(tmp_path, ray_start_regular_with_external_redis) without return types