class_docstr: Sampler that uses multiprocessing to distribute workers.
  
  The sampler need to be created either from a worker factory or from
  parameters which can construct a worker factory. See the __init__ method
  of WorkerFactory for the detail of these parameters.
  
  Args:
      agents (Policy or List[Policy]): Agent(s) to use to sample episodes.
          If a list is passed in, it must have length exactly
          `worker_factory.n_workers`, and will be spread across the
          workers.
      envs (Environment or List[Environment]): Environment from which
          episodes are sampled. If a list is passed in, it must have length
          exactly `worker_factory.n_workers`, and will be spread across the
          workers.
      worker_factory (WorkerFactory): Pickleable factory for creating
          workers. Should be transmitted to other processes / nodes where
          work needs to be done, then workers should be constructed
          there. Either this param or params after this are required to
          construct a sampler.
      max_episode_length(int): Params used to construct a worker factory.
          The maximum length episodes which will be sampled.
      is_tf_worker (bool): Whether it is workers for TFTrainer.
      seed(int): The seed to use to initialize random number generators.
      n_workers(int): The number of workers to use.
      worker_class(type): Class of the workers. Instances should implement
          the Worker interface.
      worker_args (dict or None): Additional arguments that should be passed
          to the worker.
docstr: Sample an exact number of episodes per worker.
  
  Args:
      n_eps_per_worker (int): Exact number of episodes to gather for
          each worker.
      agent_update (object): Value which will be passed into the
          `agent_update_fn` before sampling episodes. If a list is passed
          in, it must have length exactly `factory.n_workers`, and will
          be spread across the workers.
      env_update (object): Value which will be passed into the
          `env_update_fn` before sampling episodes. If a list is passed
          in, it must have length exactly `factory.n_workers`, and will
          be spread across the workers.
  
  Returns:
      EpisodeBatch: Batch of gathered episodes. Always in worker
          order. In other words, first all episodes from worker 0,
          then all episodes from worker 1, etc.
  
  Raises:
      AssertionError: On internal errors.
docstr_tok: ["Sample","an","exact","number","of","episodes","per","worker",".","Args",":","n_eps_per_worker","(","int",")",":","Exact","number","of","episodes","to","gather","for","each","worker",".","agent_update","(","object",")",":","Value","which","will","be","passed","into","the","`","agent_update_fn","`","before","sampling","episodes",".","If","a","list","is","passed","in",",","it","must","have","length","exactly","`","factory.n_workers","`",",","and","will","be","spread","across","the","workers",".","env_update","(","object",")",":","Value","which","will","be","passed","into","the","`","env_update_fn","`","before","sampling","episodes",".","If","a","list","is","passed","in",",","it","must","have","length","exactly","`","factory.n_workers","`",",","and","will","be","spread","across","the","workers",".","Returns",":","EpisodeBatch",":","Batch","of","gathered","episodes",".","Always","in","worker","order",".","In","other","words",",","first","all","episodes","from","worker","0",",","then","all","episodes","from","worker","1",",","etc",".","Raises",":","AssertionError",":","On","internal","errors","."]
code: def obtain_exact_episodes(
      self, n_eps_per_worker, agent_update, env_update=None
  ):
      """Sample an exact number of episodes per worker.
  
      Args:
          n_eps_per_worker (int): Exact number of episodes to gather for
              each worker.
          agent_update (object): Value which will be passed into the
              `agent_update_fn` before sampling episodes. If a list is passed
              in, it must have length exactly `factory.n_workers`, and will
              be spread across the workers.
          env_update (object): Value which will be passed into the
              `env_update_fn` before sampling episodes. If a list is passed
              in, it must have length exactly `factory.n_workers`, and will
              be spread across the workers.
  
      Returns:
          EpisodeBatch: Batch of gathered episodes. Always in worker
              order. In other words, first all episodes from worker 0,
              then all episodes from worker 1, etc.
  
      Raises:
          AssertionError: On internal errors.
  
      """
      self._agent_version += 1
      updated_workers = set()
      agent_ups = self._factory.prepare_worker_messages(
          agent_update, cloudpickle.dumps
      )
      env_ups = self._factory.prepare_worker_messages(
          env_update, cloudpickle.dumps
      )
      episodes = defaultdict(list)
  
      with click.progressbar(
          length=self._factory.n_workers, label="Sampling"
      ) as pbar:
          while any(
              len(episodes[i]) < n_eps_per_worker
              for i in range(self._factory.n_workers)
          ):
              self._push_updates(updated_workers, agent_ups, env_ups)
              tag, contents = self._to_sampler.get()
  
              if tag == "episode":
                  batch, version, worker_n = contents
  
                  if version == self._agent_version:
                      if len(episodes[worker_n]) < n_eps_per_worker:
                          episodes[worker_n].append(batch)
  
                      if len(episodes[worker_n]) == n_eps_per_worker:
                          pbar.update(1)
                          try:
                              self._to_worker[worker_n].put_nowait(
                                  ("stop", ())
                              )
                          except queue.Full:
                              pass
              else:
                  raise AssertionError(
                      "Unknown tag {} with contents {}".format(
                          tag, contents
                      )
                  )
  
          for q in self._to_worker:
              try:
                  q.put_nowait(("stop", ()))
              except queue.Full:
                  pass
  
      ordered_episodes = list(
          itertools.chain(
              *[episodes[i] for i in range(self._factory.n_workers)]
          )
      )
      samples = EpisodeBatch.concatenate(*ordered_episodes)
      self.total_env_steps += sum(samples.lengths)
      return samples
  
imports: from collections import defaultdict
  import itertools
  import multiprocessing
  import queue
  import click
  import cloudpickle
  import psutil
  import setproctitle
  from garage import EpisodeBatch
  from garage.experiment.deterministic import get_seed
  from garage.sampler.default_worker import DefaultWorker
  from garage.sampler.sampler import Sampler
  from garage.sampler.worker_factory import WorkerFactory
label_desc: Use image node_id 5 for calling the MultiprocessingSampler obj's underlying member method code with example usage: obj.obtain_exact_episodes(n_eps_per_worker, agent_update, env_update) and returns: samples