code: def reduce(futures, func, client=None):
      """
      Performs a cluster-wide reduction by first
      running function on worker->host->cluster. This
      function takes locality into account by first
      reducing partitions local to each worker before
      reducing partitions on each host and, finally,
      reducing the partitions across the cluster into
      a single reduced partition.
  
      Parameters
      ----------
  
      futures : array-like of dask.Future futures to reduce
      func : Python reduction function accepting list
             of objects to reduce and returning a single
             reduced object.
  
      client : dask.distributed.Client to use for scheduling
  
      Returns
      -------
  
      output : dask.Future a future containing the final reduce
          object.
      """
  
      client = get_client(client)
  
      # Make sure input futures have been assigned to worker(s)
      wait(futures)
  
      for local_reduction_func in [workers_to_parts, hosts_to_parts]:
          who_has = client.who_has(futures)
  
          workers = [(first(who_has[m.key]), m) for m in futures]
          worker_parts = local_reduction_func(workers)
  
          # Short circuit when all parts already have preferred
          # locality
          if len(worker_parts) > 1:
              # Local tree reduction for scalability
              futures = client.compute(
                  [
                      tree_reduce(p, func)
                      for w, p in worker_parts.items()
                  ]
              )
  
              wait(futures)
  
      # Merge across workers
      ret = client.compute(tree_reduce(futures, func))
      wait(ret)
  
      return ret
  
file_path: cuml/python/cuml/dask/common/func.py
imports: import cuml.internals.logger
  import dask
  from cuml.dask.common.utils import get_client
  from cuml.dask.common.part_utils import hosts_to_parts
  from cuml.dask.common.part_utils import workers_to_parts
  from dask.delayed import Delayed
  from dask.distributed import wait
  from toolz import first
label_desc: Use image node_id 1 for calling a global function with example usage: reduce(futures, func, client) and returns: ret