class_docstr: Producer/Consumer implementation to (possibly) parallelize execution.
  
  It is an iterable providing a multi-threaded producer/consumer implementation,
  where there could be multiple consumers for items produced by a producer.  Since
  in DataLad majority of time is done in IO interactions with outside git and git-annex
  processes, and since we typically operate across multiple datasets, multi-threading
  across datasets operations already provides a significant performance benefit.
  
  All results from consumers are all yielded as soon as they are produced by consumers.
  Because this implementation is based on threads, `producer` and `consumer` could
  be some "closures" within code, thus having lean interface and accessing
  data from shared "outer scope".
  
  Notes
  -----
  - with jobs > 1, results are yielded as soon as available, so order
    might not match the one provided by "producer".
  - jobs > 1, is "effective" only for Python >= 3.8.  For older versions it
    would log a warning (upon initial encounter) if jobs > 1 is specified.
  - `producer` must produce unique entries. AssertionError might be raised if
    the same entry is to be consumed.
  - `consumer` can add to the queue of items produced by producer via
    `.add_to_producer_queue`. This allows for continuous reuse of the same
    instance in recursive operations (see `get` use of ProducerConsumer).
  - if producer or consumer raise an exception, we will try to "fail gracefully",
    unless subsequent Ctrl-C is pressed, we will let already running jobs to
    finish first.
  
  Examples
  --------
  A simple and somewhat boring example could be to count lines in '*.py'
  files in parallel
  
      from glob import glob
      from pprint import pprint
      from datalad.support.parallel import ProducerConsumer
  
      def count_lines(fname):
          with open(fname) as f:
              return fname, len(f.readlines())
      pprint(dict(ProducerConsumer(glob("*.py"), count_lines)))
  
  More usage examples could be found in `test_parallel.py` and around the
  codebase `addurls.py`, `get.py`, `save.py`, etc.
