code: def train_torch_vanilla(
      *,
      config: dict,
      num_workers: int = 4,
      cpus_per_worker: int = 8,
      use_gpu: bool = False,
  ) -> Tuple[float, float, float]:
      # This function is kicked off by the main() function and subsequently kicks
      # off tasks that run train_torch_vanilla_worker() on the worker nodes.
      from benchmark_util import (
          upload_file_to_all_nodes,
          create_actors_with_options,
          run_commands_on_actors,
          run_fn_on_actors,
          get_ip_port_actors,
          get_gpu_ids_actors,
          map_ips_to_gpus,
          set_cuda_visible_devices,
      )
  
      path = os.path.abspath(__file__)
      upload_file_to_all_nodes(path)
  
      num_epochs = config["epochs"]
  
      try:
          nccl_network_interface = find_network_interface()
          runtime_env = {
              "env_vars": {"NCCL_SOCKET_IFNAME": nccl_network_interface}
          }
      except Exception:
          runtime_env = {}
  
      actors = create_actors_with_options(
          num_actors=num_workers,
          resources={
              "CPU": cpus_per_worker,
              "GPU": int(use_gpu),
          },
          runtime_env=runtime_env,
      )
  
      run_fn_on_actors(
          actors=actors,
          fn=lambda: os.environ.pop("OMP_NUM_THREADS", None),
      )
  
      # Get IPs and ports for all actors
      ip_ports = get_ip_port_actors(actors=actors)
  
      # Rank 0 is the master addr/port
      master_addr, master_port = ip_ports[0]
  
      if use_gpu:
          # Extract IPs
          actor_ips = [ipp[0] for ipp in ip_ports]
  
          # Get allocated GPU IDs for all actors
          gpu_ids = get_gpu_ids_actors(actors=actors)
  
          # Build a map of IP to all allocated GPUs on that machine
          ip_to_gpu_map = map_ips_to_gpus(ips=actor_ips, gpus=gpu_ids)
  
          # Set the environment variables on the workers
          set_cuda_visible_devices(
              actors=actors,
              actor_ips=actor_ips,
              ip_to_gpus=ip_to_gpu_map,
          )
  
          use_gpu_ids = [gi[0] for gi in gpu_ids]
      else:
          use_gpu_ids = [0] * num_workers
  
      cmds = [
          [
              "python",
              path,
              "worker",
              "--num-epochs",
              str(num_epochs),
              "--num-workers",
              str(num_workers),
              "--rank",
              str(rank),
              "--master-addr",
              master_addr,
              "--master-port",
              str(master_port),
              "--batch-size",
              str(config["batch_size"]),
          ]
          + (["--use-gpu"] if use_gpu else [])
          + (["--gpu-id", str(use_gpu_ids[rank])] if use_gpu else [])
          for rank in range(num_workers)
      ]
  
      run_fn_on_actors(
          actors=actors,
          fn=lambda: os.environ.setdefault("OMP_NUM_THREADS", "1"),
      )
  
      start_time = time.monotonic()
      run_commands_on_actors(actors=actors, cmds=cmds)
      time_taken = time.monotonic() - start_time
  
      loss = 0.0
      if os.path.exists(VANILLA_RESULT_JSON):
          with open(VANILLA_RESULT_JSON, "r") as f:
              result = json.load(f)
          loss = result["loss"]
          local_time_taken = result["local_time_taken"]
  
      return time_taken, local_time_taken, loss
  
imports: import json
  import os
  import time
  from pathlib import Path
  from typing import Dict, Tuple
  import click
  import numpy
  import torch
  from torch import nn, distributed
  from torch.utils.data import DataLoader, DistributedSampler
  from torch.utils.data.dataloader import default_collate
  from torchvision import datasets
  from torchvision.transforms import ToTensor
label_desc: Use image node_id 7 for calling a global function with example usage: train_torch_vanilla() and returns: time_taken,  local_time_taken,  loss