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","."]
docstr: Query the distributed nearest neighbors index
  
  Parameters
  ----------
  X : dask_cudf.Dataframe
      Vectors to query. If not provided, neighbors of each indexed point
      are returned.
  n_neighbors : int
      Number of neighbors to query for each row in X. If not provided,
      the n_neighbors on the model are used.
  return_distance : boolean (default=True)
      If false, only indices are returned
  
  Returns
  -------
  ret : tuple (dask_cudf.DataFrame, dask_cudf.DataFrame)
      First dask-cuDF DataFrame contains distances, second contains the
      indices.
docstr_tok: ["Query","the","distributed","nearest","neighbors","index","Parameters","--","--","--","--","--","X",":","dask_cudf.Dataframe","Vectors","to","query",".","If","not","provided",",","neighbors","of","each","indexed","point","are","returned",".","n_neighbors",":","int","Number","of","neighbors","to","query","for","each","row","in","X",".","If","not","provided",",","the","n_neighbors","on","the","model","are","used",".","return_distance",":","boolean","(","default=True",")","If","false",",","only","indices","are","returned","Returns","--","--","--","-","ret",":","tuple","(","dask_cudf.DataFrame",",","dask_cudf.DataFrame",")","First","dask-cuDF","DataFrame","contains","distances",",","second","contains","the","indices","."]
code: def kneighbors(
      self,
      X=None,
      n_neighbors=None,
      return_distance=True,
      _return_futures=False,
  ):
      """
      Query the distributed nearest neighbors index
  
      Parameters
      ----------
      X : dask_cudf.Dataframe
          Vectors to query. If not provided, neighbors of each indexed point
          are returned.
      n_neighbors : int
          Number of neighbors to query for each row in X. If not provided,
          the n_neighbors on the model are used.
      return_distance : boolean (default=True)
          If false, only indices are returned
  
      Returns
      -------
      ret : tuple (dask_cudf.DataFrame, dask_cudf.DataFrame)
          First dask-cuDF DataFrame contains distances, second contains the
          indices.
      """
      n_neighbors = self.get_neighbors(n_neighbors)
  
      query_handler = (
          self.X_handler
          if X is None
          else DistributedDataHandler.create(data=X, client=self.client)
      )
  
      if query_handler is None:
          raise ValueError(
              "Model needs to be trained using fit() "
              "before calling kneighbors()"
          )
  
      """
          Create communicator clique
          """
      comms = NearestNeighbors._build_comms(
          self.X_handler, query_handler, self.streams_per_handle
      )
  
      """
          Initialize models on workers
          """
      nn_models = self._create_models(comms)
  
      """
          Perform model query
          """
      nn_fit, out_d_futures, out_i_futures = self._query_models(
          n_neighbors, comms, nn_models, self.X_handler, query_handler
      )
  
      comms.destroy()
  
      if _return_futures:
          ret = nn_fit, out_i_futures if not return_distance else (
              nn_fit,
              out_d_futures,
              out_i_futures,
          )
      else:
          ret = (
              to_output(out_i_futures, self.datatype)
              if not return_distance
              else (
                  to_output(out_d_futures, self.datatype),
                  to_output(out_i_futures, self.datatype),
              )
          )
  
      return ret
  
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 9 for calling the NearestNeighbors obj's underlying member method code with example usage: obj.kneighbors(X, n_neighbors, return_distance, _return_futures) and returns: ret