{
    "file_diffs": [
        {
            "old_file_content": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Helpers for parallel execution\n\n\"\"\"\n\n__docformat__ = 'restructuredtext'\n\nimport concurrent.futures\nimport inspect\nimport time\nimport uuid\n\nfrom collections import defaultdict\nfrom queue import Queue, Empty\nfrom threading import Thread\n\nfrom . import ansi_colors as colors\nfrom ..log import log_progress\nfrom ..utils import path_is_subpath\n\n\ndef _count_str(count, verb, omg=False):\n    if count:\n        msg = \"{:d} {}\".format(count, verb)\n        if omg:\n            msg = colors.color_word(msg, colors.RED)\n        return msg\n\n\ndef no_parentds_in_futures(futures, path):\n    \"\"\"Return True if no path in futures keys is parentds for provided path\n\n    Assumes that the future's key is the path.\n    \"\"\"\n    # TODO: OPT.  Could benefit from smarter than linear time if not one at a time?\n    #   or may be we should only go through active futures (still linear!)?\n    return not any(path_is_subpath(path, p) for p in futures)\n\n\nclass ProducerConsumer:\n    \"\"\"\n    \"\"\"\n\n    def __init__(self,\n                 producer, consumer,\n                 njobs=None,\n                 safe_to_consume=None,\n                 reraise_immediately=False,\n                 agg=None,\n                 # TODO: \"absorb\" into some more generic \"logging\" helper\n                 # so could be used not only with consumers which yield our records\n                 # 'agg' from above could also more relate to i\n                 label=\"Total\", unit=\"items\",\n                 lgr=None,\n                 ):\n        \"\"\"\n\n        Parameters\n        ----------\n        ...\n        safe_to_consume: callable\n          A callable which gets a dict of all known futures and current producer output.\n          It should return True if we can proceed with current value from producer.\n          If unsafe - we will wait.  WARNING: outside code should make sure about provider and\n          safe_to_consume to play nicely or deadlock can happen.\n        reraise_immediately: bool, optional\n          If True, it would stop producer yielding values as soon as it detects that some\n          exception has occurred (although there might still be values in the queue to be yielded\n          which were collected before the exception was raised)\n        \"\"\"\n        self.producer = producer\n        self.consumer = consumer\n        self.njobs = njobs\n        self.safe_to_consume = safe_to_consume\n        self.reraise_immediately = reraise_immediately\n        self.agg = agg\n        self.label = label\n        self.unit = unit\n        self.lgr = lgr\n\n        self.total = None\n        self.producer_finished = None\n        self._executor = None\n        self._exc = []\n\n    def __del__(self):\n        # if we are killed while executing, we should ask executor to terminate\n        executor = getattr(self, \"_executor\")\n        if executor:\n            executor.terminate()\n\n    def __iter__(self):\n        self.producer_finished = False\n        self._exc = []\n\n        producer_queue = Queue()\n        consumer_queue = Queue()\n\n        def producer_worker():\n            \"\"\"That is the one which interrogates producer and updates .total\"\"\"\n            total = None if self.agg else 0\n            try:\n                for value in self.producer:\n                    producer_queue.put(value)\n                    if self.agg:\n                        total = (\n                            self.agg(value, total) if total is not None else self.agg(value)\n                        )\n                    else:\n                        total += 1\n                    self.total = total\n            except BaseException as e:\n                self._exc.append(e)\n            finally:\n                self.producer_finished = True\n\n        def consumer_worker(callable, *args, **kwargs):\n            \"\"\"Since jobs could return a generator and we cannot really \"inspect\" for that\n            \"\"\"\n            try:\n                res = callable(*args, **kwargs)\n                if inspect.isgenerator(res):\n                    lgr.debug(\"Got consumer worker which returned a generator %s\", res)\n                    didgood = False\n                    for r in res:\n                        didgood = True\n                        lgr.debug(\"Adding %s to queue\", r)\n                        consumer_queue.put(r)\n                    if not didgood:\n                        lgr.error(\"Nothing was obtained from %s :-(\", res)\n                else:\n                    lgr.debug(\"Got straight result %s, not a generator\", res)\n                    consumer_queue.put(res)\n            except BaseException as e:\n                self._exc.append(e)\n\n        producer_thread = Thread(target=producer_worker)\n        producer_thread.start()\n\n        # LOGGING\n        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?\n        lgr = self.lgr\n        label = self.label\n        if lgr is None:\n            from .. import lgr\n        log_progress(lgr.info, pid,\n                     \"%s: starting\", self.label,\n                     # will become known only later total=len(items),\n                     label=self.label, unit=\" \" + self.unit)\n        counts = defaultdict(int)\n\n        futures = {}\n\n        total_announced = False\n        njobs = self.njobs or 1\n        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", njobs)\n        # we will increase sleep_time when doing nothing useful\n        sleeper = Sleeper()\n        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:\n            self._executor = executor\n            # yield from the producer_queue (.total and .finished could be accessed meanwhile)\n            while True:\n                done_useful = False\n                if self.reraise_immediately and self._exc:\n                    break\n                if (self.producer_finished and\n                        not futures and\n                        consumer_queue.empty() and\n                        producer_queue.empty()):\n                    break\n\n                # LOGGING\n                if not total_announced and self.producer_finished:\n                    log_progress(\n                        lgr.info,\n                        pid,\n                        None,  # I do not think there is something valuable to announce\n                        total=self.total,\n                        # unfortunately of no effect, so we cannot inform that more items to come\n                        # unit=(\"+\" if not it.finished else \"\") + \" \" + unit,\n                        update=0  # not None, so it does not stop\n                    )\n                    total_announced = True\n\n                # important!  We are using threads, so worker threads will be sharing CPU time\n                # with this master thread. For it to become efficient, we should consume as much\n                # as possible from producer asap and push it to executor.  So drain the queue\n                while not producer_queue.empty():\n                    done_useful = True\n                    try:\n                        job_args = producer_queue.get() # timeout=0.001)\n                        if self.safe_to_consume:\n                            # Sleep a little if we are not yet ready\n                            # TODO: add some .debug level reporting based on elapsed time\n                            # IIRC I did smth like growing exponentially delays somewhere (dandi?)\n                            while not self.safe_to_consume(futures, job_args):\n                                _prune_futures(futures, lgr) or sleeper()\n                        # Current implementation, to provide depchecking, relies on unique\n                        # args for the job\n                        assert job_args not in futures\n                        lgr.debug(\"Submitting worker future for %s\", job_args)\n                        futures[job_args] = executor.submit(consumer_worker, self.consumer, job_args)\n                    except Empty:\n                        pass\n\n                # check active futures\n                if not consumer_queue.empty():\n                    done_useful = True\n\n                    # ATM we do not bother of some \"in order\" reporting\n                    # Just report as soon as any new record arrives\n                    res = consumer_queue.get()\n                    lgr.debug(\"Got %s from consumer_queue\", res)\n                    # LOGGING\n                    counts[res[\"status\"]] += 1\n                    count_strs = (_count_str(*args)\n                                  for args in [(counts[\"notneeded\"], \"skipped\", False),\n                                               (counts[\"error\"], \"failed\", True)])\n                    if counts[\"notneeded\"] or counts[\"error\"]:\n                        label = \"{} ({})\".format(\n                            self.label,\n                            \", \".join(filter(None, count_strs)))\n\n                    log_progress(\n                        lgr.error if res[\"status\"] == \"error\" else lgr.info,\n                        pid,\n                        \"%s: processed result%s\", self.label,\n                        \" for \" + res[\"path\"] if \"path\" in res else \"\",\n                        label=label, update=1, increment=True)\n\n                    yield res\n\n                done_useful |= _prune_futures(futures, lgr)\n\n                if not done_useful:  # you need some rest\n                    # TODO: same here -- progressive logging\n                    lgr.debug(\n                        \"Did nothing useful, sleeping. Have %s %s %s %s\",\n                        self.producer_finished,\n                        futures,\n                        consumer_queue.empty(),\n                        producer_queue.empty())\n                    sleeper()\n                else:\n                    sleeper.reset()\n\n        self._executor = None\n\n        # LOGGING\n        log_progress(lgr.info, pid, \"%s: done\", self.label)\n        producer_thread.join()\n        if self._exc:\n            if len(self._exc) > 1:\n                lgr.debug(\"%d exceptions were collected while performing execution in parallel. Only the first one \"\n                          \"will be reraised\", len(self._exc))\n            raise self._exc[0]\n        else:\n            assert not futures, \\\n                \"There is still %d active futures for following args: %s\" \\\n                % (len(futures), ', '.join(futures))\n\n\ndef _prune_futures(futures, lgr):\n    \"\"\"Removes .done from provided futures.\n\n    Returns\n    -------\n    bool\n      True if any future was removed\n    \"\"\"\n    done_useful = False\n    # remove futures which are done\n    for args, future in list(futures.items()):\n        if future.done():\n            done_useful = True\n            future_ = futures.pop(args)\n            msg = \"\"\n            if future_.exception():\n                msg = \" with an exception %s\" % future_.exception()\n            lgr.debug(\"Future for %r is done%s\", args, msg)\n    return done_useful\n\n\nclass Sleeper():\n    def __init__(self):\n        self.min_sleep_time = 0.001\n        # but no more than to this max\n        self.max_sleep_time = 0.1\n        self.sleep_time = self.min_sleep_time\n\n    def __call__(self):\n        time.sleep(self.sleep_time)\n        self.sleep_time = min(self.max_sleep_time, self.sleep_time * 2)\n\n    def reset(self):\n        self.sleep_time = self.min_sleep_time\n",
            "new_file_content": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Helpers for parallel execution\n\n\"\"\"\n\n__docformat__ = 'restructuredtext'\n\nimport concurrent.futures\nimport inspect\nimport time\nimport uuid\n\nfrom collections import defaultdict\nfrom queue import Queue, Empty\nfrom threading import Thread\n\nfrom . import ansi_colors as colors\nfrom ..log import log_progress\nfrom ..utils import path_is_subpath\n\n\ndef _count_str(count, verb, omg=False):\n    if count:\n        msg = \"{:d} {}\".format(count, verb)\n        if omg:\n            msg = colors.color_word(msg, colors.RED)\n        return msg\n\n\ndef no_parentds_in_futures(futures, path):\n    \"\"\"Return True if no path in futures keys is parentds for provided path\n\n    Assumes that the future's key is the path.\n    \"\"\"\n    # TODO: OPT.  Could benefit from smarter than linear time if not one at a time?\n    #   or may be we should only go through active futures (still linear!)?\n    return not any(path_is_subpath(path, p) for p in futures)\n\n\nclass ProducerConsumer:\n    \"\"\"\n    \"\"\"\n\n    def __init__(self,\n                 producer, consumer,\n                 jobs=None,\n                 safe_to_consume=None,\n                 reraise_immediately=False,\n                 agg=None,\n                 # TODO: \"absorb\" into some more generic \"logging\" helper\n                 # so could be used not only with consumers which yield our records\n                 # 'agg' from above could also more relate to i\n                 label=\"Total\", unit=\"items\",\n                 lgr=None,\n                 ):\n        \"\"\"\n\n        Parameters\n        ----------\n        ...\n        safe_to_consume: callable\n          A callable which gets a dict of all known futures and current producer output.\n          It should return True if we can proceed with current value from producer.\n          If unsafe - we will wait.  WARNING: outside code should make sure about provider and\n          safe_to_consume to play nicely or deadlock can happen.\n        reraise_immediately: bool, optional\n          If True, it would stop producer yielding values as soon as it detects that some\n          exception has occurred (although there might still be values in the queue to be yielded\n          which were collected before the exception was raised)\n        \"\"\"\n        self.producer = producer\n        self.consumer = consumer\n        self.jobs = jobs\n        self.safe_to_consume = safe_to_consume\n        self.reraise_immediately = reraise_immediately\n        self.agg = agg\n        self.label = label\n        self.unit = unit\n        self.lgr = lgr\n\n        self.total = None\n        self.producer_finished = None\n        self._executor = None\n        self._exc = []\n\n    def __del__(self):\n        # if we are killed while executing, we should ask executor to terminate\n        executor = getattr(self, \"_executor\")\n        if executor:\n            executor.terminate()\n\n    def __iter__(self):\n        self.producer_finished = False\n        self._exc = []\n\n        producer_queue = Queue()\n        consumer_queue = Queue()\n\n        def producer_worker():\n            \"\"\"That is the one which interrogates producer and updates .total\"\"\"\n            total = None if self.agg else 0\n            try:\n                for value in self.producer:\n                    producer_queue.put(value)\n                    if self.agg:\n                        total = (\n                            self.agg(value, total) if total is not None else self.agg(value)\n                        )\n                    else:\n                        total += 1\n                    self.total = total\n            except BaseException as e:\n                self._exc.append(e)\n            finally:\n                self.producer_finished = True\n\n        def consumer_worker(callable, *args, **kwargs):\n            \"\"\"Since jobs could return a generator and we cannot really \"inspect\" for that\n            \"\"\"\n            try:\n                res = callable(*args, **kwargs)\n                if inspect.isgenerator(res):\n                    lgr.debug(\"Got consumer worker which returned a generator %s\", res)\n                    didgood = False\n                    for r in res:\n                        didgood = True\n                        lgr.debug(\"Adding %s to queue\", r)\n                        consumer_queue.put(r)\n                    if not didgood:\n                        lgr.error(\"Nothing was obtained from %s :-(\", res)\n                else:\n                    lgr.debug(\"Got straight result %s, not a generator\", res)\n                    consumer_queue.put(res)\n            except BaseException as e:\n                self._exc.append(e)\n\n        producer_thread = Thread(target=producer_worker)\n        producer_thread.start()\n\n        # LOGGING\n        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?\n        lgr = self.lgr\n        label = self.label\n        if lgr is None:\n            from .. import lgr\n        log_progress(lgr.info, pid,\n                     \"%s: starting\", self.label,\n                     # will become known only later total=len(items),\n                     label=self.label, unit=\" \" + self.unit)\n        counts = defaultdict(int)\n\n        futures = {}\n\n        total_announced = False\n        jobs = self.jobs or 1\n        if jobs == \"auto\":\n            # ATM there is no \"auto\" for this operation.  We will just make it ...\n            # \"auto\" could be for some auto-scaling based on a single future time\n            # to complete, scaling up/down.  TODO\n            jobs = 5\n        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", jobs)\n        # we will increase sleep_time when doing nothing useful\n        sleeper = Sleeper()\n        with concurrent.futures.ThreadPoolExecutor(jobs) as executor:\n            self._executor = executor\n            # yield from the producer_queue (.total and .finished could be accessed meanwhile)\n            while True:\n                done_useful = False\n                if self.reraise_immediately and self._exc:\n                    break\n                if (self.producer_finished and\n                        not futures and\n                        consumer_queue.empty() and\n                        producer_queue.empty()):\n                    break\n\n                # LOGGING\n                if not total_announced and self.producer_finished:\n                    log_progress(\n                        lgr.info,\n                        pid,\n                        None,  # I do not think there is something valuable to announce\n                        total=self.total,\n                        # unfortunately of no effect, so we cannot inform that more items to come\n                        # unit=(\"+\" if not it.finished else \"\") + \" \" + unit,\n                        update=0  # not None, so it does not stop\n                    )\n                    total_announced = True\n\n                # important!  We are using threads, so worker threads will be sharing CPU time\n                # with this master thread. For it to become efficient, we should consume as much\n                # as possible from producer asap and push it to executor.  So drain the queue\n                while not producer_queue.empty():\n                    done_useful = True\n                    try:\n                        job_args = producer_queue.get() # timeout=0.001)\n                        if self.safe_to_consume:\n                            # Sleep a little if we are not yet ready\n                            # TODO: add some .debug level reporting based on elapsed time\n                            # IIRC I did smth like growing exponentially delays somewhere (dandi?)\n                            while not self.safe_to_consume(futures, job_args):\n                                _prune_futures(futures, lgr) or sleeper()\n                        # Current implementation, to provide depchecking, relies on unique\n                        # args for the job\n                        assert job_args not in futures\n                        lgr.debug(\"Submitting worker future for %s\", job_args)\n                        futures[job_args] = executor.submit(consumer_worker, self.consumer, job_args)\n                    except Empty:\n                        pass\n\n                # check active futures\n                if not consumer_queue.empty():\n                    done_useful = True\n\n                    # ATM we do not bother of some \"in order\" reporting\n                    # Just report as soon as any new record arrives\n                    res = consumer_queue.get()\n                    lgr.debug(\"Got %s from consumer_queue\", res)\n                    # LOGGING\n                    counts[res[\"status\"]] += 1\n                    count_strs = (_count_str(*args)\n                                  for args in [(counts[\"notneeded\"], \"skipped\", False),\n                                               (counts[\"error\"], \"failed\", True)])\n                    if counts[\"notneeded\"] or counts[\"error\"]:\n                        label = \"{} ({})\".format(\n                            self.label,\n                            \", \".join(filter(None, count_strs)))\n\n                    log_progress(\n                        lgr.error if res[\"status\"] == \"error\" else lgr.info,\n                        pid,\n                        \"%s: processed result%s\", self.label,\n                        \" for \" + res[\"path\"] if \"path\" in res else \"\",\n                        label=label, update=1, increment=True)\n\n                    yield res\n\n                done_useful |= _prune_futures(futures, lgr)\n\n                if not done_useful:  # you need some rest\n                    # TODO: same here -- progressive logging\n                    lgr.debug(\n                        \"Did nothing useful, sleeping. Have %s %s %s %s\",\n                        self.producer_finished,\n                        futures,\n                        consumer_queue.empty(),\n                        producer_queue.empty())\n                    sleeper()\n                else:\n                    sleeper.reset()\n\n        self._executor = None\n\n        # LOGGING\n        log_progress(lgr.info, pid, \"%s: done\", self.label)\n        producer_thread.join()\n        if self._exc:\n            if len(self._exc) > 1:\n                lgr.debug(\"%d exceptions were collected while performing execution in parallel. Only the first one \"\n                          \"will be reraised\", len(self._exc))\n            raise self._exc[0]\n        else:\n            assert not futures, \\\n                \"There is still %d active futures for following args: %s\" \\\n                % (len(futures), ', '.join(futures))\n\n\ndef _prune_futures(futures, lgr):\n    \"\"\"Removes .done from provided futures.\n\n    Returns\n    -------\n    bool\n      True if any future was removed\n    \"\"\"\n    done_useful = False\n    # remove futures which are done\n    for args, future in list(futures.items()):\n        if future.done():\n            done_useful = True\n            future_ = futures.pop(args)\n            msg = \"\"\n            if future_.exception():\n                msg = \" with an exception %s\" % future_.exception()\n            lgr.debug(\"Future for %r is done%s\", args, msg)\n    return done_useful\n\n\nclass Sleeper():\n    def __init__(self):\n        self.min_sleep_time = 0.001\n        # but no more than to this max\n        self.max_sleep_time = 0.1\n        self.sleep_time = self.min_sleep_time\n\n    def __call__(self):\n        time.sleep(self.sleep_time)\n        self.sleep_time = min(self.max_sleep_time, self.sleep_time * 2)\n\n    def reset(self):\n        self.sleep_time = self.min_sleep_time\n",
            "header": {
                "file": {
                    "path": "datalad/support/parallel.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "ad71207b1",
                "new_commit_hash": "7346b8a93",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/datalad/support/parallel.py"
            },
            "plus_file": {
                "path": "b/datalad/support/parallel.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 50,
                            "length": 7
                        },
                        "new_range": {
                            "start": 50,
                            "length": 7
                        },
                        "section": "class ProducerConsumer:"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    def __init__(self,",
                                "type": "context"
                            },
                            {
                                "content": "                 producer, consumer,",
                                "type": "context"
                            },
                            {
                                "content": "                 njobs=None,",
                                "type": "deleted"
                            },
                            {
                                "content": "                 jobs=None,",
                                "type": "added"
                            },
                            {
                                "content": "                 safe_to_consume=None,",
                                "type": "context"
                            },
                            {
                                "content": "                 reraise_immediately=False,",
                                "type": "context"
                            },
                            {
                                "content": "                 agg=None,",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "datalad/support/parallel.py",
                            "type": "class",
                            "name": "ProducerConsumer",
                            "content": "class ProducerConsumer:\n    \"\"\"\n    \"\"\"\n\n    def __init__(self,\n                 producer, consumer,\n                 njobs=None,\n                 safe_to_consume=None,\n                 reraise_immediately=False,\n                 agg=None,\n                 # TODO: \"absorb\" into some more generic \"logging\" helper\n                 # so could be used not only with consumers which yield our records\n                 # 'agg' from above could also more relate to i\n                 label=\"Total\", unit=\"items\",\n                 lgr=None,\n                 ):\n        \"\"\"\n\n        Parameters\n        ----------\n        ...\n        safe_to_consume: callable\n          A callable which gets a dict of all known futures and current producer output.\n          It should return True if we can proceed with current value from producer.\n          If unsafe - we will wait.  WARNING: outside code should make sure about provider and\n          safe_to_consume to play nicely or deadlock can happen.\n        reraise_immediately: bool, optional\n          If True, it would stop producer yielding values as soon as it detects that some\n          exception has occurred (although there might still be values in the queue to be yielded\n          which were collected before the exception was raised)\n        \"\"\"\n        self.producer = producer\n        self.consumer = consumer\n        self.njobs = njobs\n        self.safe_to_consume = safe_to_consume\n        self.reraise_immediately = reraise_immediately\n        self.agg = agg\n        self.label = label\n        self.unit = unit\n        self.lgr = lgr\n\n        self.total = None\n        self.producer_finished = None\n        self._executor = None\n        self._exc = []\n\n    def __del__(self):\n        # if we are killed while executing, we should ask executor to terminate\n        executor = getattr(self, \"_executor\")\n        if executor:\n            executor.terminate()\n\n    def __iter__(self):\n        self.producer_finished = False\n        self._exc = []\n\n        producer_queue = Queue()\n        consumer_queue = Queue()\n\n        def producer_worker():\n            \"\"\"That is the one which interrogates producer and updates .total\"\"\"\n            total = None if self.agg else 0\n            try:\n                for value in self.producer:\n                    producer_queue.put(value)\n                    if self.agg:\n                        total = (\n                            self.agg(value, total) if total is not None else self.agg(value)\n                        )\n                    else:\n                        total += 1\n                    self.total = total\n            except BaseException as e:\n                self._exc.append(e)\n            finally:\n                self.producer_finished = True\n\n        def consumer_worker(callable, *args, **kwargs):\n            \"\"\"Since jobs could return a generator and we cannot really \"inspect\" for that\n            \"\"\"\n            try:\n                res = callable(*args, **kwargs)\n                if inspect.isgenerator(res):\n                    lgr.debug(\"Got consumer worker which returned a generator %s\", res)\n                    didgood = False\n                    for r in res:\n                        didgood = True\n                        lgr.debug(\"Adding %s to queue\", r)\n                        consumer_queue.put(r)\n                    if not didgood:\n                        lgr.error(\"Nothing was obtained from %s :-(\", res)\n                else:\n                    lgr.debug(\"Got straight result %s, not a generator\", res)\n                    consumer_queue.put(res)\n            except BaseException as e:\n                self._exc.append(e)\n\n        producer_thread = Thread(target=producer_worker)\n        producer_thread.start()\n\n        # LOGGING\n        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?\n        lgr = self.lgr\n        label = self.label\n        if lgr is None:\n            from .. import lgr\n        log_progress(lgr.info, pid,\n                     \"%s: starting\", self.label,\n                     # will become known only later total=len(items),\n                     label=self.label, unit=\" \" + self.unit)\n        counts = defaultdict(int)\n\n        futures = {}\n\n        total_announced = False\n        njobs = self.njobs or 1\n        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", njobs)\n        # we will increase sleep_time when doing nothing useful\n        sleeper = Sleeper()\n        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:\n            self._executor = executor\n            # yield from the producer_queue (.total and .finished could be accessed meanwhile)\n            while True:\n                done_useful = False\n                if self.reraise_immediately and self._exc:\n                    break\n                if (self.producer_finished and\n                        not futures and\n                        consumer_queue.empty() and\n                        producer_queue.empty()):\n                    break\n\n                # LOGGING\n                if not total_announced and self.producer_finished:\n                    log_progress(\n                        lgr.info,\n                        pid,\n                        None,  # I do not think there is something valuable to announce\n                        total=self.total,\n                        # unfortunately of no effect, so we cannot inform that more items to come\n                        # unit=(\"+\" if not it.finished else \"\") + \" \" + unit,\n                        update=0  # not None, so it does not stop\n                    )\n                    total_announced = True\n\n                # important!  We are using threads, so worker threads will be sharing CPU time\n                # with this master thread. For it to become efficient, we should consume as much\n                # as possible from producer asap and push it to executor.  So drain the queue\n                while not producer_queue.empty():\n                    done_useful = True\n                    try:\n                        job_args = producer_queue.get() # timeout=0.001)\n                        if self.safe_to_consume:\n                            # Sleep a little if we are not yet ready\n                            # TODO: add some .debug level reporting based on elapsed time\n                            # IIRC I did smth like growing exponentially delays somewhere (dandi?)\n                            while not self.safe_to_consume(futures, job_args):\n                                _prune_futures(futures, lgr) or sleeper()\n                        # Current implementation, to provide depchecking, relies on unique\n                        # args for the job\n                        assert job_args not in futures\n                        lgr.debug(\"Submitting worker future for %s\", job_args)\n                        futures[job_args] = executor.submit(consumer_worker, self.consumer, job_args)\n                    except Empty:\n                        pass\n\n                # check active futures\n                if not consumer_queue.empty():\n                    done_useful = True\n\n                    # ATM we do not bother of some \"in order\" reporting\n                    # Just report as soon as any new record arrives\n                    res = consumer_queue.get()\n                    lgr.debug(\"Got %s from consumer_queue\", res)\n                    # LOGGING\n                    counts[res[\"status\"]] += 1\n                    count_strs = (_count_str(*args)\n                                  for args in [(counts[\"notneeded\"], \"skipped\", False),\n                                               (counts[\"error\"], \"failed\", True)])\n                    if counts[\"notneeded\"] or counts[\"error\"]:\n                        label = \"{} ({})\".format(\n                            self.label,\n                            \", \".join(filter(None, count_strs)))\n\n                    log_progress(\n                        lgr.error if res[\"status\"] == \"error\" else lgr.info,\n                        pid,\n                        \"%s: processed result%s\", self.label,\n                        \" for \" + res[\"path\"] if \"path\" in res else \"\",\n                        label=label, update=1, increment=True)\n\n                    yield res\n\n                done_useful |= _prune_futures(futures, lgr)\n\n                if not done_useful:  # you need some rest\n                    # TODO: same here -- progressive logging\n                    lgr.debug(\n                        \"Did nothing useful, sleeping. Have %s %s %s %s\",\n                        self.producer_finished,\n                        futures,\n                        consumer_queue.empty(),\n                        producer_queue.empty())\n                    sleeper()\n                else:\n                    sleeper.reset()\n\n        self._executor = None\n\n        # LOGGING\n        log_progress(lgr.info, pid, \"%s: done\", self.label)\n        producer_thread.join()\n        if self._exc:\n            if len(self._exc) > 1:\n                lgr.debug(\"%d exceptions were collected while performing execution in parallel. Only the first one \"\n                          \"will be reraised\", len(self._exc))\n            raise self._exc[0]\n        else:\n            assert not futures, \\\n                \"There is still %d active futures for following args: %s\" \\\n                % (len(futures), ', '.join(futures))",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 47,
                            "end_lineno": 267,
                            "parent": null
                        },
                        {
                            "file_name": "datalad/support/parallel.py",
                            "type": "method",
                            "name": "ProducerConsumer.__init__",
                            "content": "class ProducerConsumer:\n    \"\"\"\n    \"\"\"\n\n    def __init__(self,\n                 producer, consumer,\n                 njobs=None,\n                 safe_to_consume=None,\n                 reraise_immediately=False,\n                 agg=None,\n                 # TODO: \"absorb\" into some more generic \"logging\" helper\n                 # so could be used not only with consumers which yield our records\n                 # 'agg' from above could also more relate to i\n                 label=\"Total\", unit=\"items\",\n                 lgr=None,\n                 ):\n        \"\"\"\n\n        Parameters\n        ----------\n        ...\n        safe_to_consume: callable\n          A callable which gets a dict of all known futures and current producer output.\n          It should return True if we can proceed with current value from producer.\n          If unsafe - we will wait.  WARNING: outside code should make sure about provider and\n          safe_to_consume to play nicely or deadlock can happen.\n        reraise_immediately: bool, optional\n          If True, it would stop producer yielding values as soon as it detects that some\n          exception has occurred (although there might still be values in the queue to be yielded\n          which were collected before the exception was raised)\n        \"\"\"\n        self.producer = producer\n        self.consumer = consumer\n        self.njobs = njobs\n        self.safe_to_consume = safe_to_consume\n        self.reraise_immediately = reraise_immediately\n        self.agg = agg\n        self.label = label\n        self.unit = unit\n        self.lgr = lgr\n\n        self.total = None\n        self.producer_finished = None\n        self._executor = None\n        self._exc = []\n\n    def __del__(self):\n        # if we are killed while executing, we should ask executor to terminate\n        executor = getattr(self, \"_executor\")\n        if executor:\n            executor.terminate()\n\n    def __iter__(self):\n        self.producer_finished = False\n        self._exc = []\n\n        producer_queue = Queue()\n        consumer_queue = Queue()\n\n        def producer_worker():\n            \"\"\"That is the one which interrogates producer and updates .total\"\"\"\n            total = None if self.agg else 0\n            try:\n                for value in self.producer:\n                    producer_queue.put(value)\n                    if self.agg:\n                        total = (\n                            self.agg(value, total) if total is not None else self.agg(value)\n                        )\n                    else:\n                        total += 1\n                    self.total = total\n            except BaseException as e:\n                self._exc.append(e)\n            finally:\n                self.producer_finished = True\n\n        def consumer_worker(callable, *args, **kwargs):\n            \"\"\"Since jobs could return a generator and we cannot really \"inspect\" for that\n            \"\"\"\n            try:\n                res = callable(*args, **kwargs)\n                if inspect.isgenerator(res):\n                    lgr.debug(\"Got consumer worker which returned a generator %s\", res)\n                    didgood = False\n                    for r in res:\n                        didgood = True\n                        lgr.debug(\"Adding %s to queue\", r)\n                        consumer_queue.put(r)\n                    if not didgood:\n                        lgr.error(\"Nothing was obtained from %s :-(\", res)\n                else:\n                    lgr.debug(\"Got straight result %s, not a generator\", res)\n                    consumer_queue.put(res)\n            except BaseException as e:\n                self._exc.append(e)\n\n        producer_thread = Thread(target=producer_worker)\n        producer_thread.start()\n\n        # LOGGING\n        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?\n        lgr = self.lgr\n        label = self.label\n        if lgr is None:\n            from .. import lgr\n        log_progress(lgr.info, pid,\n                     \"%s: starting\", self.label,\n                     # will become known only later total=len(items),\n                     label=self.label, unit=\" \" + self.unit)\n        counts = defaultdict(int)\n\n        futures = {}\n\n        total_announced = False\n        njobs = self.njobs or 1\n        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", njobs)\n        # we will increase sleep_time when doing nothing useful\n        sleeper = Sleeper()\n        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:\n            self._executor = executor\n            # yield from the producer_queue (.total and .finished could be accessed meanwhile)\n            while True:\n                done_useful = False\n                if self.reraise_immediately and self._exc:\n                    break\n                if (self.producer_finished and\n                        not futures and\n                        consumer_queue.empty() and\n                        producer_queue.empty()):\n                    break\n\n                # LOGGING\n                if not total_announced and self.producer_finished:\n                    log_progress(\n                        lgr.info,\n                        pid,\n                        None,  # I do not think there is something valuable to announce\n                        total=self.total,\n                        # unfortunately of no effect, so we cannot inform that more items to come\n                        # unit=(\"+\" if not it.finished else \"\") + \" \" + unit,\n                        update=0  # not None, so it does not stop\n                    )\n                    total_announced = True\n\n                # important!  We are using threads, so worker threads will be sharing CPU time\n                # with this master thread. For it to become efficient, we should consume as much\n                # as possible from producer asap and push it to executor.  So drain the queue\n                while not producer_queue.empty():\n                    done_useful = True\n                    try:\n                        job_args = producer_queue.get() # timeout=0.001)\n                        if self.safe_to_consume:\n                            # Sleep a little if we are not yet ready\n                            # TODO: add some .debug level reporting based on elapsed time\n                            # IIRC I did smth like growing exponentially delays somewhere (dandi?)\n                            while not self.safe_to_consume(futures, job_args):\n                                _prune_futures(futures, lgr) or sleeper()\n                        # Current implementation, to provide depchecking, relies on unique\n                        # args for the job\n                        assert job_args not in futures\n                        lgr.debug(\"Submitting worker future for %s\", job_args)\n                        futures[job_args] = executor.submit(consumer_worker, self.consumer, job_args)\n                    except Empty:\n                        pass\n\n                # check active futures\n                if not consumer_queue.empty():\n                    done_useful = True\n\n                    # ATM we do not bother of some \"in order\" reporting\n                    # Just report as soon as any new record arrives\n                    res = consumer_queue.get()\n                    lgr.debug(\"Got %s from consumer_queue\", res)\n                    # LOGGING\n                    counts[res[\"status\"]] += 1\n                    count_strs = (_count_str(*args)\n                                  for args in [(counts[\"notneeded\"], \"skipped\", False),\n                                               (counts[\"error\"], \"failed\", True)])\n                    if counts[\"notneeded\"] or counts[\"error\"]:\n                        label = \"{} ({})\".format(\n                            self.label,\n                            \", \".join(filter(None, count_strs)))\n\n                    log_progress(\n                        lgr.error if res[\"status\"] == \"error\" else lgr.info,\n                        pid,\n                        \"%s: processed result%s\", self.label,\n                        \" for \" + res[\"path\"] if \"path\" in res else \"\",\n                        label=label, update=1, increment=True)\n\n                    yield res\n\n                done_useful |= _prune_futures(futures, lgr)\n\n                if not done_useful:  # you need some rest\n                    # TODO: same here -- progressive logging\n                    lgr.debug(\n                        \"Did nothing useful, sleeping. Have %s %s %s %s\",\n                        self.producer_finished,\n                        futures,\n                        consumer_queue.empty(),\n                        producer_queue.empty())\n                    sleeper()\n                else:\n                    sleeper.reset()\n\n        self._executor = None\n\n        # LOGGING\n        log_progress(lgr.info, pid, \"%s: done\", self.label)\n        producer_thread.join()\n        if self._exc:\n            if len(self._exc) > 1:\n                lgr.debug(\"%d exceptions were collected while performing execution in parallel. Only the first one \"\n                          \"will be reraised\", len(self._exc))\n            raise self._exc[0]\n        else:\n            assert not futures, \\\n                \"There is still %d active futures for following args: %s\" \\\n                % (len(futures), ', '.join(futures))",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 51,
                            "end_lineno": 91,
                            "parent": {
                                "file_name": "datalad/support/parallel.py",
                                "type": "class",
                                "name": "ProducerConsumer",
                                "content": "class ProducerConsumer:\n    \"\"\"\n    \"\"\"\n\n    def __init__(self,\n                 producer, consumer,\n                 njobs=None,\n                 safe_to_consume=None,\n                 reraise_immediately=False,\n                 agg=None,\n                 # TODO: \"absorb\" into some more generic \"logging\" helper\n                 # so could be used not only with consumers which yield our records\n                 # 'agg' from above could also more relate to i\n                 label=\"Total\", unit=\"items\",\n                 lgr=None,\n                 ):\n        \"\"\"\n\n        Parameters\n        ----------\n        ...\n        safe_to_consume: callable\n          A callable which gets a dict of all known futures and current producer output.\n          It should return True if we can proceed with current value from producer.\n          If unsafe - we will wait.  WARNING: outside code should make sure about provider and\n          safe_to_consume to play nicely or deadlock can happen.\n        reraise_immediately: bool, optional\n          If True, it would stop producer yielding values as soon as it detects that some\n          exception has occurred (although there might still be values in the queue to be yielded\n          which were collected before the exception was raised)\n        \"\"\"\n        self.producer = producer\n        self.consumer = consumer\n        self.njobs = njobs\n        self.safe_to_consume = safe_to_consume\n        self.reraise_immediately = reraise_immediately\n        self.agg = agg\n        self.label = label\n        self.unit = unit\n        self.lgr = lgr\n\n        self.total = None\n        self.producer_finished = None\n        self._executor = None\n        self._exc = []\n\n    def __del__(self):\n        # if we are killed while executing, we should ask executor to terminate\n        executor = getattr(self, \"_executor\")\n        if executor:\n            executor.terminate()\n\n    def __iter__(self):\n        self.producer_finished = False\n        self._exc = []\n\n        producer_queue = Queue()\n        consumer_queue = Queue()\n\n        def producer_worker():\n            \"\"\"That is the one which interrogates producer and updates .total\"\"\"\n            total = None if self.agg else 0\n            try:\n                for value in self.producer:\n                    producer_queue.put(value)\n                    if self.agg:\n                        total = (\n                            self.agg(value, total) if total is not None else self.agg(value)\n                        )\n                    else:\n                        total += 1\n                    self.total = total\n            except BaseException as e:\n                self._exc.append(e)\n            finally:\n                self.producer_finished = True\n\n        def consumer_worker(callable, *args, **kwargs):\n            \"\"\"Since jobs could return a generator and we cannot really \"inspect\" for that\n            \"\"\"\n            try:\n                res = callable(*args, **kwargs)\n                if inspect.isgenerator(res):\n                    lgr.debug(\"Got consumer worker which returned a generator %s\", res)\n                    didgood = False\n                    for r in res:\n                        didgood = True\n                        lgr.debug(\"Adding %s to queue\", r)\n                        consumer_queue.put(r)\n                    if not didgood:\n                        lgr.error(\"Nothing was obtained from %s :-(\", res)\n                else:\n                    lgr.debug(\"Got straight result %s, not a generator\", res)\n                    consumer_queue.put(res)\n            except BaseException as e:\n                self._exc.append(e)\n\n        producer_thread = Thread(target=producer_worker)\n        producer_thread.start()\n\n        # LOGGING\n        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?\n        lgr = self.lgr\n        label = self.label\n        if lgr is None:\n            from .. import lgr\n        log_progress(lgr.info, pid,\n                     \"%s: starting\", self.label,\n                     # will become known only later total=len(items),\n                     label=self.label, unit=\" \" + self.unit)\n        counts = defaultdict(int)\n\n        futures = {}\n\n        total_announced = False\n        njobs = self.njobs or 1\n        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", njobs)\n        # we will increase sleep_time when doing nothing useful\n        sleeper = Sleeper()\n        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:\n            self._executor = executor\n            # yield from the producer_queue (.total and .finished could be accessed meanwhile)\n            while True:\n                done_useful = False\n                if self.reraise_immediately and self._exc:\n                    break\n                if (self.producer_finished and\n                        not futures and\n                        consumer_queue.empty() and\n                        producer_queue.empty()):\n                    break\n\n                # LOGGING\n                if not total_announced and self.producer_finished:\n                    log_progress(\n                        lgr.info,\n                        pid,\n                        None,  # I do not think there is something valuable to announce\n                        total=self.total,\n                        # unfortunately of no effect, so we cannot inform that more items to come\n                        # unit=(\"+\" if not it.finished else \"\") + \" \" + unit,\n                        update=0  # not None, so it does not stop\n                    )\n                    total_announced = True\n\n                # important!  We are using threads, so worker threads will be sharing CPU time\n                # with this master thread. For it to become efficient, we should consume as much\n                # as possible from producer asap and push it to executor.  So drain the queue\n                while not producer_queue.empty():\n                    done_useful = True\n                    try:\n                        job_args = producer_queue.get() # timeout=0.001)\n                        if self.safe_to_consume:\n                            # Sleep a little if we are not yet ready\n                            # TODO: add some .debug level reporting based on elapsed time\n                            # IIRC I did smth like growing exponentially delays somewhere (dandi?)\n                            while not self.safe_to_consume(futures, job_args):\n                                _prune_futures(futures, lgr) or sleeper()\n                        # Current implementation, to provide depchecking, relies on unique\n                        # args for the job\n                        assert job_args not in futures\n                        lgr.debug(\"Submitting worker future for %s\", job_args)\n                        futures[job_args] = executor.submit(consumer_worker, self.consumer, job_args)\n                    except Empty:\n                        pass\n\n                # check active futures\n                if not consumer_queue.empty():\n                    done_useful = True\n\n                    # ATM we do not bother of some \"in order\" reporting\n                    # Just report as soon as any new record arrives\n                    res = consumer_queue.get()\n                    lgr.debug(\"Got %s from consumer_queue\", res)\n                    # LOGGING\n                    counts[res[\"status\"]] += 1\n                    count_strs = (_count_str(*args)\n                                  for args in [(counts[\"notneeded\"], \"skipped\", False),\n                                               (counts[\"error\"], \"failed\", True)])\n                    if counts[\"notneeded\"] or counts[\"error\"]:\n                        label = \"{} ({})\".format(\n                            self.label,\n                            \", \".join(filter(None, count_strs)))\n\n                    log_progress(\n                        lgr.error if res[\"status\"] == \"error\" else lgr.info,\n                        pid,\n                        \"%s: processed result%s\", self.label,\n                        \" for \" + res[\"path\"] if \"path\" in res else \"\",\n                        label=label, update=1, increment=True)\n\n                    yield res\n\n                done_useful |= _prune_futures(futures, lgr)\n\n                if not done_useful:  # you need some rest\n                    # TODO: same here -- progressive logging\n                    lgr.debug(\n                        \"Did nothing useful, sleeping. Have %s %s %s %s\",\n                        self.producer_finished,\n                        futures,\n                        consumer_queue.empty(),\n                        producer_queue.empty())\n                    sleeper()\n                else:\n                    sleeper.reset()\n\n        self._executor = None\n\n        # LOGGING\n        log_progress(lgr.info, pid, \"%s: done\", self.label)\n        producer_thread.join()\n        if self._exc:\n            if len(self._exc) > 1:\n                lgr.debug(\"%d exceptions were collected while performing execution in parallel. Only the first one \"\n                          \"will be reraised\", len(self._exc))\n            raise self._exc[0]\n        else:\n            assert not futures, \\\n                \"There is still %d active futures for following args: %s\" \\\n                % (len(futures), ', '.join(futures))",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 47,
                                "end_lineno": 267,
                                "parent": null
                            }
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 77,
                            "length": 7
                        },
                        "new_range": {
                            "start": 77,
                            "length": 7
                        },
                        "section": "class ProducerConsumer:"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        \"\"\"",
                                "type": "context"
                            },
                            {
                                "content": "        self.producer = producer",
                                "type": "context"
                            },
                            {
                                "content": "        self.consumer = consumer",
                                "type": "context"
                            },
                            {
                                "content": "        self.njobs = njobs",
                                "type": "deleted"
                            },
                            {
                                "content": "        self.jobs = jobs",
                                "type": "added"
                            },
                            {
                                "content": "        self.safe_to_consume = safe_to_consume",
                                "type": "context"
                            },
                            {
                                "content": "        self.reraise_immediately = reraise_immediately",
                                "type": "context"
                            },
                            {
                                "content": "        self.agg = agg",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "datalad/support/parallel.py",
                            "type": "class",
                            "name": "ProducerConsumer",
                            "content": "class ProducerConsumer:\n    \"\"\"\n    \"\"\"\n\n    def __init__(self,\n                 producer, consumer,\n                 njobs=None,\n                 safe_to_consume=None,\n                 reraise_immediately=False,\n                 agg=None,\n                 # TODO: \"absorb\" into some more generic \"logging\" helper\n                 # so could be used not only with consumers which yield our records\n                 # 'agg' from above could also more relate to i\n                 label=\"Total\", unit=\"items\",\n                 lgr=None,\n                 ):\n        \"\"\"\n\n        Parameters\n        ----------\n        ...\n        safe_to_consume: callable\n          A callable which gets a dict of all known futures and current producer output.\n          It should return True if we can proceed with current value from producer.\n          If unsafe - we will wait.  WARNING: outside code should make sure about provider and\n          safe_to_consume to play nicely or deadlock can happen.\n        reraise_immediately: bool, optional\n          If True, it would stop producer yielding values as soon as it detects that some\n          exception has occurred (although there might still be values in the queue to be yielded\n          which were collected before the exception was raised)\n        \"\"\"\n        self.producer = producer\n        self.consumer = consumer\n        self.njobs = njobs\n        self.safe_to_consume = safe_to_consume\n        self.reraise_immediately = reraise_immediately\n        self.agg = agg\n        self.label = label\n        self.unit = unit\n        self.lgr = lgr\n\n        self.total = None\n        self.producer_finished = None\n        self._executor = None\n        self._exc = []\n\n    def __del__(self):\n        # if we are killed while executing, we should ask executor to terminate\n        executor = getattr(self, \"_executor\")\n        if executor:\n            executor.terminate()\n\n    def __iter__(self):\n        self.producer_finished = False\n        self._exc = []\n\n        producer_queue = Queue()\n        consumer_queue = Queue()\n\n        def producer_worker():\n            \"\"\"That is the one which interrogates producer and updates .total\"\"\"\n            total = None if self.agg else 0\n            try:\n                for value in self.producer:\n                    producer_queue.put(value)\n                    if self.agg:\n                        total = (\n                            self.agg(value, total) if total is not None else self.agg(value)\n                        )\n                    else:\n                        total += 1\n                    self.total = total\n            except BaseException as e:\n                self._exc.append(e)\n            finally:\n                self.producer_finished = True\n\n        def consumer_worker(callable, *args, **kwargs):\n            \"\"\"Since jobs could return a generator and we cannot really \"inspect\" for that\n            \"\"\"\n            try:\n                res = callable(*args, **kwargs)\n                if inspect.isgenerator(res):\n                    lgr.debug(\"Got consumer worker which returned a generator %s\", res)\n                    didgood = False\n                    for r in res:\n                        didgood = True\n                        lgr.debug(\"Adding %s to queue\", r)\n                        consumer_queue.put(r)\n                    if not didgood:\n                        lgr.error(\"Nothing was obtained from %s :-(\", res)\n                else:\n                    lgr.debug(\"Got straight result %s, not a generator\", res)\n                    consumer_queue.put(res)\n            except BaseException as e:\n                self._exc.append(e)\n\n        producer_thread = Thread(target=producer_worker)\n        producer_thread.start()\n\n        # LOGGING\n        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?\n        lgr = self.lgr\n        label = self.label\n        if lgr is None:\n            from .. import lgr\n        log_progress(lgr.info, pid,\n                     \"%s: starting\", self.label,\n                     # will become known only later total=len(items),\n                     label=self.label, unit=\" \" + self.unit)\n        counts = defaultdict(int)\n\n        futures = {}\n\n        total_announced = False\n        njobs = self.njobs or 1\n        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", njobs)\n        # we will increase sleep_time when doing nothing useful\n        sleeper = Sleeper()\n        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:\n            self._executor = executor\n            # yield from the producer_queue (.total and .finished could be accessed meanwhile)\n            while True:\n                done_useful = False\n                if self.reraise_immediately and self._exc:\n                    break\n                if (self.producer_finished and\n                        not futures and\n                        consumer_queue.empty() and\n                        producer_queue.empty()):\n                    break\n\n                # LOGGING\n                if not total_announced and self.producer_finished:\n                    log_progress(\n                        lgr.info,\n                        pid,\n                        None,  # I do not think there is something valuable to announce\n                        total=self.total,\n                        # unfortunately of no effect, so we cannot inform that more items to come\n                        # unit=(\"+\" if not it.finished else \"\") + \" \" + unit,\n                        update=0  # not None, so it does not stop\n                    )\n                    total_announced = True\n\n                # important!  We are using threads, so worker threads will be sharing CPU time\n                # with this master thread. For it to become efficient, we should consume as much\n                # as possible from producer asap and push it to executor.  So drain the queue\n                while not producer_queue.empty():\n                    done_useful = True\n                    try:\n                        job_args = producer_queue.get() # timeout=0.001)\n                        if self.safe_to_consume:\n                            # Sleep a little if we are not yet ready\n                            # TODO: add some .debug level reporting based on elapsed time\n                            # IIRC I did smth like growing exponentially delays somewhere (dandi?)\n                            while not self.safe_to_consume(futures, job_args):\n                                _prune_futures(futures, lgr) or sleeper()\n                        # Current implementation, to provide depchecking, relies on unique\n                        # args for the job\n                        assert job_args not in futures\n                        lgr.debug(\"Submitting worker future for %s\", job_args)\n                        futures[job_args] = executor.submit(consumer_worker, self.consumer, job_args)\n                    except Empty:\n                        pass\n\n                # check active futures\n                if not consumer_queue.empty():\n                    done_useful = True\n\n                    # ATM we do not bother of some \"in order\" reporting\n                    # Just report as soon as any new record arrives\n                    res = consumer_queue.get()\n                    lgr.debug(\"Got %s from consumer_queue\", res)\n                    # LOGGING\n                    counts[res[\"status\"]] += 1\n                    count_strs = (_count_str(*args)\n                                  for args in [(counts[\"notneeded\"], \"skipped\", False),\n                                               (counts[\"error\"], \"failed\", True)])\n                    if counts[\"notneeded\"] or counts[\"error\"]:\n                        label = \"{} ({})\".format(\n                            self.label,\n                            \", \".join(filter(None, count_strs)))\n\n                    log_progress(\n                        lgr.error if res[\"status\"] == \"error\" else lgr.info,\n                        pid,\n                        \"%s: processed result%s\", self.label,\n                        \" for \" + res[\"path\"] if \"path\" in res else \"\",\n                        label=label, update=1, increment=True)\n\n                    yield res\n\n                done_useful |= _prune_futures(futures, lgr)\n\n                if not done_useful:  # you need some rest\n                    # TODO: same here -- progressive logging\n                    lgr.debug(\n                        \"Did nothing useful, sleeping. Have %s %s %s %s\",\n                        self.producer_finished,\n                        futures,\n                        consumer_queue.empty(),\n                        producer_queue.empty())\n                    sleeper()\n                else:\n                    sleeper.reset()\n\n        self._executor = None\n\n        # LOGGING\n        log_progress(lgr.info, pid, \"%s: done\", self.label)\n        producer_thread.join()\n        if self._exc:\n            if len(self._exc) > 1:\n                lgr.debug(\"%d exceptions were collected while performing execution in parallel. Only the first one \"\n                          \"will be reraised\", len(self._exc))\n            raise self._exc[0]\n        else:\n            assert not futures, \\\n                \"There is still %d active futures for following args: %s\" \\\n                % (len(futures), ', '.join(futures))",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 47,
                            "end_lineno": 267,
                            "parent": null
                        },
                        {
                            "file_name": "datalad/support/parallel.py",
                            "type": "method",
                            "name": "ProducerConsumer.__init__",
                            "content": "class ProducerConsumer:\n    \"\"\"\n    \"\"\"\n\n    def __init__(self,\n                 producer, consumer,\n                 njobs=None,\n                 safe_to_consume=None,\n                 reraise_immediately=False,\n                 agg=None,\n                 # TODO: \"absorb\" into some more generic \"logging\" helper\n                 # so could be used not only with consumers which yield our records\n                 # 'agg' from above could also more relate to i\n                 label=\"Total\", unit=\"items\",\n                 lgr=None,\n                 ):\n        \"\"\"\n\n        Parameters\n        ----------\n        ...\n        safe_to_consume: callable\n          A callable which gets a dict of all known futures and current producer output.\n          It should return True if we can proceed with current value from producer.\n          If unsafe - we will wait.  WARNING: outside code should make sure about provider and\n          safe_to_consume to play nicely or deadlock can happen.\n        reraise_immediately: bool, optional\n          If True, it would stop producer yielding values as soon as it detects that some\n          exception has occurred (although there might still be values in the queue to be yielded\n          which were collected before the exception was raised)\n        \"\"\"\n        self.producer = producer\n        self.consumer = consumer\n        self.njobs = njobs\n        self.safe_to_consume = safe_to_consume\n        self.reraise_immediately = reraise_immediately\n        self.agg = agg\n        self.label = label\n        self.unit = unit\n        self.lgr = lgr\n\n        self.total = None\n        self.producer_finished = None\n        self._executor = None\n        self._exc = []\n\n    def __del__(self):\n        # if we are killed while executing, we should ask executor to terminate\n        executor = getattr(self, \"_executor\")\n        if executor:\n            executor.terminate()\n\n    def __iter__(self):\n        self.producer_finished = False\n        self._exc = []\n\n        producer_queue = Queue()\n        consumer_queue = Queue()\n\n        def producer_worker():\n            \"\"\"That is the one which interrogates producer and updates .total\"\"\"\n            total = None if self.agg else 0\n            try:\n                for value in self.producer:\n                    producer_queue.put(value)\n                    if self.agg:\n                        total = (\n                            self.agg(value, total) if total is not None else self.agg(value)\n                        )\n                    else:\n                        total += 1\n                    self.total = total\n            except BaseException as e:\n                self._exc.append(e)\n            finally:\n                self.producer_finished = True\n\n        def consumer_worker(callable, *args, **kwargs):\n            \"\"\"Since jobs could return a generator and we cannot really \"inspect\" for that\n            \"\"\"\n            try:\n                res = callable(*args, **kwargs)\n                if inspect.isgenerator(res):\n                    lgr.debug(\"Got consumer worker which returned a generator %s\", res)\n                    didgood = False\n                    for r in res:\n                        didgood = True\n                        lgr.debug(\"Adding %s to queue\", r)\n                        consumer_queue.put(r)\n                    if not didgood:\n                        lgr.error(\"Nothing was obtained from %s :-(\", res)\n                else:\n                    lgr.debug(\"Got straight result %s, not a generator\", res)\n                    consumer_queue.put(res)\n            except BaseException as e:\n                self._exc.append(e)\n\n        producer_thread = Thread(target=producer_worker)\n        producer_thread.start()\n\n        # LOGGING\n        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?\n        lgr = self.lgr\n        label = self.label\n        if lgr is None:\n            from .. import lgr\n        log_progress(lgr.info, pid,\n                     \"%s: starting\", self.label,\n                     # will become known only later total=len(items),\n                     label=self.label, unit=\" \" + self.unit)\n        counts = defaultdict(int)\n\n        futures = {}\n\n        total_announced = False\n        njobs = self.njobs or 1\n        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", njobs)\n        # we will increase sleep_time when doing nothing useful\n        sleeper = Sleeper()\n        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:\n            self._executor = executor\n            # yield from the producer_queue (.total and .finished could be accessed meanwhile)\n            while True:\n                done_useful = False\n                if self.reraise_immediately and self._exc:\n                    break\n                if (self.producer_finished and\n                        not futures and\n                        consumer_queue.empty() and\n                        producer_queue.empty()):\n                    break\n\n                # LOGGING\n                if not total_announced and self.producer_finished:\n                    log_progress(\n                        lgr.info,\n                        pid,\n                        None,  # I do not think there is something valuable to announce\n                        total=self.total,\n                        # unfortunately of no effect, so we cannot inform that more items to come\n                        # unit=(\"+\" if not it.finished else \"\") + \" \" + unit,\n                        update=0  # not None, so it does not stop\n                    )\n                    total_announced = True\n\n                # important!  We are using threads, so worker threads will be sharing CPU time\n                # with this master thread. For it to become efficient, we should consume as much\n                # as possible from producer asap and push it to executor.  So drain the queue\n                while not producer_queue.empty():\n                    done_useful = True\n                    try:\n                        job_args = producer_queue.get() # timeout=0.001)\n                        if self.safe_to_consume:\n                            # Sleep a little if we are not yet ready\n                            # TODO: add some .debug level reporting based on elapsed time\n                            # IIRC I did smth like growing exponentially delays somewhere (dandi?)\n                            while not self.safe_to_consume(futures, job_args):\n                                _prune_futures(futures, lgr) or sleeper()\n                        # Current implementation, to provide depchecking, relies on unique\n                        # args for the job\n                        assert job_args not in futures\n                        lgr.debug(\"Submitting worker future for %s\", job_args)\n                        futures[job_args] = executor.submit(consumer_worker, self.consumer, job_args)\n                    except Empty:\n                        pass\n\n                # check active futures\n                if not consumer_queue.empty():\n                    done_useful = True\n\n                    # ATM we do not bother of some \"in order\" reporting\n                    # Just report as soon as any new record arrives\n                    res = consumer_queue.get()\n                    lgr.debug(\"Got %s from consumer_queue\", res)\n                    # LOGGING\n                    counts[res[\"status\"]] += 1\n                    count_strs = (_count_str(*args)\n                                  for args in [(counts[\"notneeded\"], \"skipped\", False),\n                                               (counts[\"error\"], \"failed\", True)])\n                    if counts[\"notneeded\"] or counts[\"error\"]:\n                        label = \"{} ({})\".format(\n                            self.label,\n                            \", \".join(filter(None, count_strs)))\n\n                    log_progress(\n                        lgr.error if res[\"status\"] == \"error\" else lgr.info,\n                        pid,\n                        \"%s: processed result%s\", self.label,\n                        \" for \" + res[\"path\"] if \"path\" in res else \"\",\n                        label=label, update=1, increment=True)\n\n                    yield res\n\n                done_useful |= _prune_futures(futures, lgr)\n\n                if not done_useful:  # you need some rest\n                    # TODO: same here -- progressive logging\n                    lgr.debug(\n                        \"Did nothing useful, sleeping. Have %s %s %s %s\",\n                        self.producer_finished,\n                        futures,\n                        consumer_queue.empty(),\n                        producer_queue.empty())\n                    sleeper()\n                else:\n                    sleeper.reset()\n\n        self._executor = None\n\n        # LOGGING\n        log_progress(lgr.info, pid, \"%s: done\", self.label)\n        producer_thread.join()\n        if self._exc:\n            if len(self._exc) > 1:\n                lgr.debug(\"%d exceptions were collected while performing execution in parallel. Only the first one \"\n                          \"will be reraised\", len(self._exc))\n            raise self._exc[0]\n        else:\n            assert not futures, \\\n                \"There is still %d active futures for following args: %s\" \\\n                % (len(futures), ', '.join(futures))",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 51,
                            "end_lineno": 91,
                            "parent": {
                                "file_name": "datalad/support/parallel.py",
                                "type": "class",
                                "name": "ProducerConsumer",
                                "content": "class ProducerConsumer:\n    \"\"\"\n    \"\"\"\n\n    def __init__(self,\n                 producer, consumer,\n                 njobs=None,\n                 safe_to_consume=None,\n                 reraise_immediately=False,\n                 agg=None,\n                 # TODO: \"absorb\" into some more generic \"logging\" helper\n                 # so could be used not only with consumers which yield our records\n                 # 'agg' from above could also more relate to i\n                 label=\"Total\", unit=\"items\",\n                 lgr=None,\n                 ):\n        \"\"\"\n\n        Parameters\n        ----------\n        ...\n        safe_to_consume: callable\n          A callable which gets a dict of all known futures and current producer output.\n          It should return True if we can proceed with current value from producer.\n          If unsafe - we will wait.  WARNING: outside code should make sure about provider and\n          safe_to_consume to play nicely or deadlock can happen.\n        reraise_immediately: bool, optional\n          If True, it would stop producer yielding values as soon as it detects that some\n          exception has occurred (although there might still be values in the queue to be yielded\n          which were collected before the exception was raised)\n        \"\"\"\n        self.producer = producer\n        self.consumer = consumer\n        self.njobs = njobs\n        self.safe_to_consume = safe_to_consume\n        self.reraise_immediately = reraise_immediately\n        self.agg = agg\n        self.label = label\n        self.unit = unit\n        self.lgr = lgr\n\n        self.total = None\n        self.producer_finished = None\n        self._executor = None\n        self._exc = []\n\n    def __del__(self):\n        # if we are killed while executing, we should ask executor to terminate\n        executor = getattr(self, \"_executor\")\n        if executor:\n            executor.terminate()\n\n    def __iter__(self):\n        self.producer_finished = False\n        self._exc = []\n\n        producer_queue = Queue()\n        consumer_queue = Queue()\n\n        def producer_worker():\n            \"\"\"That is the one which interrogates producer and updates .total\"\"\"\n            total = None if self.agg else 0\n            try:\n                for value in self.producer:\n                    producer_queue.put(value)\n                    if self.agg:\n                        total = (\n                            self.agg(value, total) if total is not None else self.agg(value)\n                        )\n                    else:\n                        total += 1\n                    self.total = total\n            except BaseException as e:\n                self._exc.append(e)\n            finally:\n                self.producer_finished = True\n\n        def consumer_worker(callable, *args, **kwargs):\n            \"\"\"Since jobs could return a generator and we cannot really \"inspect\" for that\n            \"\"\"\n            try:\n                res = callable(*args, **kwargs)\n                if inspect.isgenerator(res):\n                    lgr.debug(\"Got consumer worker which returned a generator %s\", res)\n                    didgood = False\n                    for r in res:\n                        didgood = True\n                        lgr.debug(\"Adding %s to queue\", r)\n                        consumer_queue.put(r)\n                    if not didgood:\n                        lgr.error(\"Nothing was obtained from %s :-(\", res)\n                else:\n                    lgr.debug(\"Got straight result %s, not a generator\", res)\n                    consumer_queue.put(res)\n            except BaseException as e:\n                self._exc.append(e)\n\n        producer_thread = Thread(target=producer_worker)\n        producer_thread.start()\n\n        # LOGGING\n        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?\n        lgr = self.lgr\n        label = self.label\n        if lgr is None:\n            from .. import lgr\n        log_progress(lgr.info, pid,\n                     \"%s: starting\", self.label,\n                     # will become known only later total=len(items),\n                     label=self.label, unit=\" \" + self.unit)\n        counts = defaultdict(int)\n\n        futures = {}\n\n        total_announced = False\n        njobs = self.njobs or 1\n        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", njobs)\n        # we will increase sleep_time when doing nothing useful\n        sleeper = Sleeper()\n        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:\n            self._executor = executor\n            # yield from the producer_queue (.total and .finished could be accessed meanwhile)\n            while True:\n                done_useful = False\n                if self.reraise_immediately and self._exc:\n                    break\n                if (self.producer_finished and\n                        not futures and\n                        consumer_queue.empty() and\n                        producer_queue.empty()):\n                    break\n\n                # LOGGING\n                if not total_announced and self.producer_finished:\n                    log_progress(\n                        lgr.info,\n                        pid,\n                        None,  # I do not think there is something valuable to announce\n                        total=self.total,\n                        # unfortunately of no effect, so we cannot inform that more items to come\n                        # unit=(\"+\" if not it.finished else \"\") + \" \" + unit,\n                        update=0  # not None, so it does not stop\n                    )\n                    total_announced = True\n\n                # important!  We are using threads, so worker threads will be sharing CPU time\n                # with this master thread. For it to become efficient, we should consume as much\n                # as possible from producer asap and push it to executor.  So drain the queue\n                while not producer_queue.empty():\n                    done_useful = True\n                    try:\n                        job_args = producer_queue.get() # timeout=0.001)\n                        if self.safe_to_consume:\n                            # Sleep a little if we are not yet ready\n                            # TODO: add some .debug level reporting based on elapsed time\n                            # IIRC I did smth like growing exponentially delays somewhere (dandi?)\n                            while not self.safe_to_consume(futures, job_args):\n                                _prune_futures(futures, lgr) or sleeper()\n                        # Current implementation, to provide depchecking, relies on unique\n                        # args for the job\n                        assert job_args not in futures\n                        lgr.debug(\"Submitting worker future for %s\", job_args)\n                        futures[job_args] = executor.submit(consumer_worker, self.consumer, job_args)\n                    except Empty:\n                        pass\n\n                # check active futures\n                if not consumer_queue.empty():\n                    done_useful = True\n\n                    # ATM we do not bother of some \"in order\" reporting\n                    # Just report as soon as any new record arrives\n                    res = consumer_queue.get()\n                    lgr.debug(\"Got %s from consumer_queue\", res)\n                    # LOGGING\n                    counts[res[\"status\"]] += 1\n                    count_strs = (_count_str(*args)\n                                  for args in [(counts[\"notneeded\"], \"skipped\", False),\n                                               (counts[\"error\"], \"failed\", True)])\n                    if counts[\"notneeded\"] or counts[\"error\"]:\n                        label = \"{} ({})\".format(\n                            self.label,\n                            \", \".join(filter(None, count_strs)))\n\n                    log_progress(\n                        lgr.error if res[\"status\"] == \"error\" else lgr.info,\n                        pid,\n                        \"%s: processed result%s\", self.label,\n                        \" for \" + res[\"path\"] if \"path\" in res else \"\",\n                        label=label, update=1, increment=True)\n\n                    yield res\n\n                done_useful |= _prune_futures(futures, lgr)\n\n                if not done_useful:  # you need some rest\n                    # TODO: same here -- progressive logging\n                    lgr.debug(\n                        \"Did nothing useful, sleeping. Have %s %s %s %s\",\n                        self.producer_finished,\n                        futures,\n                        consumer_queue.empty(),\n                        producer_queue.empty())\n                    sleeper()\n                else:\n                    sleeper.reset()\n\n        self._executor = None\n\n        # LOGGING\n        log_progress(lgr.info, pid, \"%s: done\", self.label)\n        producer_thread.join()\n        if self._exc:\n            if len(self._exc) > 1:\n                lgr.debug(\"%d exceptions were collected while performing execution in parallel. Only the first one \"\n                          \"will be reraised\", len(self._exc))\n            raise self._exc[0]\n        else:\n            assert not futures, \\\n                \"There is still %d active futures for following args: %s\" \\\n                % (len(futures), ', '.join(futures))",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 47,
                                "end_lineno": 267,
                                "parent": null
                            }
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 159,
                            "length": 11
                        },
                        "new_range": {
                            "start": 159,
                            "length": 16
                        },
                        "section": "class ProducerConsumer:"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        futures = {}",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "        total_announced = False",
                                "type": "context"
                            },
                            {
                                "content": "        njobs = self.njobs or 1",
                                "type": "deleted"
                            },
                            {
                                "content": "        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", njobs)",
                                "type": "deleted"
                            },
                            {
                                "content": "        jobs = self.jobs or 1",
                                "type": "added"
                            },
                            {
                                "content": "        if jobs == \"auto\":",
                                "type": "added"
                            },
                            {
                                "content": "            # ATM there is no \"auto\" for this operation.  We will just make it ...",
                                "type": "added"
                            },
                            {
                                "content": "            # \"auto\" could be for some auto-scaling based on a single future time",
                                "type": "added"
                            },
                            {
                                "content": "            # to complete, scaling up/down.  TODO",
                                "type": "added"
                            },
                            {
                                "content": "            jobs = 5",
                                "type": "added"
                            },
                            {
                                "content": "        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", jobs)",
                                "type": "added"
                            },
                            {
                                "content": "        # we will increase sleep_time when doing nothing useful",
                                "type": "context"
                            },
                            {
                                "content": "        sleeper = Sleeper()",
                                "type": "context"
                            },
                            {
                                "content": "        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:",
                                "type": "deleted"
                            },
                            {
                                "content": "        with concurrent.futures.ThreadPoolExecutor(jobs) as executor:",
                                "type": "added"
                            },
                            {
                                "content": "            self._executor = executor",
                                "type": "context"
                            },
                            {
                                "content": "            # yield from the producer_queue (.total and .finished could be accessed meanwhile)",
                                "type": "context"
                            },
                            {
                                "content": "            while True:",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "datalad/support/parallel.py",
                            "type": "class",
                            "name": "ProducerConsumer",
                            "content": "class ProducerConsumer:\n    \"\"\"\n    \"\"\"\n\n    def __init__(self,\n                 producer, consumer,\n                 njobs=None,\n                 safe_to_consume=None,\n                 reraise_immediately=False,\n                 agg=None,\n                 # TODO: \"absorb\" into some more generic \"logging\" helper\n                 # so could be used not only with consumers which yield our records\n                 # 'agg' from above could also more relate to i\n                 label=\"Total\", unit=\"items\",\n                 lgr=None,\n                 ):\n        \"\"\"\n\n        Parameters\n        ----------\n        ...\n        safe_to_consume: callable\n          A callable which gets a dict of all known futures and current producer output.\n          It should return True if we can proceed with current value from producer.\n          If unsafe - we will wait.  WARNING: outside code should make sure about provider and\n          safe_to_consume to play nicely or deadlock can happen.\n        reraise_immediately: bool, optional\n          If True, it would stop producer yielding values as soon as it detects that some\n          exception has occurred (although there might still be values in the queue to be yielded\n          which were collected before the exception was raised)\n        \"\"\"\n        self.producer = producer\n        self.consumer = consumer\n        self.njobs = njobs\n        self.safe_to_consume = safe_to_consume\n        self.reraise_immediately = reraise_immediately\n        self.agg = agg\n        self.label = label\n        self.unit = unit\n        self.lgr = lgr\n\n        self.total = None\n        self.producer_finished = None\n        self._executor = None\n        self._exc = []\n\n    def __del__(self):\n        # if we are killed while executing, we should ask executor to terminate\n        executor = getattr(self, \"_executor\")\n        if executor:\n            executor.terminate()\n\n    def __iter__(self):\n        self.producer_finished = False\n        self._exc = []\n\n        producer_queue = Queue()\n        consumer_queue = Queue()\n\n        def producer_worker():\n            \"\"\"That is the one which interrogates producer and updates .total\"\"\"\n            total = None if self.agg else 0\n            try:\n                for value in self.producer:\n                    producer_queue.put(value)\n                    if self.agg:\n                        total = (\n                            self.agg(value, total) if total is not None else self.agg(value)\n                        )\n                    else:\n                        total += 1\n                    self.total = total\n            except BaseException as e:\n                self._exc.append(e)\n            finally:\n                self.producer_finished = True\n\n        def consumer_worker(callable, *args, **kwargs):\n            \"\"\"Since jobs could return a generator and we cannot really \"inspect\" for that\n            \"\"\"\n            try:\n                res = callable(*args, **kwargs)\n                if inspect.isgenerator(res):\n                    lgr.debug(\"Got consumer worker which returned a generator %s\", res)\n                    didgood = False\n                    for r in res:\n                        didgood = True\n                        lgr.debug(\"Adding %s to queue\", r)\n                        consumer_queue.put(r)\n                    if not didgood:\n                        lgr.error(\"Nothing was obtained from %s :-(\", res)\n                else:\n                    lgr.debug(\"Got straight result %s, not a generator\", res)\n                    consumer_queue.put(res)\n            except BaseException as e:\n                self._exc.append(e)\n\n        producer_thread = Thread(target=producer_worker)\n        producer_thread.start()\n\n        # LOGGING\n        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?\n        lgr = self.lgr\n        label = self.label\n        if lgr is None:\n            from .. import lgr\n        log_progress(lgr.info, pid,\n                     \"%s: starting\", self.label,\n                     # will become known only later total=len(items),\n                     label=self.label, unit=\" \" + self.unit)\n        counts = defaultdict(int)\n\n        futures = {}\n\n        total_announced = False\n        njobs = self.njobs or 1\n        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", njobs)\n        # we will increase sleep_time when doing nothing useful\n        sleeper = Sleeper()\n        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:\n            self._executor = executor\n            # yield from the producer_queue (.total and .finished could be accessed meanwhile)\n            while True:\n                done_useful = False\n                if self.reraise_immediately and self._exc:\n                    break\n                if (self.producer_finished and\n                        not futures and\n                        consumer_queue.empty() and\n                        producer_queue.empty()):\n                    break\n\n                # LOGGING\n                if not total_announced and self.producer_finished:\n                    log_progress(\n                        lgr.info,\n                        pid,\n                        None,  # I do not think there is something valuable to announce\n                        total=self.total,\n                        # unfortunately of no effect, so we cannot inform that more items to come\n                        # unit=(\"+\" if not it.finished else \"\") + \" \" + unit,\n                        update=0  # not None, so it does not stop\n                    )\n                    total_announced = True\n\n                # important!  We are using threads, so worker threads will be sharing CPU time\n                # with this master thread. For it to become efficient, we should consume as much\n                # as possible from producer asap and push it to executor.  So drain the queue\n                while not producer_queue.empty():\n                    done_useful = True\n                    try:\n                        job_args = producer_queue.get() # timeout=0.001)\n                        if self.safe_to_consume:\n                            # Sleep a little if we are not yet ready\n                            # TODO: add some .debug level reporting based on elapsed time\n                            # IIRC I did smth like growing exponentially delays somewhere (dandi?)\n                            while not self.safe_to_consume(futures, job_args):\n                                _prune_futures(futures, lgr) or sleeper()\n                        # Current implementation, to provide depchecking, relies on unique\n                        # args for the job\n                        assert job_args not in futures\n                        lgr.debug(\"Submitting worker future for %s\", job_args)\n                        futures[job_args] = executor.submit(consumer_worker, self.consumer, job_args)\n                    except Empty:\n                        pass\n\n                # check active futures\n                if not consumer_queue.empty():\n                    done_useful = True\n\n                    # ATM we do not bother of some \"in order\" reporting\n                    # Just report as soon as any new record arrives\n                    res = consumer_queue.get()\n                    lgr.debug(\"Got %s from consumer_queue\", res)\n                    # LOGGING\n                    counts[res[\"status\"]] += 1\n                    count_strs = (_count_str(*args)\n                                  for args in [(counts[\"notneeded\"], \"skipped\", False),\n                                               (counts[\"error\"], \"failed\", True)])\n                    if counts[\"notneeded\"] or counts[\"error\"]:\n                        label = \"{} ({})\".format(\n                            self.label,\n                            \", \".join(filter(None, count_strs)))\n\n                    log_progress(\n                        lgr.error if res[\"status\"] == \"error\" else lgr.info,\n                        pid,\n                        \"%s: processed result%s\", self.label,\n                        \" for \" + res[\"path\"] if \"path\" in res else \"\",\n                        label=label, update=1, increment=True)\n\n                    yield res\n\n                done_useful |= _prune_futures(futures, lgr)\n\n                if not done_useful:  # you need some rest\n                    # TODO: same here -- progressive logging\n                    lgr.debug(\n                        \"Did nothing useful, sleeping. Have %s %s %s %s\",\n                        self.producer_finished,\n                        futures,\n                        consumer_queue.empty(),\n                        producer_queue.empty())\n                    sleeper()\n                else:\n                    sleeper.reset()\n\n        self._executor = None\n\n        # LOGGING\n        log_progress(lgr.info, pid, \"%s: done\", self.label)\n        producer_thread.join()\n        if self._exc:\n            if len(self._exc) > 1:\n                lgr.debug(\"%d exceptions were collected while performing execution in parallel. Only the first one \"\n                          \"will be reraised\", len(self._exc))\n            raise self._exc[0]\n        else:\n            assert not futures, \\\n                \"There is still %d active futures for following args: %s\" \\\n                % (len(futures), ', '.join(futures))",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 47,
                            "end_lineno": 267,
                            "parent": null
                        },
                        {
                            "file_name": "datalad/support/parallel.py",
                            "type": "method",
                            "name": "ProducerConsumer.__iter__",
                            "content": "class ProducerConsumer:\n    \"\"\"\n    \"\"\"\n\n    def __init__(self,\n                 producer, consumer,\n                 njobs=None,\n                 safe_to_consume=None,\n                 reraise_immediately=False,\n                 agg=None,\n                 # TODO: \"absorb\" into some more generic \"logging\" helper\n                 # so could be used not only with consumers which yield our records\n                 # 'agg' from above could also more relate to i\n                 label=\"Total\", unit=\"items\",\n                 lgr=None,\n                 ):\n        \"\"\"\n\n        Parameters\n        ----------\n        ...\n        safe_to_consume: callable\n          A callable which gets a dict of all known futures and current producer output.\n          It should return True if we can proceed with current value from producer.\n          If unsafe - we will wait.  WARNING: outside code should make sure about provider and\n          safe_to_consume to play nicely or deadlock can happen.\n        reraise_immediately: bool, optional\n          If True, it would stop producer yielding values as soon as it detects that some\n          exception has occurred (although there might still be values in the queue to be yielded\n          which were collected before the exception was raised)\n        \"\"\"\n        self.producer = producer\n        self.consumer = consumer\n        self.njobs = njobs\n        self.safe_to_consume = safe_to_consume\n        self.reraise_immediately = reraise_immediately\n        self.agg = agg\n        self.label = label\n        self.unit = unit\n        self.lgr = lgr\n\n        self.total = None\n        self.producer_finished = None\n        self._executor = None\n        self._exc = []\n\n    def __del__(self):\n        # if we are killed while executing, we should ask executor to terminate\n        executor = getattr(self, \"_executor\")\n        if executor:\n            executor.terminate()\n\n    def __iter__(self):\n        self.producer_finished = False\n        self._exc = []\n\n        producer_queue = Queue()\n        consumer_queue = Queue()\n\n        def producer_worker():\n            \"\"\"That is the one which interrogates producer and updates .total\"\"\"\n            total = None if self.agg else 0\n            try:\n                for value in self.producer:\n                    producer_queue.put(value)\n                    if self.agg:\n                        total = (\n                            self.agg(value, total) if total is not None else self.agg(value)\n                        )\n                    else:\n                        total += 1\n                    self.total = total\n            except BaseException as e:\n                self._exc.append(e)\n            finally:\n                self.producer_finished = True\n\n        def consumer_worker(callable, *args, **kwargs):\n            \"\"\"Since jobs could return a generator and we cannot really \"inspect\" for that\n            \"\"\"\n            try:\n                res = callable(*args, **kwargs)\n                if inspect.isgenerator(res):\n                    lgr.debug(\"Got consumer worker which returned a generator %s\", res)\n                    didgood = False\n                    for r in res:\n                        didgood = True\n                        lgr.debug(\"Adding %s to queue\", r)\n                        consumer_queue.put(r)\n                    if not didgood:\n                        lgr.error(\"Nothing was obtained from %s :-(\", res)\n                else:\n                    lgr.debug(\"Got straight result %s, not a generator\", res)\n                    consumer_queue.put(res)\n            except BaseException as e:\n                self._exc.append(e)\n\n        producer_thread = Thread(target=producer_worker)\n        producer_thread.start()\n\n        # LOGGING\n        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?\n        lgr = self.lgr\n        label = self.label\n        if lgr is None:\n            from .. import lgr\n        log_progress(lgr.info, pid,\n                     \"%s: starting\", self.label,\n                     # will become known only later total=len(items),\n                     label=self.label, unit=\" \" + self.unit)\n        counts = defaultdict(int)\n\n        futures = {}\n\n        total_announced = False\n        njobs = self.njobs or 1\n        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", njobs)\n        # we will increase sleep_time when doing nothing useful\n        sleeper = Sleeper()\n        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:\n            self._executor = executor\n            # yield from the producer_queue (.total and .finished could be accessed meanwhile)\n            while True:\n                done_useful = False\n                if self.reraise_immediately and self._exc:\n                    break\n                if (self.producer_finished and\n                        not futures and\n                        consumer_queue.empty() and\n                        producer_queue.empty()):\n                    break\n\n                # LOGGING\n                if not total_announced and self.producer_finished:\n                    log_progress(\n                        lgr.info,\n                        pid,\n                        None,  # I do not think there is something valuable to announce\n                        total=self.total,\n                        # unfortunately of no effect, so we cannot inform that more items to come\n                        # unit=(\"+\" if not it.finished else \"\") + \" \" + unit,\n                        update=0  # not None, so it does not stop\n                    )\n                    total_announced = True\n\n                # important!  We are using threads, so worker threads will be sharing CPU time\n                # with this master thread. For it to become efficient, we should consume as much\n                # as possible from producer asap and push it to executor.  So drain the queue\n                while not producer_queue.empty():\n                    done_useful = True\n                    try:\n                        job_args = producer_queue.get() # timeout=0.001)\n                        if self.safe_to_consume:\n                            # Sleep a little if we are not yet ready\n                            # TODO: add some .debug level reporting based on elapsed time\n                            # IIRC I did smth like growing exponentially delays somewhere (dandi?)\n                            while not self.safe_to_consume(futures, job_args):\n                                _prune_futures(futures, lgr) or sleeper()\n                        # Current implementation, to provide depchecking, relies on unique\n                        # args for the job\n                        assert job_args not in futures\n                        lgr.debug(\"Submitting worker future for %s\", job_args)\n                        futures[job_args] = executor.submit(consumer_worker, self.consumer, job_args)\n                    except Empty:\n                        pass\n\n                # check active futures\n                if not consumer_queue.empty():\n                    done_useful = True\n\n                    # ATM we do not bother of some \"in order\" reporting\n                    # Just report as soon as any new record arrives\n                    res = consumer_queue.get()\n                    lgr.debug(\"Got %s from consumer_queue\", res)\n                    # LOGGING\n                    counts[res[\"status\"]] += 1\n                    count_strs = (_count_str(*args)\n                                  for args in [(counts[\"notneeded\"], \"skipped\", False),\n                                               (counts[\"error\"], \"failed\", True)])\n                    if counts[\"notneeded\"] or counts[\"error\"]:\n                        label = \"{} ({})\".format(\n                            self.label,\n                            \", \".join(filter(None, count_strs)))\n\n                    log_progress(\n                        lgr.error if res[\"status\"] == \"error\" else lgr.info,\n                        pid,\n                        \"%s: processed result%s\", self.label,\n                        \" for \" + res[\"path\"] if \"path\" in res else \"\",\n                        label=label, update=1, increment=True)\n\n                    yield res\n\n                done_useful |= _prune_futures(futures, lgr)\n\n                if not done_useful:  # you need some rest\n                    # TODO: same here -- progressive logging\n                    lgr.debug(\n                        \"Did nothing useful, sleeping. Have %s %s %s %s\",\n                        self.producer_finished,\n                        futures,\n                        consumer_queue.empty(),\n                        producer_queue.empty())\n                    sleeper()\n                else:\n                    sleeper.reset()\n\n        self._executor = None\n\n        # LOGGING\n        log_progress(lgr.info, pid, \"%s: done\", self.label)\n        producer_thread.join()\n        if self._exc:\n            if len(self._exc) > 1:\n                lgr.debug(\"%d exceptions were collected while performing execution in parallel. Only the first one \"\n                          \"will be reraised\", len(self._exc))\n            raise self._exc[0]\n        else:\n            assert not futures, \\\n                \"There is still %d active futures for following args: %s\" \\\n                % (len(futures), ', '.join(futures))",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 99,
                            "end_lineno": 267,
                            "parent": {
                                "file_name": "datalad/support/parallel.py",
                                "type": "class",
                                "name": "ProducerConsumer",
                                "content": "class ProducerConsumer:\n    \"\"\"\n    \"\"\"\n\n    def __init__(self,\n                 producer, consumer,\n                 njobs=None,\n                 safe_to_consume=None,\n                 reraise_immediately=False,\n                 agg=None,\n                 # TODO: \"absorb\" into some more generic \"logging\" helper\n                 # so could be used not only with consumers which yield our records\n                 # 'agg' from above could also more relate to i\n                 label=\"Total\", unit=\"items\",\n                 lgr=None,\n                 ):\n        \"\"\"\n\n        Parameters\n        ----------\n        ...\n        safe_to_consume: callable\n          A callable which gets a dict of all known futures and current producer output.\n          It should return True if we can proceed with current value from producer.\n          If unsafe - we will wait.  WARNING: outside code should make sure about provider and\n          safe_to_consume to play nicely or deadlock can happen.\n        reraise_immediately: bool, optional\n          If True, it would stop producer yielding values as soon as it detects that some\n          exception has occurred (although there might still be values in the queue to be yielded\n          which were collected before the exception was raised)\n        \"\"\"\n        self.producer = producer\n        self.consumer = consumer\n        self.njobs = njobs\n        self.safe_to_consume = safe_to_consume\n        self.reraise_immediately = reraise_immediately\n        self.agg = agg\n        self.label = label\n        self.unit = unit\n        self.lgr = lgr\n\n        self.total = None\n        self.producer_finished = None\n        self._executor = None\n        self._exc = []\n\n    def __del__(self):\n        # if we are killed while executing, we should ask executor to terminate\n        executor = getattr(self, \"_executor\")\n        if executor:\n            executor.terminate()\n\n    def __iter__(self):\n        self.producer_finished = False\n        self._exc = []\n\n        producer_queue = Queue()\n        consumer_queue = Queue()\n\n        def producer_worker():\n            \"\"\"That is the one which interrogates producer and updates .total\"\"\"\n            total = None if self.agg else 0\n            try:\n                for value in self.producer:\n                    producer_queue.put(value)\n                    if self.agg:\n                        total = (\n                            self.agg(value, total) if total is not None else self.agg(value)\n                        )\n                    else:\n                        total += 1\n                    self.total = total\n            except BaseException as e:\n                self._exc.append(e)\n            finally:\n                self.producer_finished = True\n\n        def consumer_worker(callable, *args, **kwargs):\n            \"\"\"Since jobs could return a generator and we cannot really \"inspect\" for that\n            \"\"\"\n            try:\n                res = callable(*args, **kwargs)\n                if inspect.isgenerator(res):\n                    lgr.debug(\"Got consumer worker which returned a generator %s\", res)\n                    didgood = False\n                    for r in res:\n                        didgood = True\n                        lgr.debug(\"Adding %s to queue\", r)\n                        consumer_queue.put(r)\n                    if not didgood:\n                        lgr.error(\"Nothing was obtained from %s :-(\", res)\n                else:\n                    lgr.debug(\"Got straight result %s, not a generator\", res)\n                    consumer_queue.put(res)\n            except BaseException as e:\n                self._exc.append(e)\n\n        producer_thread = Thread(target=producer_worker)\n        producer_thread.start()\n\n        # LOGGING\n        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?\n        lgr = self.lgr\n        label = self.label\n        if lgr is None:\n            from .. import lgr\n        log_progress(lgr.info, pid,\n                     \"%s: starting\", self.label,\n                     # will become known only later total=len(items),\n                     label=self.label, unit=\" \" + self.unit)\n        counts = defaultdict(int)\n\n        futures = {}\n\n        total_announced = False\n        njobs = self.njobs or 1\n        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", njobs)\n        # we will increase sleep_time when doing nothing useful\n        sleeper = Sleeper()\n        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:\n            self._executor = executor\n            # yield from the producer_queue (.total and .finished could be accessed meanwhile)\n            while True:\n                done_useful = False\n                if self.reraise_immediately and self._exc:\n                    break\n                if (self.producer_finished and\n                        not futures and\n                        consumer_queue.empty() and\n                        producer_queue.empty()):\n                    break\n\n                # LOGGING\n                if not total_announced and self.producer_finished:\n                    log_progress(\n                        lgr.info,\n                        pid,\n                        None,  # I do not think there is something valuable to announce\n                        total=self.total,\n                        # unfortunately of no effect, so we cannot inform that more items to come\n                        # unit=(\"+\" if not it.finished else \"\") + \" \" + unit,\n                        update=0  # not None, so it does not stop\n                    )\n                    total_announced = True\n\n                # important!  We are using threads, so worker threads will be sharing CPU time\n                # with this master thread. For it to become efficient, we should consume as much\n                # as possible from producer asap and push it to executor.  So drain the queue\n                while not producer_queue.empty():\n                    done_useful = True\n                    try:\n                        job_args = producer_queue.get() # timeout=0.001)\n                        if self.safe_to_consume:\n                            # Sleep a little if we are not yet ready\n                            # TODO: add some .debug level reporting based on elapsed time\n                            # IIRC I did smth like growing exponentially delays somewhere (dandi?)\n                            while not self.safe_to_consume(futures, job_args):\n                                _prune_futures(futures, lgr) or sleeper()\n                        # Current implementation, to provide depchecking, relies on unique\n                        # args for the job\n                        assert job_args not in futures\n                        lgr.debug(\"Submitting worker future for %s\", job_args)\n                        futures[job_args] = executor.submit(consumer_worker, self.consumer, job_args)\n                    except Empty:\n                        pass\n\n                # check active futures\n                if not consumer_queue.empty():\n                    done_useful = True\n\n                    # ATM we do not bother of some \"in order\" reporting\n                    # Just report as soon as any new record arrives\n                    res = consumer_queue.get()\n                    lgr.debug(\"Got %s from consumer_queue\", res)\n                    # LOGGING\n                    counts[res[\"status\"]] += 1\n                    count_strs = (_count_str(*args)\n                                  for args in [(counts[\"notneeded\"], \"skipped\", False),\n                                               (counts[\"error\"], \"failed\", True)])\n                    if counts[\"notneeded\"] or counts[\"error\"]:\n                        label = \"{} ({})\".format(\n                            self.label,\n                            \", \".join(filter(None, count_strs)))\n\n                    log_progress(\n                        lgr.error if res[\"status\"] == \"error\" else lgr.info,\n                        pid,\n                        \"%s: processed result%s\", self.label,\n                        \" for \" + res[\"path\"] if \"path\" in res else \"\",\n                        label=label, update=1, increment=True)\n\n                    yield res\n\n                done_useful |= _prune_futures(futures, lgr)\n\n                if not done_useful:  # you need some rest\n                    # TODO: same here -- progressive logging\n                    lgr.debug(\n                        \"Did nothing useful, sleeping. Have %s %s %s %s\",\n                        self.producer_finished,\n                        futures,\n                        consumer_queue.empty(),\n                        producer_queue.empty())\n                    sleeper()\n                else:\n                    sleeper.reset()\n\n        self._executor = None\n\n        # LOGGING\n        log_progress(lgr.info, pid, \"%s: done\", self.label)\n        producer_thread.join()\n        if self._exc:\n            if len(self._exc) > 1:\n                lgr.debug(\"%d exceptions were collected while performing execution in parallel. Only the first one \"\n                          \"will be reraised\", len(self._exc))\n            raise self._exc[0]\n        else:\n            assert not futures, \\\n                \"There is still %d active futures for following args: %s\" \\\n                % (len(futures), ', '.join(futures))",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 47,
                                "end_lineno": 267,
                                "parent": null
                            }
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        },
        {
            "old_file_content": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\nfrom time import sleep, time\nfrom functools import partial\n\nfrom datalad.support import path as op\n\n# absolute import only to be able to run test without `nose` so to see progress bar\nfrom datalad.support.parallel import (\n    ProducerConsumer,\n    no_parentds_in_futures,\n)\nfrom datalad.tests.utils import (\n    assert_equal,\n    assert_repo_status,\n    assert_raises,\n    rmtree,\n    with_tempfile,\n)\n\nfrom datalad.support.exceptions import IncompleteResultsError\n\ndef test_ProducerConsumer():\n    def slowprod(n, secs=0.1):\n        for i in range(n):\n            yield i\n            sleep(secs)\n\n    def slowcons(i):\n        # so takes longer to consume than to produce and progress bar will appear\n        # after slowprod is done producing\n        #print(f\"Consuming {i}\")\n        #t0 = time()\n        sleep(0.2)\n        #print(f\"Consumed {i} in {time() - t0}\")\n        yield {\n            \"i\": i, \"status\": \"ok\" if i % 2 else \"error\"\n        }\n    assert_equal(list(ProducerConsumer(\n        slowprod(10),\n        slowcons,\n        njobs=10,\n    )), [{\"i\": i, \"status\": \"ok\" if i % 2 else \"error\"} for i in range(10)])\n\n\n@with_tempfile(mkdir=True)\ndef test_creatsubdatasets(topds_path, n=3):\n    from datalad.distribution.dataset import Dataset\n    from datalad.api import create\n    ds = Dataset(topds_path).create()\n    paths = [op.join(topds_path, \"subds%d\" % i) for i in range(n)]\n    paths.extend(op.join(topds_path, \"subds%d\" % i, \"subsub%d\" %k) for i in range(n) for k in range(2))\n    # To allow for parallel execution without hitting the problem of\n    # a lock in the super dataset, we create all subdatasets, and then\n    # save them all within their superdataset\n    create_ = partial(create,  # cfg_proc=\"yoda\",\n                      result_xfm=None, return_type='generator')\n    # if we flip the paths so to go from the end, create without --force should fail\n    # and we should get the exception (the first one encountered!)\n    # Note: reraise_immediately is of \"concern\" only for producer. since we typically\n    # rely on outside code to do the killing!\n    assert_raises(IncompleteResultsError, list, ProducerConsumer(paths[::-1], create_, njobs=5))\n    # we are in a dirty state, let's just remove all those for a clean run\n    rmtree(topds_path)\n\n    # and this one followed by save should be good IFF we provide our dependency checker\n    ds = Dataset(topds_path).create()\n    list(ProducerConsumer(paths, create_, safe_to_consume=no_parentds_in_futures, njobs=10))\n    ds.save(paths)\n    assert_repo_status(ds.repo)\n\n\nif __name__ == '__main__':\n    # test_ProducerConsumer()\n    test_creatsubdatasets()",
            "new_file_content": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\nfrom time import sleep, time\nfrom functools import partial\n\nfrom datalad.support import path as op\n\n# absolute import only to be able to run test without `nose` so to see progress bar\nfrom datalad.support.parallel import (\n    ProducerConsumer,\n    no_parentds_in_futures,\n)\nfrom datalad.tests.utils import (\n    assert_equal,\n    assert_repo_status,\n    assert_raises,\n    rmtree,\n    with_tempfile,\n)\n\nfrom datalad.support.exceptions import IncompleteResultsError\n\ndef test_ProducerConsumer():\n    def slowprod(n, secs=0.1):\n        for i in range(n):\n            yield i\n            sleep(secs)\n\n    def slowcons(i):\n        # so takes longer to consume than to produce and progress bar will appear\n        # after slowprod is done producing\n        #print(f\"Consuming {i}\")\n        #t0 = time()\n        sleep(0.2)\n        #print(f\"Consumed {i} in {time() - t0}\")\n        yield {\n            \"i\": i, \"status\": \"ok\" if i % 2 else \"error\"\n        }\n    assert_equal(list(ProducerConsumer(\n        slowprod(10),\n        slowcons,\n        jobs=10,\n    )), [{\"i\": i, \"status\": \"ok\" if i % 2 else \"error\"} for i in range(10)])\n\n\n@with_tempfile(mkdir=True)\ndef test_creatsubdatasets(topds_path, n=10):\n    from datalad.distribution.dataset import Dataset\n    from datalad.api import create\n    ds = Dataset(topds_path).create()\n    paths = [op.join(topds_path, \"subds%d\" % i) for i in range(n)]\n    paths.extend(op.join(topds_path, \"subds%d\" % i, \"subsub%d\" %k) for i in range(n) for k in range(2))\n    # To allow for parallel execution without hitting the problem of\n    # a lock in the super dataset, we create all subdatasets, and then\n    # save them all within their superdataset\n    create_ = partial(create,  # cfg_proc=\"yoda\",\n                      result_xfm=None, return_type='generator')\n    # if we flip the paths so to go from the end, create without --force should fail\n    # and we should get the exception (the first one encountered!)\n    # Note: reraise_immediately is of \"concern\" only for producer. since we typically\n    # rely on outside code to do the killing!\n    assert_raises(IncompleteResultsError, list, ProducerConsumer(paths[::-1], create_, jobs=5))\n    # we are in a dirty state, let's just remove all those for a clean run\n    rmtree(topds_path)\n\n    # and this one followed by save should be good IFF we provide our dependency checker\n    ds = Dataset(topds_path).create()\n    list(ProducerConsumer(paths, create_, safe_to_consume=no_parentds_in_futures, jobs=10))\n    ds.save(paths)\n    assert_repo_status(ds.repo)\n\n\nif __name__ == '__main__':\n    # test_ProducerConsumer()\n    test_creatsubdatasets()",
            "header": {
                "file": {
                    "path": "datalad/support/tests/test_parallel.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "c94efb70e",
                "new_commit_hash": "77f844362",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/datalad/support/tests/test_parallel.py"
            },
            "plus_file": {
                "path": "b/datalad/support/tests/test_parallel.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 46,
                            "length": 12
                        },
                        "new_range": {
                            "start": 46,
                            "length": 12
                        },
                        "section": "def test_ProducerConsumer():"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "    assert_equal(list(ProducerConsumer(",
                                "type": "context"
                            },
                            {
                                "content": "        slowprod(10),",
                                "type": "context"
                            },
                            {
                                "content": "        slowcons,",
                                "type": "context"
                            },
                            {
                                "content": "        njobs=10,",
                                "type": "deleted"
                            },
                            {
                                "content": "        jobs=10,",
                                "type": "added"
                            },
                            {
                                "content": "    )), [{\"i\": i, \"status\": \"ok\" if i % 2 else \"error\"} for i in range(10)])",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "@with_tempfile(mkdir=True)",
                                "type": "context"
                            },
                            {
                                "content": "def test_creatsubdatasets(topds_path, n=3):",
                                "type": "deleted"
                            },
                            {
                                "content": "def test_creatsubdatasets(topds_path, n=10):",
                                "type": "added"
                            },
                            {
                                "content": "    from datalad.distribution.dataset import Dataset",
                                "type": "context"
                            },
                            {
                                "content": "    from datalad.api import create",
                                "type": "context"
                            },
                            {
                                "content": "    ds = Dataset(topds_path).create()",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "datalad/support/tests/test_parallel.py",
                            "type": "function",
                            "name": "test_creatsubdatasets",
                            "content": "def test_creatsubdatasets(topds_path, n=3):\n    from datalad.distribution.dataset import Dataset\n    from datalad.api import create\n    ds = Dataset(topds_path).create()\n    paths = [op.join(topds_path, \"subds%d\" % i) for i in range(n)]\n    paths.extend(op.join(topds_path, \"subds%d\" % i, \"subsub%d\" %k) for i in range(n) for k in range(2))\n    # To allow for parallel execution without hitting the problem of\n    # a lock in the super dataset, we create all subdatasets, and then\n    # save them all within their superdataset\n    create_ = partial(create,  # cfg_proc=\"yoda\",\n                      result_xfm=None, return_type='generator')\n    # if we flip the paths so to go from the end, create without --force should fail\n    # and we should get the exception (the first one encountered!)\n    # Note: reraise_immediately is of \"concern\" only for producer. since we typically\n    # rely on outside code to do the killing!\n    assert_raises(IncompleteResultsError, list, ProducerConsumer(paths[::-1], create_, njobs=5))\n    # we are in a dirty state, let's just remove all those for a clean run\n    rmtree(topds_path)\n\n    # and this one followed by save should be good IFF we provide our dependency checker\n    ds = Dataset(topds_path).create()\n    list(ProducerConsumer(paths, create_, safe_to_consume=no_parentds_in_futures, njobs=10))\n    ds.save(paths)\n    assert_repo_status(ds.repo)",
                            "ast_type_str": "FunctionDef",
                            "start_lineno": 54,
                            "end_lineno": 77,
                            "parent": null
                        },
                        {
                            "file_name": "datalad/support/tests/test_parallel.py",
                            "type": "function",
                            "name": "test_ProducerConsumer",
                            "content": "def test_ProducerConsumer():\n    def slowprod(n, secs=0.1):\n        for i in range(n):\n            yield i\n            sleep(secs)\n\n    def slowcons(i):\n        # so takes longer to consume than to produce and progress bar will appear\n        # after slowprod is done producing\n        #print(f\"Consuming {i}\")\n        #t0 = time()\n        sleep(0.2)\n        #print(f\"Consumed {i} in {time() - t0}\")\n        yield {\n            \"i\": i, \"status\": \"ok\" if i % 2 else \"error\"\n        }\n    assert_equal(list(ProducerConsumer(\n        slowprod(10),\n        slowcons,\n        njobs=10,\n    )), [{\"i\": i, \"status\": \"ok\" if i % 2 else \"error\"} for i in range(10)])",
                            "ast_type_str": "FunctionDef",
                            "start_lineno": 30,
                            "end_lineno": 50,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 66,
                            "length": 13
                        },
                        "new_range": {
                            "start": 66,
                            "length": 13
                        },
                        "section": "def test_creatsubdatasets(topds_path, n=3):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "    # and we should get the exception (the first one encountered!)",
                                "type": "context"
                            },
                            {
                                "content": "    # Note: reraise_immediately is of \"concern\" only for producer. since we typically",
                                "type": "context"
                            },
                            {
                                "content": "    # rely on outside code to do the killing!",
                                "type": "context"
                            },
                            {
                                "content": "    assert_raises(IncompleteResultsError, list, ProducerConsumer(paths[::-1], create_, njobs=5))",
                                "type": "deleted"
                            },
                            {
                                "content": "    assert_raises(IncompleteResultsError, list, ProducerConsumer(paths[::-1], create_, jobs=5))",
                                "type": "added"
                            },
                            {
                                "content": "    # we are in a dirty state, let's just remove all those for a clean run",
                                "type": "context"
                            },
                            {
                                "content": "    rmtree(topds_path)",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    # and this one followed by save should be good IFF we provide our dependency checker",
                                "type": "context"
                            },
                            {
                                "content": "    ds = Dataset(topds_path).create()",
                                "type": "context"
                            },
                            {
                                "content": "    list(ProducerConsumer(paths, create_, safe_to_consume=no_parentds_in_futures, njobs=10))",
                                "type": "deleted"
                            },
                            {
                                "content": "    list(ProducerConsumer(paths, create_, safe_to_consume=no_parentds_in_futures, jobs=10))",
                                "type": "added"
                            },
                            {
                                "content": "    ds.save(paths)",
                                "type": "context"
                            },
                            {
                                "content": "    assert_repo_status(ds.repo)",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "datalad/support/tests/test_parallel.py",
                            "type": "function",
                            "name": "test_creatsubdatasets",
                            "content": "def test_creatsubdatasets(topds_path, n=3):\n    from datalad.distribution.dataset import Dataset\n    from datalad.api import create\n    ds = Dataset(topds_path).create()\n    paths = [op.join(topds_path, \"subds%d\" % i) for i in range(n)]\n    paths.extend(op.join(topds_path, \"subds%d\" % i, \"subsub%d\" %k) for i in range(n) for k in range(2))\n    # To allow for parallel execution without hitting the problem of\n    # a lock in the super dataset, we create all subdatasets, and then\n    # save them all within their superdataset\n    create_ = partial(create,  # cfg_proc=\"yoda\",\n                      result_xfm=None, return_type='generator')\n    # if we flip the paths so to go from the end, create without --force should fail\n    # and we should get the exception (the first one encountered!)\n    # Note: reraise_immediately is of \"concern\" only for producer. since we typically\n    # rely on outside code to do the killing!\n    assert_raises(IncompleteResultsError, list, ProducerConsumer(paths[::-1], create_, njobs=5))\n    # we are in a dirty state, let's just remove all those for a clean run\n    rmtree(topds_path)\n\n    # and this one followed by save should be good IFF we provide our dependency checker\n    ds = Dataset(topds_path).create()\n    list(ProducerConsumer(paths, create_, safe_to_consume=no_parentds_in_futures, njobs=10))\n    ds.save(paths)\n    assert_repo_status(ds.repo)",
                            "ast_type_str": "FunctionDef",
                            "start_lineno": 54,
                            "end_lineno": 77,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        }
    ],
    "old_commit_hash": "6096b64e10afc251895b2eabdf126cef2a94c33b^",
    "new_commit_hash": "6096b64e10afc251895b2eabdf126cef2a94c33b",
    "commit_message": "RF: njobs -> jobs for consistency with the rest + hardcode \"auto\" = 5 for ProducerConsumer\n\n5 is arbitrary and I would have preferred 10 ;)  But it is also not\nsensible to base it on cores since we are just\nmulti-threading. Ideally some dynamic scaling is needed.",
    "commit_date": "2020-10-19T16:00:29-04:00",
    "metadata": {}
}
{
    "repo_name": "datalad",
    "new_commit_hash": "6096b64e10afc251895b2eabdf126cef2a94c33b",
    "test_file_codes": [
        "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\nfrom time import sleep, time\nfrom functools import partial\n\nfrom datalad.support import path as op\n\n# absolute import only to be able to run test without `nose` so to see progress bar\nfrom datalad.support.parallel import (\n    ProducerConsumer,\n    no_parentds_in_futures,\n)\nfrom datalad.tests.utils import (\n    assert_equal,\n    assert_repo_status,\n    assert_raises,\n    rmtree,\n    with_tempfile,\n)\n\nfrom datalad.support.exceptions import IncompleteResultsError\n\ndef test_ProducerConsumer():\n    def slowprod(n, secs=0.1):\n        for i in range(n):\n            yield i\n            sleep(secs)\n\n    def slowcons(i):\n        # so takes longer to consume than to produce and progress bar will appear\n        # after slowprod is done producing\n        #print(f\"Consuming {i}\")\n        #t0 = time()\n        sleep(0.2)\n        #print(f\"Consumed {i} in {time() - t0}\")\n        yield {\n            \"i\": i, \"status\": \"ok\" if i % 2 else \"error\"\n        }\n    assert_equal(list(ProducerConsumer(\n        slowprod(10),\n        slowcons,\n        jobs=10,\n    )), [{\"i\": i, \"status\": \"ok\" if i % 2 else \"error\"} for i in range(10)])\n\n\n@with_tempfile(mkdir=True)\ndef test_creatsubdatasets(topds_path, n=10):\n    from datalad.distribution.dataset import Dataset\n    from datalad.api import create\n    ds = Dataset(topds_path).create()\n    paths = [op.join(topds_path, \"subds%d\" % i) for i in range(n)]\n    paths.extend(op.join(topds_path, \"subds%d\" % i, \"subsub%d\" %k) for i in range(n) for k in range(2))\n    # To allow for parallel execution without hitting the problem of\n    # a lock in the super dataset, we create all subdatasets, and then\n    # save them all within their superdataset\n    create_ = partial(create,  # cfg_proc=\"yoda\",\n                      result_xfm=None, return_type='generator')\n    # if we flip the paths so to go from the end, create without --force should fail\n    # and we should get the exception (the first one encountered!)\n    # Note: reraise_immediately is of \"concern\" only for producer. since we typically\n    # rely on outside code to do the killing!\n    assert_raises(IncompleteResultsError, list, ProducerConsumer(paths[::-1], create_, jobs=5))\n    # we are in a dirty state, let's just remove all those for a clean run\n    rmtree(topds_path)\n\n    # and this one followed by save should be good IFF we provide our dependency checker\n    ds = Dataset(topds_path).create()\n    list(ProducerConsumer(paths, create_, safe_to_consume=no_parentds_in_futures, jobs=10))\n    ds.save(paths)\n    assert_repo_status(ds.repo)\n\n\nif __name__ == '__main__':\n    # test_ProducerConsumer()\n    test_creatsubdatasets()",
        "import pytest\nimport tempfile\nimport os\n\n\n@pytest.fixture\ndef path():\n    with tempfile.TemporaryDirectory() as tmpdir:\n        yield tmpdir\n\n\n@pytest.fixture(autouse=True)\ndef setup_git_config():\n    with tempfile.TemporaryDirectory() as home:\n        old_home = os.environ.get(\"HOME\")\n        os.environ[\"HOME\"] = home\n        os.system('git config --global user.name \"DataLad Tester\"')\n        os.system('git config --global user.email \"test@example.com\"')\n        yield\n        if old_home:\n            os.environ[\"HOME\"] = old_home\n"
    ],
    "test_file_names": [
        "test_1.py",
        "conftest.py"
    ],
    "setup_res_code": 0,
    "setup_res_stdout": "Starting Datalad installation attempts...\nVerifying installation...\nDatalad version: 0.13.4.dev339\n✅ Installation successful!\nSuccessfully installed Datalad using Python 3.9\n",
    "setup_res_stderr": "Using CPython 3.9.21\nCreating virtual environment at: .venv\nActivate with: source .venv/bin/activate\nResolved 25 packages in 102ms\nInstalled 25 packages in 5.86s\n + coverage==7.6.10\n + docopt==0.6.2\n + exceptiongroup==1.2.2\n + grabbit==0.2.6\n + importlib-resources==6.4.5\n + iniconfig==2.0.0\n + nibabel==5.3.2\n + num2words==0.5.14\n + numpy==2.0.2\n + packaging==24.2\n + pandas==2.2.3\n + patsy==1.0.1\n + pluggy==1.5.0\n + pybids==0.6.5\n + pytest==8.3.4\n + pytest-cov==6.0.0\n + python-dateutil==2.9.0.post0\n + pytz==2024.2\n + scipy==1.13.1\n + setuptools==75.6.0\n + six==1.17.0\n + tomli==2.2.1\n + typing-extensions==4.12.2\n + tzdata==2024.2\n + zipp==3.21.0\nResolved 63 packages in 3.87s\nPrepared 1 package in 4.81s\nInstalled 57 packages in 333ms\n + annexremote==1.6.6\n + appdirs==1.4.4\n + backports-tarfile==1.2.0\n + beautifulsoup4==4.12.3\n + boto==2.49.0\n + certifi==2024.12.14\n + cffi==1.17.1\n + chardet==5.2.0\n + charset-normalizer==3.4.1\n + citeproc-py==0.6.0\n + cryptography==44.0.0\n + datalad==0.13.4 (from file:///home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b)\n + deprecated==1.2.15\n + distro==1.9.0\n + duecredit==0.10.2\n + exifread==3.0.0\n + fasteners==0.19\n + httpretty==1.1.4\n + humanize==4.11.0\n + idna==3.10\n + importlib-metadata==8.5.0\n + iso8601==2.1.0\n + jaraco-classes==3.4.0\n + jaraco-context==6.0.1\n + jaraco-functools==4.1.0\n + jeepney==0.8.0\n + jsmin==3.0.1\n + keyring==25.6.0\n + keyrings-alt==5.0.2\n + looseversion==1.3.0\n + lxml==5.3.0\n + more-itertools==10.5.0\n + msgpack==1.1.0\n + multidict==6.1.0\n + mutagen==1.47.0\n + nose==1.3.7\n + patool==1.15.0\n + pillow==11.0.0\n + propcache==0.2.1\n + pycparser==2.22\n + pygithub==2.5.0\n + pyjwt==2.10.1\n + pynacl==1.5.0\n + pyperclip==1.9.0\n + python-xmp-toolkit==2.0.2\n + pyyaml==6.0.2\n + requests==2.32.3\n + requests-ftp==0.3.1\n + secretstorage==3.3.3\n + simplejson==3.19.3\n + soupsieve==2.6\n + tqdm==4.67.1\n + urllib3==1.26.20\n + vcrpy==7.0.0\n + whoosh==2.7.4\n + wrapt==1.17.0\n + yarl==1.18.3\nResolved 97 packages in 196ms\nInstalled 32 packages in 345ms\n + alabaster==0.7.16\n + asv==0.6.4\n + asv-runner==0.2.1\n + babel==2.16.0\n + build==1.2.2.post1\n + distlib==0.3.9\n + docutils==0.21.2\n + filelock==3.16.1\n + gprof2dot==2024.6.6\n + imagesize==1.4.1\n + jinja2==3.1.5\n + json5==0.10.0\n + markupsafe==3.0.2\n + nose-timer==1.0.1\n + platformdirs==4.3.6\n + psutil==6.1.1\n + pygments==2.18.0\n + pympler==1.1\n + pypandoc==1.14\n + pyproject-hooks==1.2.0\n + snowballstemmer==2.2.0\n + sphinx==7.4.7\n + sphinx-rtd-theme==3.0.2\n + sphinxcontrib-applehelp==2.0.0\n + sphinxcontrib-devhelp==2.0.0\n + sphinxcontrib-htmlhelp==2.1.0\n + sphinxcontrib-jquery==4.1\n + sphinxcontrib-jsmath==1.0.1\n + sphinxcontrib-qthelp==2.0.0\n + sphinxcontrib-serializinghtml==2.0.0\n + tabulate==0.9.0\n + virtualenv==20.28.0\n",
    "new_commit_res_code": 1,
    "new_commit_res_stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.9.21, pytest-8.3.4, pluggy-1.5.0\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b\nconfigfile: pyproject.toml\nplugins: cov-6.0.0\ncollected 2 items\n\nr2e_tests/test_1.py .E                                                   [100%]\n\n==================================== ERRORS ====================================\n___________________ ERROR at setup of test_creatsubdatasets ____________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/r2e_tests/test_1.py, line 53\n  @with_tempfile(mkdir=True)\n  def test_creatsubdatasets(topds_path, n=10):\nE       fixture 'topds_path' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/r2e_tests/test_1.py:53\n==================================== PASSES ====================================\n____________________________ test_ProducerConsumer _____________________________\n----------------------------- Captured stderr call -----------------------------\n[INFO] Total: starting \n[ERROR] Total: processed result \n[INFO] Total: processed result \n[ERROR] Total: processed result \n[INFO] Total: processed result \n[ERROR] Total: processed result \n[INFO] Total: processed result \n[ERROR] Total: processed result \n[INFO] Total: processed result \n--- Logging error ---\nTraceback (most recent call last):\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/logging/__init__.py\", line 1083, in emit\n    msg = self.format(record)\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/logging/__init__.py\", line 927, in format\n    return fmt.format(record)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/datalad/log.py\", line 173, in format\n    if record.msg.startswith('| '):\nAttributeError: 'NoneType' object has no attribute 'startswith'\nCall stack:\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/runpy.py\", line 197, in _run_module_as_main\n    return _run_code(code, main_globals, None,\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/runpy.py\", line 87, in _run_code\n    exec(code, run_globals)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pytest/__main__.py\", line 9, in <module>\n    raise SystemExit(pytest.console_main())\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/config/__init__.py\", line 201, in console_main\n    code = main()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/config/__init__.py\", line 175, in main\n    ret: ExitCode | int = config.hook.pytest_cmdline_main(config=config)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_hooks.py\", line 513, in __call__\n    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_manager.py\", line 120, in _hookexec\n    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_callers.py\", line 103, in _multicall\n    res = hook_impl.function(*args)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/main.py\", line 330, in pytest_cmdline_main\n    return wrap_session(config, _main)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/main.py\", line 283, in wrap_session\n    session.exitstatus = doit(config, session) or 0\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/main.py\", line 337, in _main\n    config.hook.pytest_runtestloop(session=session)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_hooks.py\", line 513, in __call__\n    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_manager.py\", line 120, in _hookexec\n    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_callers.py\", line 103, in _multicall\n    res = hook_impl.function(*args)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/main.py\", line 362, in pytest_runtestloop\n    item.config.hook.pytest_runtest_protocol(item=item, nextitem=nextitem)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_hooks.py\", line 513, in __call__\n    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_manager.py\", line 120, in _hookexec\n    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_callers.py\", line 103, in _multicall\n    res = hook_impl.function(*args)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/runner.py\", line 113, in pytest_runtest_protocol\n    runtestprotocol(item, nextitem=nextitem)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/runner.py\", line 132, in runtestprotocol\n    reports.append(call_and_report(item, \"call\", log))\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/runner.py\", line 241, in call_and_report\n    call = CallInfo.from_call(\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/runner.py\", line 341, in from_call\n    result: TResult | None = func()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/runner.py\", line 242, in <lambda>\n    lambda: runtest_hook(item=item, **kwds), when=when, reraise=reraise\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_hooks.py\", line 513, in __call__\n    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_manager.py\", line 120, in _hookexec\n    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_callers.py\", line 103, in _multicall\n    res = hook_impl.function(*args)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/runner.py\", line 174, in pytest_runtest_call\n    item.runtest()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/python.py\", line 1627, in runtest\n    self.ihook.pytest_pyfunc_call(pyfuncitem=self)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_hooks.py\", line 513, in __call__\n    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_manager.py\", line 120, in _hookexec\n    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/pluggy/_callers.py\", line 103, in _multicall\n    res = hook_impl.function(*args)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/.venv/lib/python3.9/site-packages/_pytest/python.py\", line 159, in pytest_pyfunc_call\n    result = testfunction(**testargs)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/r2e_tests/test_1.py\", line 46, in test_ProducerConsumer\n    assert_equal(list(ProducerConsumer(\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/datalad/support/parallel.py\", line 186, in __iter__\n    log_progress(\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/datalad/log.py\", line 318, in log_progress\n    lgrcall(*args, extra=d)\nMessage: None\nArguments: ()\n[ERROR] Total: processed result \n[INFO] Total: processed result \n[INFO] Total: done \n------------------------------ Captured log call -------------------------------\nINFO     datalad:log.py:318 Total: starting\nERROR    datalad:log.py:318 Total: processed result\nINFO     datalad:log.py:318 Total: processed result\nERROR    datalad:log.py:318 Total: processed result\nINFO     datalad:log.py:318 Total: processed result\nERROR    datalad:log.py:318 Total: processed result\nINFO     datalad:log.py:318 Total: processed result\nERROR    datalad:log.py:318 Total: processed result\nINFO     datalad:log.py:318 Total: processed result\nINFO     datalad:log.py:318 None\nERROR    datalad:log.py:318 Total: processed result\nINFO     datalad:log.py:318 Total: processed result\nINFO     datalad:log.py:318 Total: done\n=========================== short test summary info ============================\nPASSED r2e_tests/test_1.py::test_ProducerConsumer\nERROR r2e_tests/test_1.py::test_creatsubdatasets\n========================== 1 passed, 1 error in 1.97s ==========================\n",
    "new_commit_res_stderr": "",
    "old_commit_res_code": 1,
    "old_commit_res_stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.9.21, pytest-8.3.4, pluggy-1.5.0\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b\nconfigfile: pyproject.toml\nplugins: cov-6.0.0\ncollected 2 items\n\nr2e_tests/test_1.py FE                                                   [100%]\n\n==================================== ERRORS ====================================\n___________________ ERROR at setup of test_creatsubdatasets ____________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/r2e_tests/test_1.py, line 53\n  @with_tempfile(mkdir=True)\n  def test_creatsubdatasets(topds_path, n=10):\nE       fixture 'topds_path' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_6096b64e10afc251895b2eabdf126cef2a94c33b/r2e_tests/test_1.py:53\n=================================== FAILURES ===================================\n____________________________ test_ProducerConsumer _____________________________\n\n    def test_ProducerConsumer():\n        def slowprod(n, secs=0.1):\n            for i in range(n):\n                yield i\n                sleep(secs)\n    \n        def slowcons(i):\n            # so takes longer to consume than to produce and progress bar will appear\n            # after slowprod is done producing\n            #print(f\"Consuming {i}\")\n            #t0 = time()\n            sleep(0.2)\n            #print(f\"Consumed {i} in {time() - t0}\")\n            yield {\n                \"i\": i, \"status\": \"ok\" if i % 2 else \"error\"\n            }\n>       assert_equal(list(ProducerConsumer(\n            slowprod(10),\n            slowcons,\n            jobs=10,\n        )), [{\"i\": i, \"status\": \"ok\" if i % 2 else \"error\"} for i in range(10)])\nE       TypeError: __init__() got an unexpected keyword argument 'jobs'\n\nr2e_tests/test_1.py:46: TypeError\n=========================== short test summary info ============================\nERROR r2e_tests/test_1.py::test_creatsubdatasets\nFAILED r2e_tests/test_1.py::test_ProducerConsumer - TypeError: __init__() got...\n========================== 1 failed, 1 error in 0.55s ==========================\n",
    "old_commit_res_stderr": ""
}
datalad/support/parallel.py datalad/support/tests/test_parallel.py
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ISSUE]

