class_docstr: Run a DataFlow in >=1 processes, with ZeroMQ for communication.
  It will fork the calling process of :meth:`reset_state()`,
  and collect datapoints from the given dataflow in each process by ZeroMQ IPC pipe.
  This is typically faster than :class:`MultiProcessRunner`.
  
  Note:
      1. (Data integrity) An iterator cannot run faster automatically -- what's happening is
         that the process will be forked ``num_proc`` times.
         There will be ``num_proc`` dataflow running in parallel and **independently**.
         As a result, we have the following guarantee on the dataflow correctness:
  
         a. When ``num_proc=1``, this dataflow produces the same data as the
            given dataflow in the same order.
         b. When ``num_proc>1``, if each sample from the given dataflow is i.i.d.,
            then this dataflow produces the **same distribution** of data as the given dataflow.
            This implies that there will be duplication, reordering, etc.
            You probably only want to use it for training.
  
            For example, if your original dataflow contains no randomness and produces the same first datapoint,
            then after parallel prefetching, the datapoint will be produced ``num_proc`` times
            at the beginning.
            Even when your original dataflow is fully shuffled, you still need to be aware of the
            `Birthday Paradox <https://en.wikipedia.org/wiki/Birthday_problem>`_
            and know that you'll likely see duplicates.
  
         To utilize parallelism with more strict data integrity, you can use
         the parallel versions of :class:`MapData`: :class:`MultiThreadMapData`, :class:`MultiProcessMapData`.
      2. `reset_state()` of the given dataflow will be called **once and only once** in the worker processes.
      3. The fork of processes happened in this dataflow's `reset_state()` method.
         Please note that forking a TensorFlow GPU session may be unsafe.
         If you're managing this dataflow on your own,
         it's better to fork before creating the session.
      4. (Fork-safety) After the fork has happened, this dataflow becomes not fork-safe.
         i.e., if you fork an already reset instance of this dataflow,
         it won't be usable in the forked process. Therefore, do not nest two `MultiProcessRunnerZMQ`.
      5. (Thread-safety) ZMQ is not thread safe. Therefore, do not call :meth:`get_data` of the same dataflow in
         more than 1 threads.
      6. This dataflow does not support windows. Use `MultiProcessRunner` which works on windows.
      7. (For Mac only) A UNIX named pipe will be created in the current directory.
         However, certain non-local filesystem such as NFS/GlusterFS/AFS doesn't always support pipes.
         You can change the directory by ``export TENSORPACK_PIPEDIR=/other/dir``.
         In particular, you can use somewhere under '/tmp' which is usually local.
  
         Note that some non-local FS may appear to support pipes and code
         may appear to run but crash with bizarre error.
         Also note that ZMQ limits the maximum length of pipe path.
         If you hit the limit, you can set the directory to a softlink
         which points to a local directory.
