code: def run_workflow(rank, devices, scheduler_address):
      from model import Sage, train_model
  
      # Below sets gpu_number
      dev_id = devices[rank]
      initalize_pytorch_worker(dev_id)
      device = torch.device(f"cuda:{dev_id}")
      # cugraph dask client initialization
      client = create_dask_client(scheduler_address)
  
      # Pytorch training worker initialization
      dist_init_method = "tcp://{master_ip}:{master_port}".format(
          master_ip="127.0.0.1", master_port="12346"
      )
  
      torch.distributed.init_process_group(
          backend="nccl",
          init_method=dist_init_method,
          world_size=len(devices),
          rank=rank,
      )
  
      print(f"rank {rank}.", flush=True)
      print("Initalized across GPUs.")
  
      event = Dask_Event("cugraph_gs_creation_event")
      if rank == 0:
          (
              gs,
              train_idx,
              valid_idx,
              test_idx,
              num_classes,
          ) = create_cugraph_graphstore_from_dgl_dataset(
              "ogbn-products", single_gpu=False
          )
          client.publish_dataset(cugraph_gs=gs)
          client.publish_dataset(train_idx=train_idx)
          client.publish_dataset(valid_idx=valid_idx)
          client.publish_dataset(test_idx=test_idx)
          client.publish_dataset(num_classes=num_classes)
          event.set()
      else:
          if event.wait(timeout=1000):
              gs = client.get_dataset("cugraph_gs")
              train_idx = client.get_dataset("train_idx")
              valid_idx = client.get_dataset("valid_idx")
              test_idx = client.get_dataset("test_idx")
              num_classes = client.get_dataset("num_classes")
          else:
              raise RuntimeError(
                  f"Fetch cugraph_gs to worker_id {rank} failed"
              )
  
      torch.distributed.barrier()
      print(
          f"Loading cugraph_store to worker {rank} is complete",
          flush=True,
      )
      dataloader = create_dataloader(gs, train_idx, device)
      print("Data Loading Complete", flush=True)
      num_feats = gs.ndata["feat"]["_N"].shape[1]
      hid_size = 256
      # Load Training example
      model = Sage(num_feats, hid_size, num_classes).to(device)
      model = torch.nn.parallel.DistributedDataParallel(
          model,
          device_ids=[device],
          output_device=device,
      )
      torch.distributed.barrier()
      n_epochs = 10
      total_st = time.time()
      opt = torch.optim.Adam(model.parameters(), lr=0.01)
      train_model(model, gs, opt, dataloader, n_epochs, rank, valid_idx)
      torch.distributed.barrier()
      total_et = time.time()
      print(
          f"Total time taken on n_epochs {n_epochs} = {total_et-total_st} s",
          f"measured by worker = {rank}",
      )
  
      # cleanup dask cluster
      if rank == 0:
          client.unpublish_dataset("cugraph_gs")
          client.unpublish_dataset("train_idx")
          client.unpublish_dataset("valid_idx")
          client.unpublish_dataset("test_idx")
          event.clear()
      print("Workflow completed")
      print("---" * 10)
      Comms.destroy()
  
file_path: cugraph/python/cugraph-dgl/examples/multi_trainer_MG_example/workflow.py
imports: import dgl
  import torch
  import time
  from distributed import Client, Event
  import tempfile
  from cugraph.dask.comms import comms
label_desc: Use image node_id 8 for calling a global function with example usage: run_workflow(rank, devices, scheduler_address) without return types