class_docstr: Runs multiple instances (workers) of a predictor in parallel.
  Exceptions are propagated from the workers.
  Note: That there is currently an issue with tqdm that will cause things
  to hang if the ParallelizedPredictor is used with tqdm and an exception
  occurs during prediction.
  https://github.com/tqdm/tqdm/issues/548
  
  Parameters
  ----------
  base_predictor
      A representable predictor that will be used
  num_workers
      Number of workers (processes) to use. If set to
      None, one worker per CPU will be used.
  chunk_size
      Number of items to pass per call
class_docstr_tok: ["Runs","multiple","instances","(","workers",")","of","a","predictor","in","parallel",".","Exceptions","are","propagated","from","the","workers",".","Note",":","That","there","is","currently","an","issue","with","tqdm","that","will","cause","things","to","hang","if","the","ParallelizedPredictor","is","used","with","tqdm","and","an","exception","occurs","during","prediction",".","https",":","\/\/github.com\/tqdm\/tqdm\/issues\/548","Parameters","--","--","--","--","--","base_predictor","A","representable","predictor","that","will","be","used","num_workers","Number","of","workers","(","processes",")","to","use",".","If","set","to","None",",","one","worker","per","CPU","will","be","used",".","chunk_size","Number","of","items","to","pass","per","call"]
code: def predict(self, dataset: Dataset, **kwargs) -> Iterator[Forecast]:
      with TemporaryDirectory() as tempdir:
          predictor_path = Path(tempdir)
          self._base_predictor.serialize(predictor_path)
  
          # TODO: Consider using shared memory for the data transfer.
  
          self._input_queues = [
              mp.Queue() for _ in range(self._num_workers)
          ]
          self._output_queue = mp.Queue()
  
          workers = []
          for worker_id, in_q in enumerate(self._input_queues):
              worker = mp.Process(
                  target=_worker_loop,
                  args=(
                      predictor_path,
                      in_q,
                      self._output_queue,
                      worker_id,
                  ),
                  kwargs=kwargs,
              )
  
              worker.daemon = True
              worker.start()
              workers.append(worker)
              self._num_running_workers += 1
  
          self._workers = workers
  
          chunked_data = self._grouper(dataset, self._chunk_size)
  
          self._send_idx = 0
          self._next_idx = 0
  
          self._data_buffer: Dict[int, Any] = {}
  
          worker_ids = list(range(self._num_workers))
  
          def receive():
              idx, worker_id, result = self._output_queue.get()
              if isinstance(idx, WorkerError):
                  self._num_running_workers -= 1
                  self.terminate()
                  raise Exception(idx.msg)
              if idx is not None:
                  self._data_buffer[idx] = result
              return idx, worker_id, result
  
          def get_next_from_buffer():
              while self._next_idx in self._data_buffer:
                  result_batch = self._data_buffer.pop(self._next_idx)
                  self._next_idx += 1
                  yield from result_batch
  
          def send(worker_id, chunk):
              q = self._input_queues[worker_id]
              q.put((self._send_idx, chunk))
              self._send_idx += 1
  
          try:
              # prime the queues
              for wid in worker_ids:
                  chunk = next(chunked_data)
                  send(wid, chunk)
  
              while True:
                  idx, wid, result = receive()
                  yield from get_next_from_buffer()
                  chunk = next(chunked_data)
                  send(wid, chunk)
          except StopIteration:
              # signal workers end of data
              for q in self._input_queues:
                  q.put((None, None))
  
          # collect any outstanding results
          while self._num_running_workers > 0:
              idx, worker_id, result = receive()
              if idx is None:
                  self._num_running_workers -= 1
                  continue
              yield from get_next_from_buffer()
          assert len(self._data_buffer) == 0
          assert self._send_idx == self._next_idx
  
imports: import itertools
  import json
  import logging
  import multiprocessing
  import sys
  import traceback
  from pathlib import Path
  from pydoc import locate
  from tempfile import TemporaryDirectory
  from typing import TYPE_CHECKING, Callable, Iterator, Optional, Dict, Any
  import numpy
  import gluonts
  from gluonts.core import fqname_for
  from gluonts.core.component import equals, from_hyperparameters
  from gluonts.core.serde import dump_json, load_json
  from gluonts.dataset.common import DataEntry, Dataset
  from gluonts.model.forecast import Forecast
label_desc: Use image node_id 4 for calling the ParallelizedPredictor obj's underlying member method code with example usage: obj.predict(dataset) and returns: idx,  worker_id,  result