class_docstr: Samples episodes in a data-parallel fashion using a Ray cluster.
  
  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 (list[Policy]): Agents to distribute across workers.
      envs (list[Environment]): Environments to distribute across workers.
      worker_factory (WorkerFactory): Used for worker behavior. 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.
class_docstr_tok: ["Samples","episodes","in","a","data-parallel","fashion","using","a","Ray","cluster",".","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","(","list","[","Policy","]",")",":","Agents","to","distribute","across","workers",".","envs","(","list","[","Environment","]",")",":","Environments","to","distribute","across","workers",".","worker_factory","(","WorkerFactory",")",":","Used","for","worker","behavior",".","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 the policy for new episodes.
  
  Args:
      itr (int): Iteration number.
      num_samples (int): Number of steps the the sampler should collect.
      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.
docstr_tok: ["Sample","the","policy","for","new","episodes",".","Args",":","itr","(","int",")",":","Iteration","number",".","num_samples","(","int",")",":","Number","of","steps","the","the","sampler","should","collect",".","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","."]
code: def obtain_samples(
      self, itr, num_samples, agent_update, env_update=None
  ):
      """Sample the policy for new episodes.
  
      Args:
          itr (int): Iteration number.
          num_samples (int): Number of steps the the sampler should collect.
          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.
  
      """
      active_workers = []
      completed_samples = 0
      batches = []
  
      # update the policy params of each worker before sampling
      # for the current iteration
      idle_worker_ids = []
      updating_workers = self._update_workers(agent_update, env_update)
  
      with click.progressbar(
          length=num_samples, label="Sampling"
      ) as pbar:
          while completed_samples < num_samples:
              # if there are workers still being updated, check
              # which ones are still updating and take the workers that
              # are done updating, and start collecting episodes on those
              # workers.
              if updating_workers:
                  updated, updating_workers = ray.wait(
                      updating_workers, num_returns=1, timeout=0.1
                  )
                  upd = [ray.get(up) for up in updated]
                  idle_worker_ids.extend(upd)
  
              # if there are idle workers, use them to collect episodes and
              # mark the newly busy workers as active
              while idle_worker_ids:
                  idle_worker_id = idle_worker_ids.pop()
                  worker = self._all_workers[idle_worker_id]
                  active_workers.append(worker.rollout.remote())
  
              # check which workers are done/not done collecting a sample
              # if any are done, send them to process the collected
              # episode if they are not, keep checking if they are done
              ready, not_ready = ray.wait(
                  active_workers, num_returns=1, timeout=0.001
              )
              active_workers = not_ready
              for result in ready:
                  ready_worker_id, episode_batch = ray.get(result)
                  idle_worker_ids.append(ready_worker_id)
                  num_returned_samples = episode_batch.lengths.sum()
                  completed_samples += num_returned_samples
                  batches.append(episode_batch)
                  pbar.update(num_returned_samples)
  
      samples = EpisodeBatch.concatenate(*batches)
      self.total_env_steps += sum(samples.lengths)
      return samples
  
imports: from collections import defaultdict
  import itertools
  import click
  import cloudpickle
  import psutil
  import ray
  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 RaySampler obj's underlying member method code with example usage: obj.obtain_samples(itr, num_samples, agent_update, env_update) and returns: samples