code: def flatten_grouped_results(
      client,
      gpu_futures,
      worker_results_map,
      getter_func=_default_part_getter,
  ):
      """
      This function is useful when a series of partitions have been grouped by
      the worker responsible for the data and the resulting partitions are
      stored on each worker as a list. This happens when a communications
      implementation is used which does not allow multiple ranks per device, so
      the partitions need to be grouped on the ranks to be processed concurrently
      using different streams.
  
      :param client: Dask client
      :param gpu_futures: [(future, part)] worker to part list of tuples
      :param worker_results_map: { rank: future } where future is a list
             of data partitions on a Dask worker
      :param getter_func: a function that takes a future and partition index
             as arguments and returns the data for a specific partitions
      :return: the ordered list of futures holding each partition on the workers
      """
      futures = []
      completed_part_map = {}
      for rank, part in gpu_futures:
          if rank not in completed_part_map:
              completed_part_map[rank] = 0
  
          f = worker_results_map[rank]
  
          futures.append(
              client.submit(getter_func, f, completed_part_map[rank])
          )
  
          completed_part_map[rank] += 1
  
      return futures
  
file_path: cuml/python/cuml/dask/common/part_utils.py
imports: from cuml.dask.common.utils import parse_host_port
  from dask_cudf.core import Series
  from cuml.internals.safe_imports import gpu_only_import_from
  from dask.dataframe import Series
  from dask.dataframe import DataFrame
  from dask.array.core import Array
  from toolz import first
  from dask.distributed import futures_of, default_client, wait
  from collections.abc import Sequence
  from tornado import gen
  from functools import reduce
  from collections import OrderedDict
  from cuml.internals.safe_imports import cpu_only_import
label_desc: Use image node_id 6 for calling a global function with example usage: flatten_grouped_results(client, gpu_futures, worker_results_map, getter_func) and returns: futures