docstr: Create or retrieve an NCCL communicator for p2p tasks.
  
  Note(Hao): this function is not thread-safe now.
  
  Args:
      comm_key: communicator key.
      my_gpu_idx: the gpu index on the current process.
      peer_rank: the rank of the destination process.
      peer_gpu_idx: the gpu index on the peer process.
  Returns:
      communicator
docstr_tok: ["Create","or","retrieve","an","NCCL","communicator","for","p2p","tasks",".","Note","(","Hao",")",":","this","function","is","not","thread-safe","now",".","Args",":","comm_key",":","communicator","key",".","my_gpu_idx",":","the","gpu","index","on","the","current","process",".","peer_rank",":","the","rank","of","the","destination","process",".","peer_gpu_idx",":","the","gpu","index","on","the","peer","process",".","Returns",":","communicator"]
code: def _get_nccl_p2p_communicator(
      self, comm_key, my_gpu_idx, peer_rank, peer_gpu_idx
  ):
      """Create or retrieve an NCCL communicator for p2p tasks.
  
      Note(Hao): this function is not thread-safe now.
  
      Args:
          comm_key: communicator key.
          my_gpu_idx: the gpu index on the current process.
          peer_rank: the rank of the destination process.
          peer_gpu_idx: the gpu index on the peer process.
      Returns:
          communicator
      """
      if not comm_key:
          raise RuntimeError("Got empty communicator key.")
  
      # TODO(Hao): lock the _dev_comm_map here.
      if comm_key in self._dev_comm_map:
          return self._dev_comm_map[comm_key]
  
      # Note (Hao): This is a bit complex so I decide to take a note here.
      # Here we need to consider three cases:
      # Case 1: src_rank != dst_rank, hence the send and recv happen on
      # different process (actors/tasks); each process makes independent
      # collective calls and manages corresponding communicators.
      # Case 2: src_rank == dst_rank, src_gpu_idx == dst_gpu_idx; for
      # this case, we simply throw a RuntimeError;
      # Case 3: src_rank == dst_rank, src_gpu_idx != dst_gpu_idx, which
      # means the send and recv will be called on the same process. We
      # DO NOT support this case for now. We need to properly scope:
      # (1) communicators creation, and
      # (2) send/recv calls
      # using groupStart(（ and groupEnd() calls to avoid deadlocks.
      if self.rank < peer_rank:
          my_p2p_rank = 0
      elif self.rank > peer_rank:
          my_p2p_rank = 1
      else:
          raise RuntimeError(
              "Send and recv happens on the same process! "
              "ray.util.collective does not support this case as of now. "
              "Alternatively, consider doing GPU to GPU memcpy?"
          )
  
      group_key = self._generate_group_key(comm_key)
      if my_p2p_rank == 0:
          nccl_uid = self._generate_nccl_uid(group_key)
      else:
          rendezvous = Rendezvous(group_key)
          rendezvous.meet()
          nccl_uid = rendezvous.get_nccl_id()
  
      # create the p2p communicators
      with nccl_util.Device(my_gpu_idx):
          comm = nccl_util.create_nccl_communicator(
              2, nccl_uid, my_p2p_rank
          )
          stream = get_stream_pool(my_gpu_idx).get_stream()
          event = cupy.cuda.Event()
  
      # TODO(Fu): lock and might need to add event
      self._dev_comm_map[comm_key] = [comm]
      self._dev_streams_map[comm_key] = [stream]
      self._dev_event_map[comm_key] = [event]
      return [comm]
  
imports: import logging
  import datetime
  import time
  import ray
  import cupy
  from ray.util.collective.const import ENV
  from ray.util.collective.collective_group import nccl_util
  from ray.util.collective.collective_group.base_collective_group import BaseGroup
  from ray.util.collective.const import get_store_name
  from ray.util.collective.types import AllReduceOptions, BarrierOptions, Backend, ReduceOptions, BroadcastOptions, AllGatherOptions, ReduceScatterOptions, SendOptions, RecvOptions
  from ray.util.collective.collective_group.cuda_stream import get_stream_pool
label_desc: Use image node_id 14 for calling the NCCLGroup obj's underlying member method code with example usage: obj._get_nccl_p2p_communicator(comm_key, my_gpu_idx, peer_rank, peer_gpu_idx) and returns: list, self