code: def run(
      args,
      dataloader,
      exe,
      program,
      fetchs,
      epoch,
      mode=Mode.TRAIN,
      lr_scheduler=None,
  ):
      """
      Execute program.
  
      Args:
          args(Namespace): Arguments obtained from ArgumentParser.
          dataloader(nvidia.dali.plugin.paddle.DALIGenericIterator):
                  Iteratable output of NVIDIA DALI pipeline,
                  please refer to dali_dataloader in dali.py for details.
          exe(paddle.static.Executor): A executor to run program.
          program(paddle.static.Program): The program to be executed.
          fetchs(dict): A dict of outputs from Program execution (included loss and measures).
          epoch(int): Current epoch id to run.
          mode(utils.Mode, optional): Train or eval mode. Default: Mode.TRAIN.
          lr_scheduler(paddle.optimizer.lr.LRScheduler, optional): A learning rate scheduler.
                                                                   Default: None.
      Returns:
          metrics (dict): A dictionary to collect values of metrics.
      """
      num_trainers = get_num_trainers()
      fetch_list = [f[0] for f in fetchs.values()]
      metric_dict = {
          "lr": AverageMeter("lr", "f", postfix=",", need_avg=False)
      }
  
      for k in fetchs:
          if fetchs[k][1] is not None:
              metric_dict[k] = fetchs[k][1]
  
      metric_dict["batch_time"] = AverageMeter(
          "batch_time", ".5f", postfix=" s,"
      )
      metric_dict["data_time"] = AverageMeter(
          "data_time", ".5f", postfix=" s,"
      )
      metric_dict["compute_time"] = AverageMeter(
          "compute_time", ".5f", postfix=" s,"
      )
  
      for m in metric_dict.values():
          m.reset()
  
      profiler = Profiler()
      tic = time.perf_counter()
  
      idx = 0
      batch_size = None
      latency = []
  
      total_benchmark_steps = (
          args.benchmark_steps + args.benchmark_warmup_steps
      )
  
      dataloader.reset()
      while True:
          # profiler.profile_setup return True only when
          # profile is enable and idx == stop steps
          if profiler.profile_setup(idx):
              break
  
          idx += 1
          try:
              batch = next(dataloader)
          except StopIteration:
              # Reset dataloader when run benchmark to fill required steps.
              if args.benchmark and (idx < total_benchmark_steps):
                  dataloader.reset()
                  # Reset tic timestamp to ignore exception handling time.
                  tic = time.perf_counter()
                  continue
              break
          except RuntimeError:
              logging.warning(
                  "Except RuntimeError when reading data from dataloader, try to read once again..."
              )
              continue
  
          reader_toc = time.perf_counter()
          metric_dict["data_time"].update(reader_toc - tic)
  
          batch_size = batch[0]["data"].shape()[0]
          feed_dict = batch[0]
  
          with profiler.profile_tag(
              idx, "Training" if mode == Mode.TRAIN else "Evaluation"
          ):
              results = exe.run(
                  program=program, feed=feed_dict, fetch_list=fetch_list
              )
  
          for name, m in zip(fetchs.keys(), results):
              if name in metric_dict:
                  metric_dict[name].update(np.mean(m), batch_size)
          metric_dict["compute_time"].update(
              time.perf_counter() - reader_toc
          )
          metric_dict["batch_time"].update(time.perf_counter() - tic)
          if mode == Mode.TRAIN:
              metric_dict["lr"].update(lr_scheduler.get_lr())
  
          if lr_scheduler is not None:
              with profiler.profile_tag(idx, "LR Step"):
                  lr_scheduler.step()
  
          tic = time.perf_counter()
  
          if idx % args.print_interval == 0:
              log_msg = dict()
              log_msg["loss"] = metric_dict["loss"].val.item()
              log_msg["top1"] = metric_dict["top1"].val.item()
              log_msg["top5"] = metric_dict["top5"].val.item()
              log_msg["data_time"] = metric_dict["data_time"].val
              log_msg["compute_time"] = metric_dict["compute_time"].val
              log_msg["batch_time"] = metric_dict["batch_time"].val
              log_msg["ips"] = (
                  batch_size
                  * num_trainers
                  / metric_dict["batch_time"].val
              )
              if mode == Mode.TRAIN:
                  log_msg["lr"] = metric_dict["lr"].val
              log_info((epoch, idx), log_msg, mode)
  
          if args.benchmark:
              latency.append(metric_dict["batch_time"].val)
              # Ignore the warmup iters
              if idx == args.benchmark_warmup_steps:
                  metric_dict["compute_time"].reset()
                  metric_dict["data_time"].reset()
                  metric_dict["batch_time"].reset()
                  latency.clear()
                  logging.info("Begin benchmark at step %d", idx + 1)
  
              if idx == total_benchmark_steps:
                  benchmark_data = dict()
                  benchmark_data["ips"] = (
                      batch_size
                      * num_trainers
                      / metric_dict["batch_time"].avg
                  )
                  if mode == mode.EVAL:
                      latency = np.array(latency) * 1000
                      quantile = np.quantile(latency, [0.9, 0.95, 0.99])
  
                      benchmark_data["latency_avg"] = np.mean(latency)
                      benchmark_data["latency_p90"] = quantile[0]
                      benchmark_data["latency_p95"] = quantile[1]
                      benchmark_data["latency_p99"] = quantile[2]
  
                  logging.info("End benchmark at epoch step %d", idx)
                  return benchmark_data
  
      epoch_data = dict()
      epoch_data["loss"] = metric_dict["loss"].avg.item()
      epoch_data["epoch_time"] = metric_dict["batch_time"].total
      epoch_data["ips"] = (
          batch_size
          * num_trainers
          * metric_dict["batch_time"].count
          / metric_dict["batch_time"].sum
      )
      if mode == Mode.EVAL:
          epoch_data["top1"] = metric_dict["top1"].avg.item()
          epoch_data["top5"] = metric_dict["top5"].avg.item()
      log_info((epoch,), epoch_data, mode)
  
      return epoch_data
  
imports: import time
  import logging
  from profile import Profiler
  import numpy
  from optimizer import build_optimizer
  from lr_scheduler import build_lr_scheduler
  from utils.misc import AverageMeter
  from utils.mode import Mode, RunScope
  from utils.utility import get_num_trainers
  import models
  import dllogger
  import paddle
  import paddle.nn.functional
  from paddle.distributed import fleet
  from paddle.distributed.fleet import DistributedStrategy
  from paddle.incubate import asp
  from paddle.distributed.fleet.meta_optimizers.common import CollectiveHelper
label_desc: Use image node_id 7 for calling a global function with example usage: run(args, dataloader, exe, program, fetchs, epoch, mode, lr_scheduler) and returns: epoch_data, benchmark_data