code: def concat_within_workers(client, ddf):
      """
      Concats all partitions within workers without transfers.
      """
      df_delayed = get_delayed_dict(ddf)
  
      result = []
      for worker, tasks in client.has_what().items():
          worker_task_list = []
  
          for task in list(tasks):
              if task in df_delayed:
                  worker_task_list.append(df_delayed[task])
          concat_tasks = delayed(concat_dfs)(worker_task_list)
          result.append(
              client.persist(collections=concat_tasks, workers=worker)
          )
  
      return dask_cudf.from_delayed(result)
  
imports: import numpy
  from dask.distributed import futures_of, default_client, wait
  from toolz import first
  import collections
  import dask_cudf
  from dask.array.core import Array
  from dask_cudf.core import DataFrame
  from dask_cudf.core import Series
  from functools import reduce
  import cugraph.dask.comms.comms
  from dask.delayed import delayed
  import cudf
label_desc: Use image node_id 17 for calling a global function with example usage: concat_within_workers(client, ddf) and returns: dask_cudf