awaiting asyncio.Future swallows StopIteration

[ikelly]

BPO	26221
Nosy	@gvanrossum, @Rosuav, @1st1
Files	no_stop_iter.patch no_stop_iter.patch

^{Note: these values reflect the state of the issue at the time it was migrated and might not reflect the current state.}

Show more details

GitHub fields:

assignee = None
closed_at = <Date 2016-03-02.16:04:43.564>
created_at = <Date 2016-01-27.17:25:33.665>
labels = ['type-bug', 'expert-asyncio']
title = 'awaiting asyncio.Future swallows StopIteration'
updated_at = <Date 2016-03-02.16:04:43.542>
user = 'https://bugs.python.org/ikelly'

bugs.python.org fields:

activity = <Date 2016-03-02.16:04:43.542>
actor = 'yselivanov'
assignee = 'none'
closed = True
closed_date = <Date 2016-03-02.16:04:43.564>
closer = 'yselivanov'
components = ['asyncio']
creation = <Date 2016-01-27.17:25:33.665>
creator = 'ikelly'
dependencies = []
files = ['41980', '41986']
hgrepos = []
issue_num = 26221
keywords = ['patch']
message_count = 20.0
messages = ['259036', '259041', '259044', '259045', '259049', '259056', '259285', '260563', '260574', '260577', '260584', '260590', '260591', '260597', '260616', '260618', '260622', '260638', '261117', '261118']
nosy_count = 5.0
nosy_names = ['gvanrossum', 'ikelly', 'python-dev', 'Rosuav', 'yselivanov']
pr_nums = []
priority = 'normal'
resolution = 'fixed'
stage = 'resolved'
status = 'closed'
superseder = None
type = 'behavior'
url = 'https://bugs.python.org/issue26221'
versions = ['Python 3.5']

[ikelly]

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.

[gvanrossum]

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?)

[ikelly]

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

[gvanrossum]

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.

[ikelly]

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)

[gvanrossum]

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.)

[ikelly]

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."

[ikelly]

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.

[gvanrossum]

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?

[Rosuav]

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"?

[gvanrossum]

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().

[Rosuav]

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

[gvanrossum]

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