      """ Return total number of cpus """
      return mp.cpu_count()
  
  
  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))
  
  
  class PoolProcess():
      """ Pool multiple processes """
      def __init__(self, method, in_queue, out_queue, *args, processes=None, **kwargs):
          self._name = method.__qualname__
          logger.debug("Initializing %s: (target: '%s', processes: %s)",
 
</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         logger.debug("Launching minibatch generator for queue (side: '%s', is_display: %s)",
                     side, is_display)
        for batch_wrapper in load_process:
            with batch_wrapper as batch:
                logger.trace("Yielding batch: (size: %s, item shapes: %s, side:  '%s', "
                             "is_display: %s)",
                             len(batch), [item.shape for item in batch], side, is_display)
                yield batch
        load_process.stop()
        logger.debug("Finished minibatch generator for queue: (side: '%s', is_display: %s)",
 </s> add         logger.debug("Loading minibatch generator: (image_count: %s, side: '%s', is_display: %s, "
                     "do_shuffle: %s)", len(images), side, is_display, do_shuffle)
        self.validate_samples(images)

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

        img_iter = _img_iter(images)
        while True:
            batch = list()
            for _ in range(batchsize):
                img_path = next(img_iter)
                data = self.process_face(img_path, side, is_display)
                batch.append(data)
            batch = list(zip(*batch))
            batch = [np.array(x, dtype="float32") for x in batch]
            logger.trace("Yielding batch: (size: %s, item shapes: %s, side:  '%s', "
                         "is_display: %s)",
                         len(batch), [item.shape for item in batch], side, is_display)
            yield batch

        logger.debug("Finished minibatch generator: (side: '%s', is_display: %s)", </s> remove     def __init__(self, generator, prefetch=1):
        threading.Thread.__init__(self)
        self.queue = Queue.Queue(maxsize=prefetch)
 </s> add     def __init__(self, generator, prefetch=1, thread_count=2,
                 queue=None, args=None, kwargs=None):
        super().__init__(target=self._run, thread_count=thread_count)
        self.queue = queue or Queue.Queue(prefetch) </s> remove         return [thread.err for thread in self._threads]
 </s> add         return [thread.err for thread in self._threads if thread.err]
      """ Return total number of cpus """
      return mp.cpu_count()
  
  




























































































































































































































































  class PoolProcess():
      """ Pool multiple processes """
      def __init__(self, method, in_queue, out_queue, *args, processes=None, **kwargs):
          self._name = method.__qualname__
          logger.debug("Initializing %s: (target: '%s', processes: %s)",
keep keep keep keep replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace 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