code: def _ddp_runner(proc_id, nprocs, g, data, args):
      mode, drop_last = args
      indices, sampler = data
      if mode == "cpu":
          device = torch.device("cpu")
      else:
          device = torch.device(proc_id)
          torch.cuda.set_device(device)
      if mode == "pure_gpu":
          g = g.to(F.cuda())
      if mode in ("cpu", "uva_cpu_indices"):
          indices = indices.cpu()
      else:
          indices = indices.cuda()
  
      dist.init_process_group(
          "nccl" if mode != "cpu" else "gloo",
          "tcp://127.0.0.1:12347",
          world_size=nprocs,
          rank=proc_id,
      )
      use_uva = mode.startswith("uva")
      batch_size = g.num_nodes()
      shuffle = False
      for num_workers in [1, 4] if mode == "cpu" else [0]:
          dataloader = dgl.dataloading.DataLoader(
              g,
              indices,
              sampler,
              device=device,
              batch_size=batch_size,  # g1.num_nodes(),
              num_workers=num_workers,
              use_uva=use_uva,
              use_ddp=True,
              drop_last=drop_last,
              shuffle=shuffle,
          )
          max_nid = [0]
          for i, (input_nodes, output_nodes, blocks) in enumerate(
              dataloader
          ):
              block = blocks[-1]
              o_src, o_dst = block.edges()
              src_nodes_id = block.srcdata[dgl.NID][o_src]
              dst_nodes_id = block.dstdata[dgl.NID][o_dst]
              max_nid.append(np.max(dst_nodes_id.cpu().numpy()))
  
          local_max = torch.tensor(np.max(max_nid))
          if torch.distributed.get_backend() == "nccl":
              local_max = local_max.cuda()
          dist.reduce(local_max, 0, op=dist.ReduceOp.MAX)
          if proc_id == 0:
              if drop_last and not shuffle and local_max > 0:
                  assert (
                      local_max.item()
                      == len(indices)
                      - len(indices) % nprocs
                      - 1
                      - (len(indices) // nprocs) % batch_size
                  )
              elif not drop_last:
                  assert local_max == len(indices) - 1
      dist.destroy_process_group()
  
imports: import os
  import unittest
  from collections.abc import Iterator, Mapping
  from functools import partial
  import backend
  import dgl
  import dgl.ops
  import numpy
  import pytest
  import torch
  import torch.distributed
  import torch.multiprocessing
  from utils import parametrize_idtype
label_desc: Use image node_id 9 for calling a global function with example usage: _ddp_runner(proc_id, nprocs, g, data, args) without return types