gh-119154: Simplify consumers by adding `.Queue.iter()` and `.Queue.iter_nowait()` by nineteendo · Pull Request #120925 · python/cpython · GitHub

Purpose

Currently consuming items from a queue can be a bit more complex than necessary. You need to

use an infinite while loop
check for a sentinel or add all worker tasks to a list
call q.task_done() after processing each item
join the queue
cancel all worker tasks
wait until all worker tasks are cancelled

By adding *.Queue.iter() and *.Queue.iter_nowait(), this becomes a bit easier, you only need to call queue.shutdown().

Overview of changes

asyncio

Calling queue.Queue.iter() returns an asynchronous generator which iterates over the queue of items
Calling queue.Queue.iter_nowait() returns a generator which iterates over the queue of items without blocking.
A private _AsyncQueueIterator has been added to handle the asynchronous iteration
The example in the documentation is greatly simplified by the new addition

queue.py

Calling queue.Queue.iter() returns a generator which iterates over the queue of items
Calling queue.Queue.iter_nowait() returns a generator which iterates over the queue of items without blocking.
An example has been added in the documentation which is a bit simpler than working with daemon threads

Examples

asyncio

Without iteration

import asyncio

async def producer(queue):
    for i in range(5):
        print(f'{i} -> queue')
        await queue.put(i)
        await asyncio.sleep(.1)

async def consumer(queue):
    while True:
        i = await queue.get()
        print(f'queue -> {i}')
        await asyncio.sleep(.3)
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    tasks = []
    for _ in range(1):
        task = asyncio.create_task(consumer(queue))
        tasks.append(task)

    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(queue))

    print('produced everything')
    await queue.join()
    for task in tasks:
        task.cancel()

    await asyncio.gather(*tasks, return_exceptions=True)
    print('consumed everything')

asyncio.run(main())

With iteration

import asyncio

async def producer(queue):
    for i in range(5):
        print(f'{i} -> queue')
        await queue.put(i)
        await asyncio.sleep(.1)

async def consumer(queue):
    async for i in queue.iter():
        print(f'queue -> {i}')
        await asyncio.sleep(.3)

async def main():
    queue = asyncio.Queue()
    async with asyncio.TaskGroup() as tg1:
        tg1.create_task(consumer(queue))
        async with asyncio.TaskGroup() as tg2:
            tg2.create_task(producer(queue))

        queue.shutdown()
        print('produced everything')

    print('consumed everything')

asyncio.run(main())

Output

0 -> queue
queue -> 0
1 -> queue
2 -> queue
queue -> 1
3 -> queue
4 -> queue
produced everything
queue -> 2
queue -> 3
queue -> 4
consumed everything

queue as final step

Without iteration

import threading
import queue

def worker(q):
    while True:
        item = q.get()
        print(f'queue -> {item}')
        q.task_done()

q = queue.Queue()
threading.Thread(target=worker, args=(q,) daemon=True).start()
for item in range(5):
    q.put(item)
    print(f'{item} -> queue')

print('produced everything')
q.join()
print('consumed everything') # Worker is still running

With iteration

import concurrent.futures
import queue

def worker(q):
    for item in q.iter():
        print(f'queue -> {item}')

q = queue.Queue()
with concurrent.futures.ThreadPoolExecutor() as tp:
    tp.submit(worker, q)
    for item in range(5):
        q.put(item)
        print(f'{item} -> queue')

    q.shutdown()
    print('produced everything')

print('consumed everything') # No worker is running

Output

0 -> queue
queue -> 0
1 -> queue
queue -> 1
queue -> 2
2 -> queue
3 -> queue
4 -> queue
produced everything
queue -> 3
queue -> 4
consumed everything

queue not as final step

Without iteration

import concurrent.futures
import queue
import time

def producer(q):
    for i in range(5):
        print(f'{i} -> queue')
        q.put(i)
        time.sleep(.1)

def consumer(q):
    while True:
        i = q.get()
        if i is None:
            break

        print(f'queue -> {i}')
        time.sleep(.3)
        q.task_done()

q = queue.Queue()
with concurrent.futures.ThreadPoolExecutor() as tp1:
    tp1.submit(consumer, q)
    with concurrent.futures.ThreadPoolExecutor() as tp2:
        tp2.submit(producer, q)

    print('produced everything')
    q.join()
    q.put(None)

