code: def test_gracefull_death():
      def assert_provides_and_raises(pc, exception, target=None):
          """Helper to get all results before exception is raised"""
          results = []
          with assert_raises(exception):
              for r in pc:
                  results.append(r)
          # results should be sorted since we do not guarantee order
          results = sorted(results)
          if target is not None:
              assert_equal(results, target)
          return results
  
      def interrupted_producer():
          yield 1
          raise ValueError()
  
      def consumer(i):
          sleep(0.001)
          yield i
  
      assert_provides_and_raises(
          ProducerConsumer(interrupted_producer(), consumer, jobs=3),
          ValueError,
          [1],
      )
  
      def faulty_consumer(i):
          sleep(0.001)
          if i == 1:
              raise ValueError()
          return i
  
      # so we do not get failed, but other parallel ones finish their job
      results = assert_provides_and_raises(
          ProducerConsumer(range(1000), faulty_consumer, jobs=5),
          ValueError,
      )
      # and analysis of futures to raise an exception can take some time etc, so
      # we could get more, but for sure we should not get all 999 and not even a 100
      if info_log_level:
          assert_greater(100, len(results))
      assert_equal(results[:4], [0, 2, 3, 4])
  
      def producer():
          for i in range(10):
              sleep(0.0003)
              yield i
          raise ValueError()
  
      # by default we do not stop upon producer failing
      assert_provides_and_raises(
          ProducerConsumer(producer(), consumer, jobs=2),
          ValueError,
          list(range(10)),
      )
      # if producer produces more than we can as quickly consume but then fails
      # ATM we do not proceed to consume other items, but fail when we finish
      # consuming until the time point when producer has failed
      # by default we do not stop upon producer failing
      results = assert_provides_and_raises(
          ProducerConsumer(
              producer(), consumer, reraise_immediately=True, jobs=2
          ),
          ValueError,
      )
      # we will get some results, seems around 4 and they should be "sequential"
      assert_equal(results, list(range(len(results))))
      try:
          assert_greater_equal(len(results), 2)
      except AssertionError:
          # Possible TODO: if tests below would start failing too, move xfail to the level
          # of the entire test
          pytest.xfail(
              f"Rarely but happens. Got only {len(results)} instead of at least 2"
          )
  
      # This test relies too much on threads scheduling to not hog up on handling
      # consumers, but if it happens so - they might actually consume all results
      # before producer decides to finally raise an exception.  As such it remains
      # flaky and thus not ran, but could be useful to test locally while
      # changing that logic.
      #
      # if info_log_level and not (on_windows or on_osx):
      #     # consumers should not be able to consume all produced items.
      #     # production of 10 should take 3 unites, while consumers 10/2 (jobs)
      #     # 5 units, so some should not have a chance.
      #     assert_greater_equal(8, len(results))
  
      # Simulate situation close to what we have when outside code consumes
      # some yielded results and then "looses interest" (on_failure="error").
      # In this case we should still exit gracefully (no GeneratorExit warnings),
      # not over-produce, and also do not kill already running consumers
      consumed = []
  
      def inner():
          def consumer(i):
              sleep(0.01)
              consumed.append(i)
              return i
  
          pc = iter(ProducerConsumer(range(1000), consumer, jobs=2))
          yield next(pc)
          yield next(pc)
  
      # typically it should be [0, 1] but it does happen some times that
      # one other worker gets ahead and we get [0, 2]. As it is not per se the
      # purpose of this test to ensure absence of such race, we just allow for any
      # two from first 3 possible.
      assert len(set(inner()).intersection({0, 1, 2})) == 2
      consumed = sorted(consumed)
      assert_equal(consumed, list(range(len(consumed))))
      assert_greater_equal(
          len(consumed), 4
      )  # we should wait for that 2nd batch to finish
      if info_log_level:
          assert_greater_equal(20, len(consumed))
  
file_path: datalad/datalad/support/tests/test_parallel.py
imports: import logging
  from functools import partial
  from time import sleep, time
  import pytest
  from datalad import lgr
  from datalad.support import path
  from datalad.support.exceptions import IncompleteResultsError
  from datalad.support.parallel import ProducerConsumer, ProducerConsumerProgressLog, no_parentds_in_futures
  from datalad.tests.utils_pytest import assert_equal, assert_greater, assert_greater_equal, assert_raises, assert_repo_status, known_failure_osx, on_osx, on_windows, rmtree, skip_if, slow, with_tempfile
label_desc: Use image node_id 6 for calling a global function with example usage: test_gracefull_death() and returns: results, i, i