Baselines: ConnectionResetError with multiprocessing in A2C

[kxgao]

I've modified train.py in A2C for cartpole in Gym, and I'm running into a ConnectionResetError while testing with two processes. I'm using python 3.5, gym 0.9.2, and tensorflow-cpu 1.3.0 on Ubuntu 14.04.

Here is the relevant portion of my version of train.py:

ncpu = num_processes
config = tf.ConfigProto(allow_soft_placement=True, intra_op_parallelism_threads=ncpu, inter_op_parallelism_threads=ncpu)
tf.Session(config=config).__enter__()
set_global_seeds(seed)

def make_env(rank):
    env = gym.make(env_id)
    env.seed(seed + rank)
    if logger.get_dir():
        env = bench.Monitor(env, os.path.join(logger.get_dir(), 'train-{}.monitor.json'.format(rank)))
    return env

env = SubprocVecEnv([make_env(i) for i in range(ncpu)])
env = VecNormalize(env)

and here is the error I get when num_processes = 2:

Process Process-1:
Traceback (most recent call last):
  File "/home/katelyng/anaconda3/lib/python3.5/multiprocessing/process.py", line 249, in _bootstrap
    self.run()
  File "/home/katelyng/anaconda3/lib/python3.5/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/home/katelyng/baselines/baselines/common/vec_env/subproc_vec_env.py", line 8, in worker
    env = env_fn_wrapper.x()
TypeError: 'Monitor' object is not callable
Process Process-2:
Traceback (most recent call last):
  File "/home/katelyng/anaconda3/lib/python3.5/multiprocessing/process.py", line 249, in _bootstrap
    self.run()
  File "/home/katelyng/anaconda3/lib/python3.5/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/home/katelyng/baselines/baselines/common/vec_env/subproc_vec_env.py", line 8, in worker
    env = env_fn_wrapper.x()
TypeError: 'Monitor' object is not callable
Traceback (most recent call last):
  File "/home/katelyng/anaconda3/lib/python3.5/runpy.py", line 184, in _run_module_as_main
    "__main__", mod_spec)
  File "/home/katelyng/anaconda3/lib/python3.5/runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "/home/katelyng/AdaptiveRL/rl-environments/examples/a2c_baselines/train.py", line 116, in <module>
    main()
  File "/home/katelyng/AdaptiveRL/rl-environments/examples/a2c_baselines/train.py", line 111, in main
    seed=args.seed,
  File "/home/katelyng/AdaptiveRL/rl-environments/examples/a2c_baselines/train.py", line 44, in train
    env = SubprocVecEnv([make_env(i) for i in range(ncpu)])
  File "/home/katelyng/baselines/baselines/common/vec_env/subproc_vec_env.py", line 49, in __init__
    observation_space, action_space = self.remotes[0].recv()
  File "/home/katelyng/anaconda3/lib/python3.5/multiprocessing/connection.py", line 250, in recv
    buf = self._recv_bytes()
  File "/home/katelyng/anaconda3/lib/python3.5/multiprocessing/connection.py", line 407, in _recv_bytes
    buf = self._recv(4)
  File "/home/katelyng/anaconda3/lib/python3.5/multiprocessing/connection.py", line 379, in _recv
    chunk = read(handle, remaining)
ConnectionResetError: [Errno 104] Connection reset by peer

I would appreciate any help with debugging this.

[wetliu]

Hi. I have the same problem as you do. Would you mind posting your solution here?
Thanks.

[kxgao]

Modifying make_env worked for me:

def make_env(rank):
    def _thunk():
        env = base.make_env(env_id, process_idx=rank, outdir=logger.get_dir())
        env.seed(seed + rank)
        if logger.get_dir():
            env = bench.Monitor(env, os.path.join(logger.get_dir(), 'train-{}.monitor.json'.format(rank)))
        return env
    return _thunk

[yukang2017]

Hi,
@kxgao
Would you please tell me more detail about “base.make_env” implementation？
I meet the same issue. Thank you inadvance.

Best~
Yukang

[zpschang]

I met the same error with a little difference in my code.
I found that the directory I passed to bench.Monitor doesn't exist, and then ConnectionResetError occurred.
After mkdir the monitor directory, the error was fixed.

[kxgao]

Sorry about the late reply. I don't know if my answer is still relevant after the code restructuring; it worked for the previous version of baselines. base.make_env is a function that returns a constructed environment.

def make_env(env_id, process_idx=0, outdir=None):
    env = gym.make(env_id)
    return env

[zfar-]

Hi ,
I have encountered the same error, but recently the repository have changed, even making sub directory "monitor" doesn't worked, is there something that I am missing ?

def make_env(env_id, env_type, subrank=0, seed=None, reward_scale=1.0, gamestate=None, wrapper_kwargs=None):
    mpi_rank = MPI.COMM_WORLD.Get_rank() if MPI else 0
    if env_type == 'atari':
        env = make_atari(env_id)
    elif env_type == 'retro':
        import retro
        gamestate = gamestate or retro.State.DEFAULT
        env = retro_wrappers.make_retro(game=env_id, max_episode_steps=10000, use_restricted_actions=retro.Actions.DISCRETE, state=gamestate)
    else:
        env = gym.make(env_id)

    env.seed(seed + subrank if seed is not None else None)

    print(" Logger dir {}".format(logger.get_dir()))    
    if logger.get_dir():
        env = Monitor(env,
            logger.get_dir() and os.path.join(logger.get_dir(), str(mpi_rank) + '.' + str(subrank)),
            allow_early_resets=True)

    if env_type == 'atari':
         return wrap_deepmind(env, **wrapper_kwargs)
    elif reward_scale != 1:
         return retro_wrappers.RewardScaler(env, reward_scale)
    else:
        return env