class_docstr_tok: ["Run","a","DataFlow","in",">","=1","processes",",","with","ZeroMQ","for","communication",".","It","will","fork","the","calling","process","of",":","meth",":","`","reset_state","(",")","`",",","and","collect","datapoints","from","the","given","dataflow","in","each","process","by","ZeroMQ","IPC","pipe",".","This","is","typically","faster","than",":","class",":","`","MultiProcessRunner","`",".","Note",":","1",".","(","Data","integrity",")","An","iterator","can","not","run","faster","automatically","--","what","'s","happening","is","that","the","process","will","be","forked","``","num_proc","``","times",".","There","will","be","``","num_proc","``","dataflow","running","in","parallel","and","*","*","independently","*","*",".","As","a","result",",","we","have","the","following","guarantee","on","the","dataflow","correctness",":","a",".","When","``","num_proc=1","``",",","this","dataflow","produces","the","same","data","as","the","given","dataflow","in","the","same","order",".","b",".","When","``","num_proc",">","1","``",",","if","each","sample","from","the","given","dataflow","is","i.i.d.",",","then","this","dataflow","produces","the","*","*","same","distribution","*","*","of","data","as","the","given","dataflow",".","This","implies","that","there","will","be","duplication",",","reordering",",","etc",".","You","probably","only","want","to","use","it","for","training",".","For","example",",","if","your","original","dataflow","contains","no","randomness","and","produces","the","same","first","datapoint",",","then","after","parallel","prefetching",",","the","datapoint","will","be","produced","``","num_proc","``","times","at","the","beginning",".","Even","when","your","original","dataflow","is","fully","shuffled",",","you","still","need","to","be","aware","of","the","`","Birthday","Paradox","<","https",":","\/\/en.wikipedia.org\/wiki\/Birthday_problem",">","`","_","and","know","that","you","'ll","likely","see","duplicates",".","To","utilize","parallelism","with","more","strict","data","integrity",",","you","can","use","the","parallel","versions","of",":","class",":","`","MapData","`",":",":","class",":","`","MultiThreadMapData","`",",",":","class",":","`","MultiProcessMapData","`",".","2",".","`","reset_state","(",")","`","of","the","given","dataflow","will","be","called","*","*","once","and","only","once","*","*","in","the","worker","processes",".","3",".","The","fork","of","processes","happened","in","this","dataflow","'s","`","reset_state","(",")","`","method",".","Please","note","that","forking","a","TensorFlow","GPU","session","may","be","unsafe",".","If","you","'re","managing","this","dataflow","on","your","own",",","it","'s","better","to","fork","before","creating","the","session",".","4",".","(","Fork-safety",")","After","the","fork","has","happened",",","this","dataflow","becomes","not","fork-safe",".","i.e.",",","if","you","fork","an","already","reset","instance","of","this","dataflow",",","it","wo","n't","be","usable","in","the","forked","process",".","Therefore",",","do","not","nest","two","`","MultiProcessRunnerZMQ","`",".","5",".","(","Thread-safety",")","ZMQ","is","not","thread","safe",".","Therefore",",","do","not","call",":","meth",":","`","get_data","`","of","the","same","dataflow","in","more","than","1","threads",".","6",".","This","dataflow","does","not","support","windows",".","Use","`","MultiProcessRunner","`","which","works","on","windows",".","7",".","(","For","Mac","only",")","A","UNIX","named","pipe","will","be","created","in","the","current","directory",".","However",",","certain","non-local","filesystem","such","as","NFS\/GlusterFS\/AFS","does","n't","always","support","pipes",".","You","can","change","the","directory","by","``","export","TENSORPACK_PIPEDIR=\/other\/dir","``",".","In","particular",",","you","can","use","somewhere","under","'\/tmp","'","which","is","usually","local",".","Note","that","some","non-local","FS","may","appear","to","support","pipes","and","code","may","appear","to","run","but","crash","with","bizarre","error",".","Also","note","that","ZMQ","limits","the","maximum","length","of","pipe","path",".","If","you","hit","the","limit",",","you","can","set","the","directory","to","a","softlink","which","points","to","a","local","directory","."]
docstr: Args:
      ds (DataFlow): input DataFlow.
      num_proc (int): number of processes to use.
      hwm (int): the zmq "high-water mark" (queue size) for both sender and receiver.
code: def __init__(self, ds, num_proc=1, hwm=50):
      """
      Args:
          ds (DataFlow): input DataFlow.
          num_proc (int): number of processes to use.
          hwm (int): the zmq "high-water mark" (queue size) for both sender and receiver.
      """
      super(MultiProcessRunnerZMQ, self).__init__()
  
      self.ds = ds
      self.num_proc = num_proc
      self._hwm = hwm
  
      if num_proc > 1:
          logger.info(
              "[MultiProcessRunnerZMQ] Will fork a dataflow more than one times. "
              "This assumes the datapoints are i.i.d."
          )
      try:
          self._size = ds.__len__()
      except NotImplementedError:
          self._size = -1
  
imports: import atexit
  import pickle
  import errno
  import traceback
  import itertools
  import multiprocessing
  import os
  import sys
  import uuid
  import weakref
  from contextlib import contextmanager
  import zmq
  from six.moves import queue, range
  from ..utils import logger
  from ..utils.concurrency import StoppableThread, enable_death_signal, ensure_proc_terminate, start_proc_mask_signal
  from ..utils.serialize import dumps_once, loads_once
  from .base import DataFlow, DataFlowReentrantGuard, DataFlowTerminated, ProxyDataFlow
label_desc: Use image node_id 1 to create a new MultiProcessRunnerZMQ object from inherited base classes: _MultiProcessZMQDataFlow with example: obj = MultiProcessRunnerZMQ(ds, num_proc, hwm)