class_docstr: Multi-node Multi-GPU NearestNeighbors Model.
  
  Parameters
  ----------
  n_neighbors : int (default=5)
      Default number of neighbors to query
  batch_size: int (optional, default 2000000)
      Maximum number of query rows processed at once. This parameter can
      greatly affect the throughput of the algorithm. The optimal setting
      of this value will vary for different layouts index to query ratios,
      but it will require `batch_size * n_features * 4` bytes of additional
      memory on each worker hosting index partitions.
  handle : cuml.Handle
      Specifies the cuml.handle that holds internal CUDA state for
      computations in this model. Most importantly, this specifies the CUDA
      stream that will be used for the model's computations, so users can
      run different models concurrently in different streams by creating
      handles in several streams.
      If it is None, a new one is created.
  verbose : int or boolean, default=False
      Sets logging level. It must be one of `cuml.common.logger.level_*`.
      See :ref:`verbosity-levels` for more info.
class_docstr_tok: ["Multi-node","Multi-GPU","NearestNeighbors","Model",".","Parameters","--","--","--","--","--","n_neighbors",":","int","(","default=5",")","Default","number","of","neighbors","to","query","batch_size",":","int","(","optional",",","default","2000000",")","Maximum","number","of","query","rows","processed","at","once",".","This","parameter","can","greatly","affect","the","throughput","of","the","algorithm",".","The","optimal","setting","of","this","value","will","vary","for","different","layouts","index","to","query","ratios",",","but","it","will","require","`","batch_size","*","n_features","*","4","`","bytes","of","additional","memory","on","each","worker","hosting","index","partitions",".","handle",":","cuml.Handle","Specifies","the","cuml.handle","that","holds","internal","CUDA","state","for","computations","in","this","model",".","Most","importantly",",","this","specifies","the","CUDA","stream","that","will","be","used","for","the","model","'s","computations",",","so","users","can","run","different","models","concurrently","in","different","streams","by","creating","handles","in","several","streams",".","If","it","is","None",",","a","new","one","is","created",".","verbose",":","int","or","boolean",",","default=False","Sets","logging","level",".","It","must","be","one","of","`","cuml.common.logger.level_","*","`",".","See",":","ref",":","`","verbosity-levels","`","for","more","info","."]
code: def _build_comms(index_handler, query_handler, streams_per_handle):
      # Communicator clique needs to include the union of workers hosting
      # query and index partitions
      workers = set(index_handler.workers)
      workers.update(query_handler.workers)
  
      comms = Comms(
          comms_p2p=True, streams_per_handle=streams_per_handle
      )
      comms.init(workers=workers)
      return comms
  
imports: from cuml.dask.common import parts_to_ranks
  from cuml.dask.common.utils import wait_and_raise_from_futures
  from cuml.dask.common import flatten_grouped_results
  from cuml.dask.common import raise_mg_import_exception
  from cuml.dask.common.base import BaseEstimator
  from dask.distributed import get_worker
  from raft_dask.common.comms import get_raft_comm_state
  from raft_dask.common.comms import Comms
  from cuml.dask.common.input_utils import to_output
  from cuml.dask.common.input_utils import DistributedDataHandler
  from uuid import uuid1
label_desc: Use image node_id 5 for calling the NearestNeighbors obj's underlying member method code with example usage: obj._build_comms(index_handler, query_handler, streams_per_handle) and returns: comms