code: def train_prog(
      exe, program, loss, node2vec_pyreader, args, train_steps
  ):
      trainer_id = int(os.getenv("PADDLE_TRAINER_ID", "0"))
      step = 0
      if not os.path.exists(args.save_path):
          os.makedirs(args.save_path)
      while True:
          try:
              begin_time = time.time()
              (loss_val,) = exe.run(program, fetch_list=[loss])
              log.info(
                  "step %s: loss %.5f speed: %.5f s/step"
                  % (step, np.mean(loss_val), time.time() - begin_time)
              )
              step += 1
          except F.core.EOFException:
              node2vec_pyreader.reset()
  
          if step % args.steps_per_save == 0 or step == train_steps:
              save_path = args.save_path
              if trainer_id == 0:
                  model_path = os.path.join(save_path, "%s" % step)
                  fleet.save_persistables(exe, model_path)
          if step == train_steps:
              break
  
file_path: PGL/legacy/examples/distribute_metapath2vec/cluster_train.py
imports: import argparse
  import time
  import os
  import math
  import numpy
  import paddle.fluid
  import paddle.fluid.layers
  from paddle.fluid.incubate.fleet.parameter_server.distribute_transpiler import fleet
  import paddle.fluid.incubate.fleet.base.role_maker
  from paddle.fluid.incubate.fleet.parameter_server.distribute_transpiler import StrategyFactory
  from pgl.utils.logger import log
  from model import Metapath2vecModel
  from graph import m2vGraph
  from utils import load_config
  from walker import multiprocess_data_generator
label_desc: Use image node_id 3 for calling a global function with example usage: train_prog(exe, program, loss, node2vec_pyreader, args, train_steps) without return types