code: def _mg_rmat(
      scale,
      num_edges,
      a,
      b,
      c,
      seed,
      clip_and_flip,
      scramble_vertex_ids,
      include_edge_weights,
      minimum_weight,
      maximum_weight,
      dtype,
      include_edge_ids,
      include_edge_types,
      min_edge_type_value,
      max_edge_type_value,
      create_using=cugraph.Graph,
  ):
      """
      Calls RMAT on multiple GPUs and uses the resulting Dask cuDF DataFrame to
      initialize and return a cugraph Graph object specified with create_using.
      If create_using is None, returns the Dask DataFrame edgelist as-is.
  
      seed is used as the initial seed for the first worker used (worker 0), then
      each subsequent worker will receive seed+<worker num> as the seed value.
      """
      client = default_client()
      worker_list = list(client.scheduler_info()["workers"].keys())
      num_workers = len(worker_list)
      num_edges_list = _calc_num_edges_per_worker(
          num_workers, num_edges
      )
      result = []
      for i, worker_num_edges in enumerate(num_edges_list):
          unique_worker_seed = seed + i
          future = client.submit(
              _call_rmat,
              Comms.get_session_id(),
              scale,
              worker_num_edges,
              a,
              b,
              c,
              unique_worker_seed,
              clip_and_flip,
              scramble_vertex_ids,
              include_edge_weights,
              minimum_weight,
              maximum_weight,
              dtype,
              include_edge_ids,
              include_edge_types,
              min_edge_type_value,
              max_edge_type_value,
              workers=worker_list[i],
          )
          result.append(future)
  
      wait(result)
  
      cudf_result = [
          client.submit(convert_to_cudf, cp_arrays)
          for cp_arrays in result
      ]
  
      wait(cudf_result)
  
      ddf = dask_cudf.from_delayed(cudf_result)
  
      if create_using is None:
          return ddf
  
      if isinstance(create_using, cugraph.Graph):
          attrs = {"directed": create_using.is_directed()}
          G = type(create_using)(**attrs)
      elif create_using in _graph_types:
          G = create_using()
      else:
          raise TypeError(
              "create_using must be a cugraph.Graph "
              "(or subclass) type or instance, got: "
              f"{type(create_using)}"
          )
  
      weights = None
      edge_id = None
      edge_type = None
  
      if "weights" in ddf.columns:
          weights = "weights"
  
      if "edge_id" in ddf.columns:
          edge_id = "edge_id"
  
      if "edge_type" in ddf.columns:
          edge_type = "edge_type"
  
      G.from_dask_cudf_edgelist(
          ddf,
          source="src",
          destination="dst",
          weight=weights,
          edge_id=edge_id,
          edge_type=edge_type,
      )
  
      return G
  
file_path: cugraph/python/cugraph/cugraph/generators/rmat.py
imports: from dask.distributed import default_client, wait
  import dask_cudf
  from cugraph.dask.comms import comms
  import cudf
  import numpy
  import cupy
  import cugraph
  from pylibcugraph import generate_rmat_edgelist
  from pylibcugraph import generate_rmat_edgelists
  from pylibcugraph import ResourceHandle
label_desc: Use image node_id 4 for calling a global function with example usage: _mg_rmat(scale, num_edges, a, b, c, seed, clip_and_flip, scramble_vertex_ids, include_edge_weights, minimum_weight, maximum_weight, dtype, include_edge_ids, include_edge_types, min_edge_type_value, max_edge_type_value, create_using) and returns: G, ddf