gh-115258: Fix failed tests on threading queue shutdown by YvesDup · Pull Request #115940 · python/cpython · GitHub

Fix infinite loop in _read_msg_thread of test_shutdown_[immediate_]all_methods_in_many_threads unittests.

Related PR: gh-96471: Add threading queue shutdown #104750

Issue: test_queue times out #115258

Thanks for the change! Had some minor comments when reviewing, please see inline

Hmm I might be missing something but it looks like I can't see the latest changes?

Hmm I might be missing something but it looks like I can't see the latest changes?

Hi, I was on vacation. The changes are scheduled for early next week.

Ah sorry for that! Enjoy your holiday :D

Test hangs when adding many time.sleep(random.random() / 10.0) calls between each statement

Diff (click to expand)

diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py
index 9dc7f62999..d223c358e0 100644
--- a/Lib/test/test_queue.py
+++ b/Lib/test/test_queue.py
@@ -320,58 +320,90 @@ def test_shutdown_immediate_all_methods_in_one_thread(self):
     def _write_msg_thread(self, q, n, results,
                             i_when_exec_shutdown, event_shutdown,
                             barrier_start):
+        time.sleep(random.random() / 10.0)
         # All `write_msg_threads`
         # put several items into the queue.
         for i in range(0, i_when_exec_shutdown//2):
+            time.sleep(random.random() / 10.0)
             q.put((i, 'LOYD'))
+        time.sleep(random.random() / 10.0)
         # Wait for the barrier to be complete.
         barrier_start.wait()
+        time.sleep(random.random() / 10.0)
 
         for i in range(i, n):
+            time.sleep(random.random() / 10.0)
             try:
                 q.put((i, "YDLO"))
             except self.queue.ShutDown:
+                time.sleep(random.random() / 10.0)
                 results.append(False)
+                time.sleep(random.random() / 10.0)
                 break
+            time.sleep(random.random() / 10.0)
 
             # Trigger queue shutdown.
             if i == i_when_exec_shutdown:
+                time.sleep(random.random() / 10.0)
                 # Only once thread do it.
                 if not event_shutdown.is_set():
+                    time.sleep(random.random() / 10.0)
                     event_shutdown.set()
+                    time.sleep(random.random() / 10.0)
                     results.append(True)
+        time.sleep(random.random() / 10.0)
         q.join()
+        time.sleep(random.random() / 10.0)
 
     def _read_msg_thread(self, q, results, barrier_start):
         # Wait for the barrier to be complete.
+        time.sleep(random.random() / 10.0)
         barrier_start.wait()
+        time.sleep(random.random() / 10.0)
         while True:
+            time.sleep(random.random() / 10.0)
             try:
                 q.get(False)
+                time.sleep(random.random() / 10.0)
                 q.task_done()
             except self.queue.ShutDown:
+                time.sleep(random.random() / 10.0)
                 results.append(True)
+                time.sleep(random.random() / 10.0)
                 break
             except self.queue.Empty:
                 pass
+            time.sleep(random.random() / 10.0)
+        time.sleep(random.random() / 10.0)
         q.join()
+        time.sleep(random.random() / 10.0)
 
     def _shutdown_thread(self, q, results, event_end, immediate):
+        time.sleep(random.random() / 10.0)
         event_end.wait()
+        time.sleep(random.random() / 10.0)
         q.shutdown(immediate)
+        time.sleep(random.random() / 10.0)
         results.append(q.qsize() == 0)
+        time.sleep(random.random() / 10.0)
         q.join()
+        time.sleep(random.random() / 10.0)
 
     def _join_thread(self, q, barrier_start):
+        time.sleep(random.random() / 10.0)
         # Wait for the barrier to be complete.
         barrier_start.wait()
+        time.sleep(random.random() / 10.0)
         q.join()
+        time.sleep(random.random() / 10.0)
 
     def _shutdown_all_methods_in_many_threads(self, immediate):
         # Run a 'multi-producers/consumers queue' use case,
         # with enough items into the queue.
         # When shutdown, all running threads will be concerned.
+        time.sleep(random.random() / 10.0)
         q = self.type2test()
+        time.sleep(random.random() / 10.0)
         ps = []
         res_puts = []
         res_gets = []
@@ -382,11 +414,14 @@ def _shutdown_all_methods_in_many_threads(self, immediate):
         nb_msgs = 1024*64
         nb_msgs_w = nb_msgs // write_threads
         when_exec_shutdown = nb_msgs_w // 2
+        time.sleep(random.random() / 10.0)
         # Use of a `threading.Barrier`` to ensure that all `_write_msg_threads`
         # put their part of items into the queue. And trigger the start of
         # other threads as `_read_msg_thread`and `_join_thread`.
         barrier_start = threading.Barrier(write_threads+read_threads+join_threads)
+        time.sleep(random.random() / 10.0)
         ev_exec_shutdown = threading.Event()
+        time.sleep(random.random() / 10.0)
         lprocs = (
             (self._write_msg_thread, write_threads, (q, nb_msgs_w, res_puts,
                                             when_exec_shutdown, ev_exec_shutdown,
@@ -395,19 +430,34 @@ def _shutdown_all_methods_in_many_threads(self, immediate):
             (self._join_thread, join_threads, (q, barrier_start)),
             (self._shutdown_thread, 1, (q, res_shutdown, ev_exec_shutdown, immediate)),
             )
+        time.sleep(random.random() / 10.0)
         # start all threads.
         for func, n, args in lprocs:
+            time.sleep(random.random() / 10.0)
             for i in range(n):
+                time.sleep(random.random() / 10.0)
                 ps.append(threading.Thread(target=func, args=args))
+                time.sleep(random.random() / 10.0)
                 ps[-1].start()
+                time.sleep(random.random() / 10.0)
+            time.sleep(random.random() / 10.0)
+        time.sleep(random.random() / 10.0)
         for thread in ps:
+            time.sleep(random.random() / 10.0)
             thread.join()
+            time.sleep(random.random() / 10.0)
+        time.sleep(random.random() / 10.0)
 
         self.assertEqual(res_puts.count(True), 1)
+        time.sleep(random.random() / 10.0)
         self.assertLessEqual(res_gets.count(True), read_threads)
+        time.sleep(random.random() / 10.0)
         if immediate:
+            time.sleep(random.random() / 10.0)
             self.assertListEqual(res_shutdown, [True])
+            time.sleep(random.random() / 10.0)
             self.assertTrue(q.empty())
+            time.sleep(random.random() / 10.0)
 
     def test_shutdown_all_methods_in_many_threads(self):
         return self._shutdown_all_methods_in_many_threads(False)

Also, perhaps look into threading_helper.join_thread(thread) and threading_helper.start_threads(threads) (from test.support) for better thread management. They should fail the test instead of hanging

Test hangs when adding many time.sleep(random.random() / 10.0) calls between each statement

When I run your modified version of the test, it hangs, but only because _read_msg_thread threads are not started.
The line (self._read_msg_thread, read_threads, (q, res_gets, barrier_start)), in the procs tuple is missing: so the barrier waits for ever, partiesis never reached. Could you check that please ?
When I add the missing line, test succeeds. I reduce number of threads and msg because all time.sleep() are too much long ;-).

The line (self._read_msg_thread, read_threads, (q, res_gets, barrier_start)), in the procs tuple is missing: so the barrier waits for ever, partiesis never reached. Could you check that please ?

@YvesDup it wasn't missing, just was outside of the diff context

test succeeds. I reduce number of threads and msg because all time.sleep() are too much long

Tests finish also for me when I instead use time.sleep(random.random() / 1000.0) (and / 10000.0). However, sometimes I get:

FAIL: test_shutdown_all_methods_in_many_threads (test.test_queue.PyPriorityQueueTest.test_shutdown_all_methods_in_many_threads)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "~/src/cpython/Lib/test/test_queue.py", line 463, in test_shutdown_all_methods_in_many_threads
    return self._shutdown_all_methods_in_many_threads(False)
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^
  File "~/src/cpython/Lib/test/test_queue.py", line 451, in _shutdown_all_methods_in_many_threads
    self.assertEqual(res_puts.count(True), 1)
    ~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError: 2 != 1

Thank for adding all these time.sleep() in the test. It's very instructive.

  File "~/src/cpython/Lib/test/test_queue.py", line 451, in _shutdown_all_methods_in_many_threads
    self.assertEqual(res_puts.count(True), 1)
    ~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError: 2 != 1

About this message, I have got this failed test once. That means here that there are 2 _write_msg_threads threads which reach the i_when_exec_shutdown. I'am going to modify it to check there is at least one True value.

That said, I am going to:

modify the test above.
modify the _read_msg_thread to ensure at least one msg was got, and check that a ShutDown was raised for each thread.
remove q.join() instructions from _write_msg_thread, _read_msg_thread and _shutdown_thread. They can cause hangs when shutdown is not immediate.
start _join_thread only when shutdow, is immediate.

I will submit a new version of this test.

of 'test_shutdown_[immediate_]all_methods_in_many_threads' unittests

…_many_threads` methods, with a code refactoring. Add a `results` list to the `_shutdown_thread` method. Add tests. Fix nit.

Start `join_thread` only when shutdown is immediate. Update tests.

Please don't force-push. It makes reviewing harder. We squash-merge ultimately anyway. Unless you're not ready to have this reviewed, in which case the PR should be in Draft mode.

I believe I am now following what this does. Let me summarize, please check my understanding.

We start 4 writer threads, each of which is going to put 16*1024 items into the queue.
(Actually, one more per thread, probably by mistake.)
When the first of these threads reaches half that number, it calls shutdown().
All writer threads also have to reach halfway before passing the barrier.
There are also 6 reader threads. These each get at least one item, wait for the barrier, and then busy-wait until the queue gets shut down.
There are two variants, one for shutdown(True), one for shutdown(False).
For shutdown(immediate=True), we also start a join thread.
All threads must complete for the test to pass.

I am a little unclear on what the test is trying to prove. It seems to be stress testing concurrent reading and writing of the queue plus shutdown. Is that what it is after?

I am a little unclear on what the test is trying to prove. It seems to be stress testing concurrent reading and writing of the queue plus shutdown. Is that what it is after?

Your understanding is correct. The aim is to have a queue not empty and check that all running threads stop correctly when queue shutdows.

There are 3 threads types which used each one of methods of Queue class (put, couple get+task_done and join). Each king of thread is started several times. All threads are really doing operations before the shutdown.
At this end of test case, we check:

all threads are stopped,
all read_msg threads threads raised of ShutDown exception,
and queue is empty when shutdown is immediate.

if you or/and @EpicWink think this test case is not really useful, I'll let you decide what to do with it.

all read_msg threads threads raised of ShutDown exception,

Then shouldn't the assert be

        self.assertEqual(res_gets.count(True), read_threads)

(instead of assertLessEqual)?

Then shouldn't the assert be
        self.assertEqual(res_gets.count(True), read_threads)
(instead of assertLessEqual)?

Yes

Fix start value of range. Change `self.assertLessEqual` to `self.assertEqual`.

Okay, LGTM.

@EpicWink Do you agree that this test is worth having and that this version looks bullet-proof?

🤖 New build scheduled with the buildbot fleet by @gvanrossum for commit 349d08a 🤖

If you want to schedule another build, you need to add the 🔨 test-with-buildbots label again.

I've run the modified tests ~50 times, both with and without GIL, both single-process and 15-processes, both with and without the time.sleep()s; after updating with main. No failures, so the test fix looks good to me.

I agree the purpose of the test is not very indicative of any real-world scenario, but it does seem to test invariants so I didn't push back against its inclusion.

However, the NEWS entry still hasn't been removed.

Thanks @EpicWink for the mention in the whatsnew3.13.

…on#115940) This reinstates `test_shutdown_immediate_all_methods_in_many_threads` and improves `test_shutdown_all_methods_in_many_threads`.

YvesDup requested a review from rhettinger as a code owner February 26, 2024 10:19

bedevere-app Bot added the awaiting review label Feb 26, 2024

bedevere-app Bot mentioned this pull request Feb 26, 2024

test_queue times out #115258

Closed

bedevere-app Bot added the tests Tests in the Lib/test dir label Feb 26, 2024

rhettinger removed their request for review February 26, 2024 12:55

Jason-Y-Z reviewed Mar 2, 2024

View reviewed changes

Comment thread Lib/test/test_queue.py Outdated

Comment thread Lib/test/test_queue.py

Comment thread Lib/test/test_queue.py Outdated

Comment thread Lib/test/test_queue.py Outdated

Comment thread Lib/test/test_queue.py Outdated

YvesDup force-pushed the Threading-queue-shutdown-test-failed branch from 8b24985 to f246bf8 Compare March 9, 2024 12:01

YvesDup force-pushed the Threading-queue-shutdown-test-failed branch from f246bf8 to fd811b5 Compare March 11, 2024 09:36

YvesDup mentioned this pull request Mar 12, 2024

gh-96471: Add threading queue shutdown #104750

Merged

EpicWink mentioned this pull request Mar 13, 2024

gh-115258: Fix thread queue shutdown test #115898

Closed

EpicWink suggested changes Mar 13, 2024

View reviewed changes

Comment thread Misc/NEWS.d/next/Tests/2024-02-26-15-04-57.gh-issue-115258.p__Nfv.rst Outdated

bedevere-app Bot added awaiting core review and removed awaiting review labels Mar 13, 2024

YvesDup and others added 5 commits March 13, 2024 16:22

Fix infinite loop in '_read_msg_thread'

6188472

of 'test_shutdown_[immediate_]all_methods_in_many_threads' unittests

📜🤖 Added by blurb_it.

deb0b85

Update 2024-02-26-15-04-57.gh-issue-115258.p__Nfv.rst

63a545f

Update 2024-02-26-15-04-57.gh-issue-115258.p__Nfv.rst

5a09f7d

Add comments to the _write_msg_thread and `_shutdown_all_methods_in…

b9ee958

…_many_threads` methods, with a code refactoring. Add a `results` list to the `_shutdown_thread` method. Add tests. Fix nit.

YvesDup force-pushed the Threading-queue-shutdown-test-failed branch from 50bddd3 to b9ee958 Compare March 13, 2024 15:22

Fix hangs by removing q.join() instructions from threads.

e0be6b5

Start `join_thread` only when shutdown is immediate. Update tests.

gvanrossum reviewed Mar 13, 2024

View reviewed changes

Comment thread Lib/test/test_queue.py Outdated

Comment thread Lib/test/test_queue.py Outdated

Comment thread Lib/test/test_queue.py Outdated

Comment thread Lib/test/test_queue.py Outdated

Comment thread Misc/NEWS.d/next/Tests/2024-02-26-15-04-57.gh-issue-115258.p__Nfv.rst Outdated

YvesDup added 2 commits March 15, 2024 15:02

Update comments.

e4e880b

Fix start value of range. Change `self.assertLessEqual` to `self.assertEqual`.

Add newline at end of file.

349d08a

gvanrossum approved these changes Mar 15, 2024

View reviewed changes

bedevere-app Bot added awaiting merge and removed awaiting core review labels Mar 15, 2024

gvanrossum added the 🔨 test-with-buildbots Test PR w/ buildbots; report in status section label Mar 15, 2024

bedevere-bot removed the 🔨 test-with-buildbots Test PR w/ buildbots; report in status section label Mar 15, 2024

Remove unnecessary news entry

7c1935d

gvanrossum approved these changes Mar 18, 2024

View reviewed changes

gvanrossum added the skip news label Mar 18, 2024

gvanrossum enabled auto-merge (squash) March 18, 2024 15:51

gvanrossum merged commit 7707b14 into python:main Mar 18, 2024

bedevere-app Bot removed the awaiting merge label Mar 18, 2024

YvesDup deleted the Threading-queue-shutdown-test-failed branch October 21, 2024 20:26

gh-115258: Fix failed tests on threading queue shutdown by YvesDup · Pull Request #115940 · python/cpython · GitHub

Uh oh!

Uh oh!

gh-115258: Fix failed tests on threading queue shutdown - #115940

gh-115258: Fix failed tests on threading queue shutdown#115940gvanrossum merged 9 commits intopython:mainpython/cpython:mainfrom YvesDup:Threading-queue-shutdown-test-failedYvesDup/cpython:Threading-queue-shutdown-test-failedCopy head branch name to clipboard

Conversation

YvesDup commented Feb 26, 2024 • edited by gvanrossum Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Jason-Y-Z left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Jason-Y-Z commented Mar 9, 2024

Uh oh!

YvesDup commented Mar 9, 2024

Uh oh!

Jason-Y-Z commented Mar 9, 2024

Uh oh!

EpicWink left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

YvesDup commented Mar 13, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

EpicWink commented Mar 13, 2024

Uh oh!

YvesDup commented Mar 13, 2024

Uh oh!

gvanrossum commented Mar 13, 2024

Uh oh!

gvanrossum left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

YvesDup commented Mar 14, 2024

Uh oh!

gvanrossum commented Mar 14, 2024

Uh oh!

YvesDup commented Mar 15, 2024

Uh oh!

gvanrossum left a comment

Choose a reason for hiding this comment

Uh oh!

bedevere-bot commented Mar 15, 2024

Uh oh!

EpicWink commented Mar 18, 2024

Uh oh!

YvesDup commented Mar 18, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

gh-115258: Fix failed tests on threading queue shutdown#115940
gvanrossum merged 9 commits into
python:mainfrom
YvesDup:Threading-queue-shutdown-test-failed

YvesDup commented Feb 26, 2024 •
edited by gvanrossum
Loading

YvesDup commented Mar 13, 2024 •
edited
Loading