Skip to content

gh-83055: Prevent get/put[_nowait] item in asyncio.Queue when there are woken tasks - #160983

Draft
YvesDup wants to merge 3 commits into
python:mainfrom
YvesDup:queue-asyncio-keep-insertion-order-fix
Draft

YvesDup wants to merge 3 commits into
python:mainfrom
YvesDup:queue-asyncio-keep-insertion-order-fix

Conversation

@YvesDup

@YvesDup YvesDup commented Oct 30, 2026 •

Copy link
Copy Markdown
Contributor

The proposal fixes 2 main problems when tasks are woken just after the retrieval of an item, or after its insertion when queue is empty

  1. Just after the retrieval of an item, a new slot is free and a pending putter is awaken (case without pending putters is not concern). So this free slot that can be stole by put[_nowait] statement. Moreover this issue can break the FIFO's order of the queue.
    In this case, a QueueFull exception is raised with an explicit message

  2. Just after the insertion of an item in an empty queue with pending getters, one of the pending getter is awaken to get this fresh item at the next iteration. But this item can be get via a get[_nowait] statement before the awaken task.
    In this case, a QueueEmpty exception is raised with an explicit message.

+ add 3 private attributes,
+ add new "QueueWithPendingTasks" exception,
+ update put_nowait, getnowait methods, could raise this new exception,
+ update _wakeup_next method,
+ update __repr__,
+ add 2 unit tests,
@YvesDup YvesDup changed the title gh-83055: Prevent get/put item in asyncio.Queue when there are pending task or woken tasks gh-83055: Prevent get/put item in asyncio.Queue when there are pending task or woken tasks Oct 30, 2026
@YvesDup YvesDup changed the title gh-83055: Prevent get/put item in asyncio.Queue when there are pending task or woken tasks gh-83055: Prevent get/put item in asyncio.Queue when there are pending task or woken tasks Oct 30, 2026
@YvesDup

YvesDup commented Oct 30, 2026 •

Copy link
Copy Markdown
Contributor Author

@asvetlov : A draft PR with an another proposal.

This PR also fixes this issue for the following example:

async def simplier_get_case():
    async def get_and_print(q):
        item = await q.get()
        print(f"\tGET: {item}, {str(q) = }")
 
    q = asyncio.Queue(1)
    g = 3
    for n in range(g):
        asyncio.create_task(get_and_print(q))
    await asyncio.sleep(0.0)

    # queue full: 1 item in queue, 2 items waiting into the _putters
    n = 1
    print(f'PUT: {await q.put(n) or n}')
    n += 1
    # queue not empty: 1 item in queue, 2 tasks the _putters
 
    # steal the free solt
    try:
        print(f"\tGET_NOWAIT steal a free slot: {q.get_nowait()}")
    except:
        g -= 1
        print("Failed to GET_NOWAIT item".center(60, '-'))
  
    # get all items
    for _ in range(g):
        print(f'PUT: {await q.put(n) or n}')
        n += 1
    await asyncio.sleep(0.0)

asyncio.run(simplier_get_case())

@YvesDup YvesDup changed the title gh-83055: Prevent get/put item in asyncio.Queue when there are pending task or woken tasks gh-83055: Prevent get/put[_nowait] item in asyncio.Queue when there woken tasks Nov 1, 2026
@YvesDup YvesDup changed the title gh-83055: Prevent get/put[_nowait] item in asyncio.Queue when there woken tasks gh-83055: Prevent get/put[_nowait] item in asyncio.Queue when there are woken tasks Nov 1, 2026
+ in Queue class
  - remove ``QueueWithpendingTasks`` exception and replace with existing ``QueueFull`` and ``QueueEmpty`` with a specific message.
  - simplify new tests that raise these exceptions.
  - fix update to woken counters.
+ rewrite tests.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant