Add a shutdown method to threading and asyncio Queue classes

@asvetlov

BPO	37334
Nosy	@asvetlov, @cjrh, @1st1, @tecki, @eamanu
PRs	bpo-37334: Add a cancel method to asyncio Queues #14227

^{Note: these values reflect the state of the issue at the time it was migrated and might not reflect the current state.}

Show more details

GitHub fields:

assignee = None
closed_at = None
created_at = <Date 2019-06-19.07:41:48.477>
labels = ['3.11', 'type-feature', 'expert-asyncio']
title = 'Add a cancel method to asyncio Queues'
updated_at = <Date 2021-05-12.12:06:13.863>
user = 'https://github.com/tecki'

bugs.python.org fields:

activity = <Date 2021-05-12.12:06:13.863>
actor = 'Martin.Teichmann'
assignee = 'none'
closed = False
closed_date = None
closer = None
components = ['asyncio']
creation = <Date 2019-06-19.07:41:48.477>
creator = 'Martin.Teichmann'
dependencies = []
files = []
hgrepos = []
issue_num = 37334
keywords = ['patch']
message_count = 16.0
messages = ['346023', '346026', '346113', '346268', '346276', '346391', '357204', '357211', '357270', '357283', '357303', '357305', '357307', '357376', '357381', '393505']
nosy_count = 5.0
nosy_names = ['asvetlov', 'cjrh', 'yselivanov', 'Martin.Teichmann', 'eamanu']
pr_nums = ['14227']
priority = 'normal'
resolution = None
stage = 'patch review'
status = 'open'
superseder = None
type = 'enhancement'
url = 'https://bugs.python.org/issue37334'
versions = ['Python 3.11']

When working with queues, it is not uncommon that at some point the producer stops producing data for good, or the consumer stops consuming, for example because a network connection broke down or some user simply closed the session.

In this situation it is very useful to simply cancel all the waiting getters and putters. A simple method can do that, Queue.cancel.

Sounds like .close() is better name for described behavior.

I also thought about .close() but then found .cancel() more intuitive. But intuition is not universal, so I am open to any wording.

I'm interested in how this change would affect the pattern of shutting down a queue-processing task.

