code: def run(self):
      while not self.broadcasting_container.stop_flag():
          self._update_policy_if_available()
  
          # limit the step count according to the task specification
          self.env.set_max_episode_steps(self.aux_data["max_steps"])
          if self.aux_data["normalization_stats"] is not None:
              self.env.set_normalization_statistics(
                  self.aux_data["normalization_stats"]
              )
  
          try:
              signal.signal(signal.SIGUSR1, self._abort_handler)
              with torch.no_grad():
                  if self.is_eval_worker:
                      result = self.env.generate_evaluation(self.policy)
                  else:
                      result = self.env.generate_training(
                          self.policy,
                          noise_stddev=self.aux_data["noise_stddev"],
                      )
  
              # Ignore abort during communication
              signal.signal(signal.SIGUSR1, signal.SIG_IGN)
              self.output_queue.put(
                  (self.policy_version_counter, result)
              )
          except ESAbortException:
              if self.env.abort:
                  self.env.clear_abort()
              else:
                  raise
  
file_path: maze/maze/train/trainers/es/distributed/es_subproc_worker.py
imports: import multiprocessing
  import signal
  import cloudpickle
  import torch
  from maze.core.wrappers.log_stats_wrapper import LogStatsWrapper
  from maze.core.wrappers.time_limit_wrapper import TimeLimitWrapper
  from maze.train.parallelization.broadcasting_container import BroadcastingContainer
  from maze.train.trainers.es.distributed.es_rollout_wrapper import ESRolloutWorkerWrapper, ESAbortException
  from maze.train.trainers.es.es_shared_noise_table import SharedNoiseTable
label_desc: Use image node_id 2 for calling the ESSubprocWorker obj's underlying member method code with example usage: obj.run() without return types