code: def test_not_killing_workers_that_own_objects(shutdown_only):
      # Set the small interval for worker capping
      # so that we can easily trigger it.
      ray.init(
          num_cpus=1,
          _system_config={
              "kill_idle_workers_interval_ms": 10,
              "worker_lease_timeout_milliseconds": 0,
          },
      )
  
      expected_num_workers = 6
      # Create a nested tasks to start 8 workers each of which owns an object.
  
      @ray.remote
      def nested(i):
          # The task owns an object.
          if i >= expected_num_workers - 1:
              return [ray.put(np.ones(1 * 1024 * 1024, dtype=np.uint8))]
          else:
              return [
                  ray.put(np.ones(1 * 1024 * 1024, dtype=np.uint8))
              ] + ray.get(nested.remote(i + 1))
  
      ref = ray.get(nested.remote(0))
      num_workers = len(get_workers())
  
      # Wait for worker capping. worker capping should be triggered
      # every 10 ms, but we wait long enough to avoid a flaky test.
      time.sleep(1)
      ref2 = ray.get(nested.remote(0))
  
      # New workers shouldn't be registered because we reused the
      # previous workers that own objects.
      cur_num_workers = len(get_workers())
      # TODO(ekl) ideally these would be exactly equal, however the test is
      # occasionally flaky with that check.
      assert abs(num_workers - cur_num_workers) < 2, (
          num_workers,
          cur_num_workers,
      )
      assert len(ref2) == expected_num_workers
      assert len(ref) == expected_num_workers
  
file_path: ray/python/ray/tests/test_multi_tenancy.py
imports: import os
  import sys
  import time
  import pytest
  import numpy
  import ray
  from ray.core.generated import common_pb2
  from ray.core.generated import node_manager_pb2, node_manager_pb2_grpc
  from ray._private.test_utils import wait_for_condition, run_string_as_driver, run_string_as_driver_nonblocking
  from ray._private.utils import init_grpc_channel
label_desc: Use image node_id 9 for calling a global function with example usage: test_not_killing_workers_that_own_objects(shutdown_only) and returns: list, unknown