code: def benchmark_cugraph_bulk_sampling(
      dataset,
      output_path,
      seed,
      batch_size,
      seeds_per_call,
      fanout,
      sampling_target_framework,
      reverse_edges=True,
      dataset_dir=".",
      replication_factor=1,
      num_labels=256,
      labeled_percentage=0.001,
      add_edge_types=False,
      num_epochs=1,
  ):
      """
      Entry point for the benchmark.
  
      Parameters
      ----------
      dataset: str
          The dataset to sample.  Can be rmat_{scale}_{edgefactor}, or the name of an ogb dataset.
      output_path: str
          The output path, where samples and metadata will be stored.
      seed: int
          The random seed.
      batch_size: int
          The batch size (number of input seeds in a single sampling batch).
      seeds_per_call: int
          The number of input seeds in a single sampling call.
      fanout: list[int]
          The fanout.
      reverse_edges: bool
          Whether to reverse edges when constructing the graph.
      dataset_dir: str
          The directory where datasets are stored (only for ogb datasets)
      replication_factor: int
          The number of times to replicate the dataset.
      num_labels: int
          The number of random labels to generate (only for rmat datasets)
      labeled_percentage: float
          The percentage of the data that is labeled (only for rmat datasets)
          Defaults to 0.001 to match papers100M
      add_edge_types: bool
          Whether to add edge types to the edgelist.
          Defaults to False.
      sampling_target_framework: str
          The framework to sample for.
      num_epochs: int
          The number of epochs to sample for.
      """
  
      logger = logging.getLogger("__main__")
      logger.info(str(dataset))
      if dataset[0:4] == "rmat":
          (
              dask_edgelist_df,
              dask_label_df,
              node_offsets,
              edge_offsets,
              total_num_nodes,
          ) = generate_rmat_dataset(
              dataset,
              reverse_edges=reverse_edges,
              seed=seed,
              labeled_percentage=labeled_percentage,
              num_labels=num_labels,
              add_edge_types=add_edge_types,
          )
  
      else:
          (
              dask_edgelist_df,
              dask_label_df,
              node_offsets,
              edge_offsets,
              total_num_nodes,
          ) = load_disk_dataset(
              dataset,
              dataset_dir=dataset_dir,
              reverse_edges=reverse_edges,
              replication_factor=replication_factor,
              add_edge_types=add_edge_types,
          )
  
      num_input_edges = len(dask_edgelist_df)
      logger.info(f"Number of input edges = {num_input_edges:,}")
  
      G = construct_graph(dask_edgelist_df)
      del dask_edgelist_df
      logger.info("constructed graph")
  
      input_memory = (
          G.edgelist.edgelist_df.memory_usage().sum().compute()
      )
      logger.info(f"input memory: {input_memory}")
  
      output_subdir = os.path.join(
          output_path,
          f"{dataset}[{replication_factor}]_b{batch_size}_f{fanout}",
      )
      os.makedirs(output_subdir)
  
      if sampling_target_framework == "cugraph_dgl_csr":
          sampling_kwargs = {
              "deduplicate_sources": True,
              "prior_sources_behavior": "carryover",
              "renumber": True,
              "compression": "CSR",
              "compress_per_hop": True,
              "use_legacy_names": False,
              "include_hop_column": False,
          }
      else:
          # FIXME: Update these arguments when CSC mode is fixed in cuGraph-PyG (release 24.02)
          sampling_kwargs = {
              "deduplicate_sources": True,
              "prior_sources_behavior": "exclude",
              "renumber": True,
              "compression": "COO",
              "compress_per_hop": False,
              "use_legacy_names": False,
              "include_hop_column": True,
          }
  
      batches_per_partition = 600_000 // batch_size
      execution_time, allocation_counts = sample_graph(
          G=G,
          label_df=dask_label_df,
          output_path=output_subdir,
          num_epochs=num_epochs,
          seed=seed,
          batch_size=batch_size,
          seeds_per_call=seeds_per_call,
          batches_per_partition=batches_per_partition,
          fanout=fanout,
          sampling_kwargs=sampling_kwargs,
      )
  
      output_meta = {
          "dataset": dataset,
          "dataset_dir": dataset_dir,
          "seed": seed,
          "node_offsets": node_offsets,
          "edge_offsets": {
              "__".join(k): v for k, v in edge_offsets.items()
          },
          "total_num_nodes": total_num_nodes,
          "total_num_edges": num_input_edges,
          "batch_size": batch_size,
          "seeds_per_call": seeds_per_call,
          "batches_per_partition": batches_per_partition,
          "fanout": fanout,
          "replication_factor": replication_factor,
          "num_sampling_gpus": len(G._plc_graph),
          "execution_time": execution_time,
      }
  
      with open(
          os.path.join(output_subdir, "output_meta.json"), "w"
      ) as f:
          json.dump(output_meta, f, indent="\t")
  
      logger.info("allocation counts b:")
      logger.info(allocation_counts.values())
  
      (
          input_to_peak_ratio,
          output_to_peak_ratio,
          input_memory_per_worker,
          peak_allocation_across_workers,
      ) = get_memory_statistics(
          allocation_counts=allocation_counts, input_memory=input_memory
      )
      logger.info(
          f"Number of edges in final graph = {G.number_of_edges():,}"
      )
      logger.info("-" * 80)
      return (
          num_input_edges,
          input_to_peak_ratio,
          output_to_peak_ratio,
          input_memory_per_worker,
          peak_allocation_across_workers,
      )
  
imports: import logging
  import warnings
  import argparse
  import traceback
  from cugraph.testing.mg_utils import generate_edgelist_rmat, get_allocation_counts_dask_lazy, sizeof_fmt, get_peak_output_ratio_across_workers, start_dask_client, stop_dask_client, enable_spilling
  from cugraph.structure.symmetrize import symmetrize
  from cugraph.gnn import BulkSampler
  import cugraph
  import json
  import re
  import os
  import gc
  from time import sleep, perf_counter
  from math import ceil
  import pandas
  import numpy
  import cupy
  import cudf
  import dask_cudf
  import dask.dataframe
  from dask.distributed import default_client
  from cugraph.dask import get_n_workers
  from typing import Optional, Union, Dict
label_desc: Use image node_id 10 for calling a global function with example usage: benchmark_cugraph_bulk_sampling(dataset, output_path, seed, batch_size, seeds_per_call, fanout, sampling_target_framework, reverse_edges, dataset_dir, replication_factor, num_labels, labeled_percentage, add_edge_types, num_epochs) and returns: num_input_edges,  input_to_peak_ratio,  output_to_peak_ratio,  input_memory_per_worker,  peak_allocation_across_workers