code: def _distcpu_infer(self):
      log.info("distcpu infer data generator")
      client_id = os.getpid()
      graph = DistGraphClient(
          self.config,
          self.config.shard_num,
          self.ip_list_file,
          client_id,
      )
  
      # total number rank = total fleet workers * dataloader workers
      nrank = self._worker_info.num_workers * fleet.worker_num()
      rank = (
          self._worker_info.fid
          + self._worker_info.num_workers * fleet.worker_index()
      )
      log.info("rank %s in total nrank %s" % (rank, nrank))
  
      pipeline = []
      pipeline.append(
          NodeGenerator(
              self.config,
              graph,
              rank=rank,
              nrank=nrank,
              gen_mode="infer_node_generator",
          )
      )
      pipeline.append(
          WalkGenerator(
              self.config,
              graph,
              rank=rank,
              nrank=nrank,
              gen_mode="infer_walk_generator",
          )
      )
      pipeline.append(
          EgoGraphGenerator(
              self.config,
              graph,
              rank=rank,
              nrank=nrank,
              sample_list=self.config.infer_sample_num_list,
          )
      )
  
      generator = Generator()
      for p in pipeline:
          generator = generator.apply(p)
  
      cc = 0
      for walks in generator():
          for egos in walks:
              for ego in egos:
                  yield [ego, ego]
                  cc += 1
  
imports: import os
  import sys
  import time
  import warnings
  import functools
  import numpy
  from collections import defaultdict, OrderedDict
  import paddle.distributed.fleet
  import pgl
  from pgl.graph_kernel import skip_gram_gen_pair
  from pgl.utils.logger import log
  from pgl.distributed import DistGraphClient, DistGraphServer
  from pgl.utils.data import Dataloader, StreamDataset
  from pgl.distributed import helper
  from pgl.utils.logger import log
  from utils.config import prepare_config
  from datasets.node import NodeGenerator
  from datasets.walk import WalkGenerator
  from datasets.ego_graph import ego_graph_sample, EgoGraphGenerator
  from datasets.pair import PairGenerator
label_desc: Use image node_id 5 for calling the SageDataset obj's underlying member method code with example usage: obj._distcpu_infer() without return types