code: def test_edge_dataloader(
      idtype, sampler_name, neg_sampler, mode, use_ddp
  ):
      if mode != "cpu" and F.ctx() == F.cpu():
          pytest.skip("UVA and GPU sampling require a GPU.")
      if mode == "uva" and isinstance(
          neg_sampler, dgl.dataloading.negative_sampler.GlobalUniform
      ):
          pytest.skip("GlobalUniform don't support UVA yet.")
      if use_ddp:
          if os.name == "nt":
              pytest.skip(
                  "PyTorch 1.13.0+ has problems in Windows DDP..."
              )
          dist.init_process_group(
              "gloo" if F.ctx() == F.cpu() else "nccl",
              "tcp://127.0.0.1:12347",
              world_size=1,
              rank=0,
          )
      g1 = dgl.graph(([0, 0, 0, 1, 1], [1, 2, 3, 3, 4])).astype(idtype)
      g1.ndata["feat"] = F.copy_to(F.randn((5, 8)), F.cpu())
      if mode == "pure_gpu":
          g1 = g1.to(F.cuda())
  
      sampler = {
          "full": dgl.dataloading.MultiLayerFullNeighborSampler(2),
          "neighbor": dgl.dataloading.MultiLayerNeighborSampler([3, 3]),
      }[sampler_name]
  
      # no negative sampler
      edge_sampler = dgl.dataloading.as_edge_prediction_sampler(sampler)
      dataloader = dgl.dataloading.DataLoader(
          g1,
          g1.edges(form="eid"),
          edge_sampler,
          device=F.ctx(),
          batch_size=g1.num_edges(),
          use_uva=(mode == "uva"),
          use_ddp=use_ddp,
      )
      for input_nodes, pos_pair_graph, blocks in dataloader:
          _check_device(input_nodes)
          _check_device(pos_pair_graph)
          _check_device(blocks)
  
      # negative sampler
      edge_sampler = dgl.dataloading.as_edge_prediction_sampler(
          sampler, negative_sampler=neg_sampler
      )
      dataloader = dgl.dataloading.DataLoader(
          g1,
          g1.edges(form="eid"),
          edge_sampler,
          device=F.ctx(),
          batch_size=g1.num_edges(),
          use_uva=(mode == "uva"),
          use_ddp=use_ddp,
      )
      for (
          input_nodes,
          pos_pair_graph,
          neg_pair_graph,
          blocks,
      ) in dataloader:
          _check_device(input_nodes)
          _check_device(pos_pair_graph)
          _check_device(neg_pair_graph)
          _check_device(blocks)
  
      g2 = dgl.heterograph(
          {
              ("user", "follow", "user"): (
                  [0, 0, 0, 1, 1, 1, 2],
                  [1, 2, 3, 0, 2, 3, 0],
              ),
              ("user", "followed-by", "user"): (
                  [1, 2, 3, 0, 2, 3, 0],
                  [0, 0, 0, 1, 1, 1, 2],
              ),
              ("user", "play", "game"): (
                  [0, 1, 1, 3, 5],
                  [0, 1, 2, 0, 2],
              ),
              ("game", "played-by", "user"): (
                  [0, 1, 2, 0, 2],
                  [0, 1, 1, 3, 5],
              ),
          }
      ).astype(idtype)
      for ntype in g2.ntypes:
          g2.nodes[ntype].data["feat"] = F.copy_to(
              F.randn((g2.num_nodes(ntype), 8)), F.cpu()
          )
      if mode == "pure_gpu":
          g2 = g2.to(F.cuda())
  
      batch_size = max(g2.num_edges(ety) for ety in g2.canonical_etypes)
      sampler = {
          "full": dgl.dataloading.MultiLayerFullNeighborSampler(2),
          "neighbor": dgl.dataloading.MultiLayerNeighborSampler(
              [{etype: 3 for etype in g2.etypes}] * 2
          ),
      }[sampler_name]
  
      # no negative sampler
      edge_sampler = dgl.dataloading.as_edge_prediction_sampler(sampler)
      dataloader = dgl.dataloading.DataLoader(
          g2,
          {
              ety: g2.edges(form="eid", etype=ety)
              for ety in g2.canonical_etypes
          },
          edge_sampler,
          device=F.ctx(),
          batch_size=batch_size,
          use_uva=(mode == "uva"),
          use_ddp=use_ddp,
      )
      for input_nodes, pos_pair_graph, blocks in dataloader:
          _check_device(input_nodes)
          _check_device(pos_pair_graph)
          _check_device(blocks)
  
      # negative sampler
      edge_sampler = dgl.dataloading.as_edge_prediction_sampler(
          sampler, negative_sampler=neg_sampler
      )
      dataloader = dgl.dataloading.DataLoader(
          g2,
          {
              ety: g2.edges(form="eid", etype=ety)
              for ety in g2.canonical_etypes
          },
          edge_sampler,
          device=F.ctx(),
          batch_size=batch_size,
          use_uva=(mode == "uva"),
          use_ddp=use_ddp,
      )
  
      assert isinstance(iter(dataloader), Iterator)
      for (
          input_nodes,
          pos_pair_graph,
          neg_pair_graph,
          blocks,
      ) in dataloader:
          _check_device(input_nodes)
          _check_device(pos_pair_graph)
          _check_device(neg_pair_graph)
          _check_device(blocks)
  
      if use_ddp:
          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 11 for calling a global function with example usage: test_edge_dataloader(idtype, sampler_name, neg_sampler, mode, use_ddp) without return types