**Title:** ProducerConsumer class does not accept `jobs` parameter, causing TypeError

**Description:**
When initializing the `ProducerConsumer` class with the `jobs` keyword argument, a `TypeError` is raised indicating that the `__init__` method does not recognize the `jobs` parameter.

**Example Code:**
```python
consumer = ProducerConsumer(producer, consumer_func, jobs=10)
```

**Expected Behavior:**
The `ProducerConsumer` class should accept the `jobs` parameter to specify the number of parallel jobs, allowing users to configure the level of concurrency.

**Actual Behavior:**
A `TypeError` is raised:
```
TypeError: __init__() got an unexpected keyword argument 'jobs'
```
This error occurs because the `ProducerConsumer` class is expecting a different parameter name (`njobs`), leading to inconsistencies and failures when attempting to use the `jobs` argument.

[/ISSUE]
--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.

--- BEGIN FILE ---
```
### datalad/support/parallel.py
# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-
# ex: set sts=4 ts=4 sw=4 noet:
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
#
#   See COPYING file distributed along with the datalad package for the
#   copyright and license terms.
#
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
"""Helpers for parallel execution

"""

__docformat__ = 'restructuredtext'

import concurrent.futures
import inspect
import time
import uuid

from collections import defaultdict
from queue import Queue, Empty
from threading import Thread

from . import ansi_colors as colors
from ..log import log_progress
from ..utils import path_is_subpath


def _count_str(count, verb, omg=False):
    if count:
        msg = "{:d} {}".format(count, verb)
        if omg:
            msg = colors.color_word(msg, colors.RED)
        return msg


def no_parentds_in_futures(futures, path):
    """Return True if no path in futures keys is parentds for provided path

    Assumes that the future's key is the path.
    """
    # TODO: OPT.  Could benefit from smarter than linear time if not one at a time?
    #   or may be we should only go through active futures (still linear!)?
    return not any(path_is_subpath(path, p) for p in futures)


class ProducerConsumer:
    """
    """

    def __init__(self,
                 producer, consumer,
                 njobs=None,
                 safe_to_consume=None,
                 reraise_immediately=False,
                 agg=None,
                 # TODO: "absorb" into some more generic "logging" helper
                 # so could be used not only with consumers which yield our records
                 # 'agg' from above could also more relate to i
                 label="Total", unit="items",
                 lgr=None,
                 ):
        """

        Parameters
        ----------
        ...
        safe_to_consume: callable
          A callable which gets a dict of all known futures and current producer output.
          It should return True if we can proceed with current value from producer.
          If unsafe - we will wait.  WARNING: outside code should make sure about provider and
          safe_to_consume to play nicely or deadlock can happen.
        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)
        """
        self.producer = producer
        self.consumer = consumer
        self.njobs = njobs
        self.safe_to_consume = safe_to_consume
        self.reraise_immediately = reraise_immediately
        self.agg = agg
        self.label = label
        self.unit = unit
        self.lgr = lgr

        self.total = None
        self.producer_finished = None
        self._executor = None
        self._exc = []

    def __del__(self):
        # if we are killed while executing, we should ask executor to terminate
        executor = getattr(self, "_executor")
        if executor:
            executor.terminate()

    def __iter__(self):
        self.producer_finished = False
        self._exc = []

        producer_queue = Queue()
        consumer_queue = Queue()

        def producer_worker():
            """That is the one which interrogates producer and updates .total"""
            total = None if self.agg else 0
            try:
                for value in self.producer:
                    producer_queue.put(value)
                    if self.agg:
                        total = (
                            self.agg(value, total) if total is not None else self.agg(value)
                        )
                    else:
                        total += 1
                    self.total = total
            except BaseException as e:
                self._exc.append(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
            """
            try:
                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)
            except BaseException as e:
                self._exc.append(e)

        producer_thread = Thread(target=producer_worker)
        producer_thread.start()

        # LOGGING
        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?
        lgr = self.lgr
        label = self.label
        if lgr is None:
            from .. import lgr
        log_progress(lgr.info, pid,
                     "%s: starting", self.label,
                     # will become known only later total=len(items),
                     label=self.label, unit=" " + self.unit)
        counts = defaultdict(int)

        futures = {}

        total_announced = False
        njobs = self.njobs or 1
        lgr.debug("Initiating ThreadPoolExecutor with %d jobs", njobs)
        # we will increase sleep_time when doing nothing useful
        sleeper = Sleeper()
        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:
            self._executor = executor
            # yield from the producer_queue (.total and .finished could be accessed meanwhile)
            while True:
                done_useful = False
                if self.reraise_immediately and self._exc:
                    break
                if (self.producer_finished and
                        not futures and
                        consumer_queue.empty() and
                        producer_queue.empty()):
                    break

                # LOGGING
                if not total_announced and self.producer_finished:
                    log_progress(
                        lgr.info,
                        pid,
                        None,  # I do not think there is something valuable to announce
                        total=self.total,
                        # unfortunately of no effect, so we cannot inform that more items to come
                        # unit=("+" if not it.finished else "") + " " + unit,
                        update=0  # not None, so it does not stop
                    )
                    total_announced = True

                # 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():
                    done_useful = True
                    try:
                        job_args = producer_queue.get() # timeout=0.001)
                        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_args):
                                _prune_futures(futures, lgr) or sleeper()
                        # Current implementation, to provide depchecking, relies on unique
                        # args for the job
                        assert job_args not in futures
                        lgr.debug("Submitting worker future for %s", job_args)
                        futures[job_args] = 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)
                    # LOGGING
                    counts[res["status"]] += 1
                    count_strs = (_count_str(*args)
                                  for args in [(counts["notneeded"], "skipped", False),
                                               (counts["error"], "failed", True)])
                    if counts["notneeded"] or counts["error"]:
                        label = "{} ({})".format(
                            self.label,
                            ", ".join(filter(None, count_strs)))

                    log_progress(
                        lgr.error if res["status"] == "error" else lgr.info,
                        pid,
                        "%s: processed result%s", self.label,
                        " for " + res["path"] if "path" in res else "",
                        label=label, update=1, increment=True)

                    yield res

                done_useful |= _prune_futures(futures, lgr)

                if not done_useful:  # you need some rest
                    # TODO: same here -- progressive logging
                    lgr.debug(
                        "Did nothing useful, sleeping. Have %s %s %s %s",
                        self.producer_finished,
                        futures,
                        consumer_queue.empty(),
                        producer_queue.empty())
                    sleeper()
                else:
                    sleeper.reset()

        self._executor = None

        # LOGGING
        log_progress(lgr.info, pid, "%s: done", self.label)
        producer_thread.join()
        if self._exc:
            if len(self._exc) > 1:
                lgr.debug("%d exceptions were collected while performing execution in parallel. Only the first one "
                          "will be reraised", len(self._exc))
            raise self._exc[0]
        else:
            assert not futures, \
                "There is still %d active futures for following args: %s" \
                % (len(futures), ', '.join(futures))


def _prune_futures(futures, lgr):
    """Removes .done from provided futures.

    Returns
    -------
    bool
      True if any future was removed
    """
    done_useful = False
    # remove futures which are done
    for args, future in list(futures.items()):
        if future.done():
            done_useful = True
            future_ = futures.pop(args)
            msg = ""
            if future_.exception():
                msg = " with an exception %s" % future_.exception()
            lgr.debug("Future for %r is done%s", args, msg)
    return done_useful


class Sleeper():
    def __init__(self):
        self.min_sleep_time = 0.001
        # but no more than to this max
        self.max_sleep_time = 0.1
        self.sleep_time = self.min_sleep_time

    def __call__(self):
        time.sleep(self.sleep_time)
        self.sleep_time = min(self.max_sleep_time, self.sleep_time * 2)

    def reset(self):
        self.sleep_time = self.min_sleep_time

```
--- END FILE ---