print('consumed everything')

With iteration

import concurrent.futures
import queue
import time

def producer(q):
    for i in range(5):
        print(f'{i} -> queue')
        q.put(i)
        time.sleep(.1)

def consumer(q):
    for i in q.iter():
        print(f'queue -> {i}')
        time.sleep(.3)

q = queue.Queue()
with concurrent.futures.ThreadPoolExecutor() as tp1:
    tp1.submit(consumer, q)
    with concurrent.futures.ThreadPoolExecutor() as tp2:
        tp2.submit(producer, q)

    q.shutdown()
    print('produced everything')

print('consumed everything')

Output

0 -> queue
queue -> 0
1 -> queue
2 -> queue
queue -> 1
3 -> queue
4 -> queue
produced everything
queue -> 2
queue -> 3
queue -> 4
consumed everything

Issue: Add asyncio.Queue.__aiter__ #119154

📚 Documentation preview 📚: library/queue.html library/asyncio-queue.html

Co-authored-by: Guido van Rossum <guido@python.org>

Co-authored-by: Zac Hatfield-Dodds <zac.hatfield.dodds@gmail.com>

…AiErS.rst

pythongh-120924: Simplify consumers by adding `*.Queue.iter()` and `*.Queue.iter_nowait()`

Sorry, I didn't intend to request a review. I merged the pull requests to leave it at just 2.

I’m sorry, I am currently on vacation and dealing with health issues. I will not be a available to review this.

…9-35-14.gh-issue-119154.7CXFaX.rst

Note, I (with Erlend Aasland concurring) marked #120503 as rejected 2 weeks ago. Rather than accept the result from the Queue module maintainer, the OP immediately opened this new PR with substantially the same proposal and code (plus other additions). No mention was made of the prior rejection or reasons for it. This seems deceptive and inappropriate.

I first asked whether this would be a better API: #120503 (comment). And clarified how Empty exceptions would be handled: #120503 (comment). You haven't given your opinion yet.

I'll attempt tot address the previous rejection.

The code is too trivial and there is almost no benefit (the edit to the example saves only one line).1

By supporting this by default we "support such design patterns without subclassing or a generator helper function"2.
You're no longer forced to use daemon threads or sentinels (in which case more than 1 line is saved). Having daemon threads spinning in the background when all work is done is undesirable.

With shutdown being new, it is better to keep it explicit.1

If you want, we can delay this change until Python 3.15.

Currently the get() is an obvious marker of where blocking occurs. Code will be harder to review with it hidden behind an abstraction.1

Is it clearer with q.iter() and q.iter_nowait()? That's should be as clear as get() and get_nowait().

Current the get provides non-blocking and timeout options which become inaccessible with this.1
It is perfectly reasonable to wait to see how people use shutdown() and whether they typically need a timeout which would be inaccessible in this abstraction.3

q.iter() provides the same options now.

The queue API is foundational and everything done there propagates to other concurrency APIs. Ideally, this API should be as compact, orthogonal, and non-opinionated as possible.1

We haven't done that with Shutdown either. This simply makes it more powerful.

I disagree with the premise, "Currently consuming items from a queue is very complex." Most consumers are simple. They only add task_done if they need to track and join on task completion. The PR does very little, only combining a while/get step with a try/except that isn't always needed.1

According to my GitHub search, only a minority of 24.5% are "simple" consumers:4

116k files create a queue: /(?-i)(?:from|import) queue/ AND /(?-i)Queue\(\)/
28.5k files also use daemon threads: /(?-i)(?:from|import) queue/ AND /(?-i)Queue\(\)/ AND /daemon ?= ?True/

"performance is not the only golden rule for CPython. Code readability is critical for maintenance"5

Minor respellings typically don't benefit users. In this case, I think it makes them worse-off being hiding two essential elements that should be explicit (the get call and catching the exception).1

q.iter() makes this a bit more explicit. I think which exceptions are caught should make sense.

queue.Queue.iter(block=True, timeout=None)
What would this do with an Empty exception? Would it eat the exception and terminate the for-loop treating it exactly the same as a Shutdown? This seems like bug bait.6

