Skip to content

Commit

Permalink
Fix hangs by removing q.join() instructions from threads.
Browse files Browse the repository at this point in the history
Start `join_thread` only when shutdown is immediate.
Update tests.
  • Loading branch information
YvesDup committed Mar 13, 2024
1 parent b9ee958 commit e0be6b5
Showing 1 changed file with 17 additions and 11 deletions.
28 changes: 17 additions & 11 deletions Lib/test/test_queue.py
Original file line number Diff line number Diff line change
Expand Up @@ -340,9 +340,11 @@ def _write_msg_thread(self, q, n, results,
if not event_shutdown.is_set():
event_shutdown.set()
results.append(True)
q.join()

def _read_msg_thread(self, q, results, barrier_start):
# Get at least one item.
q.get(True)
q.task_done()
# Wait for the barrier to be complete.
barrier_start.wait()
while True:
Expand All @@ -354,13 +356,11 @@ def _read_msg_thread(self, q, results, barrier_start):
break
except self.queue.Empty:
pass
q.join()

def _shutdown_thread(self, q, results, event_end, immediate):
event_end.wait()
q.shutdown(immediate)
results.append(q.qsize() == 0)
q.join()

def _join_thread(self, q, barrier_start):
# Wait for the barrier to be complete.
Expand All @@ -382,19 +382,25 @@ 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
# 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)
# Use of a `threading.Barrier`` to ensure that
# all `_write_msg_threads`put their part of items into the queue
# all `_read_msg_thread` get at least one itme from the queue,
# and keep on running until shutdown.
# The `_join_thread` is started only when shutdown is emmediate.
nparties = write_threads + read_threads
if immediate:
nparties += join_threads
barrier_start = threading.Barrier(nparties)
ev_exec_shutdown = threading.Event()
lprocs = (
lprocs = [
(self._write_msg_thread, write_threads, (q, nb_msgs_w, res_puts,
when_exec_shutdown, ev_exec_shutdown,
barrier_start)),
(self._read_msg_thread, read_threads, (q, res_gets, barrier_start)),
(self._join_thread, join_threads, (q, barrier_start)),
(self._shutdown_thread, 1, (q, res_shutdown, ev_exec_shutdown, immediate)),
)
]
if immediate:
lprocs.append((self._join_thread, join_threads, (q, barrier_start)))
# start all threads.
for func, n, args in lprocs:
for i in range(n):
Expand All @@ -403,7 +409,7 @@ def _shutdown_all_methods_in_many_threads(self, immediate):
for thread in ps:
thread.join()

self.assertEqual(res_puts.count(True), 1)
self.assertTrue(True in res_puts)
self.assertLessEqual(res_gets.count(True), read_threads)
if immediate:
self.assertListEqual(res_shutdown, [True])
Expand Down

0 comments on commit e0be6b5

Please sign in to comment.