code: def get_local_state_client():
      hostname = ray.worker._global_node.gcs_address
  
      gcs_channel = ray._private.utils.init_grpc_channel(
          hostname, ray_constants.GLOBAL_GRPC_OPTIONS, asynchronous=True
      )
  
      gcs_aio_client = gcs_utils.GcsAioClient(
          address=hostname, nums_reconnect_retry=0
      )
      client = StateDataSourceClient(gcs_channel, gcs_aio_client)
      for node in ray.nodes():
          node_id = node["NodeID"]
          ip = node["NodeManagerAddress"]
          port = int(node["NodeManagerPort"])
          runtime_env_agent_port = int(node["RuntimeEnvAgentPort"])
          client.register_raylet_client(
              node_id, ip, port, runtime_env_agent_port
          )
          client.register_agent_client(node_id, ip, port)
  
      return client
  
imports: from math import ceil
  import sys
  import time
  import pytest
  import ray
  from ray._private import ray_constants
  import ray._private.gcs_utils
  from ray._private.test_utils import wait_for_condition, raw_metrics
  import numpy
  from ray._private.utils import get_system_memory
  from ray._private.utils import get_used_memory
  from ray._private.state_api_test_utils import verify_failed_task
  from ray.util.state.state_manager import StateDataSourceClient
label_desc: Use image node_id 1 for calling a global function with example usage: get_local_state_client() and returns: client