code: def _fit(self, model_func, data):
      n_cols = data[0].shape[1]
  
      data = DistributedDataHandler.create(
          data=data, client=self.client
      )
      self.datatype = data.datatype
  
      comms = Comms(comms_p2p=False)
      comms.init(workers=data.workers)
  
      data.calculate_parts_to_sizes(comms)
      self.ranks = data.ranks
  
      worker_info = comms.worker_info(comms.worker_addresses)
      parts_to_sizes, _ = parts_to_ranks(
          self.client, worker_info, data.gpu_futures
      )
  
      lin_models = dict(
          [
              (
                  data.worker_info[worker_data[0]]["rank"],
                  self.client.submit(
                      model_func,
                      comms.sessionId,
                      self.datatype,
                      **self.kwargs,
                      pure=False,
                      workers=[worker_data[0]],
                  ),
              )
              for worker, worker_data in enumerate(
                  data.worker_to_parts.items()
              )
          ]
      )
  
      fit_func = self._func_fit
      lin_fit = dict(
          [
              (
                  worker_data[0],
                  self.client.submit(
                      fit_func,
                      lin_models[
                          data.worker_info[worker_data[0]]["rank"]
                      ],
                      worker_data[1],
                      data.total_rows,
                      n_cols,
                      parts_to_sizes,
                      data.worker_info[worker_data[0]]["rank"],
                      pure=False,
                      workers=[worker_data[0]],
                  ),
              )
              for worker, worker_data in enumerate(
                  data.worker_to_parts.items()
              )
          ]
      )
  
      wait_and_raise_from_futures(list(lin_fit.values()))
  
      comms.destroy()
      return lin_models
  
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 SyncFitMixinLinearModel obj's underlying member method code with example usage: obj._fit(model_func, data) and returns: lin_models