An error is raised if it's the first element. "This ensures the queue is not empty (without risking race conditions). That seems like the only sensible behaviour as queues aren't infinite (why would you otherwise want to iterate over them without waiting).7

def iter(self, block=True, timeout=None):
    try:
        yield self.get(block=block, timeout=timeout)
    except ShutDown:
        return

    try:
        while True:
            yield self.get(block=block, timeout=timeout)
    except (Empty, ShutDown):
        return

Also, it distances the user from adding any other behavior in the except-clause, possibly making a log entry or finalizing a resource.6

You can do that after the for loop:8

for ... in q.iter():
    # do something

cleanup(q)

And as noted above it, distances the user from adding timeouts and non-blocking behaviors which typically need special handling distinct from the case of a shutdown.6

In that case, create a subclass.8

I also don't see how this would work well with code that wants to join() a queue where a worker needs to call task_done().6

It's safer to call q.task_done() explicitly, see the discussion here: #119154.9

In short, users that want to do anything other than one specific case are made worse-off by trying to use the for-loop variant. It won't even be obvious that they have options.6

I think that in most cases a subclass won't be necessary.

We discussed this at the sprint today and agreed this (or any variation of it) should not be done. Multiple reasons were given.

The most important was that this is only one of the multitude of reasonable ways to use existing queue API: the new shutdown exception, previously existing custom exceptions, waiting for timeouts, multiple producers, multiple consumers, having a second queue for two-way communication, etc.

Another reason was that a senior concurrency developer had become averse to trend widening existing APIs to support the various dunder method protocols (iterator, context manager etc).

Other devs agreed that a for-loop front-end would hide (make implicit) details that we really want to be explicit for code review (i.e. when blocking happens and what exception will be caught).

This senior concurrency dev suggested that I write a small edit to the docs showing a user could easily write a wrapper like this. I'll do that in a separate PR.

Does this decision apply to #120491 as well? In that case both the issue and PR should be closed.

nineteendo and others added 30 commits June 14, 2024 11:31

Add asyncio.Queue.__aiter__()

e9c1a3e

Co-authored-by: Guido van Rossum <guido@python.org>

Reduce diff

3cd59ee

Add test

16fc429

Simplify code

f13ddb3

Add to __all__

afd7ad4

📜🤖 Added by blurb_it.

214548c

Simplify example

9824b63

Fix indentation

6358b8e

Improve order

316b800

Test StopAsyncIteration

92f3295

Document AsyncQueueIterator

a18767d

Add queue.Queue.__iter__()

e826a01

Add asyncio.Queue.__iter__()

c8cbe4d

📜🤖 Added by blurb_it.

940e3c0

Fix lint

8fca288

Wait

499457c

Update 2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst

7f206d9

Apply suggestions from code review

320691c

Co-authored-by: Zac Hatfield-Dodds <zac.hatfield.dodds@gmail.com>

Unpack for loop

f43bdb1

Insert newline

4ca4a93

Simplify example

fffacb2

Indent correctly

92d7b16

Document __aiter__()

ce23fce

Fix indent attempt 1

640d2a8

Move while loop inside try/except block

3cad181

test iteration

cc9f7ef

Reduce diff

92d7dc0

Run test

a92e031

Reduce diff

3621cc1

Adapt example from asyncio

98252c6

nineteendo added 2 commits June 24, 2024 09:18

Delete Misc/NEWS.d/next/Library/2024-06-23-20-31-59.gh-issue-120924.2…

461d1e1

…AiErS.rst

Merge pull request #16 from nineteendo/add-asyncio.Queue.iter

48783bb

pythongh-120924: Simplify consumers by adding `*.Queue.iter()` and `*.Queue.iter_nowait()`

nineteendo requested review from 1st1, asvetlov, gvanrossum, kumaraditya303 and willingc as code owners June 24, 2024 07:18

nineteendo changed the title gh-120924: Simplify consumers by adding queue.Queue.iter() and queue.Queue.iter_nowait() gh-120924: Simplify consumers by adding *.Queue.iter() and *.Queue.iter_nowait() Jun 24, 2024

Update 2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst

4d7e55b

nineteendo marked this pull request as draft June 24, 2024 07:35

bedevere-app Bot removed the awaiting review label Jun 24, 2024

Remove parentheses

888ae9e

gvanrossum removed their request for review June 24, 2024 08:16

