class_docstr: BaseRandomForestModel defines functions used in both Random Forest
  Classifier and Regressor for Multi Node and Multi GPU models. The common
  functions are defined here and called from the main Random Forest Multi
  Node Multi GPU APIs. The functions defined here are not meant to be used
  as a part of the public API.
code: def _fit(self, model, dataset, convert_dtype, broadcast_data):
      data = DistributedDataHandler.create(dataset, client=self.client)
      self.active_workers = data.workers
      self.datatype = data.datatype
      if self.datatype == "cudf":
          has_float64 = (dataset[0].dtypes == np.float64).any()
      else:
          has_float64 = dataset[0].dtype == np.float64
      if has_float64:
          raise TypeError(
              "To use Dask RF data should have dtype float32."
          )
  
      labels = self.client.persist(dataset[1])
      if self.datatype == "cudf":
          self.num_classes = len(labels.unique())
      else:
          self.num_classes = len(dask.array.unique(labels).compute())
  
      combined_data = (
          list(map(lambda x: x[1], data.gpu_futures))
          if broadcast_data
          else None
      )
  
      futures = list()
      for idx, (worker, worker_data) in enumerate(
          data.worker_to_parts.items()
      ):
          futures.append(
              self.client.submit(
                  _func_fit,
                  model[worker],
                  combined_data if broadcast_data else worker_data,
                  convert_dtype,
                  workers=[worker],
                  pure=False,
              )
          )
  
      self.n_active_estimators_per_worker = []
      for worker in data.worker_to_parts.keys():
          n = self.workers.index(worker)
          n_est = self.n_estimators_per_worker[n]
          self.n_active_estimators_per_worker.append(n_est)
  
      if len(self.workers) > len(self.active_workers):
          if self.ignore_empty_partitions:
              curent_estimators = (
                  self.n_estimators
                  / len(self.workers)
                  * len(self.active_workers)
              )
              warn_text = (
                  f"Data was not split among all workers "
                  f"using only {self.active_workers} workers to fit."
                  f"This will only train {curent_estimators}"
                  f" estimators instead of the requested "
                  f"{self.n_estimators}"
              )
              warnings.warn(warn_text)
          else:
              raise ValueError(
                  "Data was not split among all workers. "
                  "Re-run the code or "
                  "use ignore_empty_partitions=True"
                  " while creating model"
              )
      wait_and_raise_from_futures(futures)
      return self
  
imports: from cuml.fil.fil import TreeliteModel
  from cuml.dask.common.utils import get_client, wait_and_raise_from_futures
  from cuml.dask.common.input_utils import DistributedDataHandler, concatenate
  from dask.distributed import Future
  from collections.abc import Iterable
  from cuml import using_output_type
  import warnings
  from cuml.internals.safe_imports import gpu_only_import
  import dask
  import json
  import math
  from cuml.internals.safe_imports import cpu_only_import
label_desc: Use image node_id 3 for calling the BaseRandomForestModel obj's underlying member method code with example usage: obj._fit(model, dataset, convert_dtype, broadcast_data) and returns: self