bpo-37334: Add a cancel method to asyncio Queues

[tecki]

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.close.

https://bugs.python.org/issue37334

[mangrisano]

/cc @asvetlov @1st1

[asvetlov]

Do you want to allow calling queue.close() multiple times?
Call get() / put() after close()?

[tecki]

While I think it does not make much sense, I do not see a good reason to forbid calling close several times. You can also call cancel several times on a future. Also I do not think we should forbid calling get or put after a close. We would have to artificially add some flags.
Currently, what close does is pretty simple: it cancels all waiting getters and putters. Not more, not less. I think users will be happy to have this very simple interface, as then they know what's going on.

[asvetlov]

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

@1st1 we need your opinion here.

[tecki]

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

Is that so? I am not aware of such a behavior, but if there is one, I'm happy to fix it. Could you give an example?

[jpetkau]

@tecki Also I do not think we should forbid calling get or put after a close. We would have to artificially add some flags.
Currently, what close does is pretty simple: it cancels all waiting getters and putters. Not more, not less.

What about a consumers or producers doing something like:

async def consumer(queue):
    while not_done:
      job = await queue.get()
      result = await do_stuff(job)

async def producer(queue):
    while not_done:
      job = await make_job()
      await queue.put(job)

(This is the actual case I had, that led me to wish queues had cancellation, that led me to this issue).

In either case, they may not be waiting on the queue itself at the moment close (or cancel) is called, but they should still be shut down. Semantically, I think it makes the most sense of close means: "no more items can be put on this queue, so trying to do so raises an exception, and trying to get once the queue is empty also raises an exception."

[tecki]

@jpetkau your case can actually be solved pretty easily by setting your not_done variable to False once you cancel/close the queue. Sure, this feature could also be added to the cancel/close itself, but I am not sure whether it is worth the additional complexity in the standard library vs. additional work for users.

So, I would add this "close is close forever" functionality if there is a general agreement about it, but if not I would rather let it be as it is.

This PR has been open for almost two years now, it would be cool if it could be finished...

[gvanrossum]

I propose to close this in favor of a shutdown() method with slightly different behavior -- it would allow the readers to continue draining the queue until it is empty. It would still stop writers whenever they attempt to write (put).