Vector env enable select worker

[duburcqa]

This is just a preliminary step before implementing async sample. Related to #103 but do not fix it.

• Modify step signature to enable run step for a given worker
• reset now returns only the observation of the concerned workers (same for new step method)

---

• Fix __getattr__ for ray vector env
• Fix support of raw list of dict for Batch item assignment

[youkaichao]

@duburcqa Do you want to implement this with asyncio or with loop query by poll mechanism?

I prefer loop query with poll mechanism to change the signature of step, which may look like this:

def step(self,
            action: Optional[np.ndarray],
            id: Optional[Union[int, List[int]]] = None)
    if id is None:
        id = range(self.env_num)
    elif np.isscalar(id):
        id = [id]
    if action is not None: # if action is None, it means to get unfinished step() calls
        assert len(action) == len(id)
            for i, j in enumerate(id):
                self.parent_remote[j].send(['step', action[i]])
                self.waiting_pool.append(self.parent_remote[j])
    result = []
    while len(self.waiting_pool) > 0 and len(result) < N:
        ready_conns = multiprocessing.connection.wait(self.waiting_pool)
        result = [x.recv() for x in ready_conns]
        self.waiting_pool.remove(ready_conns)
    obs, rew, done, info, ready_id = map(np.stack, zip(*result))
    return obs, rew, done, info, ready_id

# then in collectors
running_id = # all id of evs
obs # init observation
while not enough samples:
    action = policy(obs)
    obs, rew, done, info, finish_id = env.step(action, running_id)
    # append data to cache buffer according to ready_id
    running_id.remove(finish_id)
    if some env is done:
        synchronously reset them

I prefer loop query because I assume only step() is needed to be asynchronous, and the main thread can do nothing but wait when no environments are ready.

Implementation with asyncio would be nice, too. But I think it has bad compatibility. Tianshou supports python 3.6 - 3.8, but the APIs of asyncio in 3.6-3.8 are rather different. In addition, asynchronous code is somewhat harder to understand for most people.

What's your opinion, @magicly ?

[duburcqa]

Do you want to implement this with asyncio or with loop query by poll mechanism?

I do prefer loop query with poll mechanism too.

which may look like this: [...] I prefer loop query because I assume only step() is needed to be asynchronous, and the main thread can do nothing but wait when no environments are ready.

Yes I was working on something very similar. (See PR #134)

Implementation with asyncio would be nice, too. But I think it has bad compatibility. Tianshou supports python 3.6 - 3.8, but the APIs of asyncio in 3.6-3.8 are rather different. In addition, asynchronous code is somewhat harder to understand for most people.

I have the same feeling than you.

It seems this time we are pretty much on the same wavelength 😄

[duburcqa]

@Trinkle23897 Ready to be merged if you agree with the changes.