gh-96471: Add threading queue shutdown

[EpicWink]

Alternate implementation of #104225, where all queue items are consumed immediately in Queue.shutdown when immediate=True is passed (see the comparison for what's changed).

• task_done and join will now not raise after calling shutdown(immediate=True), rather they'll behave as if on a finished queue (join returns, task_done raises the standard ValueError)
• Potentially public is_shutdown boolean attribute

This PR includes and modified changes from #104225.

• Issue: Queue shutdown #96471

---

📚 Documentation preview 📚: https://cpython-previews--104750.org.readthedocs.build/

[gvanrossum]

Great progress. I mostly have some markup nits. But the tests seem to hang. Or was I just impatient?

[gvanrossum]

Yeah, the test is definitely hanging for me. Have you figured that out yet?

[gvanrossum]

I'd also recommend a merge. :-)

[EpicWink]

Yeah, the test is definitely hanging for me. Have you figured that out yet?

@gvanrossum It's intermittently hanging in the test test_shutdown_immediate_put_join. q.join is blocking, after shutting down the queue, because unfinished_tasks becomes -1 (when zero is required). This is because the test calls task_done before taking items from the queue, but the implementation of shutdown() reduces unfinished_tasks by the number of items in the queue.

I think the solution is to simply ensure unfinished_tasks doesn't go negative. I don't set it straight to zero because there may still be queue consumer who will successfully complete their task and call task_done, but they will get a ValueError if unfinished_tasks was immediately set to zero. There is a problem here however if some consumers still call task_done without getting items from the queue.

Perhaps it's simpler to set unfinished_tasks to zero, then skip the ValueError raise if the queue is shut down.

Edit: I just noticed that this pull request's description says that task_done should raise ValueError on a shut down queue. Hmmm.

---

It's also intermittently hanging in the test test_shutdown_put_join, simply because of a logic error: 2 items are put on the queue (sometimes), then it's shut down, then one task is marked as done, then the queue is waited for (aka joined on). If only one item was put on the queue, then the test simply fails during assertion.

Solution here (which fixes the assertion for test_shutdown_immediate_put_join as well) is to shut down the queue before running the put-then-join.

---

Edit: and of course now it's failing only in Windows (CI only succeeds because it re-runs test_queue)

[gvanrossum]

LG, assuming you believe the test is stable. I have one nit, let me know how you feel about that.

Oh, one more thing. If you feel like writing documentation, could you update Doc/whatsnew/3.13.rst? There's a section about modified modules. A few lines there will go a long way.

[EpicWink]

LG, assuming you believe the test is stable.

Test is not stable, as test_shutdown_all_methods_in_many_threads (and the corresponding immediate version) hangs intermittently in Windows. It also fails consistently when adding debugging print-statements on my machine. I'm investigating

[gvanrossum]

Excellent! All that remains is merging it (which I will take care of), closing the alternative PR for queue.py, and then we can have another look at the asyncio and multiprocessing queues. (For the latter we still need to find a reviewer.)

[gvanrossum]

@EpicWink Sorry, it looks like we have to revert this until the Windows free-threading hang has been resolved. (Although it's possible that you've uncovered a Windows-related bug in free-threading, which is a developing feature.)

[gvanrossum]

FWIW, having had only a quick look at the failing test, is it possible that it relies on timeouts too much? I see that there's a 0.1 msec delay that is used for sleeping, and _read_msg_thread() makes some non-blocking get() calls, ignoring Empty exceptions. Could there be a scenario (especially when running without GIL) where different threads run in a different order than you anticipated, and it simply doesn't read enough queue items, so the join() hangs forever?

[gvanrossum]

From #104228 (comment): it looks like Queue.join() never raises ShutDown, contradicting the docs.

[YvesDup]

Maybe I'm waking up a little too late, but despite having worked on the development 1 year ago, I have the impression that functionalities are missing in this merged PR.

In the initial version of the feature (#104225), the basic shutdown just forbade put operations, also released all threads blocked in put. On the other hand, it left get, join and task_done operations possible.

The immediate shutdown prohibited all possible previous operations and had to release all threads blocked in get, and join. And the queue was purged.

Currently, in the following methods (put, get, join and task_done), there is no specific behavior depending on the type of the shutdown. It's not even noted in an attribute.

I don't want to hurt anyone's feelings, but I think this PR needs to be reworked (including documentation).

PS: initial comment was done at the bad PR. Sorry

[EpicWink]

Maybe I'm waking up a little too late, but despite having worked on the development 1 year ago, I have the impression that functionalities are missing in this merged PR.

In the initial version of the feature (#104225), the basic shutdown just forbade put operations, also released all threads blocked in put. On the other hand, it left get, join and task_done operations possible.

The immediate shutdown prohibited all possible previous operations and had to release all threads blocked in get, and join. And the queue was purged.

Currently, in the following methods (put, get, join and task_done), there is no specific behavior depending on the type of the shutdown. It's not even noted in an attribute.

@YvesDup see comment in the issue: #96471 (comment)

Basically, the goal is to have waiters (callers of queue methods) not be blocked on a queue which has been shut-down. Making the shutdown immediate simply means gets won't take anything more from the queue (and instead except).

There is a problem with this PR though, in that q.task_done() and q.join() are documented as raising ShutDown when they don't; see #115838 for the fix for that.

[YvesDup]

@EpicWink , thank you for this explanation. I ve missed that :-(
Features really change from the initial version. Now it seems more efficient and simpler.
As we wrote all tests for the initial version, I 'd like to review this part because features are changed. May be we will have to remove some of tests, to modify others, or add new ones ?
Do you agree ?

[YvesDup]

FYI, I fixed the failling test in _read_msg_thread() method. Loop was infinite, I insert a break on queue.Shutdow exception. .... (and I removed all time.sleep())

[gvanrossum]

FYI, I fixed the failling test in _read_msg_thread() method. Loop was infinite, I insert a break on queue.Shutdow exception. .... (and I removed all time.sleep())

Where can we see this?

[YvesDup]

FYI, I fixed the failling test in _read_msg_thread() method. Loop was infinite, I insert a break on queue.Shutdow exception. .... (and I removed all time.sleep())

Where can we see this?

https://github.com/YvesDup/cpython/blob/Threading-queue-shutdown-test-failed/Lib/test/test_queue.py#L350
Debug logs r and w will be removed.
See now https://github.com/YvesDup/cpython/blob/Threading-queue-shutdown-test-failed/Lib/test/test_queue.py#L370

[gvanrossum]

Thanks, I'll wait for the pull request.

[YvesDup]

Thanks, I'll wait for the pull request.

PR is ready on this issue (#115940). Sorry if I didn't understand that you wait for a PR linked to this one. Please let me know if I have to change.

[gvanrossum]

@YvesDup, I am a little confused. There is indeed #115940, by you, but there is also gh-115898 by @EpicWink that also claims to fix these tests. How do the two relate to each other?

[YvesDup]

How do the two relate to each other?

There are related because there are both about on the same issue. But #115898 is only a draft.
I had informed @EpicWink that I had fixed the bug on the test, which is why I had submitted a new PR.
Maybe, should we wait for his feedback before going head ?

[gvanrossum]

@EpicWink: We're waiting for you! ^^

[EpicWink]

I'll review the new PR fixing the test today.

My PR is another attempt, but unfinished. Perhaps in the future it can be added as another test.

Edit: reviewed!

[YvesDup]

'll review the new PR fixing the test today.

My PR is another attempt, but unfinished. Perhaps in the future it can be added as another test.

Edit: reviewed!

Thank for the review.