class_docstr: Class to centralize distributed data management. Functionalities include:
  - Data colocation
  - Worker information extraction
  - GPU futures extraction,
  
  Additional functionality can be added as needed. This class **does not**
  contain the actual data, just the metadata necessary to handle it,
  including common pieces of code that need to be performed to call
  Dask functions.
  
  The constructor is not meant to be used directly, but through the factory
  method DistributedDataHandler.create
code: def calculate_worker_and_rank_info(self, comms):
      self.worker_info = comms.worker_info(comms.worker_addresses)
      self.ranks = dict()
  
      for w, futures in self.worker_to_parts.items():
          self.ranks[w] = self.worker_info[w]["rank"]
  
imports: import dask.dataframe
  from functools import reduce
  from toolz import first
  from dask.distributed import default_client
  from dask.distributed import wait
  from cuml.dask.common.part_utils import _extract_partitions
  from cuml.dask.common.dask_arr_utils import validate_dask_array
  from cuml.dask.common.dask_df_utils import to_dask_cudf
  from cuml.dask.common.utils import get_client
  from dask_cudf.core import Series
  from dask.dataframe import Series
  from dask.dataframe import DataFrame
  from cudf import Series
  from cuml.internals.safe_imports import gpu_only_import_from
  from collections import OrderedDict
  from cuml.internals.memory_utils import with_cupy_rmm
  from collections.abc import Sequence
  import dask.array
  from cuml.internals.safe_imports import cpu_only_import
  import cuml.internals.logger
  from cuml.internals.safe_imports import gpu_only_import
label_desc: Use image node_id 4 for calling the DistributedDataHandler obj's underlying member method code with example usage: obj.calculate_worker_and_rank_info(comms) without return types