Class EnvRunnerManager

examples/DeepQNetwork/expreplay.py:199–254

A class which manages a list of :class:`EnvRunner`. Its job is to execute them possibly in parallel and aggregate their results.

class EnvRunnerManager(object):
    """
    A class which manages a list of :class:`EnvRunner`.
    Its job is to execute them possibly in parallel and aggregate their results.
    """
    def __init__(self, env_runners, maximum_staleness):
        """
        Args:
            env_runners (list[EnvRunner]):
            maximum_staleness (int): when >1 environments run in parallel,
                the actual stepping of an environment may happen several steps
                after calls to `EnvRunnerManager.step()`, in order to achieve better throughput.
        """
        assert len(env_runners) > 0
        self._runners = env_runners

        if len(self._runners) > 1:
            # Only use threads when having >1 runners.
            self._populate_job_queue = queue.Queue(maxsize=maximum_staleness)
            self._threads = [self._create_simulator_thread(i) for i in range(len(self._runners))]
            for t in self._threads:
                t.start()

    def _create_simulator_thread(self, idx):
        # spawn a separate thread to run policy
        def populate_job_func():
            exp = self._populate_job_queue.get()
            self._runners[idx].step(exp)

        th = ShareSessionThread(LoopThread(populate_job_func, pausable=False))
        th.name = "SimulatorThread-{}".format(idx)
        return th

    def step(self, exploration):
        """
        Execute one step in any of the runners.
        """
        if len(self._runners) > 1:
            self._populate_job_queue.put(exploration)
        else:
            self._runners[0].step(exploration)

    def reset_stats(self):
        """
        Returns:
            mean, max: two stats of the runners, to be added to backend
        """
        scores = list(itertools.chain.from_iterable([v.total_scores for v in self._runners]))
        for v in self._runners:
            v.total_scores.clear()

        try:
            return np.mean(scores), np.max(scores)
        except Exception:
            logger.exception("Cannot compute total scores in EnvRunner.")
            return None, None