Please first localize the bug based on the issue statement, and then generate *SEARCH/REPLACE* edits to fix the issue.

Every *SEARCH/REPLACE* edit must use this format:
1. The file path
2. The start of search block: <<<<<<< SEARCH
3. A contiguous chunk of lines to search for in the existing source code
4. The dividing line: =======
5. The lines to replace into the source code
6. The end of the replace block: >>>>>>> REPLACE

Here is an example:

```python
### mathweb/flask/app.py
<<<<<<< SEARCH
from flask import Flask
=======
import math
from flask import Flask
>>>>>>> REPLACE
```
</solution>

Please note that the *SEARCH/REPLACE* edit REQUIRES PROPER INDENTATION. If you would like to add the line '        print(x)', you must fully write that out, with all those spaces before the code!
Wrap each *SEARCH/REPLACE* edit in a code block as shown in the example above. If you have multiple *SEARCH/REPLACE* edits, use a separate code block for each one.
[ISSUE]

**Title:** ProducerConsumer class does not accept `jobs` parameter, causing TypeError

**Description:**
When initializing the `ProducerConsumer` class with the `jobs` keyword argument, a `TypeError` is raised indicating that the `__init__` method does not recognize the `jobs` parameter.

**Example Code:**
```python
consumer = ProducerConsumer(producer, consumer_func, jobs=10)
```

