Repository: ray-project/ray
File: rllib/algorithms/tests/test_algorithm.py (Python)

Code before:
        # Make sure that EnvRunners have received the correct mapping fn.
        mapped_pols = [
            algo.env_runner.config.policy_mapping_fn(0, None) for _ in range(100)
        ]
        self.assertTrue(all(f"p{i}" in mapped_pols for i in mapped))
        self.assertTrue(not any(f"p{i}" in mapped_pols for i in not_mapped))

    def test_evaluation_in_parallel_to_training(self):
        SECONDS_TO_SLEEP = 2

        class SluggishEnv(gym.Env):
            def __init__(self, config):
                self.action_space = gym.spaces.Discrete(2)
                self.observation_space = gym.spaces.Box(-1, 1, dtype=np.float32)

            def step(self, action):
                time.sleep(SECONDS_TO_SLEEP)
                return self.observation_space.sample(), 1, True, False, {}

            def reset(self, *, seed=None, options=None):
                super().reset(seed=seed)
                return self.observation_space.sample(), {}

        config = (
            ppo.PPOConfig()
            .environment(env=SluggishEnv)
            .evaluation(
                evaluation_parallel_to_training=True,
                evaluation_interval=1,
                evaluation_num_env_runners=1,
                evaluation_duration=1,
                evaluation_duration_unit="timesteps",
            )
            .training(train_batch_size=1, minibatch_size=1)  # Speed things up
        )
        algo = config.build()
        metrics = algo.train()
        # This can only be true if we do not execute training and evaluation in sequence
        assert metrics[TIMERS][TRAINING_ITERATION_TIMER] < SECONDS_TO_SLEEP * 2
        algo.stop()


if __name__ == "__main__":
    import sys

    import pytest

    sys.exit(pytest.main(["-v", __file__]))


Review comment (suggestion):
Could you make a second config where `evaluation_parallel_to_training=False` as both a positive and negative version of the test

Code after:
            algo.env_runner.config.policy_mapping_fn(0, None) for _ in range(100)
        ]
        self.assertTrue(all(f"p{i}" in mapped_pols for i in mapped))
        self.assertTrue(not any(f"p{i}" in mapped_pols for i in not_mapped))

    def test_evaluation_in_parallel_to_training(self):
        SECONDS_TO_SLEEP = 2

        class SluggishEnv(gym.Env):
            def __init__(self, config):
                self.action_space = gym.spaces.Discrete(2)
                self.observation_space = gym.spaces.Box(-1, 1, dtype=np.float32)

            def step(self, action):
                time.sleep(SECONDS_TO_SLEEP)
                return self.observation_space.sample(), 1, True, False, {}

            def reset(self, *, seed=None, options=None):
                super().reset(seed=seed)
                return self.observation_space.sample(), {}

        config = (
            ppo.PPOConfig()
            .environment(env=SluggishEnv)
            .evaluation(
                evaluation_parallel_to_training=True,
                evaluation_interval=1,
                evaluation_num_env_runners=1,
                evaluation_duration=1,
                evaluation_duration_unit="timesteps",
            )
            .training(train_batch_size=1, minibatch_size=1)  # Speed things up
        )
        algo = config.build()
        metrics = algo.train()
        # This can only be true if we do not execute training and evaluation in sequence
        assert metrics["time_this_iter_s"] < SECONDS_TO_SLEEP * 2
        assert metrics["time_this_iter_s"] > SECONDS_TO_SLEEP
        algo.stop()

        config.evaluation(evaluation_parallel_to_training=False)
        algo_2 = config.build()
        metrics_2 = algo_2.train()
        # This must be true if we execute training and evaluation in sequence
        assert metrics_2["time_this_iter_s"] > SECONDS_TO_SLEEP * 2
        algo_2.stop()


if __name__ == "__main__":
    import sys