code: def worker(
      num_layers,
      batch_info,
      graph_wrappers,
      samples,
      feed_name_list,
      use_pyreader,
      graph,
      predict,
  ):
      """worker"""
      pid = os.getppid()
      np.random.seed((int(time.time() * 10000) + pid) % 65535)
  
      graphs = [graph, graph]
  
      def work():
          """work"""
          feed_dict = {}
          ind = 0
          perm = np.arange(0, len(batch_info))
          np.random.shuffle(perm)
          for p in perm:
              batch_src, batch_dst, batch_eid = batch_info[p]
              ind += 1
              ind_start = time.time()
              try:
                  nodes = start_nodes = np.concatenate(
                      [batch_src, batch_dst], 0
                  )
                  eids = []
                  layer_nodes, layer_eids = [], []
                  for layer_idx in reversed(range(num_layers)):
                      if len(start_nodes) == 0:
                          layer_nodes = [nodes] + layer_nodes
                          layer_eids = [eids] + layer_eids
                          continue
                      pred_nodes, pred_eids = graphs[
                          layer_idx
                      ].sample_predecessor(
                          start_nodes,
                          samples[layer_idx],
                          return_eids=True,
                      )
                      last_nodes = nodes
                      nodes, eids = flat_node_and_edge(
                          [nodes, pred_nodes], [eids, pred_eids]
                      )
                      layer_nodes = [nodes] + layer_nodes
                      layer_eids = [eids] + layer_eids
                      # Find new nodes
                      start_nodes = list(set(nodes) - set(last_nodes))
                  if predict is False:
                      eids = (batch_eid * 2 + 1).tolist() + (
                          batch_eid * 2
                      ).tolist()
                      layer_eids[0] = list(
                          set(layer_eids[0]) - set(eids)
                      )
  
                  # layer_nodes[0]: use first layer nodes as all subgraphs' nodes
                  subgraph = graphs[0].subgraph(
                      nodes=layer_nodes[0], eid=layer_eids[0]
                  )
                  node_feat = np.array(layer_nodes[0], dtype="int64")
                  subgraph.node_feat["index"] = node_feat
  
              except Exception as e:
                  print(e)
                  if len(feed_dict) > 0:
                      yield feed_dict
                  continue
              feed_dict = graph_wrappers[0].to_feed(subgraph)
  
              # only reindex from first subgraph
              sub_src_idx = subgraph.reindex_from_parrent_nodes(
                  batch_src
              )
              sub_dst_idx = subgraph.reindex_from_parrent_nodes(
                  batch_dst
              )
  
              feed_dict["src_index"] = sub_src_idx.astype("int64")
              feed_dict["dst_index"] = sub_dst_idx.astype("int64")
              if predict:
                  feed_dict["node_id"] = batch_src.astype("int64")
  
              if use_pyreader:
                  yield [feed_dict[name] for name in feed_name_list]
              else:
                  yield feed_dict
  
      return work
  
file_path: PGL/legacy/examples/unsup_graphsage/reader.py
imports: import os
  import numpy
  import pickle
  import paddle
  import paddle.fluid
  import pgl
  import time
  from pgl.utils.logger import log
  from pgl.utils import mp_reader
label_desc: Use image node_id 5 for calling a global function with example usage: worker(num_layers, batch_info, graph_wrappers, samples, feed_name_list, use_pyreader, graph, predict) and returns: work