code: def replicate_edgelist(
      edgelist_ddf: Union[dask_cudf.DataFrame, cudf.DataFrame] = None,
      source="src",
      destination="dst",
      weight=None,
      edge_id=None,
      edge_type=None,
  ) -> dask_cudf.DataFrame:
      """
      Replicate edges across all GPUs
  
      Parameters
      ----------
  
      edgelist_ddf: cudf.DataFrame or dask_cudf.DataFrame
          A DataFrame that contains edge information.
  
      source : str or array-like
              source column name or array of column names
  
      destination : str or array-like
          destination column name or array of column names
  
      weight : str, optional (default=None)
          Name of the weight column in the input dataframe.
  
      edge_id : str, optional (default=None)
          Name of the edge id column in the input dataframe.
  
      edge_type : str, optional (default=None)
          Name of the edge type column in the input dataframe.
  
      Returns
      -------
      df : dask_cudf.DataFrame
          A distributed dataframe where each partition contains the
          combined edgelist from all GPUs. If a cudf.DataFrame was passed
          as input, the edgelist will be replicated across all the other
          GPUs in the cluster. If as dask_cudf.DataFrame was passed as input,
          each partition will be filled with the edges of all partitions
          in the dask_cudf.DataFrame.
  
      """
  
      _client = default_client()
  
      if isinstance(edgelist_ddf, cudf.DataFrame):
          edgelist_ddf = dask_cudf.from_cudf(
              edgelist_ddf, npartitions=len(Comms.get_workers())
          )
      col_names = [source, destination]
  
      if weight is not None:
          col_names.append(weight)
      if edge_id is not None:
          col_names.append(edge_id)
      if edge_type is not None:
          col_names.append(edge_type)
  
      if not (set(col_names).issubset(set(edgelist_ddf.columns))):
          raise ValueError(
              "Invalid column names were provided: valid columns names are "
              f"{edgelist_ddf.columns}"
          )
  
      edgelist_ddf = persist_dask_df_equal_parts_per_worker(
          edgelist_ddf, _client
      )
      edgelist_ddf = get_persisted_df_worker_map(edgelist_ddf, _client)
  
      ddf = _mg_call_plc_replicate(
          _client,
          Comms.get_session_id(),
          edgelist_ddf,
          "edgelist",
          col_names,
      )
  
      return ddf
  
file_path: cugraph/python/cugraph/cugraph/structure/replicate_edgelist.py
imports: import dask_cudf
  import cudf
  from dask.distributed import wait, default_client
  import numpy
  from pylibcugraph import ResourceHandle, replicate_edgelist
  from cugraph.dask.common.part_utils import get_persisted_df_worker_map, persist_dask_df_equal_parts_per_worker
  import dask
  import cupy
  import cugraph.dask.comms.comms
  from typing import Union, Tuple
label_desc: Use image node_id 6 for calling a global function with example usage: replicate_edgelist(edgelist_ddf, source, destination, weight, edge_id, edge_type) and returns: ddf