class_docstr: Internal class for `parallel_map`, arguments match the constructor's.
class_docstr_tok: ["Internal","class","for","`","parallel_map","`",",","arguments","match","the","constructor","'s","."]
docstr: Performs all the parallel mapping logic.
  
  Arguments
  ---------
  executor: concurrent.futures.Executor
      The executor to `.invoke` all jobs on. The executor is NOT shut down
      at the end of processing.
docstr_tok: ["Performs","all","the","parallel","mapping","logic",".","Arguments","--","--","--","--","-","executor",":","concurrent.futures.Executor","The","executor","to","`",".invoke","`","all","jobs","on",".","The","executor","is","NOT","shut","down","at","the","end","of","processing","."]
code: def _map_all(self):
      """Performs all the parallel mapping logic.
  
      Arguments
      ---------
      executor: concurrent.futures.Executor
          The executor to `.invoke` all jobs on. The executor is NOT shut down
          at the end of processing.
      """
  
      # initial queue fill
      for _ in range(self.queue_size):
          if not self._enqueue_job():
              break
  
      # consume & requeue logic
      while (not self.depleted_source) or (
          len(self.future_chunks) != 0
      ):
          with self.cv:
              # if `cv.notify` was called by a worker _after_ the `with cv`
              # block last iteration, then `just_finished_count` would be
              # incremented, but this `cv.wait` would not wake up -- skip it.
              while self.just_finished_count == 0:
                  # wait to be woken up by a worker thread, which could mean:
                  # - that a chunk was processed: try to yield any
                  # - that a call failed with an exception: raise it
                  # - nothing; it could be a spurious CV wakeup: keep looping
                  self.cv.wait()
  
              if self.remote_exception is not None:
                  raise self.remote_exception
  
              # store the amount to requeue, avoiding data races
              to_queue_count = self.just_finished_count
              self.just_finished_count = 0
  
          # try to enqueue as many jobs as there were just finished.
          # when the input is finished, the queue will not be refilled.
          for _ in range(to_queue_count):
              if not self._enqueue_job():
                  break
  
          # yield from left to right as long as there is enough ready
          # e.g. | done | done | !done | done | !done | !done
          # would yield from the first two. we might deplete the entire queue
          # at that point, the `depleted_source` loop check is needed as such.
          while (
              len(self.future_chunks) != 0
              and self.future_chunks[0].done()
          ):
              yield from self.future_chunks.popleft().result()
  
      if self.pbar is not None:
          self.pbar.close()
  
file_path: speechbrain/speechbrain/utils/parallel.py
imports: import itertools
  import multiprocessing
  from collections import deque
  from concurrent.futures import Executor, ProcessPoolExecutor
  from threading import Condition
  from typing import Any, Callable, Iterable, Optional
  from tqdm.auto import tqdm
label_desc: Use image node_id 5 for calling the _ParallelMapper obj's underlying member method code with example usage: obj._map_all() without return types