Flash vs flagship: Gemini 3.6 Flash / 3.5 Flash-Lite vs Opus 4.8 vs GPT-5.6 Terra (HN 48993414)

Identification of the Deadlock

The deadlock occurs because await asyncio.gather(*workers) is called inside the loop after putting a single item. gather() waits for all worker tasks to return/finish, but the workers run infinite while True loops waiting for items on the queue (await queue.get()). The main loop cannot enque/process the rest of the items because it is blocked waiting for workers to exit, and workers cannot exit because they are waiting for more items from the main loop.

Corrected Code

import asyncio

async def worker(queue, results):
    while True:
        item = await queue.get()
        results.append(item * 2)
        queue.task_done()

async def process_all(items, n_workers=4):
    queue = asyncio.Queue(maxsize=n_workers)
    results = []
    workers = [asyncio.create_task(worker(queue, results)) for _ in range(n_workers)]

    # Feed items into the queue (backpressure is enforced by queue maxsize)
    for item in items:
        await queue.put(item)

    # Wait for all enqueued items to be processed
    await queue.join()

    # Cleanly cancel and gather workers
    for w in workers:
        w.cancel()
    await asyncio.gather(*workers, return_exceptions=True)

    return results

Explanation of the Fix

• Removed gather from the feeding loop: Items are now pushed continuously into the queue without awaiting worker completion per item, letting await queue.put() handle bounded backpressure naturally via queue.maxsize.
• Synchronized completion with queue.join(): Replaced the incorrect per-item wait with await queue.join(), ensuring the main task waits until every enqueued item has been processed and marked as done with task_done().
• Clean worker shutdown: After all work is complete, worker tasks are explicitly cancelled and awaited with return_exceptions=True to prevent unhandled CancelledError warnings and guarantee leak-free resource cleanup.

The Deadlock

The deadlock occurs because await asyncio.gather(*workers) is placed inside the item-enqueuing loop. Since the workers run in an infinite while True: loop, gather will never complete. Consequently, the main coroutine suspends indefinitely on the very first iteration, preventing any subsequent items from being queued and leaving the queue with insufficient items to unblock the workers.

Corrected Version

import asyncio

async def worker(queue, results):
    while True:
        item = await queue.get()
        try:
            results.append(item * 2)
        finally:
            queue.task_done()

async def process_all(items, n_workers=4):
    queue = asyncio.Queue(maxsize=n_workers)
    results = []
    workers = [asyncio.create_task(worker(queue, results)) for _ in range(n_workers)]
    
    for item in items:
        await queue.put(item)  # Bounded backpressure via maxsize
        
    await queue.join()         # Wait until all items are processed
    
    for w in workers:
        w.cancel()
    
    # Suppress CancelledError noise from shutting down idle workers
    await asyncio.gather(*workers, return_exceptions=True)
    return results

Explanation of the Fix

• Removed Blocking Gather: Deleted await asyncio.gather(*workers) from inside the item loop, allowing items to be enqueued continuously while workers process them concurrently.
• Proper Backpressure & Synchronization: Relied on await queue.put(item) for bounded backpressure (blocking when the queue reaches n_workers) and await queue.join() to wait until all items are fully consumed.
• Clean Termination: Moved await asyncio.gather(*workers, return_exceptions=True) to the end after cancellation to safely reap the infinite-loop workers without leaving dangling tasks or raising unhandled CancelledError exceptions.

The Deadlock

The killer line is:

await queue.put(item)
await asyncio.gather(*workers)   # <-- deadlock

gather(*workers) waits for the worker tasks to complete. But the workers run while True: — they never finish. So after enqueuing the very first item, process_all blocks forever waiting on workers that are designed to run indefinitely.

(Secondary latent bug: even if you fixed that, the workers only ever queue.get(); there's no sentinel/cancellation-driven exit path baked into the loop, so queue.join() + cancel() is the only way out — which is fine, but the gather makes it moot.)

Corrected Version

import asyncio

async def worker(queue, results):
    while True:
        item = await queue.get()
        try:
            results.append(item * 2)
        finally:
            queue.task_done()   # always mark done, even on error

async def process_all(items, n_workers=4):
    queue = asyncio.Queue(maxsize=n_workers)
    results = []
    workers = [asyncio.create_task(worker(queue, results)) for _ in range(n_workers)]

    # Producer: put() blocks when the queue is full -> bounded backpressure
    for item in items:
        await queue.put(item)

    # Wait until every enqueued item has been marked done
    await queue.join()

    # Now it's safe to tear down the idle workers
    for w in workers:
        w.cancel()
    await asyncio.gather(*workers, return_exceptions=True)  # await cancellation cleanly

    return results

The Fix in 3 Bullets

• Removed await asyncio.gather(*workers) from the enqueue loop. Those workers never return, so gathering them mid-loop blocks forever. Backpressure is instead provided naturally by await queue.put(item), which suspends the producer whenever the bounded queue (maxsize=n_workers) is full.
• Used await queue.join() to detect completion. join() returns only when every put() has a matching task_done(), guaranteeing all items are processed before teardown — no premature exit, no lost work.
• Cancelled workers and then awaited them with return_exceptions=True. Cancelling schedules the CancelledError; awaiting the gather lets each worker actually unwind so you terminate cleanly without "task was destroyed but it is pending" warnings. (Also wrapped task_done() in finally so a bad item can't stall join().)

import asyncio

_STOP = object()

async def worker(queue, results):
    while True:
        item = await queue.get()
        try:
            if item is _STOP:
                return

            results.append(item * 2)
        finally:
            queue.task_done()

async def process_all(items, n_workers=4):
    queue = asyncio.Queue(maxsize=n_workers)
    results = []

    workers = [
        asyncio.create_task(worker(queue, results))
        for _ in range(n_workers)
    ]

    # Bounded backpressure: put() waits when the queue is full.
    for item in items:
        await queue.put(item)

    # Tell every worker to exit after all normal queued work.
    for _ in workers:
        await queue.put(_STOP)

    # Wait until both work items and stop markers have been acknowledged.
    await queue.join()

    # Workers exit naturally after consuming their stop markers.
    await asyncio.gather(*workers)

    return results

• await asyncio.gather(*workers) inside the producer loop never completes: workers are infinite loops, so after processing the first item they wait for more work instead of finishing.
• Remove the per-item gather; Queue(maxsize=n_workers) already provides bounded backpressure because queue.put() blocks whenever the queue is full.
• Send one sentinel per worker, have each worker call task_done() in a finally block, then queue.join() and gather() so every queued item is processed and every worker exits cleanly.