asyncio.TaskGroup when one fetch fails

Tried your second stop with an empty weight fetch, curious which part would complain first.

async def fetch_weights():
    return []

async def main():
    async with asyncio.TaskGroup() as tg:
        m = tg.create_task(fetch_measurements())
        w = tg.create_task(fetch_weights())
    xs, ws = m.result(), w.result()
    return sum(x * y for x, y in zip(xs, ws)) / sum(ws)

The group exits cleanly. No cancellation, nothing propagates. Then ZeroDivisionError comes out of the last line, after the group is gone, so the traceback points at the arithmetic and says nothing about which fetch came back empty. I'd half expected Julia to fail the same way, and it doesn't. sum(w .* x) / sum(w) over empty Float64 vectors is 0.0 / 0.0, which is NaN, and the score just carries on.

Two empty collections might be a legitimate empty calculation or a required input that went missing. Nothing in the task group can tell those apart, so what empty means has to be decided per input, before the division.

Two fetches. That's the whole population, so a worker pool here would be ceremony.

The part worth tracing is the fetch that's still running. Its sibling raises, the group cancels it, and then the group waits. Cancelling is a request. The fetch might be halfway through a socket read or a cleanup block when it lands, and it's done only when the task actually finishes. Go has the same shape with errgroup.WithContext. The context gets cancelled on the first error, and Wait still blocks until every goroutine returns.

So the requirement I care about is one line. The combined request doesn't leave its scope while any member is still running. If somebody wants the measurements that did arrive, write down whether they can be used after the group has reported failure.

The word is shared and the handoff isn't. Before the comparison gets a row per language, write down who is in charge of waiting in each one. In the Python version it's the async with block. In a Lua version it would be whatever loop the host wrote, and that loop can just as easily forget a sibling or never cancel it.

So check each language's facility against what TaskGroup does in your calculation, waiting for its members and cancelling on failure, and don't let the shared word carry that over.

This might be a good spot for an F# union. Ready weights and Unavailable as their own cases, and an empty collection is free to mean valid empty input.

I'd push back a little for two fetches. The first exception ends the request anyway, and asyncio.gather already gives you that without a new block to learn. TaskGroup makes sense once there are a dozen tasks. For two, it's one more thing a beginner has to read past.

Not the same result. gather hands you the first exception and leaves the other fetch running.

async def main():
    slow = asyncio.create_task(fetch_measurements())
    try:
        await asyncio.gather(slow, fetch_weights())
    except ValueError:
        print("measurements done?", slow.done())

Prints measurements done? False. Nobody cancels it and nobody waits for it.

Okay, the orphan is real, thanks for showing it. Cancelling it by hand is two lines, though, and for a beginner I still find two visible lines easier to follow than a new block.

Depends on timing. The measurements that arrived can be read after the failure, but only some of the time, and the code has to check which time it is.

try:
    async with asyncio.TaskGroup() as tg:
        m = tg.create_task(fetch_measurements())
        w = tg.create_task(fetch_weights())
except* ValueError:
    if m.done() and not m.cancelled():
        print("kept", m.result())
    else:
        print("measurements were cancelled")

With the measurement fetch finishing before the weight fetch fails, this prints kept [1.0, 2.0]. Make the measurement fetch slower than the failure and the same code prints measurements were cancelled, because the group cancelled it on the way out. Both are legal outcomes of one program. except* needs 3.11, same as the group.

So yes, they can be used, if they finished first. Whether they should be is the part I'm less sure of. A score built from whichever half happened to win the race seems hard to explain to anyone reading it later.

That race is my objection. If the answer depends on which fetch won, it isn't a result I'd hand to a caller.

Max's line about cancellation being a request is the one that matters on the failure path. The cleanup I worried about in the third stop is fine. It catches Exception, and CancelledError has been a BaseException since 3.8, so it goes straight past. The retry helper around the weight fetch is not.

async def fetch_weights():
    for attempt in range(3):
        try:
            return await get_weights()
        except BaseException:
            print("retrying after", attempt)

When the measurement fetch raises early, the group cancels the weight fetch, the helper catches the cancellation, prints retrying after 0 and starts a fresh attempt. The group then waits for that attempt to finish before raising, so the failed request takes the full length of a second weight fetch to report anything. Nothing hangs, which is almost worse, because it only looks slow.

Catching asyncio.CancelledError and re-raising it ahead of the broad handler removes the delay. I haven't decided whether a retry helper should catch anything that broad at all.

For the comparison with other languages this goes on the list right next to cancellation. A Go version with errgroup has the same wait, since Wait returns only after every goroutine has, and a retry loop that ignores the context would stretch it the same way.

Retry helpers that catch BaseException are the bug. Re-raise cancellation first, retry the rest.

Okay, real question, and max8's line about cancellation being a request is what set it off! My weight source is a C extension call that blocks, so fetch_weights is just await asyncio.to_thread(load_weights). When the measurement fetch fails and the group cancels the weight task, what happens to that thread? Does the C call get interrupted somehow? Or does the group sit there until it returns on its own? I was hoping for the first, well, almost hoping, because a C call cut off in the middle of its own cleanup sounds worse. I just can't tell which one the group is waiting on, the task or the thread.

Thread keeps running. Cancel hits the await, not the thread, and Python has no way to stop a C call from outside. Task ends right away, group exits, the call finishes in the background and its result goes nowhere.

vik_o has the group right, and I checked the part after it, since the program still has to finish somewhere. With load_weights sleeping for a second and the measurement fetch failing early, I got this order.

import asyncio, time

def load_weights():
    time.sleep(1.0)
    print("thread finished")
    return [0.5, 0.5]

async def fetch_weights():
    return await asyncio.to_thread(load_weights)

The except* ValueError around the group ran after about a tenth of a second. thread finished printed later, and asyncio.run returned only after the full second, because on its way out it shuts down the default executor and waits for calls still running there. So the group's guarantee covers its tasks and stops there. The thread is outside it.

For the score this matters at one stage. Anything load_weights does besides returning, writing a cache file for instance, still happens after the request has been reported as failed.

Small tangent, because the orphaned thread made me wonder what else it keeps hold of. It keeps a worker in the default executor. to_thread runs on that pool, and the pool has a fixed size. I shrank it to one worker so the effect would be hard to miss.

import asyncio, time
from concurrent.futures import ThreadPoolExecutor

def slow_c_call():
    time.sleep(1.0)

async def main():
    loop = asyncio.get_running_loop()
    loop.set_default_executor(ThreadPoolExecutor(max_workers=1))
    t = asyncio.create_task(asyncio.to_thread(slow_c_call))
    await asyncio.sleep(0.1)
    t.cancel()
    t0 = time.monotonic()
    await asyncio.to_thread(print, "next call runs")
    print(f"waited {time.monotonic() - t0:.1f}s")

asyncio.run(main())

With slow_c_call sleeping for a second, it prints waited 0.9s. The cancelled task is long gone, and the next to_thread still queues behind its thread. With the default pool size it takes several failed requests in quick succession before anything visibly slows down, which probably makes it harder to spot rather than easier.