code: def test_delete_refs_on_disconnect(ray_start_cluster):
      cluster = ray_start_cluster
      with ray_start_cluster_client_server_pair(
          cluster.address
      ) as pair:
          ray, server = pair
  
          @ray.remote
          def f(x):
              return x + 2
  
          thing1 = f.remote(6)  # noqa
          thing2 = ray.put("Hello World")  # noqa
  
          # One put, one function -- the function result thing1 is
          # in a different category, according to the raylet.
          # But we're maintaining the reference
          assert server_object_ref_count(server, 3)()
          # And can get the data
          assert ray.get(thing1) == 8
  
          # Close the client.
          ray.close()
  
          wait_for_condition(
              server_object_ref_count(server, 0), timeout=5
          )
  
          # Connect to the real ray again, since we disconnected
          # upon num_clients = 0.
          real_ray.init(
              address=cluster.address,
              namespace="default_test_namespace",
          )
  
          def test_cond():
              return object_memory_usage() == 0
  
          wait_for_condition(test_cond, timeout=5)
  
file_path: ray/python/ray/tests/test_client_references.py
imports: from concurrent.futures import Future
  import pytest
  import ray
  from ray._private.test_utils import object_memory_usage, wait_for_condition
  from ray._raylet import ActorID, ObjectRef
  from ray.util.client import _ClientContext
  from ray.util.client.common import ClientActorRef, ClientObjectRef
  from ray.util.client.ray_client_helpers import ray_start_client_server, ray_start_client_server_pair, ray_start_cluster_client_server_pair
label_desc: Use image node_id 5 for calling a global function with example usage: test_delete_refs_on_disconnect(ray_start_cluster) and returns: unknown, unknown