.. image:: http://www.repostatus.org/badges/latest/active.svg
:target: http://www.repostatus.org/#active
:alt: Project Status: Active — The project has reached a stable, usable
state and is being actively developed.
.. image:: https://github.com/jwodder/interleave/workflows/Test/badge.svg?branch=master
:target: https://github.com/jwodder/interleave/actions?workflow=Test
:alt: CI Status
.. image:: https://codecov.io/gh/jwodder/interleave/branch/master/graph/badge.svg
:target: https://codecov.io/gh/jwodder/interleave
.. image:: https://img.shields.io/pypi/pyversions/interleave.svg
:target: https://pypi.org/project/interleave/
.. image:: https://img.shields.io/conda/vn/conda-forge/interleave.svg
:target: https://anaconda.org/conda-forge/interleave
:alt: Conda Version
.. image:: https://img.shields.io/github/license/jwodder/interleave.svg
:target: https://opensource.org/licenses/MIT
:alt: MIT License
`GitHub <https://github.com/jwodder/interleave>`_
| `PyPI <https://pypi.org/project/interleave/>`_
| `Issues <https://github.com/jwodder/interleave/issues>`_
| `Changelog <https://github.com/jwodder/interleave/blob/master/CHANGELOG.md>`_
The ``interleave`` package provides a function of the same name that takes a
number of iterators, runs them in separate threads, and yields the values
produced as soon as each one is available.
Installation
============
``interleave`` requires Python 3.7 or higher. Just use `pip
<https://pip.pypa.io>`_ for Python 3 (You have pip, right?) to install
``interleave`` and its dependencies::
python3 -m pip install interleave
Example
=======
>>> from time import sleep, strftime
>>> from interleave import interleave
>>>
>>> def sleeper(idno, delays):
... for i, d in enumerate(delays):
... sleep(d)
... yield (idno, i)
...
>>> with interleave(
... [
... sleeper(0, [0, 1, 2]),
... sleeper(1, [2, 2, 2]),
... sleeper(2, [5, 2, 1]),
... ]
... ) as it:
... for x in it:
... print(strftime("%H:%M:%S"), x)
...
22:08:39 (0, 0)
22:08:40 (0, 1)
22:08:41 (1, 0)
22:08:42 (0, 2)
22:08:43 (1, 1)
22:08:44 (2, 0)
22:08:45 (1, 2)
22:08:46 (2, 1)
22:08:47 (2, 2)
API
===
.. code:: python
interleave.interleave(
iterators: Iterable[Iterator[T]],
*,
max_workers: Optional[int] = None,
thread_name_prefix: str = "",
queue_size: Optional[int] = None,
onerror: interleave.OnError = interleave.STOP,
) -> interleave.Interleaver[T]
``interleave()`` runs the given iterators in separate threads and returns an
iterator that yields the values yielded by them as they become available. (See
below for details on the ``Interleaver`` class.)
The ``max_workers`` and ``thread_name_prefix`` parameters are passed through to
the underlying |ThreadPoolExecutor|_ (q.v.). ``max_workers`` determines the
maximum number of iterators to run at one time.
.. |ThreadPoolExecutor| replace:: ``concurrent.futures.ThreadPoolExecutor``
.. _ThreadPoolExecutor:
https://docs.python.org/3/library/concurrent.futures.html
#concurrent.futures.ThreadPoolExecutor
The ``queue_size`` parameter sets the maximum size of the queue used internally
to pipe values yielded by the iterators; when the queue is full, any iterator
with a value to yield will block waiting for the next value to be dequeued by a
call to the interleaver's ``__next__``. When ``queue_size`` is ``None`` (the
default), ``interleave()`` uses a ``queue.SimpleQueue``, which has no maximum
size. When ``queue_size`` is non-``None`` (including zero, signifying no
maximum size), ``interleave()`` uses a ``queue.Queue``, whose ``get()`` method
is uninterruptible (including by ``KeyboardInterrupt``) on Windows.
The ``onerror`` parameter is an enum that determines how ``interleave()``
should behave if one of the iterators raises an exception. The possible values
are:
``STOP``
*(default)* Stop iterating over all iterators, cancel any outstanding
iterators that haven't been started yet, wait for all running threads to
finish, and reraise the exception. Note that, due to the inability to stop
an iterator between yields, the "waiting" step involves waiting for each
currently-running iterator to yield its next value before stopping. This
can deadlock if the queue fills up in the interim.
``DRAIN``
Like ``STOP``, but any remaining values yielded by the iterators before
they finish are yielded by the interleaver before raising the exception
``FINISH_ALL``
Continue running as normal and reraise the exception once all iterators
have finished
``FINISH_CURRENT``
Like ``FINISH_ALL``, except that only currently-running iterators are run
to completion; any iterators whose threads haven't yet been started when
the exception is raised will have their jobs cancelled
Regardless of the value of ``onerror``, any later exceptions raised by
iterators after the initial exception are discarded.
.. code:: python
class Interleaver(Generic[T]):
def __init__(
self,
max_workers: Optional[int] = None,
thread_name_prefix: str = "",
queue_size: Optional[int] = None,
onerror: OnError = STOP,
)
An iterator and context manager. As an iterator, it yields the values
generated by the iterators passed to the corresponding ``interleave()`` call as
they become available. As a context manager, it returns itself on entry and,
on exit, cleans up any unfinished threads by calling the
``shutdown(wait=True)`` method (see below).
An ``Interleaver`` can be instantiated either by calling ``interleave()`` or by
calling the constructor directly. The constructor takes the same arguments as
``interleave()``, minus ``iterators``, and produces a new ``Interleaver`` that
is not yet running any iterators. Iterators are submitted to a new
``Interleaver`` via the ``submit()`` method; once all desired iterators have
been submitted, the ``finalize()`` method **must** be called so that the
``Interleaver`` can tell when everything's finished.
An ``Interleaver`` will shut down its ``ThreadPoolExecutor`` and wait for the
threads to finish after yielding its final value (specifically, when a call is
made to ``__next__``/``get()`` that would result in ``StopIteration`` or
another exception being raised). In the event that an ``Interleaver`` is
abandoned before iteration completes, the associated resources may not be
properly cleaned up, and threads may continue running indefinitely. For this
reason, it is strongly recommended that you wrap any iteration over an
``Interleaver`` in the context manager in order to handle a premature end to
iteration (including from a ``KeyboardInterrupt``).
Besides the iterator and context manager APIs, an ``Interleaver`` has the
following public methods:
.. code:: python
Interleaver.submit(it: Iterator[T]) -> None
*New in version 0.2.0*
Add an iterator to the ``Interleaver``.
If the ``Interleaver`` was returned from ``interleave()`` or has already had
``finalize()`` called on it, calling ``submit()`` will result in a
``ValueError``.
.. code:: python
Interleave.finalize() -> None
*New in version 0.2.0*
Notify the ``Interleaver`` that all iterators have been registered. This
method must be called in order for the ``Interleaver`` to detect the end of
iteration; if this method has not been called and all submitted iterators have
finished & had their values retrieved, then a subsequent call to ``next(it)``
will end up hanging indefinitely.
.. code:: python
Interleaver.get(block: bool = True, timeout: Optional[float] = None) -> T
*New in version 0.2.0*
Fetch the next value generated by the iterators. If all iterators have
finished and all values have been retrieved, raises
``interleaver.EndOfInputError``. If ``block`` is ``False`` and no values are
immediately available, raises ``queue.Empty``. If ``block`` is ``True``, waits
up to ``timeout`` seconds (or indefinitely, if ``timeout`` is ``None``) for the
next value to become available or for all iterators to end; if nothing happens
before the timeout expires, raises ``queue.Empty``.
``it.get(block=True, timeout=None)`` is equivalent to ``next(it)``, except that
the latter converts an ``EndOfInputError`` to ``StopIteration``.
**Note:** When ``onerror=STOP`` and a timeout is set, if an iterator raises an
exception, the timeout may be exceeded as the ``Interleaver`` waits for all
remaining threads to shut down.
.. code:: python
Interleaver.shutdown(wait: bool = True) -> None
Call ``finalize()`` if it hasn't been called yet, tell all running iterators to
stop iterating, cancel any outstanding iterators that haven't been started yet,
and shut down the ``ThreadPoolExecutor``. The ``wait`` parameter is passed
through to the call to ``ThreadPoolExecutor.shutdown()``.
The ``Interleaver`` can continue to be iterated over after calling
``shutdown()`` and will yield any remaining values produced by the iterators
before they stopped completely.
Raw data
{
"_id": null,
"home_page": "https://github.com/jwodder/interleave",
"name": "interleave",
"maintainer": "",
"docs_url": null,
"requires_python": "~=3.7",
"maintainer_email": "",
"keywords": "interleaving,iterators,multithreading",
"author": "John Thorvald Wodder II",
"author_email": "interleave@varonathe.org",
"download_url": "https://files.pythonhosted.org/packages/8c/80/1c45cd6dcd72c0669c3e11c74fbb1459cea5aa3d80960ffee47b66d1640d/interleave-0.2.1.tar.gz",
"platform": null,
"description": ".. image:: http://www.repostatus.org/badges/latest/active.svg\n :target: http://www.repostatus.org/#active\n :alt: Project Status: Active \u2014 The project has reached a stable, usable\n state and is being actively developed.\n\n.. image:: https://github.com/jwodder/interleave/workflows/Test/badge.svg?branch=master\n :target: https://github.com/jwodder/interleave/actions?workflow=Test\n :alt: CI Status\n\n.. image:: https://codecov.io/gh/jwodder/interleave/branch/master/graph/badge.svg\n :target: https://codecov.io/gh/jwodder/interleave\n\n.. image:: https://img.shields.io/pypi/pyversions/interleave.svg\n :target: https://pypi.org/project/interleave/\n\n.. image:: https://img.shields.io/conda/vn/conda-forge/interleave.svg\n :target: https://anaconda.org/conda-forge/interleave\n :alt: Conda Version\n\n.. image:: https://img.shields.io/github/license/jwodder/interleave.svg\n :target: https://opensource.org/licenses/MIT\n :alt: MIT License\n\n`GitHub <https://github.com/jwodder/interleave>`_\n| `PyPI <https://pypi.org/project/interleave/>`_\n| `Issues <https://github.com/jwodder/interleave/issues>`_\n| `Changelog <https://github.com/jwodder/interleave/blob/master/CHANGELOG.md>`_\n\nThe ``interleave`` package provides a function of the same name that takes a\nnumber of iterators, runs them in separate threads, and yields the values\nproduced as soon as each one is available.\n\nInstallation\n============\n``interleave`` requires Python 3.7 or higher. Just use `pip\n<https://pip.pypa.io>`_ for Python 3 (You have pip, right?) to install\n``interleave`` and its dependencies::\n\n python3 -m pip install interleave\n\n\nExample\n=======\n\n>>> from time import sleep, strftime\n>>> from interleave import interleave\n>>>\n>>> def sleeper(idno, delays):\n... for i, d in enumerate(delays):\n... sleep(d)\n... yield (idno, i)\n...\n>>> with interleave(\n... [\n... sleeper(0, [0, 1, 2]),\n... sleeper(1, [2, 2, 2]),\n... sleeper(2, [5, 2, 1]),\n... ]\n... ) as it:\n... for x in it:\n... print(strftime(\"%H:%M:%S\"), x)\n...\n22:08:39 (0, 0)\n22:08:40 (0, 1)\n22:08:41 (1, 0)\n22:08:42 (0, 2)\n22:08:43 (1, 1)\n22:08:44 (2, 0)\n22:08:45 (1, 2)\n22:08:46 (2, 1)\n22:08:47 (2, 2)\n\n\nAPI\n===\n\n.. code:: python\n\n interleave.interleave(\n iterators: Iterable[Iterator[T]],\n *,\n max_workers: Optional[int] = None,\n thread_name_prefix: str = \"\",\n queue_size: Optional[int] = None,\n onerror: interleave.OnError = interleave.STOP,\n ) -> interleave.Interleaver[T]\n\n``interleave()`` runs the given iterators in separate threads and returns an\niterator that yields the values yielded by them as they become available. (See\nbelow for details on the ``Interleaver`` class.)\n\nThe ``max_workers`` and ``thread_name_prefix`` parameters are passed through to\nthe underlying |ThreadPoolExecutor|_ (q.v.). ``max_workers`` determines the\nmaximum number of iterators to run at one time.\n\n.. |ThreadPoolExecutor| replace:: ``concurrent.futures.ThreadPoolExecutor``\n.. _ThreadPoolExecutor:\n https://docs.python.org/3/library/concurrent.futures.html\n #concurrent.futures.ThreadPoolExecutor\n\nThe ``queue_size`` parameter sets the maximum size of the queue used internally\nto pipe values yielded by the iterators; when the queue is full, any iterator\nwith a value to yield will block waiting for the next value to be dequeued by a\ncall to the interleaver's ``__next__``. When ``queue_size`` is ``None`` (the\ndefault), ``interleave()`` uses a ``queue.SimpleQueue``, which has no maximum\nsize. When ``queue_size`` is non-``None`` (including zero, signifying no\nmaximum size), ``interleave()`` uses a ``queue.Queue``, whose ``get()`` method\nis uninterruptible (including by ``KeyboardInterrupt``) on Windows.\n\nThe ``onerror`` parameter is an enum that determines how ``interleave()``\nshould behave if one of the iterators raises an exception. The possible values\nare:\n\n``STOP``\n *(default)* Stop iterating over all iterators, cancel any outstanding\n iterators that haven't been started yet, wait for all running threads to\n finish, and reraise the exception. Note that, due to the inability to stop\n an iterator between yields, the \"waiting\" step involves waiting for each\n currently-running iterator to yield its next value before stopping. This\n can deadlock if the queue fills up in the interim.\n\n``DRAIN``\n Like ``STOP``, but any remaining values yielded by the iterators before\n they finish are yielded by the interleaver before raising the exception\n\n``FINISH_ALL``\n Continue running as normal and reraise the exception once all iterators\n have finished\n\n``FINISH_CURRENT``\n Like ``FINISH_ALL``, except that only currently-running iterators are run\n to completion; any iterators whose threads haven't yet been started when\n the exception is raised will have their jobs cancelled\n\nRegardless of the value of ``onerror``, any later exceptions raised by\niterators after the initial exception are discarded.\n\n.. code:: python\n\n class Interleaver(Generic[T]):\n def __init__(\n self,\n max_workers: Optional[int] = None,\n thread_name_prefix: str = \"\",\n queue_size: Optional[int] = None,\n onerror: OnError = STOP,\n )\n\nAn iterator and context manager. As an iterator, it yields the values\ngenerated by the iterators passed to the corresponding ``interleave()`` call as\nthey become available. As a context manager, it returns itself on entry and,\non exit, cleans up any unfinished threads by calling the\n``shutdown(wait=True)`` method (see below).\n\nAn ``Interleaver`` can be instantiated either by calling ``interleave()`` or by\ncalling the constructor directly. The constructor takes the same arguments as\n``interleave()``, minus ``iterators``, and produces a new ``Interleaver`` that\nis not yet running any iterators. Iterators are submitted to a new\n``Interleaver`` via the ``submit()`` method; once all desired iterators have\nbeen submitted, the ``finalize()`` method **must** be called so that the\n``Interleaver`` can tell when everything's finished.\n\nAn ``Interleaver`` will shut down its ``ThreadPoolExecutor`` and wait for the\nthreads to finish after yielding its final value (specifically, when a call is\nmade to ``__next__``/``get()`` that would result in ``StopIteration`` or\nanother exception being raised). In the event that an ``Interleaver`` is\nabandoned before iteration completes, the associated resources may not be\nproperly cleaned up, and threads may continue running indefinitely. For this\nreason, it is strongly recommended that you wrap any iteration over an\n``Interleaver`` in the context manager in order to handle a premature end to\niteration (including from a ``KeyboardInterrupt``).\n\nBesides the iterator and context manager APIs, an ``Interleaver`` has the\nfollowing public methods:\n\n.. code:: python\n\n Interleaver.submit(it: Iterator[T]) -> None\n\n*New in version 0.2.0*\n\nAdd an iterator to the ``Interleaver``.\n\nIf the ``Interleaver`` was returned from ``interleave()`` or has already had\n``finalize()`` called on it, calling ``submit()`` will result in a\n``ValueError``.\n\n.. code:: python\n\n Interleave.finalize() -> None\n\n*New in version 0.2.0*\n\nNotify the ``Interleaver`` that all iterators have been registered. This\nmethod must be called in order for the ``Interleaver`` to detect the end of\niteration; if this method has not been called and all submitted iterators have\nfinished & had their values retrieved, then a subsequent call to ``next(it)``\nwill end up hanging indefinitely.\n\n.. code:: python\n\n Interleaver.get(block: bool = True, timeout: Optional[float] = None) -> T\n\n*New in version 0.2.0*\n\nFetch the next value generated by the iterators. If all iterators have\nfinished and all values have been retrieved, raises\n``interleaver.EndOfInputError``. If ``block`` is ``False`` and no values are\nimmediately available, raises ``queue.Empty``. If ``block`` is ``True``, waits\nup to ``timeout`` seconds (or indefinitely, if ``timeout`` is ``None``) for the\nnext value to become available or for all iterators to end; if nothing happens\nbefore the timeout expires, raises ``queue.Empty``.\n\n``it.get(block=True, timeout=None)`` is equivalent to ``next(it)``, except that\nthe latter converts an ``EndOfInputError`` to ``StopIteration``.\n\n**Note:** When ``onerror=STOP`` and a timeout is set, if an iterator raises an\nexception, the timeout may be exceeded as the ``Interleaver`` waits for all\nremaining threads to shut down.\n\n.. code:: python\n\n Interleaver.shutdown(wait: bool = True) -> None\n\nCall ``finalize()`` if it hasn't been called yet, tell all running iterators to\nstop iterating, cancel any outstanding iterators that haven't been started yet,\nand shut down the ``ThreadPoolExecutor``. The ``wait`` parameter is passed\nthrough to the call to ``ThreadPoolExecutor.shutdown()``.\n\nThe ``Interleaver`` can continue to be iterated over after calling\n``shutdown()`` and will yield any remaining values produced by the iterators\nbefore they stopped completely.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Yield from multiple iterators as values become available",
"version": "0.2.1",
"project_urls": {
"Bug Tracker": "https://github.com/jwodder/interleave/issues",
"Homepage": "https://github.com/jwodder/interleave",
"Source Code": "https://github.com/jwodder/interleave"
},
"split_keywords": [
"interleaving",
"iterators",
"multithreading"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "f8ff0d393596b592e911e57f9970faa799c54d70eed3b92c2064f4a19c94445f",
"md5": "5ed254fec93bda034f571d9ab94855de",
"sha256": "3345cedac12b8663e787cb2e0a80a3c1e450387b66b152ed32b951bb7c3d9e7e"
},
"downloads": -1,
"filename": "interleave-0.2.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "5ed254fec93bda034f571d9ab94855de",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "~=3.7",
"size": 11271,
"upload_time": "2022-07-02T22:48:16",
"upload_time_iso_8601": "2022-07-02T22:48:16.315873Z",
"url": "https://files.pythonhosted.org/packages/f8/ff/0d393596b592e911e57f9970faa799c54d70eed3b92c2064f4a19c94445f/interleave-0.2.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "8c801c45cd6dcd72c0669c3e11c74fbb1459cea5aa3d80960ffee47b66d1640d",
"md5": "45a4a68aeccac12d1d3e9cbc00560c9e",
"sha256": "210969c607828ecafc5e425532a6b6c10d09f1343e5976238ca15e8a07fb5118"
},
"downloads": -1,
"filename": "interleave-0.2.1.tar.gz",
"has_sig": false,
"md5_digest": "45a4a68aeccac12d1d3e9cbc00560c9e",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "~=3.7",
"size": 16058,
"upload_time": "2022-07-02T22:48:17",
"upload_time_iso_8601": "2022-07-02T22:48:17.989667Z",
"url": "https://files.pythonhosted.org/packages/8c/80/1c45cd6dcd72c0669c3e11c74fbb1459cea5aa3d80960ffee47b66d1640d/interleave-0.2.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2022-07-02 22:48:17",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "jwodder",
"github_project": "interleave",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"tox": true,
"lcname": "interleave"
}
JSON
