Fix infinite loop in '_read_msg_thread'

of 'test_shutdown_[immediate_]all_methods_in_many_threads' unittests

Author: YvesDup

Lib/test/test_queue.py

@@ -317,38 +317,39 @@ def test_shutdown_all_methods_in_one_thread(self):
     def test_shutdown_immediate_all_methods_in_one_thread(self):
         return self._shutdown_all_methods_in_one_thread(True)
 
-    def _write_msg_thread(self, q, n, results, delay,
-                            i_when_exec_shutdown,
-                            event_start, event_end):
-        event_start.wait()
+    def _write_msg_thread(self, q, n, results,
+                            i_when_exec_shutdown, event_shutdown,
+                            event_start):
+        put_atleast = i_when_exec_shutdown//2
         for i in range(1, n+1):
             try:
                 q.put((i, "YDLO"))
-                results.append(True)
             except self.queue.ShutDown:
                 results.append(False)
-            # triggers shutdown of queue
+                break
+
+            # Be sure that all write_threads
+            # put few items into the queue.
+            if i == put_atleast:
+                event_start.wait()
+
+            # Triggers shutdown of queue.
             if i == i_when_exec_shutdown:
-                event_end.set()
-            time.sleep(delay)
-        # end of all puts
+                if not event_shutdown.is_set():
+                    event_shutdown.set()
+                    results.append(True)
         q.join()
 
-    def _read_msg_thread(self, q, nb, results, delay, event_start):
-        event_start.wait()
-        block = True
-        while nb:
-            time.sleep(delay)
+    def _read_msg_thread(self, q, results, event_start):
+        nbr = 0
+        while True:
             try:
-                # Get at least one message
-                q.get(block)
-                block = False
+                q.get(False)
                 q.task_done()
-                results.append(True)
-                nb -= 1
+                nbr += 1
             except self.queue.ShutDown:
-                results.append(False)
-                nb -= 1
+                results.append(True)
+                break
             except self.queue.Empty:
                 pass
         q.join()
@@ -358,9 +359,8 @@ def _shutdown_thread(self, q, event_end, immediate):
         q.shutdown(immediate)
         q.join()
 
-    def _join_thread(self, q, delay, event_start):
+    def _join_thread(self, q, event_start):
         event_start.wait()
-        time.sleep(delay)
         q.join()
 
     def _shutdown_all_methods_in_many_threads(self, immediate):
@@ -370,44 +370,35 @@ def _shutdown_all_methods_in_many_threads(self, immediate):
         ev_exec_shutdown = threading.Event()
         res_puts = []
         res_gets = []
-        delay = 1e-4
-        read_process = 4
-        nb_msgs = read_process * 16
-        nb_msgs_r = nb_msgs // read_process
-        when_exec_shutdown = nb_msgs // 2
+        write_threads = 4
+        read_threads = 16
+        nb_msgs = 1024*4
+        nb_msgs_w = nb_msgs // write_threads
+        when_exec_shutdown = nb_msgs_w // 2
         lprocs = (
-            (self._write_msg_thread, 1,  (q, nb_msgs, res_puts, delay,
-                                            when_exec_shutdown,
-                                            ev_start, ev_exec_shutdown)),
-            (self._read_msg_thread, read_process, (q, nb_msgs_r,
-                                                    res_gets, delay*2,
+            (self._write_msg_thread, write_threads, (q, nb_msgs_w, res_puts,
+                                            when_exec_shutdown, ev_exec_shutdown,
+                                            ev_start)),
+            (self._read_msg_thread, read_threads, (q, res_gets,
                                                     ev_start)),
-            (self._join_thread, 2, (q, delay*2, ev_start)),
+            (self._join_thread, 2, (q, ev_start)),
             (self._shutdown_thread, 1, (q, ev_exec_shutdown, immediate)),
             )
-        # start all threds
+        # start all thredas
         for func, n, args in lprocs:
             for i in range(n):
                 ps.append(threading.Thread(target=func, args=args))
                 ps[-1].start()
-        # set event in order to run q.shutdown()
         ev_start.set()
-
-        if not immediate:
-            assert(len(res_gets) == len(res_puts))
-            assert(res_gets.count(True) == res_puts.count(True))
-        else:
-            assert(len(res_gets) <= len(res_puts))
-            assert(res_gets.count(True) <= res_puts.count(True))
-
-        for thread in ps[1:]:
+        for thread in ps:
             thread.join()
 
-    @unittest.skip("test times out (gh-115258)")
+        self.assertEqual(res_puts.count(True), 1)
+        self.assertLessEqual(res_gets.count(True), read_threads)
+
     def test_shutdown_all_methods_in_many_threads(self):
         return self._shutdown_all_methods_in_many_threads(False)
 
-    @unittest.skip("test times out (gh-115258)")
     def test_shutdown_immediate_all_methods_in_many_threads(self):
         return self._shutdown_all_methods_in_many_threads(True)