Class MultiProcessRunner

tensorpack/dataflow/parallel.py:142–238

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

class MultiProcessRunner(ProxyDataFlow):
    """
    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&#x27;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&#x27;s not forkable.
    """

    class _Worker(mp.Process):
        def __init__(self, ds, queue, idx):
            super(MultiProcessRunner._Worker, self).__init__()
            self.ds = ds
            self.queue = queue
            self.idx = idx

        def run(self):
            enable_death_signal(_warn=self.idx == 0)
            # reset all ds so each process will produce different data
            self.ds.reset_state()
            while True:
                for dp in self.ds:
                    self.queue.put(dp)

    def __init__(self, ds, num_prefetch, num_proc):
        """
        Args:
            ds (DataFlow): input DataFlow.
            num_prefetch (int): size of the queue to hold prefetched datapoints.