I was playing around with this class for adapting regular iterators to async iterators using BaseEventLoop.run_in_executor:


import asyncio

class AsyncIteratorWrapper:

    def __init__(self, iterable, loop=None, executor=None):
        self._iterator = iter(iterable)
        self._loop = loop or asyncio.get_event_loop()
        self._executor = executor

    async def __aiter__(self):
        return self

    async def __anext__(self):
        try:
            return await self._loop.run_in_executor(
                    self._executor, next, self._iterator)
        except StopIteration:
            raise StopAsyncIteration


Unfortunately this fails because when next raises StopIteration, run_in_executor swallows the exception and just returns None back to the coroutine, resulting in an infinite iterator of Nones.

What are you trying to do here? Can you post a simple example of an iterator that you would like to use with this? Without that it just raises my hackles -- it seems totally wrong to run an iterator in another thread. (Or is the iterator really a coroutine/future?)

The idea is that the wrapped iterator is something potentially blocking, like a database cursor that doesn't natively support asyncio. Usage would be something like this:


async def get_data():
    cursor.execute('select * from stuff')
    async for row in AsyncIteratorWrapper(cursor):
        process(row)


Investigating this further, I think the problem is actually in await, not run_in_executor:

>>> async def test():
...     fut = asyncio.Future()
...     fut.set_exception(StopIteration())
...     print(await fut)
... 
>>> loop.run_until_complete(test())
None

StopIteration has a special meaning. Don't use set_exception() with it.

You probably need a more roundabout way to do this.

Instead of submitting each __next__() call to the executor separately, you should submit something to the executor that pulls the items from the iterator and sticks them into a queue; then on the asyncio side you pull them out of the queue.

You can use an asyncio.Queue as the queue, and use loop.call_soon_threadsafe() to put things into that queue from the tread.

Fair enough. I think there should be some documentation though to the effect that coroutines aren't robust to passing StopIteration across coroutine boundaries. It's particularly surprising with PEP-492 coroutines, since those aren't even iterators and intuitively should ignore StopIteration like normal functions do.

As it happens, this variation (moving the try-except into the executor thread) does turn out to work but is probably best avoided for the same reason. I don't think it's obviously bad code though:


class AsyncIteratorWrapper:

    def __init__(self, iterable, loop=None, executor=None):
        self._iterator = iter(iterable)
        self._loop = loop or asyncio.get_event_loop()
        self._executor = executor

    async def __aiter__(self):
        return self

    async def __anext__(self):
        def _next(iterator):
            try:
                return next(iterator)
            except StopIteration:
                raise StopAsyncIteration
        return await self._loop.run_in_executor(
                self._executor, _next, self._iterator)

Can you suggest a sentence to insert into the docs and a place where
to insert it? (As you can imagine I'm pretty blind for such issues
myself.)

The place I'd expect to find it is in https://docs.python.org/3/library/asyncio-task.html#coroutines, in the list of "things a coroutine can do". The first two bullets in the list say that any exceptions raised will be propagated. Maybe there should be a note after the bullet list to the effect that "StopIteration carries special meaning to coroutines and will not be propagated if raised by an awaited coroutine or future."

Chris Angelico suggested on python-list that another possibly useful thing to do would be to add a "from __future__ import generator_stop" to asyncio/futures.py. This would at least have the effect of causing "await future" to raise a RuntimeError instead of silently returning None if a StopIteration is set on the future. Future.__iter__ is the only generator in the file, so this change shouldn't have any other effects.

Chris, can you help out here? I still don't understand the issue here. Since "from __future__ import generator_stop" only works in 3.5+ it would not work in Python 3.3/3.4 (supported by upstream asyncio with literally the same source code currently). If there's no use case for f.set_exception(StopIteration) maybe we should just complain about that?

Ultimately, it's the exact same thing that PEP 479 is meant to deal with - raising StopIteration is functionally identical to returning. I don't use asyncio enough to be certain, but I'm not aware of any good reason to inject a StopIteration into it; maybe an alternative solution is to add a check in set_exception "if isinstance(exception, StopIteration): raise DontBeAFool"?

OK, since eventually there won't be a way to inject StopIteration into
a Future anyway (it'll always raise RuntimeError once PEP 479 is the
default behavior) we should just reject this in set_exception().

POC patch, no tests. Is TypeError right? Should it be ValueError, since the notional type is "Exception"?

I think TypeError is fine. I would make the message a bit longer to
explain carefully what's the matter.

How about "StopException interacts badly with generators and cannot be raised into a Future"?

S.G.T.M.

On Sunday, February 21, 2016, Chris Angelico <report@bugs.python.org> wrote:

>
> Chris Angelico added the comment:
>
> How about "StopException interacts badly with generators and cannot be
> raised into a Future"?
>
> ----------
>
> _______________________________________
> Python tracker <report@bugs.python.org <javascript:;>>
> <http://bugs.python.org/issue26221>
> _______________________________________
>

Wording changed, and a simple test added. I'm currently seeing failures in test_site, but that probably means I've messed something up on my system.

Would you mind reworking this as a PR for github.com/python/asyncio ?
That's still considered "upstream" for asyncio.

--Guido

On Sun, Feb 21, 2016 at 8:17 AM, Chris Angelico <report@bugs.python.org> wrote:
>
> Chris Angelico added the comment:
>
> Wording changed, and a simple test added. I'm currently seeing failures in test_site, but that probably means I've messed something up on my system.
>
> ----------
> Added file: http://bugs.python.org/file41986/no_stop_iter.patch
>
> _______________________________________
> Python tracker <report@bugs.python.org>
> <http://bugs.python.org/issue26221>
> _______________________________________

Opened https://github.com/python/asyncio/pull/322

New changeset ef5265bc07bb by Yury Selivanov in branch '3.5':
asyncio: Prevent StopIteration from being thrown into a Future
https://hg.python.org/cpython/rev/ef5265bc07bb

New changeset 5e2f7e51af51 by Yury Selivanov in branch 'default':
Merge 3.5 (issue #26221)
https://hg.python.org/cpython/rev/5e2f7e51af51

Merged.