code: def send_dataflow_zmq(df, addr, hwm=50, format=None, bind=False):
      """
      Run DataFlow and send data to a ZMQ socket addr.
      It will serialize and send each datapoint to this address with a PUSH socket.
      This function never returns.
  
      Args:
          df (DataFlow): Will infinitely loop over the DataFlow.
          addr: a ZMQ socket endpoint.
          hwm (int): ZMQ high-water mark (buffer size)
          format (str): The serialization format.
               Default format uses :mod:`utils.serialize`.
               This format works with :class:`dataflow.RemoteDataZMQ`.
               An alternate format is 'zmq_ops', used by https://github.com/tensorpack/zmq_ops
               and :class:`input_source.ZMQInput`.
          bind (bool): whether to bind or connect to the endpoint address.
      """
      assert format in [None, "zmq_op", "zmq_ops"]
      if format is None:
          dump_fn = dumps
      else:
          from zmq_ops import dump_arrays
  
          dump_fn = dump_arrays
  
      ctx = zmq.Context()
      socket = ctx.socket(zmq.PUSH)
      socket.set_hwm(hwm)
      if bind:
          socket.bind(addr)
      else:
          socket.connect(addr)
      try:
          df.reset_state()
          logger.info(
              "Serving data to {} with {} format ...".format(
                  addr, "default" if format is None else "zmq_ops"
              )
          )
          INTERVAL = 200
          q = deque(maxlen=INTERVAL)
  
          try:
              total = len(df)
          except NotImplementedError:
              total = 0
          tqdm_args = get_tqdm_kwargs(leave=True, smoothing=0.8)
          tqdm_args["bar_format"] = (
              tqdm_args["bar_format"] + "{postfix}"
          )
          while True:
              with tqdm.trange(total, **tqdm_args) as pbar:
                  for dp in df:
                      start = time.time()
                      socket.send(dump_fn(dp), copy=False)
                      q.append(time.time() - start)
                      pbar.update(1)
                      if pbar.n % INTERVAL == 0:
                          avg = "{:.3f}".format(sum(q) / len(q))
                          pbar.set_postfix({"AvgSendLat": avg})
      finally:
          logger.info("Exiting send_dataflow_zmq ...")
          socket.setsockopt(zmq.LINGER, 0)
          socket.close()
          if not ctx.closed:
              ctx.destroy(0)
  
file_path: tensorpack/tensorpack/dataflow/remote.py
imports: import multiprocessing
  import time
  from collections import deque
  import tqdm
  from ..utils import logger
  from ..utils.concurrency import DIE
  from ..utils.serialize import dumps, loads
  from ..utils.utils import get_tqdm_kwargs
  from .base import DataFlow, DataFlowReentrantGuard
label_desc: Use image node_id 1 for calling a global function with example usage: send_dataflow_zmq(df, addr, hwm, format, bind) without return types