Sure, you can make it safe with:
from contextlib import aclosing
async with aclosing(run("python", "foo.py")) as items:
async for chunk, name in items: ...
So long as that aclosing() context manager is used, you should be fine. Async generators are otherwise generally unsafe IMHO.
Side note: it’s a bloody shame async generators can’t be natively used as async context managers, but oh well.
I thought about this further, and I recognize a problem: any iteration of async for chunk, name in items:could exit the task group, causing problems if you wedge another task group between the async with and async for. A safer alternative would therefore be:
import asyncio
from contextlib import asynccontextmanager
from subprocess import PIPE
from anyio import (
create_memory_object_stream,
open_process,
create_task_group,
IncompleteRead,
)
from anyio.streams.buffered import BufferedByteReceiveStream
async def read_stream(stream, name, sender):
async with BufferedByteReceiveStream(stream) as buffered, sender:
while True:
try:
chunk = await buffered.receive_until(b"\n", 65536)
except IncompleteRead:
if buffered.buffer:
await sender.send((buffered.buffer.decode("utf-8"), name))
break
else:
await sender.send((chunk.decode("utf-8"), name))
@asynccontextmanager
async def run(*cmd):
send, receive = create_memory_object_stream()
async with (
await open_process(cmd, stdout=PIPE, stderr=PIPE) as process,
create_task_group() as tg,
):
tg.start_soon(read_stream, process.stdout, "stdout", send.clone())
tg.start_soon(read_stream, process.stderr, "stderr", send)
yield receive
async def main():
async with run("python", "foo.py") as items:
async for chunk, name in items:
...
asyncio.run(main())
Ouch. This is now getting sufficiently complex that I’m no longer sure I feel comfortable maintaining it 
I don’t think that’s the fault of anyio, but rather of the whole async generator mess.
I think at this point, I’m in a position where I need some form of prepackaged solution. The two contenders seem to be the queue-based approach from @Tinche and the low-level implementation by @mikeshardmind. Neither is available in a released library yet, so I’d have to copy them into my code, which gives the edge to the second one as it’s more self-contained. But the fact that there’s no real consensus here on the best approach makes me feel like whatever I do involves some level of risk. Which sucks, TBH (speaking as a non-expert who just wants to use asyncio).
Luckily(?), my app is extremely niche, and there won’t be any serious consequences if a bug causes it to crash. So I guess I can live with the situation for now. But I’d certainly be reluctant to advocate for asyncio in a more critical application, when the level of asyncio expertise needed to write robust code is this high…
Ouch. This is now getting sufficiently complex that I’m no longer sure I feel comfortable maintaining it
Well, luckily the latest snippet I showed you only uses an async generator to make an async context manager which is a tried and true pattern. But yes, nobody said async programming was easy 
Agreed. But while the change needed is simple enough, it affects the code structure in a way I’ll need to adapt for my real application - and I don’t really understand why it’s needed, so there’s a high chance I’ll get something wrong when making that change.
Lol, yeah. The trouble is, everyone says async programming is a good idea, without warning you that it’s not easy!
(And to be fair, I started down this road because by existing thread-based code was messy and complicated, and async streams looked like a simpler approach. Looks like the complexity just moved, rather than disappearing!)
The general rule of thumb is to avoid async generators if you possibly can, as they’re riddled with gotchas. Do that, and use structured concurrency everywhere, and you likely won’t go wrong.
Also, I don’t imagine this being much easier with threads either.
I’ve got slightly more to think about on it, but I’m up for PRing an implementation to CPython. Since earlier discussion, there is now a release on pypi now that has an implementation, though there are two other modes of operation still not in the public API of that library yet, but the essence of that library is that it should be possible to lift anything from it for vendoring or upstreaming relatively easily.
There’s essentially 3 different sensible error handling behaviors for this. The one that’s public is the one that was initially refined as a result of this discussion, and the behavior is such that all further iteration is stopped across all merged generators, available values are yielded, and then encountered errors are raised.
In the private api, there are 2 other behaviors implemented that I’m likely exposing in the future.
The other implementations people have suggested which rely on task groups have some issues with what happens if they are cancelled.
Intentionally, none of the implementations I’ve written use async context managers or implicit cancellation tied to task group semantics, avoiding library side issues with each of those and having behavior that doesn’t rely on non-guaranteed behavior around cancellation order.
I don’t think this is a useful statement on the matter, and async generators are plenty useful. The only gotcha they have that results in unexpected behavior is an issue with composing them with context managers, and that’s an issue with those, not with async generators (specifically, certain uses of yield while in a context manager).
It’s my opinion that statements that conflate the actual places where things don’t work well and make highly opinionated statements about one right way to use it that aren’t backed entirely by technical reasons, especially when there are logical reasons to structure certain applications otherwise is part of why so many people end up writing off asyncio.
I have to admit I’m struggling to see how this advice works alongside the fact that (async) streams are a pretty basic and robust construct. After all, any stream can generally be used as an iterable. Or is it the distinction between iterators and generators that matters here? Because I’m more than happy to just merge two iterators to form an (async) iterator, if that helps.
It sucks if defining an async function containing yield is unsafe, and should be avoided, simply because technically that creates a generator rather than an iterator. In my experience, 99% of the time when people create a generator like this, all they actually want is an iterator.
Because I’m more than happy to just merge two iterators to form an (async) iterator, if that helps.
Isn’t that basically this is all about? You’ll still need tasks to forward values from them to the joined stream, whatever form that takes.
It sucks if defining an async function containing
yieldis unsafe, and should be avoided, simply because technically that creates a generator rather than an iterator. In my experience, 99% of the time when people create a generator like this, all they actually want is an iterator.
The essential considerations here are:
I believe my example is the only one given here that accomplishes all of this. But if correctness and robustness are not among your goals, then feel free to ignore everything I’ve said.
Async generators don’t play well with structured concurrency because generators are not part of the normal call stack. Suppose you create a task group in the generator, and a task in that group raises an exception, causing the task group to propagate the exception, where does that exception go? It could be that no task is consuming the generator when that happens. This is precisely the reason for PEP 789.
If you believe the one I linked doesn’t, I’d appreciate a specific reason why.
It’s interesting that you lean on 789, but yield from within a task group yourself, one of the behaviors that 789 says is broken currently. In fact, the cannonical example used to demonstrate this is combining iterators and the issue is the implementation of task groups is faulty and requires that yielding from them not happen. The solution 789 proposes is a way to block this, not to fix the implementation of the context manager.
The issue isn’t generators, it’s contextmanagers that were written with bad assumptions about control flow that treat a specific view of structured concurency as maximally correct, ignoring how people write correct and working code in other ways.
Composable iterators and generators are a feature of python, and the constant push for people to use task groups, even in places they aren’t appropriate due to shortcomings in their implementation to chase some ideal of structured concurrency isn’t helpful.
The implementation I linked properly raises errors, and doesn’t drop them on the floor so to speak. It doesn’t use task groups to do this, as they aren’t an appropriate tool for this.
At this point I believe we are hijacking the thread, but okay.
If you believe the one I linked doesn’t, I’d appreciate a specific reason why.
You’re using asyncio.gather(…, return_exceptions=True). If any of the generators being closed raises an exception during the aclose(), what happens to it? What if the task gets cancelled in the finally: block?
It’s interesting that you lean on 789, but yield from within a task group yourself, one of the behaviors that 789 says is broken currently.
Doing this from inside an @asynccontextmanager is one of the few tried-and-true patterns where this is not a problem, since the use as a context manager guarantees the proper closure of all related resources. This is in fact explicitly pointed out in the PEP itself (see combined_iterators in the example code).
The issue isn’t generators, it’s contextmanagers that were written with bad assumptions about control flow that treat a specific view of structured concurency as maximally correct, ignoring how people write correct and working code in other ways.
I’m interested to hear what alternative views of structured concurrency you’ve heard of that would better fit async generators. I’m genuinely curious. Writing correct code without SC is certainly possible, but quite error-prone. It’s like C vs Rust: You can certainly write code with no memory leaks or memory corruption errors in C, but even the best minds fail at this constantly. Therefore it makes sense to upgrade your tooling into something that categorically prevent multiple classes of mistakes. Concurrency is difficult overall, and SC makes it easier to reason about it too. But I understand not everybody is onboard, although you’ve provided mostly just accusations and little actual justification against it. Are you arguing against the idea of SC in general, or a particular Python implementation of it?
This is only for scheduling gen.aclose() prior to the user getting back control, rather than it happening implicitly by the asyncgen hooks for consistent errors if the user tries to resume a generator after cancellation. The entire block could be removed, and the same aclose would be called implicitly at some later point in time.
I don’t believe this is fully correct, as the pep continues on to how this interacts when you don’t control the other side of the generator, but I’m willing to concede this portion as it requires the other side to be doing something not quite right.
Stepping outside of it when neccessary, but only doing so when neccessary. Making more things act as implicit structure.
One of the issues people have with asyncio.create_task over task group use is that when the event loop closes, you may drop tasks. The current solution people have proposed involves passing a task group around, but there’s no reason why we couldn’t instead have .join() on the event loop, including with a timeout parameter that would cancel any tasks that didnt finish within that time, returning or raising errors that occured.
I’m not inherently against structured concurrency, I’m just willing to have a different view of acceptable structure. The implementation I wrote internally tracks it’s own tasks and ensures errors that would propogate had they been iterated by the user continue to do so, and is therefore a self-contained unit of structure.
FWIW, here’s my take on the async generator merger:
from collections.abc import AsyncGenerator, AsyncIterator
from contextlib import asynccontextmanager
from anyio import create_task_group, run, create_memory_object_stream
from anyio.abc import ObjectSendStream
@asynccontextmanager
async def merge[T](*gens: AsyncGenerator[T]) -> AsyncGenerator[AsyncIterator[T]]:
async def read_from_generator(
gen: AsyncGenerator[T], stream: ObjectSendStream[T]
) -> None:
async with stream:
async for item in gen:
await stream.send(item)
send, receive = create_memory_object_stream[T]()
async with create_task_group() as tg:
for gen in gens:
tg.start_soon(read_from_generator, gen, send.clone())
send.close()
yield receive
async def main() -> None:
async def generate_numbers(start: int):
for i in range(start, start + 10):
yield i
gen1 = generate_numbers(2)
gen2 = generate_numbers(7)
gen3 = generate_numbers(15)
async with merge(gen1, gen2, gen3) as numbers:
async for num in numbers:
print(num)
run(main)
FWIW, in my own application code, I essentially do this already with manual event loop management, because it’s not reasonable in my opinion to require all task spawning apis to need a task group passed in, it’s not ergonomic, and it’s worth a little extra work done myself, but I also do try to avoid writing library code that has to assume such.
One of the issues people have with asyncio.create_task over task group use is that when the event loop closes, you may drop tasks. The current solution people have proposed involves passing a task group around, but there’s no reason why we couldn’t instead have
.join()on the event loop, including with a timeout parameter that would cancel any tasks that didnt finish within that time, returning or raising errors that occured.
Do I understand correctly that you propose that any exceptions that occur will have to wait until the event loop is shut down to be discovered?
Let’s keep in mind the two main points of task groups:
I’m not inherently against structured concurrency, I’m just willing to have a different view of acceptable structure. The implementation I wrote internally tracks it’s own tasks and ensures errors that would propogate had they been iterated by the user continue to do so, and is therefore a self-contained unit of structure.
You may not realize it yourself, but I at least find your code much harder to understand, and the resources and tasks need to be managed manually rather than via convenient context managers. Figuring out how to do this, and correctly, can be pretty overwhelming for people who aren’t already experts in concurrency and async programming. Enabling people to write correct async code easier is my main goal in this space.
No, not at all. I actually think create_task is slightly flawed for use to launch background tasks as-is, but that people aren’t wrong for wanting background tasks that are scoped to the event loop without needing to pass a top level object around to have such a lifetime associated.
a dedicated asyncio.spawn_background_task that handled logging of exceptions via done callback paired with the ability to wait on tasks that aren’t finished when closing the event loop probably covers both aspects here.
That’s probably better served by one of the other ongoing discussions, but I figured it was a reasonable example of allowing people to structure code differently while still maintaining the benefits of structured concurrency.
Oh no, I agree completely. This isn’t simple to follow without a deeper understanding of both coroutines and the event loop than we typically expect of non-experts.
The original implementation used a queue and was simpler to follow, though this has some advantages in controlling the order in which errors are propogated and available values aren’t ever dropped by interruption due to a sibling generator raising an exception.
I don’t think most people should have to write these lower level more intricate pieces though, the goal should be to have enough composable pieces like this in the standard library such that user code gets the benefits of being able to compose existing constructs while not having to write code like this.
(Just responding to this one point for now, as I have to go and do something else).
Speaking as the person who started this thread, I will say that I don’t feel that this is hijacking the thread. I have seen 3 different suggestions so far for solutions, and my big remaining difficulty is that I don’t feel like I have enough knowledge to judge which is “right”, or even to just understand the tradeoffs.
Watching the people who developed the different solutions discuss why they made the choices they did, and why this is such a hard area of asyncio to get right, is very much on topic for me. It may leave me with the conclusion that “asyncio is a mess, and not something that a non-expert can use”, which would be a depressing conclusion to come to, but it would still be a valid resolution for this thread (for me, at least).
I’ve been watching this thread, and I feel like I have an okay understanding of async, though I also hesitate to claim expertise here, I wouldn’t have thought of a couple of the issues people have mentioned explicitly even if I landed on a safe implementation.
I think right now the best solution for user code is to not bother with abstracting this to a generic version and just write the composition you need.
@tinche, @mikeshardmind and @jamesdow21 all gave easy to follow implementations for inlining this that don’t take an expert I think.
It’s writing the generic abstraction where the complexity seems to kick in, and I think that’s okay? At least, I think it’s reasonable for user code to use the simplest version available when writing it themselves, but want for a library version that’s more capable.
It would take me a while to fully review this particular implementation, but trusting that any issues with it are ironed out, if you were to contribute this to asyncio, do you think it should be separate functions, or should controlling the error behavior be a parameter like other functions in asyncio?
I believe it’s better as separate functions in theory, but there’s an argument towards consistency in API design that’s reasonable to make here.
There are a few other possible changes to make here that I’d like to test before contributing this.
regarding the generator.aclose bit and why a gather that looks more suspicious on exception handling than it is in pactice is currently used, it would be nice if aclose was actually a sync function that set a cheap flag then returned a coroutine rather than being a coroutine function, but that’s a separate possible improvement on cleanup semantics.
Trio offers a free function, spawn_system_task(). If a system task crashes, it takes the whole event loop with it, and this is by design. Is this what you’re after? Otherwise, I don’t think I understand how the proposed spawn_background_task()differs from create_task(). The biggest problem with create_task()is that if the task crashes, it’s simply logged to stderr and life elsewhere in the event loop goes on, potentially making the app misbehave in odd ways. The biggest selling point in SC is that it forces uses to deal with errors explicitly…or else.