How would one decide between whether to cancel the queue or the task? (I'm asking for real, this is not an objection to the PR). For example, looking at the two tests in the PR:

    def test_cancel_get(self):
        queue = asyncio.Queue(loop=self.loop)

        getter = self.loop.create_task(queue.get())
        test_utils.run_briefly(self.loop)
        queue.cancel()                       # <---- HERE
        test_utils.run_briefly(self.loop)
        with self.assertRaises(asyncio.CancelledError):
            self.loop.run_until_complete(getter)

This test would work exactly the same if the getter task was cancelled instead right? Like this:

    def test_cancel_get(self):
        queue = asyncio.Queue(loop=self.loop)

        getter = self.loop.create_task(queue.get())
        test_utils.run_briefly(self.loop)
        getter.cancel()                       # <---- HERE
        test_utils.run_briefly(self.loop)
        with self.assertRaises(asyncio.CancelledError):
            self.loop.run_until_complete(getter)

So my initial reaction is that I'm not sure under what conditions it would be more useful to cancel the queue instead of the task. I am very used to applying cancellation to tasks rather than the queues they contain, so I might lack imagination in this area. The idiom I've been using so far for consuming queues looks roughly something like this:

async def consumer(q: asyncio.Queue):
    while True:
        try:
            data = await q.get()
        except asyncio.CancelledError:
            q.put_nowait(None) # ignore QueueFull for this discussion
            continue

    try:
        if not data:
            logging.info('Queue shut down cleanly')
            return     # <------ The only way to leave the coro
        <process data>
    except Exception:
        logging.exception('Unexpected exception:')
        continue
    finally:
        q.task_done()

^^ With this pattern, I can shut down the consumer task either by cancelling the task (internally it'll put a None on the queue) or by placing a None on the queue outright from anywhere else. The key point is that in either case, existing items on the queue will still get processed before the None is consumed, terminating the task from the inside.

(A) If the queue itself is cancelled (as in the proposed PR), would it still be possible to catch the CancelledError and continue processing whatever items have already been placed onto the queue? (and in this case, I think I'd still need to place a sentinel onto the queue to mark the "end"...is that correct?)

(B) The task_done() is important for app shutdown so that the application shutdown process waits for all currently-pending queue items to be processed before proceeding with the next shutdown step. So, if the queue itself is cancelled (as in the proposed PR), what happens to the application-level call to await queue.join() during the shutdown sequence, if a queue was cancelled while there were still unprocessed items on the queue for which task_done() had not been called?

It would be great to have an example of how the proposed queue.cancel() would be used idiomatically, w.r.t. the two questions above. It might be intended that the idiomatic usage of queue.cancel() is for situations where one doesn't care about dropping items previously placed on the queue. Is that the case?

Agree with Caleb.

The more I think the more I doubt about the proposal.
Cancellation is for tasks, not for queues or locks.

When should I cancel a queue but cannot cancel a task?

Given the reactions I gather "close" is a better name for the method, so I changed it accordingly.

In the current implementation, items that had been put on the queue but not processed yet still get processed after the close, and I think this is the desired behavior. I added a test such that this won't unexpectedly change in the future.

To be precise, the current implementation of the queue does not even put new items on the queue if there is already a waiting consumer. The item will directly be handed over to said consumer, which may hang around on the event loop for a bit longer, but during this time the item is not in the queue. This also answers the questions about catching the CancelledError: if there are waiting consumers, there is nothing on the queue, so the problem of processing leftover items does not exist. The same holds for the task_done.

As for the "why don't I just cancel the task?", well, if you know it. There may be many consumer or producer tasks waiting for their turn. Sure, you can keep a list of all those tasks. But that's exactly the point of the proposed change: the Queue already knows all the waiting tasks, no need to keep another list up-to-date!

This seems like a useful idea. I recommend to write a test implementation and play with it.

Andrew:

I think the proposal makes the queues API more error-prone: concurrent put() and close() produces unpredictable result on get() side.

How? Can you elaborate?

Caleb:

I'm interested in how this change would affect the pattern of shutting down a queue-processing task.

Agree, this can be useful for that.

Martin:

Given the reactions I gather "close" is a better name for the method, so I changed it accordingly.

Not sure I like "close" since it will *cancel* all getters and putters & discard all items in the queue AND allow further operation on the queue. The latter part is really questionable -- what's the point of losing the data in the queue and resuming it? Seems like a mechanism for writing unreliable code, but perhaps you can give us an example where this is necessary.

Suppose we have 2 concurrent producers, a single queue and a consumer.
The first producer puts several items into the queue and then calls q.close()
The second producer also puts several items. It doesn't matter the second producer closes the queue at the end or not.
The consumer gets items from the queue and prints them.

What items are printed and what are canceled/skipped? With the proposed PR it depends on timings of putting and getting data items and the queue size.
The output can vary from zero to all pushed data.
That's why I think that the idea is not reliable.

Hi Andrew,

I still don't get your point. First, this is an extension to the asyncio library, so concurrency is not an issue. And sure, if you call random methods of an object without any reason the outcome won't be anything useful, why, in your example, should the one task call close at all?

There is, however, a strong use case: If you have many producers but just one consumer and the consumer stops consuming, it should cancel all producers waiting on the queue. The same way, if you have one producer but many consumers, once the single producer stops producing, it should cancel all waiting consumers. In these situations, the outcome is very clear.

Whether it should be possible to "resurrect" a closed/cancelled queue I don't care much, as I neither see a use case in doing so nor a compelling argument why it should be artificially prohibited. So I simply went for the most simple code.

The proposed code does something very simple, something a user can grasp. That is IMHO a better protection for users than some code trying to artificially stop users from harming themselves.

Ok, I see now. The improvement with only a single producer/consumer might be marginal, but the proposition that queue.cancel() might simplify the situation with multiple producers and/or consumers is more compelling.

Usually, assuming M producers and N threads (and 1 queue), one would typically stop the M producers outright, and then place N None values on the queue, and then each consumer shuts itself down when it receives a None. It's clunky but it works.

This from experience with threaded code, but the difference with asyncio is that we have cancellation available to us, whereas we could not cancel threads. I think this is why I've carried over my queue-idioms from threads to asyncio. So this is an interesting idea: if we introduce cancellation to queue handling, do things get better for the MxN producers and consumers design?

Rehashing, what I might expect to happen when I call queue.cancel() is that

A CancelledError (or maybeQueueCancelled?) exception is raised in all producers and consumers ) - this gives a producer a chance to handle the error and do something with the waiting item that could not be put()
Items currently on the queue still get processed in the consumers before the consumers exit.

I think (1) is easy, but I expected (2) to be more tricky. You said it already works that way in your PR. I didn't believe it so I wrote a silly program to check, and it does! All pending items on the queue are still consumed by consumers even after the queue._getters futures are cancelled.

So yes, both (1) and (2) appear to work.

As for the "why don't I just cancel the task?", well, if you know it. There may be many consumer or producer tasks waiting for their turn. Sure, you can keep a list of all those tasks. But that's exactly the point of the proposed change: the Queue already knows all the waiting tasks, no need to keep another list up-to-date!

Finally - while I think the MxN producers/consumers case might be simplified by this, it's worth noting that usually in shutdown, all the currently-pending tasks are cancelled anyway. And as I said before, in an MxN queue scenario, one might place N None values on the queue, and then just send CancelledError to everything anyway (consumers will ignore the cancellation and just wait for the Nones to exit). This works well.

Thus, if queue.cancel() were available to me right now, the primary difference as I see it would be that during shutdown, instead of placing N None values on the queue, I would instead call queue.cancel(). I agree that's a bit neater. (It however will still necessary to absorb CancelledError in the consumers, e.g. what is raised by asyncio.run() during shutdown, so that's unchanged).

I agree with Yury that I don't like queue.close. "Cancel" seems better after all.

I disagree with Yury that items are discarded - I checked that already-present items on the queue will still be consumed by consumers, before the queue.close cancellation is actually raised.

Yes, in the one-producer-many-consumers situation on can indeed to the trick with the None. But this is just a clumsy hack, cancelling the tasks is IMHO more in line with asyncio.

In the many-producers-one-consumer scenario this does not work. The one dead consumer cannot feed back to the producers. Sure, there are still many hacks imaginable, but a closing or cancelling the queue is a very clear way of doing things.

As for the naming: I personally don't care, close() or cancel() are both fine with me.

My point is that closing (or cancellation) should be one-way ticket.

Is there a case where continuing after stopping?

Lists of producers/consumers are just lists of future objects.

I don't think that we need to expose futures in high-level public API.

I do not think that exposing the lists of futures does any good. I cannot come up with a semantics that could be implemented with that other than the one proposed.
Also I think that close() or cancel() is something a reader intuitively understands, while looping over a list of futures is not immediately obvious.

This is a years old issue, unfortunately it never got neither merged nor rejected. I just rebased it and hope somebody could finish the review.

@tecki

I brought up something similar in https://discuss.python.org/t/queue-termination/18386/22. Looking at @tecki's PR GH-14227, I'm not sure that is what I was looking for. The proposed close() method stops both readers and writers immediately, abandoning items still in the queue. While that's probably the best we can do if the consumer decides to stop reading, it's not what I was looking for in that Discourse issue -- I want the producer(s) to be able to indicate they're not going to produce anything more, and let the consumers drain the queue.

Does anyone mind if I just close the PR as "not what is needed"? Note that the shutdown() I propose will still cause all following put calls to fail immediately, so it can also be used by the consumer to signal to the producers that it is done.

We need a new PR though, and an additional one for the regular thread-based queue.Queue class.

Superseded by #96471

Navigation Menu

Repository navigation

Add a shutdown method to threading and asyncio Queue classes #81515

1 remaining item

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Add a shutdown method to threading and asyncio Queue classes #81515

Description

Activity

tecki commented on Jun 19, 2019

asvetlov commented on Jun 19, 2019

tecki commented on Jun 20, 2019

cjrh commented on Jun 22, 2019

asvetlov commented on Jun 22, 2019

tecki commented on Jun 24, 2019

1st1 commented on Nov 21, 2019

asvetlov commented on Nov 21, 2019

tecki commented on Nov 22, 2019

cjrh commented on Nov 22, 2019

tecki commented on Nov 22, 2019

asvetlov commented on Nov 22, 2019

1 remaining item

asvetlov commented on Nov 23, 2019

tecki commented on Nov 23, 2019

tecki commented on May 12, 2021

gvanrossum commented on Aug 28, 2022

kumaraditya303 commented on Apr 19, 2023

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions