code: def test_gcs_fd_usage(shutdown_only):
      ray.init(
          _system_config={
              "prestart_worker_first_driver": False,
              "enable_worker_prestart": False,
          },
      )
      gcs_process = ray._private.worker._global_node.all_processes[
          "gcs_server"
      ][0]
      gcs_process = psutil.Process(gcs_process.process.pid)
      print("GCS connections", len(gcs_process.connections()))
  
      @ray.remote(runtime_env={"env_vars": {"Hello": "World"}})
      class A:
          def f(self):
              return os.environ.get("Hello")
  
      # In case there are still some pre-start workers, consume all of them
      aa = [A.remote() for _ in range(32)]
      for a in aa:
          assert ray.get(a.f.remote()) == "World"
      base_fd_num = len(gcs_process.connections())
      print("GCS connections", base_fd_num)
  
      bb = [A.remote() for _ in range(16)]
      for b in bb:
          assert ray.get(b.f.remote()) == "World"
      new_fd_num = len(gcs_process.connections())
      print("GCS connections", new_fd_num)
      # each worker has two connections:
      #   GCS -> CoreWorker
      #   CoreWorker -> GCS
      # Sometimes, there is one more sockets opened. The reason
      # is still unknown.
      assert (new_fd_num - base_fd_num) <= len(bb) * 2 + 1
  
imports: import os
  import sys
  import time
  import pytest
  import ray
  import ray._private.ray_constants
  from ray._private.test_utils import Semaphore, enable_external_redis, client_test_enabled, run_string_as_driver, wait_for_condition, get_gcs_memory_used, run_string_as_driver_nonblocking
  from ray.experimental.internal_kv import _internal_kv_list
  from ray.tests.conftest import call_ray_start
  import subprocess
  import psutil
label_desc: Use image node_id 16 for calling a global function with example usage: test_gcs_fd_usage(shutdown_only) and returns: os