code: def get_memory_info_reply(
      state, node_manager_address=None, node_manager_port=None
  ):
      """Returns global memory info."""
  
      from ray.core.generated import (
          node_manager_pb2,
          node_manager_pb2_grpc,
      )
  
      # We can ask any Raylet for the global memory info, that Raylet internally
      # asks all nodes in the cluster for memory stats.
      if node_manager_address is None or node_manager_port is None:
          # We should ask for a raylet that is alive.
          raylet = None
          for node in state.node_table():
              if node["Alive"]:
                  raylet = node
                  break
          assert raylet is not None, "Every raylet is dead"
          raylet_address = "{}:{}".format(
              raylet["NodeManagerAddress"], raylet["NodeManagerPort"]
          )
      else:
          raylet_address = "{}:{}".format(
              node_manager_address, node_manager_port
          )
  
      channel = utils.init_grpc_channel(
          raylet_address,
          options=[
              ("grpc.max_send_message_length", MAX_MESSAGE_LENGTH),
              ("grpc.max_receive_message_length", MAX_MESSAGE_LENGTH),
          ],
      )
  
      stub = node_manager_pb2_grpc.NodeManagerServiceStub(channel)
      reply = stub.FormatGlobalMemoryInfo(
          node_manager_pb2.FormatGlobalMemoryInfoRequest(
              include_memory_info=False
          ),
          timeout=60.0,
      )
      return reply
  
file_path: ray/python/ray/_private/internal_api.py
imports: import ray
  import ray._private.profiling
  import ray._private.services
  import ray._private.utils
  import ray._private.worker
  from ray._private import ray_constants
  from ray._private.state import GlobalState
  from ray._raylet import GcsClientOptions
label_desc: Use image node_id 4 for calling a global function with example usage: get_memory_info_reply(state, node_manager_address, node_manager_port) and returns: reply