class_docstr: Running a DataFlow in >=1 processes using Python multiprocessing utilities.
  It will fork the process that calls :meth:`__init__`, collect datapoints from `ds` in each
  process by a Python :class:`multiprocessing.Queue`.
  
  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. This has more serialization overhead than :class:`MultiProcessRunnerZMQ` when data is large.
      3. You can nest like this: ``MultiProcessRunnerZMQ(MultiProcessRunner(df, num_proc=a), num_proc=b)``.
         A total of ``a`` instances of ``df`` worker processes will be created.
      4. Fork happens in `__init__`. `reset_state()` is a no-op.
         DataFlow in the worker processes will be reset at the time of fork.
      5. This DataFlow does support windows. However, Windows requires more strict picklability on processes,
         which means that some code that's forkable on Linux may not be forkable on Windows. If that happens you'll
         need to re-organize some part of code that's not forkable.
class_docstr_tok: ["Running","a","DataFlow","in",">","=1","processes","using","Python","multiprocessing","utilities",".","It","will","fork","the","process","that","calls",":","meth",":","`","__init__","`",",","collect","datapoints","from","`","ds","`","in","each","process","by","a","Python",":","class",":","`","multiprocessing.Queue","`",".","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",".","This","has","more","serialization","overhead","than",":","class",":","`","MultiProcessRunnerZMQ","`","when","data","is","large",".","3",".","You","can","nest","like","this",":","``","MultiProcessRunnerZMQ","(","MultiProcessRunner","(","df",",","num_proc=a",")",",","num_proc=b",")","``",".","A","total","of","``","a","``","instances","of","``","df","``","worker","processes","will","be","created",".","4",".","Fork","happens","in","`","__init__","`",".","`","reset_state","(",")","`","is","a","no-op",".","DataFlow","in","the","worker","processes","will","be","reset","at","the","time","of","fork",".","5",".","This","DataFlow","does","support","windows",".","However",",","Windows","requires","more","strict","picklability","on","processes",",","which","means","that","some","code","that","'s","forkable","on","Linux","may","not","be","forkable","on","Windows",".","If","that","happens","you'll","need","to","re-organize","some","part","of","code","that","'s","not","forkable","."]
docstr: Args:
      ds (DataFlow): input DataFlow.
      num_prefetch (int): size of the queue to hold prefetched datapoints.
          Required.
      num_proc (int): number of processes to use. Required.
code: def __init__(self, ds, num_prefetch, num_proc):
      """
      Args:
          ds (DataFlow): input DataFlow.
          num_prefetch (int): size of the queue to hold prefetched datapoints.
              Required.
          num_proc (int): number of processes to use. Required.
      """
      # https://docs.python.org/3.6/library/multiprocessing.html?highlight=process#the-spawn-and-forkserver-start-methods
      if os.name == "nt":
          logger.warn(
              "MultiProcessRunner does support Windows. \
  However, Windows requires more strict picklability on processes, which may \
  lead of failure on some of the code."
          )
      super(MultiProcessRunner, self).__init__(ds)
      try:
          self._size = len(ds)
      except NotImplementedError:
          self._size = -1
      assert num_proc > 0, num_proc
      assert num_prefetch > 0, num_prefetch
      self.num_proc = num_proc
      self.num_prefetch = num_prefetch
  
      if num_proc > 1:
          logger.info(
              "[MultiProcessRunner] Will fork a dataflow more than one times. "
              "This assumes the datapoints are i.i.d."
          )
  
      self.queue = mp.Queue(self.num_prefetch)
      self.procs = [
          MultiProcessRunner._Worker(self.ds, self.queue, idx)
          for idx in range(self.num_proc)
      ]
      ensure_proc_terminate(self.procs)
      self._reset_done = False
  
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 MultiProcessRunner object from inherited base classes: ProxyDataFlow with example: obj = MultiProcessRunner(ds, num_prefetch, num_proc)