code: def main(args):
      paddle.set_device("cpu")
      paddle.enable_static()
  
      fleet.init()
  
      if args.num_nodes is None:
          num_nodes = load(args.dataset).num_nodes
      else:
          num_nodes = args.num_nodes
  
      loss = StaticSkipGramModel(
          num_nodes, args.neg_num, args.embed_size, sparse=True
      )
  
      optimizer = paddle.optimizer.Adam(
          args.learning_rate, lazy_mode=True
      )
      dist_strategy = fleet.DistributedStrategy()
      dist_strategy.a_sync = True
      optimizer = fleet.distributed_optimizer(optimizer, dist_strategy)
      optimizer.minimize(loss)
  
      # init and run server or worker
      if fleet.is_server():
          fleet.init_server()
          fleet.run_server()
  
      if fleet.is_worker():
          place = paddle.CPUPlace()
          exe = paddle.static.Executor(place)
          exe.run(paddle.static.default_startup_program())
          fleet.init_worker()
  
          graph = load(args.dataset)
          # bind gen
          train_ds = ShardedDataset(graph.nodes)
          collate_fn = BatchNode2vecWalk(
              graph,
              args.walk_len,
              args.win_size,
              args.neg_num,
              args.neg_sample_type,
              args.p,
              args.q,
          )
          data_loader = Dataloader(
              train_ds,
              batch_size=args.batch_size,
              shuffle=True,
              num_workers=args.sample_workers,
              collate_fn=collate_fn,
          )
  
          cpu_num = int(os.environ.get("CPU_NUM", 1))
          if int(cpu_num) > 1:
              parallel_places = [paddle.CPUPlace()] * cpu_num
              exec_strategy = paddle.static.ExecutionStrategy()
              exec_strategy.num_threads = int(cpu_num)
              build_strategy = paddle.static.BuildStrategy()
              build_strategy.reduce_strategy = (
                  paddle.static.BuildStrategy.ReduceStrategy.Reduce
              )
              compiled_prog = paddle.static.CompiledProgram(
                  paddle.static.default_main_program()
              ).with_data_parallel(
                  loss_name=loss.name,
                  places=parallel_places,
                  build_strategy=build_strategy,
                  exec_strategy=exec_strategy,
              )
          else:
              compiled_prog = paddle.static.default_main_program()
  
          for epoch in range(args.epoch):
              train_loss = train(exe, compiled_prog, data_loader, loss)
              log.info(
                  "Runing epoch:%s\t train_loss:%.6f", epoch, train_loss
              )
          fleet.stop_worker()
  
          if fleet.is_first_worker():
              fleet.save_persistables(
                  exe, "./model", paddle.static.default_main_program()
              )
  
file_path: PGL/examples/node2vec/train_distributed_cpu.py
imports: import argparse
  import time
  import os
  import math
  import numpy
  import paddle
  import paddle.distributed.fleet
  import pgl
  from pgl.utils.logger import log
  import yaml
  from easydict import EasyDict
  from pgl.utils.data import Dataloader
  from model import SkipGramModel
  from dataset import ShardedDataset
  from dataset import BatchNode2vecWalk
label_desc: Use image node_id 4 for calling a global function with example usage: main(args) without return types