Link to github issue

a413d13

nineteendo changed the title gh-120924: Simplify consumers by adding *.Queue.iter() and *.Queue.iter_nowait() gh-120947: Simplify consumers by adding *.Queue.iter() and *.Queue.iter_nowait() Jun 24, 2024

bedevere-app Bot mentioned this pull request Jun 24, 2024

Simplify consumers with iteration over *.Queue #120947

Closed

nineteendo changed the title gh-120947: Simplify consumers by adding *.Queue.iter() and *.Queue.iter_nowait() gh-119154: Simplify consumers by adding *.Queue.iter() and *.Queue.iter_nowait() Jun 24, 2024

bedevere-app Bot mentioned this pull request Jun 24, 2024

Add asyncio.Queue.__aiter__ #119154

Closed

nineteendo added 2 commits June 25, 2024 10:22

Rename 2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst to 2024-06-23-1…

dcdd17e

…9-35-14.gh-issue-119154.7CXFaX.rst

Update issue number

525effe

rhettinger added the sprint label Sep 24, 2024

rhettinger closed this Sep 24, 2024

nineteendo deleted the add-Queue.iter branch September 24, 2024 07:08

gh-119154: Simplify consumers by adding `.Queue.iter()` and `.Queue.iter_nowait()` by nineteendo · Pull Request #120925 · python/cpython · GitHub

Repository navigation

gh-119154: Simplify consumers by adding .Queue.iter() and .Queue.iter_nowait() - #120925

gh-119154: Simplify consumers by adding .Queue.iter() and .Queue.iter_nowait()#120925
nineteendo wants to merge 90 commits into
python:mainfrom
nineteendo:add-Queue.iter

nineteendo commented Jun 23, 2024 •
edited
Loading

Uh oh!

Uh oh!

nineteendo commented Jun 24, 2024

Uh oh!

gvanrossum commented Jun 24, 2024

Uh oh!

rhettinger commented Jul 5, 2024 •
edited
Loading

Uh oh!

Uh oh!

nineteendo commented Jul 6, 2024 •
edited
Loading

Uh oh!

Uh oh!

nineteendo commented Jul 6, 2024

Uh oh!

rhettinger commented Sep 24, 2024 •
edited
Loading

Uh oh!

Uh oh!

nineteendo commented Sep 24, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

gh-119154: Simplify consumers by adding `*.Queue.iter()` and `*.Queue.iter_nowait()` by nineteendo · Pull Request #120925 · python/cpython · GitHub

Repository navigation

gh-119154: Simplify consumers by adding *.Queue.iter() and *.Queue.iter_nowait() - #120925

gh-119154: Simplify consumers by adding *.Queue.iter() and *.Queue.iter_nowait()#120925nineteendo wants to merge 90 commits intopython:mainpython/cpython:mainfrom nineteendo:add-Queue.iternineteendo/cpython:add-Queue.iterCopy head branch name to clipboard

Conversation

nineteendo commented Jun 23, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Overview of changes

asyncio

queue.py

Examples

asyncio

Without iteration

With iteration

Output

queue as final step

Without iteration

With iteration

Output

queue not as final step

Without iteration

With iteration

Output

Uh oh!

nineteendo commented Jun 24, 2024

Uh oh!

gvanrossum commented Jun 24, 2024

Uh oh!

rhettinger commented Jul 5, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

nineteendo commented Jul 6, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

nineteendo commented Jul 6, 2024

Footnotes

Uh oh!

rhettinger commented Sep 24, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

nineteendo commented Sep 24, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

gh-119154: Simplify consumers by adding `.Queue.iter()` and `.Queue.iter_nowait()` by nineteendo · Pull Request #120925 · python/cpython · GitHub

gh-119154: Simplify consumers by adding .Queue.iter() and .Queue.iter_nowait() - #120925

gh-119154: Simplify consumers by adding .Queue.iter() and .Queue.iter_nowait()#120925
nineteendo wants to merge 90 commits into
python:mainfrom
nineteendo:add-Queue.iter

nineteendo commented Jun 23, 2024 •
edited
Loading

rhettinger commented Jul 5, 2024 •
edited
Loading

nineteendo commented Jul 6, 2024 •
edited
Loading

rhettinger commented Sep 24, 2024 •
edited
Loading