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.
docstr: Parameters
  ----------
  producer: iterable
    Provides items to feed a consumer with
  consumer: callable
    Is provided with items produced by producer.  Multiple consumers might
    operate in parallel threads if jobs > 1
  jobs: int, optional
    If None or "auto", 'datalad.runtime.max-jobs' configuration variable is
    consulted.  With jobs=0 there is no threaded execution whatsoever.  With
    jobs=1 there is a separate thread for the producer, so in effect with jobs=1
    some parallelization between producer (if it is a generator) and consumer
    could be achieved, while there is only a single thread available for consumers.
  safe_to_consume: callable, optional
    A callable which gets a dict of all known futures and current item from producer.
    It should return `True` if executor can proceed with current value from producer.
    If not (unsafe to consume) - we will wait.
    WARNING: outside code should make sure about provider and `safe_to_consume` to
    play nicely or a very suboptimal behavior or possibly even a deadlock can happen.
  producer_future_key: callable, optional
    A key function for a value from producer which will be used as a key in futures
    dictionary and output of which is passed to safe_to_consume.
  reraise_immediately: bool, optional
    If True, it would stop producer yielding values as soon as it detects that some
    exception has occurred (although there might still be values in the queue to be yielded
    which were collected before the exception was raised).
  agg: callable, optional
    Should be a callable with two arguments: (item, prior total) and return a new total
    which will get assigned to .total of this object.  If not specified, .total is
    just a number of items produced by the producer.
code: def __init__(
      self,
      producer,
      consumer,
      *,
      jobs=None,
      safe_to_consume=None,
      producer_future_key=None,
      reraise_immediately=False,
      agg=None,
  ):
      """
  
      Parameters
      ----------
      producer: iterable
        Provides items to feed a consumer with
      consumer: callable
        Is provided with items produced by producer.  Multiple consumers might
        operate in parallel threads if jobs > 1
      jobs: int, optional
        If None or "auto", 'datalad.runtime.max-jobs' configuration variable is
        consulted.  With jobs=0 there is no threaded execution whatsoever.  With
        jobs=1 there is a separate thread for the producer, so in effect with jobs=1
        some parallelization between producer (if it is a generator) and consumer
        could be achieved, while there is only a single thread available for consumers.
      safe_to_consume: callable, optional
        A callable which gets a dict of all known futures and current item from producer.
        It should return `True` if executor can proceed with current value from producer.
        If not (unsafe to consume) - we will wait.
        WARNING: outside code should make sure about provider and `safe_to_consume` to
        play nicely or a very suboptimal behavior or possibly even a deadlock can happen.
      producer_future_key: callable, optional
        A key function for a value from producer which will be used as a key in futures
        dictionary and output of which is passed to safe_to_consume.
      reraise_immediately: bool, optional
        If True, it would stop producer yielding values as soon as it detects that some
        exception has occurred (although there might still be values in the queue to be yielded
        which were collected before the exception was raised).
      agg: callable, optional
        Should be a callable with two arguments: (item, prior total) and return a new total
        which will get assigned to .total of this object.  If not specified, .total is
        just a number of items produced by the producer.
      """
      self.producer = producer
      self.consumer = consumer
      self.jobs = jobs
      self.safe_to_consume = safe_to_consume
      self.producer_future_key = producer_future_key
      self.reraise_immediately = reraise_immediately
      self.agg = agg
  
      self.total = None if self.agg else 0
      self._jobs = None  # actual "parallel" jobs used
      # Relevant only for _iter_threads
      self._producer_finished = None
      self._producer_queue = None
      self._producer_exception = None
      self._producer_interrupt = None
      # so we could interrupt more or less gracefully
      self._producer_thread = None
      self._executor = None
      self._futures = {}
      self._interrupted = False
  
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 1 to create a new ProducerConsumer object  with example: obj = ProducerConsumer(producer, consumer)