class_docstr: A Tensor wrapper subclass that is used to trigger a call to wait
  prior to first use of the underlying tensor.
  Use it inside functional collective pytorch wrappers like the following:
  def functional_collective(self, group, tag):
      tag, rankset, group_size = _expand_group(group, tag)
      tensor = torch.ops.c10d_functional.{collective}(self, tag, rankset, group_size)
      return _maybe_wrap_tensor(tensor)
class_docstr_tok: ["A","Tensor","wrapper","subclass","that","is","used","to","trigger","a","call","to","wait","prior","to","first","use","of","the","underlying","tensor",".","Use","it","inside","functional","collective","pytorch","wrappers","like","the","following",":","def","functional_collective","(","self",",","group",",","tag",")",":","tag",",","rankset",",","group_size","=","_expand_group","(","group",",","tag",")","tensor","=","torch.ops.c10d_functional",".","{","collective","}","(","self",",","tag",",","rankset",",","group_size",")","return","_maybe_wrap_tensor","(","tensor",")"]
code: def __torch_dispatch__(cls, func, types, args=(), kwargs=None):
      if func == torch.ops.aten.view.default:
          # Fast handle aten.view as a lot of view related op goes to aten.view
          # eventually, this avoids pytree slowdown
          res = func(args[0].elem, args[1])
          wrapper_res = AsyncCollectiveTensor(res)
          _register_tensor_wrapper(wrapper_res)
          return wrapper_res
  
      is_view_op = _is_view_op(func)
  
      def unwrap(e: AsyncCollectiveTensor):
          # wait_tensor is idepotent and will do stream sync only once
          if not is_view_op:
              wait_tensor(e.elem)
          return e.elem
  
      def wrap(e: torch.Tensor):
          # wait_tensor is idepotent and will do stream sync only once
          assert not isinstance(e, AsyncCollectiveTensor)
          res = AsyncCollectiveTensor(e)
          _register_tensor_wrapper(res)
          return res
  
      unwrapped_args = tree_map_only(
          AsyncCollectiveTensor, unwrap, args
      )
      unwrapped_kwargs = tree_map_only(
          AsyncCollectiveTensor, unwrap, kwargs
      )
  
      # we don't wrap the result as it doesn't need to be waited on.
      out = func(*unwrapped_args, **unwrapped_kwargs)
  
      # View ops dont require a sync, so we should re-wrap the outputs.
      if is_view_op:
          out = tree_map_only(torch.Tensor, wrap, out)
  
      return out
  
imports: import sys
  import warnings
  from typing import cast, List, Optional, Tuple, TYPE_CHECKING, Union
  import torch
  import torch.distributed
  import torch.distributed.distributed_c10d
  from torch._custom_ops import impl_abstract
  from torch.distributed.device_mesh import DeviceMesh
  from torch.fx.experimental.proxy_tensor import get_innermost_proxy_mode
  from .None import _functional_collectives_impl
  from ._functional_collectives_impl import _register_tensor_wrapper
  from torch.distributed.distributed_c10d import all_gather_into_tensor, all_reduce, reduce_scatter_tensor
label_desc: Use image node_id 9 for calling the AsyncCollectiveTensor obj's underlying member method code with example usage: obj.__torch_dispatch__(cls, func, types, args, kwargs) and returns: out, wrapper_res, e, res