code: def initialize(
      comms=None, p2p=False, prows=None, pcols=None, partition_type=1
  ):
      """
      Initialize a communicator for multi-node/multi-gpu communications.  It is
      expected to be called right after client initialization for running
      multi-GPU algorithms (this wraps raft comms that manages underlying NCCL
      and UCX comms handles across the workers of a Dask cluster).
  
      It is recommended to also call `destroy()` when the comms are no longer
      needed so the underlying resources can be cleaned up.
  
      Parameters
      ----------
      comms : raft Comms, optional (default=None)
          A pre-initialized raft communicator. If provided, this is used for mnmg
          communications. If not provided, default comms are initialized as per
          client information.
  
      p2p : bool, optional (default=False)
          Initialize UCX endpoints if True.
  
      prows : int, optional (default=None)
          Specifies the number of rows when performing a 2D partitioning of the
          input graph. If specified, this must be a factor of the total number of
          parallel processes. When specified with pcols, prows*pcols should be
          equal to the total number of parallel processes.
  
      pcols : int, optional (default=None)
          Specifies the number of columns when performing a 2D partitioning of
          the input graph. If specified, this must be a factor of the total
          number of parallel processes. When specified with prows, prows*pcols
          should be equal to the total number of parallel processes.
  
      partition_type : int, optional (default=1)
          Valid values are currently 1 or any int other than 1. A value of 1 (the
          default) represents a partitioning resulting in prows*pcols
          partitions. A non-1 value currently results in a partitioning of
          p*pcols partitions, where p is the number of GPUs.
  
      Examples
      --------
      >>> from dask.distributed import Client
      >>> from dask_cuda import LocalCUDACluster
      >>> import cugraph.dask.comms as Comms
      >>> cluster = LocalCUDACluster()
      >>> client = Client(cluster)
      >>> Comms.initialize(p2p=True)
      >>> # DO WORK HERE
      >>> # All done, clean up
      >>> Comms.destroy()
      >>> client.close()
      >>> cluster.close()
  
      """
  
      global __instance
      if __instance is None:
          global __default_handle
          __default_handle = None
          if comms is None:
              # Initialize communicator
              if not p2p:
                  raise Exception(
                      "Set p2p to True for running mnmg algorithms"
                  )
              __instance = raftComms(comms_p2p=p2p)
              __instance.init()
              # Initialize subcommunicator
              subcomm_init(prows, pcols, partition_type)
          else:
              __instance = comms
      else:
          raise Exception("Communicator is already initialized")
  
file_path: cugraph/python/cugraph/cugraph/dask/comms/comms.py
imports: from cugraph.dask.common.read_utils import MissingUCXPy
  from pylibraft.common.handle import Handle
  from cugraph.dask.comms.comms_wrapper import init_subcomms
  from dask.distributed import default_client, get_worker
  from cugraph.dask.common import read_utils
  import math
label_desc: Use image node_id 4 for calling a global function with example usage: initialize(comms, p2p, prows, pcols, partition_type) without return types