  
      @property
      def errors(self):
          """ Return a list of thread errors """
          return [thread.err for thread in self._threads]
  
      def check_and_raise_error(self):
          """ Checks for errors in thread and raises them in caller """
          if not self.has_error:
              return
 
</s> remove class ConsumerBuffer():
    """ Memory buffer for consuming """
    def __init__(self, dispatcher, index, data):
        logger.trace("Initializing %s: (dispatcher: '%s', index: %s, data: %s)",
                     self.__class__.__name__, dispatcher, index, data)
        self._data = data
        self._id = index
        self._dispatcher = dispatcher
        logger.trace("Initialized %s", self.__class__.__name__)

    def get(self):
        """ Return Data """
        return self._data

    def free(self):
        """ Return Free """
        self._dispatcher.free(self._id)

    def __enter__(self):
        """ On Enter """
        return self.get()

    def __exit__(self, *args):
        """ On Exit """
        self.free()


class WorkerBuffer():
    """ Memory buffer for working """
    def __init__(self, index, data, stop_event, queue):
        logger.trace("Initializing %s: (index: '%s', data: %s, stop_event: %s, queue: %s)",
                     self.__class__.__name__, index, data, stop_event, queue)
        self._id = index
        self._data = data
        self._stop_event = stop_event
        self._queue = queue
        logger.trace("Initialized %s", self.__class__.__name__)

    def get(self):
        """ Return Data """
        return self._data

    def ready(self):
        """ Worker Ready """
        if self._stop_event.is_set():
            return
        self._queue.put(self._id)

    def __enter__(self):
        """ On Enter """
        return self.get()

    def __exit__(self, *args):
        """ On Exit """
        self.ready()


