code: def run(args, device, data):
      """
      Train and evaluate DistSAGE.
  
      Parameters
      ----------
      args : argparse.Args
          Arguments for train and evaluate.
      device : torch.Device
          Target device for train and evaluate.
      data : Packed Data
          Packed data includes train/val/test IDs, feature dimension,
          number of classes, graph.
      """
      train_nid, val_nid, test_nid, in_feats, n_classes, g = data
      sampler = dgl.dataloading.NeighborSampler(
          [int(fanout) for fanout in args.fan_out.split(",")]
      )
      dataloader = dgl.dataloading.DistNodeDataLoader(
          g,
          train_nid,
          sampler,
          batch_size=args.batch_size,
          shuffle=True,
          drop_last=False,
      )
      model = DistSAGE(
          in_feats,
          args.num_hidden,
          n_classes,
          args.num_layers,
          F.relu,
          args.dropout,
      )
      model = model.to(device)
      if args.num_gpus == 0:
          model = th.nn.parallel.DistributedDataParallel(model)
      else:
          model = th.nn.parallel.DistributedDataParallel(
              model, device_ids=[device], output_device=device
          )
      loss_fcn = nn.CrossEntropyLoss()
      loss_fcn = loss_fcn.to(device)
      optimizer = optim.Adam(model.parameters(), lr=args.lr)
  
      # Training loop.
      iter_tput = []
      epoch = 0
      epoch_time = []
      test_acc = 0.0
      for _ in range(args.num_epochs):
          epoch += 1
          tic = time.time()
          # Various time statistics.
          sample_time = 0
          forward_time = 0
          backward_time = 0
          update_time = 0
          num_seeds = 0
          num_inputs = 0
          start = time.time()
          step_time = []
  
          with model.join():
              for step, (input_nodes, seeds, blocks) in enumerate(
                  dataloader
              ):
                  tic_step = time.time()
                  sample_time += tic_step - start
                  # Slice feature and label.
                  batch_inputs = g.ndata["features"][input_nodes]
                  batch_labels = g.ndata["labels"][seeds].long()
                  num_seeds += len(blocks[-1].dstdata[dgl.NID])
                  num_inputs += len(blocks[0].srcdata[dgl.NID])
                  # Move to target device.
                  blocks = [block.to(device) for block in blocks]
                  batch_inputs = batch_inputs.to(device)
                  batch_labels = batch_labels.to(device)
                  # Compute loss and prediction.
                  start = time.time()
                  batch_pred = model(blocks, batch_inputs)
                  loss = loss_fcn(batch_pred, batch_labels)
                  forward_end = time.time()
                  optimizer.zero_grad()
                  loss.backward()
                  compute_end = time.time()
                  forward_time += forward_end - start
                  backward_time += compute_end - forward_end
  
                  optimizer.step()
                  update_time += time.time() - compute_end
  
                  step_t = time.time() - tic_step
                  step_time.append(step_t)
                  iter_tput.append(
                      len(blocks[-1].dstdata[dgl.NID]) / step_t
                  )
                  if (step + 1) % args.log_every == 0:
                      acc = compute_acc(batch_pred, batch_labels)
                      gpu_mem_alloc = (
                          th.cuda.max_memory_allocated() / 1000000
                          if th.cuda.is_available()
                          else 0
                      )
                      sample_speed = np.mean(
                          iter_tput[-args.log_every :]
                      )
                      mean_step_time = np.mean(
                          step_time[-args.log_every :]
                      )
                      print(
                          f"Part {g.rank()} | Epoch {epoch:05d} | Step {step:05d}"
                          f" | Loss {loss.item():.4f} | Train Acc {acc.item():.4f}"
                          f" | Speed (samples/sec) {sample_speed:.4f}"
                          f" | GPU {gpu_mem_alloc:.1f} MB | "
                          f"Mean step time {mean_step_time:.3f} s"
                      )
                  start = time.time()
  
          toc = time.time()
          print(
              f"Part {g.rank()}, Epoch Time(s): {toc - tic:.4f}, "
              f"sample+data_copy: {sample_time:.4f}, forward: {forward_time:.4f},"
              f" backward: {backward_time:.4f}, update: {update_time:.4f}, "
              f"#seeds: {num_seeds}, #inputs: {num_inputs}"
          )
          epoch_time.append(toc - tic)
  
          if epoch % args.eval_every == 0 or epoch == args.num_epochs:
              start = time.time()
              val_acc, test_acc = evaluate(
                  model.module,
                  g,
                  g.ndata["features"],
                  g.ndata["labels"],
                  val_nid,
                  test_nid,
                  args.batch_size_eval,
                  device,
              )
              print(
                  f"Part {g.rank()}, Val Acc {val_acc:.4f}, "
                  f"Test Acc {test_acc:.4f}, time: {time.time() - start:.4f}"
              )
  
      return (
          np.mean(epoch_time[-int(args.num_epochs * 0.8) :]),
          test_acc,
      )
  
file_path: dgl/examples/distributed/graphsage/node_classification.py
imports: import argparse
  import socket
  import time
  import dgl
  import dgl.nn.pytorch
  import numpy
  import torch
  import torch.nn
  import torch.nn.functional
  import torch.optim
  import tqdm
label_desc: Use image node_id 3 for calling a global function with example usage: run(args, device, data) and returns: np,  test_acc