← 输入框|github.com|刷新
Skip to content

gh-83055: Prevent get/put item in asyncio.Queue when there are pending task or woken tasks - #160983

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

YvesDup wants to merge 2 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 here needs a new exception that would be replaced with QueueEmpty/QueueFull with a specific message

  • add 3 private attributes,
  • add new QueueWithPendingTasks exception,
  • update put_nowait, get_nowait methods, could raise this new exception,
  • update _wakeup_next method,
  • update __repr__,
  • add 2 unit tests,

+ 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 asyncio.QueueWithPendingTasks:
        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())

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