docstr: Runs a function embarrassingly parallel on a set of workers while
  reusing instances of models and constraining the number of
  tasks that can execute concurrently on each worker.
  
  Note that this mixin assumes the subclass has been trained and
  includes a `self._get_internal_model()` function containing a subclass
  of `cuml.Base`.
  
  This is intended to abstract functions like predict, transform, and
  score, which can execute embarrassingly parallel but need addition
  execution constraints which result from the more limited GPU
  resources.
  
  Parameters
  ----------
  func : dask.delayed function to propagate to the workers to execute
         embarrassingly parallel, shared between tasks on each worker
  
  X : Dask cuDF dataframe  or CuPy backed Dask Array (n_rows, n_features)
      Distributed dense matrix (floats or doubles) of shape
      (n_samples, n_features).
  
  delayed : bool return a lazy (delayed) object?
  
  output_futures : bool returns the futures pointing the to the resuls
                   of the parallel function executions on the workers,
                   rather than a dask collection object.
  
  output_collection_type : None or a string in {'cupy', 'cudf'}
      Choose to return the resulting collection as a CuPy backed
      dask.array or a dask_cudf.DataFrame. If None, will use the same
      collection type as used in the input of fit.
      Unused if output_futures=True.
  
  Returns
  -------
  y : dask cuDF (n_rows, 1)
docstr_tok: ["Runs","a","function","embarrassingly","parallel","on","a","set","of","workers","while","reusing","instances","of","models","and","constraining","the","number","of","tasks","that","can","execute","concurrently","on","each","worker",".","Note","that","this","mixin","assumes","the","subclass","has","been","trained","and","includes","a","`","self._get_internal_model","(",")","`","function","containing","a","subclass","of","`","cuml.Base","`",".","This","is","intended","to","abstract","functions","like","predict",",","transform",",","and","score",",","which","can","execute","embarrassingly","parallel","but","need","addition","execution","constraints","which","result","from","the","more","limited","GPU","resources",".","Parameters","--","--","--","--","--","func",":","dask.delayed","function","to","propagate","to","the","workers","to","execute","embarrassingly","parallel",",","shared","between","tasks","on","each","worker","X",":","Dask","cuDF","dataframe","or","CuPy","backed","Dask","Array","(","n_rows",",","n_features",")","Distributed","dense","matrix","(","floats","or","doubles",")","of","shape","(","n_samples",",","n_features",")",".","delayed",":","bool","return","a","lazy","(","delayed",")","object","?","output_futures",":","bool","returns","the","futures","pointing","the","to","the","resuls","of","the","parallel","function","executions","on","the","workers",",","rather","than","a","dask","collection","object",".","output_collection_type",":","None","or","a","string","in","{","'cupy","'",",","'cudf","'","}","Choose","to","return","the","resulting","collection","as","a","CuPy","backed","dask.array","or","a","dask_cudf.DataFrame",".","If","None",",","will","use","the","same","collection","type","as","used","in","the","input","of","fit",".","Unused","if","output_futures=True",".","Returns","--","--","--","-","y",":","dask","cuDF","(","n_rows",",","1",")"]
code: def _run_parallel_func(
      self,
      func,
      X,
      n_dims=1,
      delayed=True,
      output_futures=False,
      output_dtype=None,
      output_collection_type=None,
      **kwargs,
  ):
      """
      Runs a function embarrassingly parallel on a set of workers while
      reusing instances of models and constraining the number of
      tasks that can execute concurrently on each worker.
  
      Note that this mixin assumes the subclass has been trained and
      includes a `self._get_internal_model()` function containing a subclass
      of `cuml.Base`.
  
      This is intended to abstract functions like predict, transform, and
      score, which can execute embarrassingly parallel but need addition
      execution constraints which result from the more limited GPU
      resources.
  
      Parameters
      ----------
      func : dask.delayed function to propagate to the workers to execute
             embarrassingly parallel, shared between tasks on each worker
  
      X : Dask cuDF dataframe  or CuPy backed Dask Array (n_rows, n_features)
          Distributed dense matrix (floats or doubles) of shape
          (n_samples, n_features).
  
      delayed : bool return a lazy (delayed) object?
  
      output_futures : bool returns the futures pointing the to the resuls
                       of the parallel function executions on the workers,
                       rather than a dask collection object.
  
      output_collection_type : None or a string in {'cupy', 'cudf'}
          Choose to return the resulting collection as a CuPy backed
          dask.array or a dask_cudf.DataFrame. If None, will use the same
          collection type as used in the input of fit.
          Unused if output_futures=True.
  
      Returns
      -------
      y : dask cuDF (n_rows, 1)
      """
      X_d = X.to_delayed()
  
      if output_collection_type is None:
          output_collection_type = self.datatype
  
      model_delayed = dask.delayed(
          self._get_internal_model(), pure=True, traverse=False
      )
  
      func = dask.delayed(func, pure=False, nout=1)
      if isinstance(X, dcDataFrame):
          preds = [func(model_delayed, part, **kwargs) for part in X_d]
          dtype = (
              first(X.dtypes) if output_dtype is None else output_dtype
          )
      elif isinstance(X, dcSeries):
          preds = [func(model_delayed, part, **kwargs) for part in X_d]
          dtype = X.dtype if output_dtype is None else output_dtype
      else:
          preds = [func(model_delayed, part[0]) for part in X_d]
          dtype = X.dtype if output_dtype is None else output_dtype
  
      # TODO: Put the following conditionals in a
      #  `to_delayed_output()` function
      # TODO: Add eager path back in
  
      if output_collection_type == "cupy":
          # todo: add parameter for option of not checking directly
          shape = (np.nan,) * n_dims
          preds_arr = [
              dask.array.from_delayed(
                  pred,
                  meta=cp.zeros(1, dtype=dtype),
                  shape=shape,
                  dtype=dtype,
              )
              for pred in preds
          ]
  
          if output_futures:
              return self.client.compute(preds)
          else:
              output = dask.array.concatenate(
                  preds_arr, axis=0, allow_unknown_chunksizes=True
              )
              return output if delayed else output.persist()
  
      elif output_collection_type == "cudf":
          if output_futures:
              return self.client.compute(preds)
          else:
              output = dask_cudf.from_delayed(preds)
              return output if delayed else output.persist()
      else:
          raise ValueError(
              "Expected cupy or cudf but found %s"
              % (output_collection_type)
          )
  
imports: from distributed.client import Future
  from functools import wraps
  from dask_cudf.core import Series
  from cuml.internals.safe_imports import gpu_only_import_from
  from cuml.internals.base import Base
  from cuml.internals import BaseMetaClass
  from cuml.dask.common import parts_to_ranks
  from cuml.dask.common.input_utils import DistributedDataHandler
  from raft_dask.common.comms import Comms
  from cuml.dask.common.utils import wait_and_raise_from_futures
  from cuml.internals.array import CumlArray
  from cuml.dask.common.utils import get_client
  from collections.abc import Iterable
  from toolz import first
  from cuml.internals.safe_imports import cpu_only_import
  import dask
  import cudf.comm.serialize
  from cuml.internals.safe_imports import gpu_only_import
label_desc: Use image node_id 1 for calling the DelayedParallelFunc obj's underlying member method code with example usage: obj._run_parallel_func(func, X, n_dims, delayed, output_futures, output_dtype, output_collection_type) and returns: self, unknown, self, unknown