code: def dist_link_neighbor_sampler(
      world_size: int,
      rank: int,
      master_port: int,
      disjoint: bool = False,
  ):
      dist_data, data = create_data(rank, world_size)
  
      current_ctx = DistContext(
          rank=rank,
          global_rank=rank,
          world_size=world_size,
          global_world_size=world_size,
          group_name="dist-sampler-test",
      )
  
      dist_sampler = DistNeighborSampler(
          data=dist_data,
          current_ctx=current_ctx,
          num_neighbors=[-1, -1],
          shuffle=False,
          disjoint=disjoint,
      )
  
      # Close RPC & worker group at exit:
      atexit.register(shutdown_rpc)
  
      init_rpc(
          current_ctx=current_ctx,
          master_addr="localhost",
          master_port=master_port,
      )
      dist_sampler.init_sampler_instance()
      dist_sampler.register_sampler_rpc()
      dist_sampler.event_loop = ConcurrentEventLoop(2)
      dist_sampler.event_loop.start_loop()
  
      if rank == 0:  # Seed nodes:
          input_row = torch.tensor([1, 6], dtype=torch.int64)
          input_col = torch.tensor([2, 7], dtype=torch.int64)
      else:
          input_row = torch.tensor([4, 9], dtype=torch.int64)
          input_col = torch.tensor([5, 0], dtype=torch.int64)
  
      inputs = EdgeSamplerInput(
          input_id=None,
          row=input_row,
          col=input_col,
          input_type=None,
      )
  
      # evaluate distributed edge sample function
      out_dist = dist_sampler.event_loop.run_task(
          coro=dist_sampler.edge_sample(
              inputs, dist_sampler.node_sample, data.num_nodes, disjoint
          )
      )
  
      sampler = NeighborSampler(
          data=data, num_neighbors=[-1, -1], disjoint=disjoint
      )
  
      # Evaluate edge sample function:
      out = edge_sample(
          inputs,
          sampler._sample,
          data.num_nodes,
          disjoint,
          node_time=None,
          neg_sampling=None,
      )
  
      # Compare distributed output with single machine output:
      assert torch.equal(out_dist.node, out.node)
      assert torch.equal(out_dist.row, out.row)
      assert torch.equal(out_dist.col, out.col)
      if disjoint:
          assert torch.equal(out_dist.batch, out.batch)
      assert out_dist.num_sampled_nodes == out.num_sampled_nodes
      assert out_dist.num_sampled_edges == out.num_sampled_edges
  
file_path: pytorch_geometric/test/distributed/test_dist_link_neighbor_sampler.py
imports: import atexit
  import socket
  from typing import Optional
  import pytest
  import torch
  from torch_geometric.data import Data
  from torch_geometric.distributed import DistNeighborSampler, LocalFeatureStore, LocalGraphStore
  from torch_geometric.distributed.dist_context import DistContext
  from torch_geometric.distributed.event_loop import ConcurrentEventLoop
  from torch_geometric.distributed.rpc import init_rpc, shutdown_rpc
  from torch_geometric.sampler import EdgeSamplerInput, NeighborSampler
  from torch_geometric.sampler.neighbor_sampler import edge_sample
  from torch_geometric.testing import onlyLinux, withPackage
label_desc: Use image node_id 2 for calling a global function with example usage: dist_link_neighbor_sampler(world_size, rank, master_port, disjoint) without return types