code: def _iter_threads(self, jobs):
      self._interrupted = False
      self._producer_finished = False
      self._producer_exception = None
      self._producer_interrupt = None
  
      # To allow feeding producer queue with more entries, possibly from consumer!
      self._producer_queue = producer_queue = Queue()
      consumer_queue = Queue()
  
      def producer_worker():
          """That is the one which interrogates producer and updates .total"""
          try:
              for value in self._producer_iter:
                  if self._producer_interrupt:
                      raise InterruptedError(
                          "Producer thread was interrupted due to %s"
                          % self._producer_interrupt
                      )
                  self.add_to_producer_queue(value)
          except InterruptedError:
              pass  # There is some outside exception which will be raised
          except BaseException as e:
              self._producer_exception = e
          finally:
              self._producer_finished = True
  
      def consumer_worker(callable, *args, **kwargs):
          """Since jobs could return a generator and we cannot really "inspect" for that"""
          res = callable(*args, **kwargs)
          if inspect.isgenerator(res):
              lgr.debug(
                  "Got consumer worker which returned a generator %s",
                  res,
              )
              didgood = False
              for r in res:
                  didgood = True
                  lgr.debug("Adding %s to queue", r)
                  consumer_queue.put(r)
              if not didgood:
                  lgr.error("Nothing was obtained from %s :-(", res)
          else:
              lgr.debug("Got straight result %s, not a generator", res)
              consumer_queue.put(res)
  
      self._producer_thread = Thread(target=producer_worker)
      self._producer_thread.start()
      self._futures = futures = {}
  
      lgr.debug("Initiating ThreadPoolExecutor with %d jobs", jobs)
      # we will increase sleep_time when doing nothing useful
      sleeper = Sleeper()
      interrupted_by_exception = None
      with concurrent.futures.ThreadPoolExecutor(jobs) as executor:
          self._executor = executor
          # yield from the producer_queue (.total and .finished could be accessed meanwhile)
          while True:
              try:
                  done_useful = False
                  if (
                      self.reraise_immediately
                      and self._producer_exception
                      and not interrupted_by_exception
                  ):
                      # so we have a chance to exit gracefully
                      # No point to reraise if there is already an exception which was raised
                      # which might have even been this one
                      lgr.debug(
                          "Reraising an exception from producer as soon as we found it"
                      )
                      raise self._producer_exception
                  if (
                      self._producer_finished
                      and not futures
                      and consumer_queue.empty()
                      and producer_queue.empty()
                  ):
                      # This will let us not "escape" the while loop and reraise any possible exception
                      # within the loop if we have any.
                      # Otherwise we might see "RuntimeError: generator ignored GeneratorExit"
                      # when e.g. we did continue upon interrupted_by_exception, and then
                      # no other subsequent exception was raised and we left the loop
                      raise _FinalShutdown()
  
                  # important!  We are using threads, so worker threads will be sharing CPU time
                  # with this master thread. For it to become efficient, we should consume as much
                  # as possible from producer asap and push it to executor.  So drain the queue
                  while not (
                      producer_queue.empty() or interrupted_by_exception
                  ):
                      done_useful = True
                      try:
                          job_args = (
                              producer_queue.get()
                          )  # timeout=0.001)
                          job_key = (
                              self.producer_future_key(job_args)
                              if self.producer_future_key
                              else job_args
                          )
                          if self.safe_to_consume:
                              # Sleep a little if we are not yet ready
                              # TODO: add some .debug level reporting based on elapsed time
                              # IIRC I did smth like growing exponentially delays somewhere (dandi?)
                              while not self.safe_to_consume(
                                  futures, job_key
                              ):
                                  self._pop_done_futures(
                                      lgr
                                  ) or sleeper()
                          # Current implementation, to provide depchecking, relies on unique
                          # args for the job
                          assert job_key not in futures
                          lgr.debug(
                              "Submitting worker future for %s",
                              job_args,
                          )
                          futures[job_key] = executor.submit(
                              consumer_worker, self.consumer, job_args
                          )
                      except Empty:
                          pass
  
                  # check active futures
                  if not consumer_queue.empty():
                      done_useful = True
                      # ATM we do not bother of some "in order" reporting
                      # Just report as soon as any new record arrives
                      res = consumer_queue.get()
                      lgr.debug("Got %s from consumer_queue", res)
                      yield res
  
                  done_useful |= self._pop_done_futures(lgr)
  
                  if not done_useful:  # you need some rest
                      # TODO: same here -- progressive logging
                      lgr.log(
                          5,
                          "Did nothing useful, sleeping. Have "
                          "producer_finished=%s producer_queue.empty=%s futures=%s consumer_queue.empty=%s",
                          self._producer_finished,
                          producer_queue.empty(),
                          futures,
                          consumer_queue.empty(),
                      )
                      sleeper()
                  else:
                      sleeper.reset()
              except (_FinalShutdown, GeneratorExit):
                  self.shutdown(
                      force=True,
                      exception=self._producer_exception
                      or interrupted_by_exception,
                  )
                  break  # if there were no exception to raise
              except BaseException as exc:
                  ce = CapturedException(exc)
                  self._interrupted = True
                  if interrupted_by_exception:
                      # so we are here again but now it depends why we are here
                      if isinstance(exc, KeyboardInterrupt):
                          lgr.warning(
                              "Interrupted via Ctrl-C.  Forcing the exit"
                          )
                          self.shutdown(force=True, exception=exc)
                      else:
                          lgr.warning(
                              "One more exception was received while "
                              "trying to finish gracefully: %s",
                              ce,
                          )
                          # and we go back into the loop until we finish or there is Ctrl-C
                  else:
                      interrupted_by_exception = exc
                      lgr.warning(
                          "Received an exception %s. Canceling not-yet "
                          "running jobs and waiting for completion of "
                          "running. You can force earlier forceful exit "
                          "by Ctrl-C.",
                          ce,
                      )
                      self.shutdown(force=False, exception=exc)
  
imports: import concurrent.futures
  import inspect
  import sys
  import time
  import uuid
  from collections import defaultdict
  from queue import Queue, Empty
  from threading import Thread
  from .None import ansi_colors
  from ..log import log_progress
  from ..utils import path_is_subpath
  from datalad.support.exceptions import CapturedException
  import logging
label_desc: Use image node_id 10 for calling the ProducerConsumer obj's underlying member method code with example usage: obj._iter_threads(jobs) without return types