**Expected Behavior:**
The `ProducerConsumer` class should accept the `jobs` parameter to specify the number of parallel jobs, allowing users to configure the level of concurrency.

**Actual Behavior:**
A `TypeError` is raised:
```
TypeError: __init__() got an unexpected keyword argument 'jobs'
```
This error occurs because the `ProducerConsumer` class is expecting a different parameter name (`njobs`), leading to inconsistencies and failures when attempting to use the `jobs` argument.

[/ISSUE]
{
    "test_ProducerConsumer": "PASSED",
    "test_creatsubdatasets": "ERROR"
}
diff --git a/datalad/support/parallel.py b/datalad/support/parallel.py
index ad71207b1..7346b8a93 100644
--- a/datalad/support/parallel.py
+++ b/datalad/support/parallel.py
@@ -50,7 +50,7 @@ class ProducerConsumer:
 
     def __init__(self,
                  producer, consumer,
-                 njobs=None,
+                 jobs=None,
                  safe_to_consume=None,
                  reraise_immediately=False,
                  agg=None,
@@ -77,7 +77,7 @@ class ProducerConsumer:
         """
         self.producer = producer
         self.consumer = consumer
-        self.njobs = njobs
+        self.jobs = jobs
         self.safe_to_consume = safe_to_consume
         self.reraise_immediately = reraise_immediately
         self.agg = agg
@@ -159,11 +159,16 @@ class ProducerConsumer:
         futures = {}
 
         total_announced = False
-        njobs = self.njobs or 1
-        lgr.debug("Initiating ThreadPoolExecutor with %d jobs", njobs)
+        jobs = self.jobs or 1
+        if jobs == "auto":
+            # ATM there is no "auto" for this operation.  We will just make it ...
+            # "auto" could be for some auto-scaling based on a single future time
+            # to complete, scaling up/down.  TODO
+            jobs = 5
+        lgr.debug("Initiating ThreadPoolExecutor with %d jobs", jobs)
         # we will increase sleep_time when doing nothing useful
         sleeper = Sleeper()
-        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:
+        with concurrent.futures.ThreadPoolExecutor(jobs) as executor:
             self._executor = executor
             # yield from the producer_queue (.total and .finished could be accessed meanwhile)
             while True:

{"datalad/support/parallel.py": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Helpers for parallel execution\n\n\"\"\"\n\n__docformat__ = 'restructuredtext'\n\nimport concurrent.futures\nimport inspect\nimport time\nimport uuid\n\nfrom collections import defaultdict\nfrom queue import Queue, Empty\nfrom threading import Thread\n\nfrom . import ansi_colors as colors\nfrom ..log import log_progress\nfrom ..utils import path_is_subpath\n\n\ndef _count_str(count, verb, omg=False):\n    if count:\n        msg = \"{:d} {}\".format(count, verb)\n        if omg:\n            msg = colors.color_word(msg, colors.RED)\n        return msg\n\n\ndef no_parentds_in_futures(futures, path):\n    \"\"\"Return True if no path in futures keys is parentds for provided path\n\n    Assumes that the future's key is the path.\n    \"\"\"\n    # TODO: OPT.  Could benefit from smarter than linear time if not one at a time?\n    #   or may be we should only go through active futures (still linear!)?\n    return not any(path_is_subpath(path, p) for p in futures)\n\n\nclass ProducerConsumer:\n    \"\"\"\n    \"\"\"\n\n    def __init__(self,\n                 producer, consumer,\n                 njobs=None,\n                 safe_to_consume=None,\n                 reraise_immediately=False,\n                 agg=None,\n                 # TODO: \"absorb\" into some more generic \"logging\" helper\n                 # so could be used not only with consumers which yield our records\n                 # 'agg' from above could also more relate to i\n                 label=\"Total\", unit=\"items\",\n                 lgr=None,\n                 ):\n        \"\"\"\n\n        Parameters\n        ----------\n        ...\n        safe_to_consume: callable\n          A callable which gets a dict of all known futures and current producer output.\n          It should return True if we can proceed with current value from producer.\n          If unsafe - we will wait.  WARNING: outside code should make sure about provider and\n          safe_to_consume to play nicely or deadlock can happen.\n        reraise_immediately: bool, optional\n          If True, it would stop producer yielding values as soon as it detects that some\n          exception has occurred (although there might still be values in the queue to be yielded\n          which were collected before the exception was raised)\n        \"\"\"\n        self.producer = producer\n        self.consumer = consumer\n        self.njobs = njobs\n        self.safe_to_consume = safe_to_consume\n        self.reraise_immediately = reraise_immediately\n        self.agg = agg\n        self.label = label\n        self.unit = unit\n        self.lgr = lgr\n\n        self.total = None\n        self.producer_finished = None\n        self._executor = None\n        self._exc = []\n\n    def __del__(self):\n        # if we are killed while executing, we should ask executor to terminate\n        executor = getattr(self, \"_executor\")\n        if executor:\n            executor.terminate()\n\n    def __iter__(self):\n        self.producer_finished = False\n        self._exc = []\n\n        producer_queue = Queue()\n        consumer_queue = Queue()\n\n        def producer_worker():\n            \"\"\"That is the one which interrogates producer and updates .total\"\"\"\n            total = None if self.agg else 0\n            try:\n                for value in self.producer:\n                    producer_queue.put(value)\n                    if self.agg:\n                        total = (\n                            self.agg(value, total) if total is not None else self.agg(value)\n                        )\n                    else:\n                        total += 1\n                    self.total = total\n            except BaseException as e:\n                self._exc.append(e)\n            finally:\n                self.producer_finished = True\n\n        def consumer_worker(callable, *args, **kwargs):\n            \"\"\"Since jobs could return a generator and we cannot really \"inspect\" for that\n            \"\"\"\n            try:\n                res = callable(*args, **kwargs)\n                if inspect.isgenerator(res):\n                    lgr.debug(\"Got consumer worker which returned a generator %s\", res)\n                    didgood = False\n                    for r in res:\n                        didgood = True\n                        lgr.debug(\"Adding %s to queue\", r)\n                        consumer_queue.put(r)\n                    if not didgood:\n                        lgr.error(\"Nothing was obtained from %s :-(\", res)\n                else:\n                    lgr.debug(\"Got straight result %s, not a generator\", res)\n                    consumer_queue.put(res)\n            except BaseException as e:\n                self._exc.append(e)\n\n        producer_thread = Thread(target=producer_worker)\n        producer_thread.start()\n\n        # LOGGING\n        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?\n        lgr = self.lgr\n        label = self.label\n        if lgr is None:\n            from .. import lgr\n        log_progress(lgr.info, pid,\n                     \"%s: starting\", self.label,\n                     # will become known only later total=len(items),\n                     label=self.label, unit=\" \" + self.unit)\n        counts = defaultdict(int)\n\n        futures = {}\n\n        total_announced = False\n        njobs = self.njobs or 1\n        lgr.debug(\"Initiating ThreadPoolExecutor with %d jobs\", njobs)\n        # we will increase sleep_time when doing nothing useful\n        sleeper = Sleeper()\n        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:\n            self._executor = executor\n            # yield from the producer_queue (.total and .finished could be accessed meanwhile)\n            while True:\n                done_useful = False\n                if self.reraise_immediately and self._exc:\n                    break\n                if (self.producer_finished and\n                        not futures and\n                        consumer_queue.empty() and\n                        producer_queue.empty()):\n                    break\n\n                # LOGGING\n                if not total_announced and self.producer_finished:\n                    log_progress(\n                        lgr.info,\n                        pid,\n                        None,  # I do not think there is something valuable to announce\n                        total=self.total,\n                        # unfortunately of no effect, so we cannot inform that more items to come\n                        # unit=(\"+\" if not it.finished else \"\") + \" \" + unit,\n                        update=0  # not None, so it does not stop\n                    )\n                    total_announced = True\n\n                # important!  We are using threads, so worker threads will be sharing CPU time\n                # with this master thread. For it to become efficient, we should consume as much\n                # as possible from producer asap and push it to executor.  So drain the queue\n                while not producer_queue.empty():\n                    done_useful = True\n                    try:\n                        job_args = producer_queue.get() # timeout=0.001)\n                        if self.safe_to_consume:\n                            # Sleep a little if we are not yet ready\n                            # TODO: add some .debug level reporting based on elapsed time\n                            # IIRC I did smth like growing exponentially delays somewhere (dandi?)\n                            while not self.safe_to_consume(futures, job_args):\n                                _prune_futures(futures, lgr) or sleeper()\n                        # Current implementation, to provide depchecking, relies on unique\n                        # args for the job\n                        assert job_args not in futures\n                        lgr.debug(\"Submitting worker future for %s\", job_args)\n                        futures[job_args] = executor.submit(consumer_worker, self.consumer, job_args)\n                    except Empty:\n                        pass\n\n                # check active futures\n                if not consumer_queue.empty():\n                    done_useful = True\n\n                    # ATM we do not bother of some \"in order\" reporting\n                    # Just report as soon as any new record arrives\n                    res = consumer_queue.get()\n                    lgr.debug(\"Got %s from consumer_queue\", res)\n                    # LOGGING\n                    counts[res[\"status\"]] += 1\n                    count_strs = (_count_str(*args)\n                                  for args in [(counts[\"notneeded\"], \"skipped\", False),\n                                               (counts[\"error\"], \"failed\", True)])\n                    if counts[\"notneeded\"] or counts[\"error\"]:\n                        label = \"{} ({})\".format(\n                            self.label,\n                            \", \".join(filter(None, count_strs)))\n\n                    log_progress(\n                        lgr.error if res[\"status\"] == \"error\" else lgr.info,\n                        pid,\n                        \"%s: processed result%s\", self.label,\n                        \" for \" + res[\"path\"] if \"path\" in res else \"\",\n                        label=label, update=1, increment=True)\n\n                    yield res\n\n                done_useful |= _prune_futures(futures, lgr)\n\n                if not done_useful:  # you need some rest\n                    # TODO: same here -- progressive logging\n                    lgr.debug(\n                        \"Did nothing useful, sleeping. Have %s %s %s %s\",\n                        self.producer_finished,\n                        futures,\n                        consumer_queue.empty(),\n                        producer_queue.empty())\n                    sleeper()\n                else:\n                    sleeper.reset()\n\n        self._executor = None\n\n        # LOGGING\n        log_progress(lgr.info, pid, \"%s: done\", self.label)\n        producer_thread.join()\n        if self._exc:\n            if len(self._exc) > 1:\n                lgr.debug(\"%d exceptions were collected while performing execution in parallel. Only the first one \"\n                          \"will be reraised\", len(self._exc))\n            raise self._exc[0]\n        else:\n            assert not futures, \\\n                \"There is still %d active futures for following args: %s\" \\\n                % (len(futures), ', '.join(futures))\n\n\ndef _prune_futures(futures, lgr):\n    \"\"\"Removes .done from provided futures.\n\n    Returns\n    -------\n    bool\n      True if any future was removed\n    \"\"\"\n    done_useful = False\n    # remove futures which are done\n    for args, future in list(futures.items()):\n        if future.done():\n            done_useful = True\n            future_ = futures.pop(args)\n            msg = \"\"\n            if future_.exception():\n                msg = \" with an exception %s\" % future_.exception()\n            lgr.debug(\"Future for %r is done%s\", args, msg)\n    return done_useful\n\n\nclass Sleeper():\n    def __init__(self):\n        self.min_sleep_time = 0.001\n        # but no more than to this max\n        self.max_sleep_time = 0.1\n        self.sleep_time = self.min_sleep_time\n\n    def __call__(self):\n        time.sleep(self.sleep_time)\n        self.sleep_time = min(self.max_sleep_time, self.sleep_time * 2)\n\n    def reset(self):\n        self.sleep_time = self.min_sleep_time\n"}
A user will ask you to solve a task. You should first draft your thinking process (inner monologue). Then, generate the solution.

Your response format must follow the template below:
<think>
Your thoughts or/and draft, like working through an exercise on scratch paper. Be as casual and as long as you want until you are confident to generate a correct solution.
</think>
<solution>
Final solution presented to the user.
</solution>
system
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ISSUE]

**Title:** ProducerConsumer class does not accept `jobs` parameter, causing TypeError

**Description:**
When initializing the `ProducerConsumer` class with the `jobs` keyword argument, a `TypeError` is raised indicating that the `__init__` method does not recognize the `jobs` parameter.

**Example Code:**
```python
consumer = ProducerConsumer(producer, consumer_func, jobs=10)
```

**Expected Behavior:**
The `ProducerConsumer` class should accept the `jobs` parameter to specify the number of parallel jobs, allowing users to configure the level of concurrency.

**Actual Behavior:**
A `TypeError` is raised:
```
TypeError: __init__() got an unexpected keyword argument 'jobs'
```
This error occurs because the `ProducerConsumer` class is expecting a different parameter name (`njobs`), leading to inconsistencies and failures when attempting to use the `jobs` argument.

[/ISSUE]
--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.

--- BEGIN FILE ---
```
### datalad/support/parallel.py
# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-
# ex: set sts=4 ts=4 sw=4 noet:
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
#
#   See COPYING file distributed along with the datalad package for the
#   copyright and license terms.
#
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
"""Helpers for parallel execution

"""

__docformat__ = 'restructuredtext'

import concurrent.futures
import inspect
import time
import uuid

from collections import defaultdict
from queue import Queue, Empty
from threading import Thread

from . import ansi_colors as colors
from ..log import log_progress
from ..utils import path_is_subpath


def _count_str(count, verb, omg=False):
    if count:
        msg = "{:d} {}".format(count, verb)
        if omg:
            msg = colors.color_word(msg, colors.RED)
        return msg


def no_parentds_in_futures(futures, path):
    """Return True if no path in futures keys is parentds for provided path

    Assumes that the future's key is the path.
    """
    # TODO: OPT.  Could benefit from smarter than linear time if not one at a time?
    #   or may be we should only go through active futures (still linear!)?
    return not any(path_is_subpath(path, p) for p in futures)


class ProducerConsumer:
    """
    """

    def __init__(self,
                 producer, consumer,
                 njobs=None,
                 safe_to_consume=None,
                 reraise_immediately=False,
                 agg=None,
                 # TODO: "absorb" into some more generic "logging" helper
                 # so could be used not only with consumers which yield our records
                 # 'agg' from above could also more relate to i
                 label="Total", unit="items",
                 lgr=None,
                 ):
        """

        Parameters
        ----------
        ...
        safe_to_consume: callable
          A callable which gets a dict of all known futures and current producer output.
          It should return True if we can proceed with current value from producer.
          If unsafe - we will wait.  WARNING: outside code should make sure about provider and
          safe_to_consume to play nicely or deadlock can happen.
        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)
        """
        self.producer = producer
        self.consumer = consumer
        self.njobs = njobs
        self.safe_to_consume = safe_to_consume
        self.reraise_immediately = reraise_immediately
        self.agg = agg
        self.label = label
        self.unit = unit
        self.lgr = lgr

        self.total = None
        self.producer_finished = None
        self._executor = None
        self._exc = []

    def __del__(self):
        # if we are killed while executing, we should ask executor to terminate
        executor = getattr(self, "_executor")
        if executor:
            executor.terminate()

    def __iter__(self):
        self.producer_finished = False
        self._exc = []

        producer_queue = Queue()
        consumer_queue = Queue()

        def producer_worker():
            """That is the one which interrogates producer and updates .total"""
            total = None if self.agg else 0
            try:
                for value in self.producer:
                    producer_queue.put(value)
                    if self.agg:
                        total = (
                            self.agg(value, total) if total is not None else self.agg(value)
                        )
                    else:
                        total += 1
                    self.total = total
            except BaseException as e:
                self._exc.append(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
            """
            try:
                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)
            except BaseException as e:
                self._exc.append(e)

        producer_thread = Thread(target=producer_worker)
        producer_thread.start()

        # LOGGING
        pid = str(uuid.uuid4())  # could be based on PID and time may be to be informative?
        lgr = self.lgr
        label = self.label
        if lgr is None:
            from .. import lgr
        log_progress(lgr.info, pid,
                     "%s: starting", self.label,
                     # will become known only later total=len(items),
                     label=self.label, unit=" " + self.unit)
        counts = defaultdict(int)

        futures = {}

        total_announced = False
        njobs = self.njobs or 1
        lgr.debug("Initiating ThreadPoolExecutor with %d jobs", njobs)
        # we will increase sleep_time when doing nothing useful
        sleeper = Sleeper()
        with concurrent.futures.ThreadPoolExecutor(njobs) as executor:
            self._executor = executor
            # yield from the producer_queue (.total and .finished could be accessed meanwhile)
            while True:
                done_useful = False
                if self.reraise_immediately and self._exc:
                    break
                if (self.producer_finished and
                        not futures and
                        consumer_queue.empty() and
                        producer_queue.empty()):
                    break

                # LOGGING
                if not total_announced and self.producer_finished:
                    log_progress(
                        lgr.info,
                        pid,
                        None,  # I do not think there is something valuable to announce
                        total=self.total,
                        # unfortunately of no effect, so we cannot inform that more items to come
                        # unit=("+" if not it.finished else "") + " " + unit,
                        update=0  # not None, so it does not stop
                    )
                    total_announced = True

                # 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():
                    done_useful = True
                    try:
                        job_args = producer_queue.get() # timeout=0.001)
                        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_args):
                                _prune_futures(futures, lgr) or sleeper()
                        # Current implementation, to provide depchecking, relies on unique
                        # args for the job
                        assert job_args not in futures
                        lgr.debug("Submitting worker future for %s", job_args)
                        futures[job_args] = 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)
                    # LOGGING
                    counts[res["status"]] += 1
                    count_strs = (_count_str(*args)
                                  for args in [(counts["notneeded"], "skipped", False),
                                               (counts["error"], "failed", True)])
                    if counts["notneeded"] or counts["error"]:
                        label = "{} ({})".format(
                            self.label,
                            ", ".join(filter(None, count_strs)))

                    log_progress(
                        lgr.error if res["status"] == "error" else lgr.info,
                        pid,
                        "%s: processed result%s", self.label,
                        " for " + res["path"] if "path" in res else "",
                        label=label, update=1, increment=True)

                    yield res

                done_useful |= _prune_futures(futures, lgr)

                if not done_useful:  # you need some rest
                    # TODO: same here -- progressive logging
                    lgr.debug(
                        "Did nothing useful, sleeping. Have %s %s %s %s",
                        self.producer_finished,
                        futures,
                        consumer_queue.empty(),
                        producer_queue.empty())
                    sleeper()
                else:
                    sleeper.reset()

        self._executor = None

        # LOGGING
        log_progress(lgr.info, pid, "%s: done", self.label)
        producer_thread.join()
        if self._exc:
            if len(self._exc) > 1:
                lgr.debug("%d exceptions were collected while performing execution in parallel. Only the first one "
                          "will be reraised", len(self._exc))
            raise self._exc[0]
        else:
            assert not futures, \
                "There is still %d active futures for following args: %s" \
                % (len(futures), ', '.join(futures))


