fixbug(zlx): Modify envpooltest

.github/workflows/envpool_test.yml

@@ -0,0 +1,28 @@
+# This workflow will check pytest
+# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions
+
+name: envpool_test
+
+on: [push, pull_request]
+
+jobs:
+  test_envpooltest:
+    runs-on: ubuntu-latest
+    if: "!contains(github.event.head_commit.message, 'ci skip')"
+    strategy:
+      matrix:
+        python-version: [3.7, 3.8] # Envpool only supports python>=3.7
+
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: do_envpool_test
+        run: |
+          python -m pip install .
+          python -m pip install ".[test,k8s]"
+          python -m pip install ".[envpool]"
+          ./ding/scripts/install-k8s-tools.sh
+          make envpooltest

Makefile

@@ -33,6 +33,10 @@ cudatest:
 	pytest ${TEST_DIR} \
 		-sv -m cudatest
 
+envpooltest:
+	pytest ${TEST_DIR} \
+		-sv -m envpooltest
+
 dockertest:
 	./ding/scripts/docker-test-entry.sh
 

ding/envs/env_manager/__init__.py

@@ -1,4 +1,4 @@
 from .base_env_manager import BaseEnvManager, create_env_manager, get_env_manager_cls
-from .envpool_env_manager import PoolEnvManager
 from .subprocess_env_manager import AsyncSubprocessEnvManager, SyncSubprocessEnvManager
 from .gym_vector_env_manager import GymVectorEnvManager
+# Do not import PoolEnvManager, because it depends on installation of `envpool`

ding/envs/env_manager/envpool_env_manager.py

@@ -1,4 +1,3 @@
-from tkinter.messagebox import NO
 import gym
 from easydict import EasyDict
 from copy import deepcopy
@@ -11,10 +10,10 @@
 except ImportError:
     import sys
     logging.warning("Please install envpool first, use 'pip install envpool'")
-    sys.exit(1)
+    envpool = None
 
 from ding.envs import BaseEnvTimestep
-from ding.utils import ENV_MANAGER_REGISTRY
+from ding.utils import ENV_MANAGER_REGISTRY, deep_merge_dicts
 
 
 @ENV_MANAGER_REGISTRY.register('env_pool')
@@ -37,7 +36,7 @@ def default_config(cls) -> EasyDict:
         type='envpool',
         # Sync  mode: batch_size == env_num
         # Async mode: batch_size <  env_num
-        env_num=16,
+        env_num=8,
         batch_size=8,
         # Unlike other env managers, envpool's seed should be specified in config.
         seed=0,
@@ -48,7 +47,7 @@ def __init__(self, cfg: EasyDict) -> None:
         self._env_num = cfg.env_num
         self._batch_size = cfg.batch_size
         self._seed = cfg.seed
-        self._ready_obs = None
+        self._ready_obs = {}
         self._closed = True
 
     def launch(self) -> None:
@@ -61,10 +60,11 @@ def launch(self) -> None:
 
     def reset(self) -> None:
         self._envs.async_reset()
-        obs, _, _, _ = self._envs.recv()
-        # obs = self._envs.reset()
+        obs, _, _, info = self._envs.recv()
+        env_id = info['env_id']
+        print(env_id)
         obs = obs.astype(np.float32)
-        self._ready_obs = {i: o for i, o in enumerate(obs)}
+        self._ready_obs = {i: o for i, o in zip(env_id, obs)}
 
     def step(self, action) -> Dict[int, namedtuple]:
         env_id = np.array(list(action.keys()))

ding/envs/env_manager/tests/test_envpool_env_manager.py

@@ -5,15 +5,18 @@
 
 from ..envpool_env_manager import PoolEnvManager
 
+env_num_args = [[16, 8]]
 
-@pytest.mark.unittest
+
+@pytest.mark.envpooltest
+@pytest.mark.parametrize('env_num, batch_size', env_num_args)
 class TestPoolEnvManager:
 
-    def test_naive(self):
+    def test_naive(self, env_num, batch_size):
         env_manager_cfg = EasyDict({
             'env_id': 'Pong-v5',
-            'env_num': 16,
-            'batch_size': 8,
+            'env_num': env_num,
+            'batch_size': batch_size,
             'seed': 3,
         })
         env_manager = PoolEnvManager(env_manager_cfg)

ding/worker/collector/sample_serial_collector.py

@@ -239,10 +239,6 @@ def collect(self,
             # TODO(nyz) vectorize this for loop
             for env_id, timestep in timesteps.items():
                 with self._timer:
-                    # In async mode, this env is reset successfully just now.
-                    if self._obs_pool[env_id] is None:
-                        self._obs_pool.update({env_id: timestep.obs})
-                        continue
                     if timestep.info.get('abnormal', False):
                         # If there is an abnormal timestep, reset all the related variables(including this env).
                         # suppose there is no reset param, just reset this env

dizoo/atari/config/serial/pong/pong_dqn_envpool_config.py

@@ -3,7 +3,7 @@
 pong_dqn_envpool_config = dict(
     exp_name='pong_dqn_envpool',
     env=dict(
-        collector_env_num=16,
+        collector_env_num=8,
         collector_batch_size=8,
         evaluator_env_num=8,
         evaluator_batch_size=8,

dizoo/atari/entry/pong_dqn_envpool_main.py

@@ -7,10 +7,8 @@
 from ding.envs import PoolEnvManager
 from ding.policy import DQNPolicy
 from ding.model import DQN
-from ding.utils import set_pkg_seed, deep_merge_dicts
+from ding.utils import set_pkg_seed
 from ding.rl_utils import get_epsilon_greedy_fn
-from dizoo.atari.envs import AtariEnv
-# from dizoo.atari.config.serial.pong.pong_dqn_envpool_config import pong_dqn_envpool_config
 from dizoo.atari.config.serial import pong_dqn_envpool_config
 
 

pytest.ini

@@ -6,6 +6,7 @@ markers =
     cudatest
     algotest
     benchmark
+    envpooltest
     tmp
 
 norecursedirs = ding/hpc_rl/tests