code: def test_nested_fork(setup_queue_actor):
      queue = setup_queue_actor
  
      @ray.remote
      def fork(queue, key, num_items):
          x = None
          for item in range(num_items):
              x = queue.enqueue.remote(key, item)
          return ray.get(x)
  
      @ray.remote
      def nested_fork(queue, key, num_items):
          # Pass the actor into a nested task.
          ray.get(fork.remote(queue, key + 1, num_items))
          x = None
          for item in range(num_items):
              x = queue.enqueue.remote(key, item)
          return ray.get(x)
  
      # Fork num_iters times.
      num_forks = 10
      num_items_per_fork = 100
  
      # Submit some tasks on new actor handles.
      forks = [
          nested_fork.remote(queue, i, num_items_per_fork)
          for i in range(0, num_forks, 2)
      ]
      ray.get(forks)
      # Check that all tasks from all handles have completed.
      items = ray.get(queue.read.remote())
      for i in range(num_forks):
          filtered_items = [item[1] for item in items if item[0] == i]
          assert filtered_items == list(range(num_items_per_fork))
  
imports: import os
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.gcs_utils
  from ray.util.state import list_actors
  import ray.cluster_utils
  from ray._private.test_utils import SignalActor, convert_actor_state, kill_actor_and_wait_for_failure, make_global_state_accessor, run_string_as_driver, wait_for_condition, wait_for_pid_to_exit
  from ray._private.ray_constants import gcs_actor_scheduling_enabled
  from ray.experimental.internal_kv import _internal_kv_get, _internal_kv_put
label_desc: Use image node_id 13 for calling a global function with example usage: test_nested_fork(setup_queue_actor) and returns: ray, ray