def _prune_futures(futures, lgr):
    """Removes .done from provided futures.

    Returns
    -------
    bool
      True if any future was removed
    """
    done_useful = False
    # remove futures which are done
    for args, future in list(futures.items()):
        if future.done():
            done_useful = True
            future_ = futures.pop(args)
            msg = ""
            if future_.exception():
                msg = " with an exception %s" % future_.exception()
            lgr.debug("Future for %r is done%s", args, msg)
    return done_useful


class Sleeper():
    def __init__(self):
        self.min_sleep_time = 0.001
        # but no more than to this max
        self.max_sleep_time = 0.1
        self.sleep_time = self.min_sleep_time

    def __call__(self):
        time.sleep(self.sleep_time)
        self.sleep_time = min(self.max_sleep_time, self.sleep_time * 2)

    def reset(self):
        self.sleep_time = self.min_sleep_time

```
--- END FILE ---

Please first localize the bug based on the issue statement, and then generate *SEARCH/REPLACE* edits to fix the issue.

Every *SEARCH/REPLACE* edit must use this format:
1. The file path
2. The start of search block: <<<<<<< SEARCH
3. A contiguous chunk of lines to search for in the existing source code
4. The dividing line: =======
5. The lines to replace into the source code
6. The end of the replace block: >>>>>>> REPLACE

Here is an example:

```python
### mathweb/flask/app.py
<<<<<<< SEARCH
from flask import Flask
=======
import math
from flask import Flask
>>>>>>> REPLACE
```
</solution>

Please note that the *SEARCH/REPLACE* edit REQUIRES PROPER INDENTATION. If you would like to add the line '        print(x)', you must fully write that out, with all those spaces before the code!
Wrap each *SEARCH/REPLACE* edit in a code block as shown in the example above. If you have multiple *SEARCH/REPLACE* edits, use a separate code block for each one.
user