class FixedProducerDispatcher():
    """
    Runs the given method in N subprocesses
    and provides fixed size shared memory to the method.
    This class is designed for endless running worker processes
    filling the provided memory with data,
    like preparing trainingsdata for neural network training.

    As soon as one worker finishes all worker are shutdown.

    Example:
        # Producer side
        def do_work(memory_gen):
            for memory_wrap in memory_gen:
                # alternative memory_wrap.get and memory_wrap.ready can be used
                with memory_wrap as memory:
                    input, exp_result = prepare_batch(...)
                    memory[0][:] = input
                    memory[1][:] = exp_result

        # Consumer side
        batch_size = 64
        height = width = 256
        batch_shapes = (batch_size, height, width, 3)
        dispatcher = FixedProducerDispatcher(do_work, shapes=[batch_shapes, batch_shapes])
        for batch_wrapper in dispatcher:
            # alternative batch_wrapper.get and batch_wrapper.free can be used
            with batch_wrapper as batch:
                send_batch_to_trainer(batch)
    """
    CTX = mp.get_context("spawn")
    EVENT = CTX.Event

    def __init__(self, method, shapes, in_queue, out_queue,
                 args=tuple(), kwargs={}, ctype=c_float, workers=1, buffers=None):
        logger.debug("Initializing %s: (method: '%s', shapes: %s, ctype: %s, workers: %s, "
                     "buffers: %s)", self.__class__.__name__, method, shapes, ctype, workers,
                     buffers)
        logger.trace("args: %s, kwargs: %s", args, kwargs)
        if buffers is None:
            buffers = workers * 2
        else:
            assert buffers >= 2 and buffers > workers
        self.name = "%s_FixedProducerDispatcher" % str(method)
        self._target_func = method
        self._shapes = shapes
        self._stop_event = self.EVENT()
        self._buffer_tokens = in_queue
        for i in range(buffers):
            self._buffer_tokens.put(i)
        self._result_tokens = out_queue
        worker_data, self.data = self._create_data(shapes, ctype, buffers)
        proc_args = {
            'data': worker_data,
            'stop_event': self._stop_event,
            'target': self._target_func,
            'buffer_tokens': self._buffer_tokens,
            'result_tokens': self._result_tokens,
            'dtype': np.dtype(ctype),
            'shapes': shapes,
            'log_queue': LOG_QUEUE,
            'log_level': logger.getEffectiveLevel(),
            'args': args,
            'kwargs': kwargs
        }
        self._worker = tuple(self._create_worker(proc_args) for _ in range(workers))
        self._open_worker = len(self._worker)
        logger.debug("Initialized %s", self.__class__.__name__)

    @staticmethod
    def _np_from_shared(shared, shapes, dtype):
        """ Numpy array from shared memory """
        arrs = []
        offset = 0
        np_data = np.frombuffer(shared, dtype=dtype)
        for shape in shapes:
            count = np.prod(shape)
            arrs.append(np_data[offset:offset+count].reshape(shape))
            offset += count
        return arrs

    def _create_data(self, shapes, ctype, buffers):
        """ Create data """
        buffer_size = int(sum(np.prod(x) for x in shapes))
        dtype = np.dtype(ctype)
        data = tuple(RawArray(ctype, buffer_size) for _ in range(buffers))
        np_data = tuple(self._np_from_shared(arr, shapes, dtype) for arr in data)
        return data, np_data

    def _create_worker(self, kwargs):
        """ Create Worker """
        return self.CTX.Process(target=self._runner, kwargs=kwargs)

    def free(self, index):
        """ Free memory """
        if self._stop_event.is_set():
            return
        if isinstance(index, ConsumerBuffer):
            index = index.index
        self._buffer_tokens.put(index)

    def __iter__(self):
        """ Iterator """
        return self

    def __next__(self):
        """ Next item """
        return self.next()

    def next(self, block=True, timeout=None):
        """
        Yields ConsumerBuffer filled by the worker.
        Will raise StopIteration if no more elements are available OR any worker is finished.
        Will raise queue.Empty when block is False and no element is available.

        The returned data is safe until ConsumerBuffer.free() is called or the
        with context is left. If you plan to hold on to it after that make a copy.

        This method is thread safe.
        """
        if self._stop_event.is_set():
            raise StopIteration
        i = self._result_tokens.get(block=block, timeout=timeout)
        if i is None:
            self._open_worker -= 1
            raise StopIteration
        if self._stop_event.is_set():
            raise StopIteration
        return ConsumerBuffer(self, i, self.data[i])

    def start(self):
        """ Start Workers """
        for process in self._worker:
            process.start()
        _launched_processes.add(self)

    def is_alive(self):
        """ Check workers are alive """
        for worker in self._worker:
            if worker.is_alive():
                return True
        return False

    def join(self):
        """ Join Workers """
        self.stop()
        while self._open_worker:
            if self._result_tokens.get() is None:
                self._open_worker -= 1
        while True:
            try:
                self._buffer_tokens.get(block=False, timeout=0.01)
            except Queue.Empty:
                break
        for worker in self._worker:
            worker.join()

    def stop(self):
        """ Stop Workers """
        self._stop_event.set()
        for _ in range(self._open_worker):
            self._buffer_tokens.put(None)

    def is_shutdown(self):
        """ Check if stop event is set """
        return self._stop_event.is_set()

    @classmethod
    def _runner(cls, data=None, stop_event=None, target=None,
                buffer_tokens=None, result_tokens=None, dtype=None,
                shapes=None, log_queue=None, log_level=None,
                args=None, kwargs=None):
        """ Shared Memory Object runner """
        # Fork inherits the queue handler, so skip registration with "fork"
        set_root_logger(log_level, queue=log_queue)
        logger.debug("FixedProducerDispatcher worker for %s started", str(target))
        np_data = [cls._np_from_shared(d, shapes, dtype) for d in data]

        def get_free_slot():
            while not stop_event.is_set():
                i = buffer_tokens.get()
                if stop_event.is_set() or i is None or i == "EOF":
                    break
                yield WorkerBuffer(i, np_data[i], stop_event, result_tokens)

        args = tuple((get_free_slot(),)) + tuple(args)
        try:
            target(*args, **kwargs)
        except Exception as ex:
            logger.exception(ex)
            stop_event.set()
        result_tokens.put(None)
        logger.debug("FixedProducerDispatcher worker for %s shutdown", str(target))


 </s> add  </s> remove         queue_in, queue_out = self.make_queues(side, is_preview, is_timelapse)
        training_size = self.training_opts.get("training_size", 256)
        batch_shape = list((
            (batchsize, training_size, training_size, 3),  # sample images
            (batchsize, self.model_input_size, self.model_input_size, 3)))  # Training Image
        # Target images
        batch_shape.extend(tuple([(batchsize, ) + shape for shape in self.model_output_shapes]))
        logger.debug("Batch shapes: %s", batch_shape)

        self.fixed_producer_dispatcher = FixedProducerDispatcher(
            method=self.load_batches,
            shapes=batch_shape,
            in_queue=queue_in,
            out_queue=queue_out,
            args=(images, side, is_display, do_shuffle, batchsize))
        self.fixed_producer_dispatcher.start()
        logger.debug("Batching to queue: (side: '%s', is_display: %s)", side, is_display)
        return self.minibatch(side, is_display, self.fixed_producer_dispatcher)

    def join_subprocess(self):
        """ Join the FixedProduceerDispatcher subprocess from outside this module """
        logger.debug("Joining FixedProducerDispatcher")
        if self.fixed_producer_dispatcher is None:
            logger.debug("FixedProducerDispatcher not yet initialized. Exiting")
            return
        self.fixed_producer_dispatcher.join()
        logger.debug("Joined FixedProducerDispatcher")

    @staticmethod
    def make_queues(side, is_preview, is_timelapse):
        """ Create the buffer token queues for Fixed Producer Dispatcher """
        q_name = "_{}".format(side)
        if is_preview:
            q_name = "{}{}".format("preview", q_name)
        elif is_timelapse:
            q_name = "{}{}".format("timelapse", q_name)
        else:
            q_name = "{}{}".format("train", q_name)
        q_names = ["{}_{}".format(q_name, direction) for direction in ("in", "out")]
        logger.debug(q_names)
        queues = [queue_manager.get_queue(queue) for queue in q_names]
        return queues

    def load_batches(self, mem_gen, images, side, is_display,
                     do_shuffle=True, batchsize=0):
        """ Load the warped images and target images to queue """
        logger.debug("Loading batch: (image_count: %s, side: '%s', is_display: %s, "
                     "do_shuffle: %s)", len(images), side, is_display, do_shuffle)
        self.validate_samples(images)
        # Intialize this for each subprocess
        self._nearest_landmarks = dict()

        def _img_iter(imgs):
            while True:
                if do_shuffle:
                    shuffle(imgs)
                for img in imgs:
                    yield img

        img_iter = _img_iter(images)
        epoch = 0
        for memory_wrapper in mem_gen:
            memory = memory_wrapper.get()
            logger.trace("Putting to batch queue: (side: '%s', is_display: %s)",
                         side, is_display)
            for i, img_path in enumerate(img_iter):
                imgs = self.process_face(img_path, side, is_display)
                for j, img in enumerate(imgs):
                    memory[j][i][:] = img
                epoch += 1
                if i == batchsize - 1:
                    break
            memory_wrapper.ready()
        logger.debug("Finished batching: (epoch: %s, side: '%s', is_display: %s)",
                     epoch, side, is_display)
 </s> add         args = (images, side, is_display, do_shuffle, batchsize)
        batcher = BackgroundGenerator(self.minibatch, thread_count=2, args=args)
        return batcher.iterator() </s> remove             reached max size => this makes 2 prefetched items! One in the
            queue, one waiting for insertion! """
        for item in self.generator:
            self.queue.put(item)
        self.queue.put(None)
 </s> add             reached max size => this makes prefetch + thread_count prefetched items!
            N in the the queue, one waiting for insertion per thread! """
        try:
            for item in self.generator(*self._gen_args, **self._gen_kwargs):
                self.queue.put(item)
            self.queue.put(None)
        except Exception:
            self.queue.put(None)
            raise </s> remove     @staticmethod
    def minibatch(side, is_display, load_process):
 </s> add     def minibatch(self, images, side, is_display, do_shuffle, batchsize):
        return [thread.err for thread in self._threads if thread.err]
  
      @property
      def errors(self):
          """ Return a list of thread errors """
        return [thread.err for thread in self._threads if thread.err]
  
      def check_and_raise_error(self):
          """ Checks for errors in thread and raises them in caller """
          if not self.has_error:
              return
keep keep keep keep replace keep keep keep keep keep
 de-MultiProcess Training

* Replaced multiprocessing in training_data with threading
* Fix cpu affinity issue
* Using multiple threads for BackgroundGenerator 
lib/multithreading.py