code: def test_dist_dataloader(
      num_server, num_workers, drop_last, num_groups
  ):
      reset_envs()
      # No multiple partitions on single machine for
      # multiple client groups in case of race condition.
      if num_groups > 1:
          num_server = 1
      with tempfile.TemporaryDirectory() as test_dir:
          ip_config = "ip_config.txt"
          generate_ip_config(ip_config, num_server, num_server)
  
          g = CitationGraphDataset("cora")[0]
          print(g.idtype)
          num_parts = num_server
          num_hops = 1
  
          orig_nid, orig_eid = partition_graph(
              g,
              "test_sampling",
              num_parts,
              test_dir,
              num_hops=num_hops,
              part_method="metis",
              return_mapping=True,
          )
  
          part_config = os.path.join(test_dir, "test_sampling.json")
          pserver_list = []
          ctx = mp.get_context("spawn")
          for i in range(num_server):
              p = ctx.Process(
                  target=start_server,
                  args=(
                      i,
                      ip_config,
                      part_config,
                      num_server > 1,
                      num_workers + 1,
                  ),
              )
              p.start()
              time.sleep(1)
              pserver_list.append(p)
  
          os.environ["DGL_DIST_MODE"] = "distributed"
          os.environ["DGL_NUM_SAMPLER"] = str(num_workers)
          ptrainer_list = []
          num_trainers = 1
          for trainer_id in range(num_trainers):
              for group_id in range(num_groups):
                  p = ctx.Process(
                      target=start_dist_dataloader,
                      args=(
                          trainer_id,
                          ip_config,
                          part_config,
                          num_server,
                          drop_last,
                          orig_nid,
                          orig_eid,
                          group_id,
                      ),
                  )
                  p.start()
                  time.sleep(
                      1
                  )  # avoid race condition when instantiating DistGraph
                  ptrainer_list.append(p)
  
          for p in ptrainer_list:
              p.join()
              assert p.exitcode == 0
          for p in pserver_list:
              p.join()
              assert p.exitcode == 0
  
imports: import multiprocessing
  import os
  import tempfile
  import time
  import unittest
  import backend
  import dgl
  import numpy
  import pytest
  import torch
  from dgl.data import CitationGraphDataset
  from dgl.distributed import DistDataLoader, DistGraph, DistGraphServer, load_partition, partition_graph
  from scipy import sparse
  from utils import generate_ip_config, reset_envs
label_desc: Use image node_id 6 for calling a global function with example usage: test_dist_dataloader(num_server, num_workers, drop_last, num_groups) without return types