Multiprocessing OpenAI ROS environments w/ python multiprocessing library

edit

Hello,

I am trying to train a Turtlebot agent using the OpenAI_ROS package. While my goal is forking multiple processes to create multiple gazebo environments, I created an only single process, to start from a simple scenario.

OpenAI baseline package provides SubprocVecEnv(), which depends on the python multiprocessing library. I was able to initialize a TurtleBot3WorldEnv using SubprocVecEnv().

However, for some reason, the Turtlebot3 spawned in the world cannot subscribe to messages even though they are all available in ROS. For example, I can echo /odom and /scan, but I cannot subscribe to them from TurtleBot3Env. Instead, I get an error message: " Current /odom not ready yet, retrying for getting doom".

I use ROS noetic w/ Ubuntu 2020 Does anyone know how to solve this issue?

Best,

env_fn below is lambda: make_env() which is a simple wrapper of gym.make()

import multiprocessing as mp
import os
import numpy as np
from .vec_env import VecEnv, CloudpickleWrapper, clear_mpi_env_vars
import time

class SubprocVecEnv(VecEnv):
def __init__(self, env_fns, spaces=None, context='fork'):  #env_fns: iterable of callables -  functions that create environments to run in subprocesses. Need to be cloud-pickleable

    self.waiting = False
    self.closed = False
    nenvs = len(env_fns)
    ctx = mp.get_context(context)
    self.remotes, self.work_remotes = zip(*[ctx.Pipe() for _ in range(nenvs)]) # remote (parent) and work_remote (child) are connected by a pipe
    self.ps = [ctx.Process(target=worker, args=(work_remote, remote, CloudpickleWrapper(env_fn)))
               for (work_remote, remote, env_fn) in zip(self.work_remotes, self.remotes, env_fns)]
    for p in self.ps:
        p.daemon = True  # if the main process crashes, we should not cause things to hang
        with clear_mpi_env_vars():
            p.start()
    for remote in self.work_remotes:
        remote.close()
    self.remotes[0].send(('get_spaces_spec', None))

    observation_space, action_space, self.spec = self.remotes[0].recv()

    self.viewer = None
    VecEnv.__init__(self, len(env_fns), observation_space, action_space)

def worker(remote, parent_remote, env_fn_wrapper):
parent_remote.close()
env = env_fn_wrapper.x()
try:
    while True:
        cmd, data = remote.recv()
        if cmd == 'step':
            ob, reward, done, info = env.step(data)
            if done:
                ob = env.reset()
            remote.send((ob, reward, done, info))
        elif cmd == 'reset':
            ob = env.reset()
            remote.send(ob)
        elif cmd == 'render':
            remote.send(env.render(mode='rgb_array'))
        elif cmd == 'close':
            remote.close()
            break
        elif cmd == 'get_spaces_spec':
            remote.send((env.observation_space, env.action_space, env.spec))
        else:
            raise NotImplementedError
except KeyboardInterrupt:
    print('SubprocVecEnv worker: got KeyboardInterrupt')
finally:
    env.close()