(self, policy_fn, obsfeat_fn, cfg, maxtasksperchild=200)
return TrajBatch.FromTrajs(trajs)
Multiprocessed simulation
        Not thread safe! But why would you want this to be thread safe anyway?
Multiprocessed simulation
        Not thread safe! But why would you want this to be thread safe anyway?
Multiprocessed simulation Not thread safe! But why would you want this to be thread safe anyway?
def sim_mp(self, policy_fn, obsfeat_fn, cfg, maxtasksperchild=200):
        '''
        Multiprocessed simulation
        Not thread safe! But why would you want this to be thread safe anyway?
        '''
        num_processes = cfg.batch_size if cfg.batch_size is not None else multiprocessing.cpu_count()//2

        # Bypass multiprocessing if only using one process
        if num_processes == 1:
            trajs = []
            num_sa = 0
            while True:
                t = self.sim_single(policy_fn, obsfeat_fn, cfg.max_traj_len)
                trajs.append(t)
                num_sa += len(t)
                if len(trajs) >= cfg.min_num_trajs and num_sa >= cfg.min_total_sa:
                    break
            return TrajBatch.FromTrajs(trajs)

        global _global_sim_info
        _global_sim_info = (self, policy_fn, obsfeat_fn, cfg.max_traj_len)

        trajs = []
        num_sa = 0

        with set_mkl_threads(1):
            # Thanks John
            pool = multiprocessing.Pool(processes=num_processes, maxtasksperchild=maxtasksperchild)
            pending = []
            done = False
            while True:
                if len(pending) < num_processes and not done:
                    pending.append(pool.apply_async(_rollout))
                stillpending = []
                for job in pending:
                    if job.ready():
                        traj = job.get()
                        trajs.append(traj)
                        num_sa += len(traj)
                    else:
                        stillpending.append(job)
                pending = stillpending
                if len(trajs) >= cfg.min_num_trajs and num_sa >= cfg.min_total_sa:
                    done = True
                    if len(pending) == 0:
                        break
                sleep(.001)
            pool.close()

        assert len(trajs) >= cfg.min_num_trajs and sum(len(traj) for traj in trajs) >= cfg.min_total_sa
        return TrajBatch.FromTrajs(trajs)
def sim_mp ( self , policy_fn , obsfeat_fn , cfg , maxtasksperchild = 200 ) : num_processes = cfg . batch_size if cfg . batch_size is not None else multiprocessing . cpu_count ( ) // 2 # Bypass multiprocessing if only using one process if num_processes == 1 : trajs = [ ] num_sa = 0 while True : t = self . sim_single ( policy_fn , obsfeat_fn , cfg . max_traj_len ) trajs . append ( t ) num_sa += len ( t ) if len ( trajs ) >= cfg . min_num_trajs and num_sa >= cfg . min_total_sa : break return TrajBatch . FromTrajs ( trajs ) global _global_sim_info _global_sim_info = ( self , policy_fn , obsfeat_fn , cfg . max_traj_len ) trajs = [ ] num_sa = 0 with set_mkl_threads ( 1 ) : # Thanks John pool = multiprocessing . Pool ( processes = num_processes , maxtasksperchild = maxtasksperchild ) pending = [ ] done = False while True : if len ( pending ) < num_processes and not done : pending . append ( pool . apply_async ( _rollout ) ) stillpending = [ ] for job in pending : if job . ready ( ) : traj = job . get ( ) trajs . append ( traj ) num_sa += len ( traj ) else : stillpending . append ( job ) pending = stillpending if len ( trajs ) >= cfg . min_num_trajs and num_sa >= cfg . min_total_sa : done = True if len ( pending ) == 0 : break sleep ( .001 ) pool . close ( ) assert len ( trajs ) >= cfg . min_num_trajs and sum ( len ( traj ) for traj in trajs ) >= cfg . min_total_sa return TrajBatch . FromTrajs ( trajs )