code: def worker(batch_info, graph_wrapper, samples):
      """Worker"""
  
      def work():
          """work"""
          redis_configs = [
              {
                  "host": socket.gethostbyname(socket.gethostname()),
                  "port": 7430,
              },
          ]
          graph = redis_graph.RedisGraph("sub_graph", redis_configs, 64)
          first = True
          for batch_train_samples, batch_train_labels in batch_info:
              start_nodes = batch_train_samples
              nodes = start_nodes
              eids = []
              eid2edges = {}
              for max_deg in samples:
                  pred, pred_eid = graph.sample_predecessor(
                      start_nodes, max_degree=max_deg, return_eids=True
                  )
                  for _dst, _srcs, _eids in zip(
                      start_nodes, pred, pred_eid
                  ):
                      for _src, _eid in zip(_srcs, _eids):
                          eid2edges[_eid] = (_src, _dst)
  
                  last_nodes = nodes
                  nodes = [nodes, pred]
                  eids = [eids, pred_eid]
                  nodes, eids = flat_node_and_edge(nodes, eids)
                  # Find new nodes
                  start_nodes = list(set(nodes) - set(last_nodes))
                  if len(start_nodes) == 0:
                      break
  
              subgraph = graph.subgraph(
                  nodes=nodes,
                  eid=eids,
                  edges=[eid2edges[e] for e in eids],
              )
              sub_node_index = subgraph.reindex_from_parrent_nodes(
                  batch_train_samples
              )
              feed_dict = graph_wrapper.to_feed(subgraph)
              feed_dict["node_label"] = np.expand_dims(
                  np.array(batch_train_labels, dtype="int64"), -1
              )
              feed_dict["node_index"] = sub_node_index
              yield feed_dict
  
      return work
  
file_path: PGL/legacy/examples/distribute_graphsage/reader.py
imports: import os
  import sys
  import numpy
  import pickle
  import paddle
  import paddle.fluid
  import socket
  import pgl
  import time
  from pgl.utils import mp_reader
  from pgl.utils.logger import log
  from pgl import redis_graph
label_desc: Use image node_id 4 for calling a global function with example usage: worker(batch_info, graph_wrapper, samples) and returns: work