code: def rpc_async(
      to, func, args=None, kwargs=None, timeout=UNSET_RPC_TIMEOUT
  ):
      r"""
      Make a non-blocking RPC call to run function ``func`` on worker ``to``. RPC
      messages are sent and received in parallel to execution of Python code. This
      method is thread-safe. This method will immediately return a
      :class:`~torch.futures.Future` that can be awaited on.
  
      Args:
          to (str or WorkerInfo or int): name/rank/``WorkerInfo`` of the destination worker.
          func (Callable): a callable function, such as Python callables, builtin
                           operators (e.g. :meth:`~torch.add`) and annotated
                           TorchScript functions.
          args (tuple): the argument tuple for the ``func`` invocation.
          kwargs (dict): is a dictionary of keyword arguments for the ``func``
                         invocation.
          timeout (float, optional): timeout in seconds to use for this RPC. If
                                     the RPC does not complete in this amount of
                                     time, an exception indicating it has
                                     timed out will be raised. A value of 0
                                     indicates an infinite timeout, i.e. a timeout
                                     error will never be raised. If not provided,
                                     the default value set during initialization
                                     or with ``_set_rpc_timeout`` is used.
  
  
      Returns:
          Returns a :class:`~torch.futures.Future` object that can be waited
          on. When completed, the return value of ``func`` on ``args`` and
          ``kwargs`` can be retrieved from the :class:`~torch.futures.Future`
          object.
  
      .. warning ::
          Using GPU tensors as arguments or return values of ``func`` is not
          supported since we don't support sending GPU tensors over the wire. You
          need to explicitly copy GPU tensors to CPU before using them as
          arguments or return values of ``func``.
  
      .. warning ::
          The ``rpc_async`` API does not copy storages of argument tensors until
          sending them over the wire, which could be done by a different thread
          depending on the RPC backend type. The caller should make sure that the
          contents of those tensors stay intact until the returned
          :class:`~torch.futures.Future` completes.
  
      Example::
          Make sure that ``MASTER_ADDR`` and ``MASTER_PORT`` are set properly
          on both workers. Refer to :meth:`~torch.distributed.init_process_group`
          API for more details. For example,
  
          export MASTER_ADDR=localhost
          export MASTER_PORT=5678
  
          Then run the following code in two different processes:
  
          >>> # xdoctest: +SKIP
          >>> # On worker 0:
          >>> import torch
          >>> import torch.distributed.rpc as rpc
          >>> rpc.init_rpc("worker0", rank=0, world_size=2)
          >>> fut1 = rpc.rpc_async("worker1", torch.add, args=(torch.ones(2), 3))
          >>> fut2 = rpc.rpc_async("worker1", min, args=(1, 2))
          >>> result = fut1.wait() + fut2.wait()
          >>> rpc.shutdown()
  
          >>> # On worker 1:
          >>> import torch.distributed.rpc as rpc
          >>> rpc.init_rpc("worker1", rank=1, world_size=2)
          >>> rpc.shutdown()
  
          Below is an example of running a TorchScript function using RPC.
  
          >>> # On both workers:
          >>> @torch.jit.script
          >>> def my_script_add(tensor: torch.Tensor, scalar: int):
          >>>    return torch.add(tensor, scalar)
  
          >>> # On worker 0:
          >>> import torch.distributed.rpc as rpc
          >>> rpc.init_rpc("worker0", rank=0, world_size=2)
          >>> fut = rpc.rpc_async("worker1", my_script_add, args=(torch.ones(2), 3))
          >>> ret = fut.wait()
          >>> rpc.shutdown()
  
          >>> # On worker 1:
          >>> import torch.distributed.rpc as rpc
          >>> rpc.init_rpc("worker1", rank=1, world_size=2)
          >>> rpc.shutdown()
      """
      torch._C._log_api_usage_once("torch.distributed.rpc_async")
      fut = _invoke_rpc(
          to, func, RPCExecMode.ASYNC, args, kwargs, timeout
      )
      if hasattr(_thread_local_var, "future_list"):
          _thread_local_var.future_list.append(fut)
      return fut
  
imports: import collections
  import contextlib
  import functools
  import inspect
  import logging
  import threading
  from typing import Dict, Generic, TypeVar, Set, Any, TYPE_CHECKING
  import torch
  from torch.futures import Future
  from torch._C._distributed_rpc import PyRRef, RemoteProfilerManager, WorkerInfo, TensorPipeAgent, get_rpc_timeout, _cleanup_python_rpc_handler, _delete_all_user_and_unforked_owner_rrefs, _destroy_rref_context, _get_current_rpc_agent, _invoke_remote_builtin, _invoke_remote_python_udf, _invoke_remote_torchscript, _invoke_rpc_builtin, _invoke_rpc_python_udf, _invoke_rpc_torchscript, _is_current_rpc_agent_set, _reset_current_rpc_agent, _set_and_start_rpc_agent
  from .internal import PythonUDF, RPCExecMode, _internal_rpc_pickler, _build_rpc_profiling_key
  from .constants import DEFAULT_SHUTDOWN_TIMEOUT, UNSET_RPC_TIMEOUT
  from ._utils import _group_membership_management, _update_group_membership
label_desc: Use image node_id 20 for calling a global function with example usage: rpc_async(to, func, args, kwargs, timeout) and returns: fut