multistream api _train_epoch

_train_epoch.py

    def _train_epoch(self, data_iterables, cur_epoch=0, total_examples=None,
                     total_words=None, queue_factor=2, report_delay=1.0):
        """Train one epoch."""
        _reset_performance_metrics()

        job_queue = Queue(maxsize=queue_factor * self.workers)
        progress_queue = Queue(maxsize=(queue_factor + 1) * self.workers)

        workers = [
            threading.Thread(
                target=self._worker_loop,
                args=(job_queue, progress_queue,))
            for _ in xrange(self.workers)
        ]

        workers.extend(
            threading.Thread(
                target=self._job_producer,
                args=(data_iterable, job_queue),
                kwargs={'cur_epoch': cur_epoch, 'total_examples': total_examples, 'total_words': total_words}
            ) for data_iterable in data_iterables
        )

        for thread in workers:
            thread.daemon = True  # make interrupting the process with ctrl+c easier
            thread.start()

        trained_word_count, raw_word_count, job_tally = self._log_epoch_progress(
            progress_queue, job_queue, cur_epoch=cur_epoch, total_examples=total_examples, total_words=total_words,
            report_delay=report_delay)

        return trained_word_count, raw_word_count, job_tally