From df3e53d86b2ad67da9ac2b5a3f56257d1f394982 Mon Sep 17 00:00:00 2001 From: Sam Gross Date: Thu, 18 Nov 2021 03:51:30 -0500 Subject: bpo-45835: Fix race condition in test_queue (#29601) Some of the tests in test_queue had a race condition in which a non-sentinel value could be enqueued after the final sentinel value leading to not all the inputs being processed (and test failures). This changes feed() to enqueue a sentinel once the inputs are exhausted, which guarantees that the final queued object is a sentinel. This requires the number of feeder threads to match the number of consumer threads, but that's already the case in the relevant tests. --- Lib/test/test_queue.py | 22 +++++++++++----------- .../Tests/2021-11-17-14-28-08.bpo-45835.Mgyhjx.rst | 1 + 2 files changed, 12 insertions(+), 11 deletions(-) create mode 100644 Misc/NEWS.d/next/Tests/2021-11-17-14-28-08.bpo-45835.Mgyhjx.rst diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index 9bb5181..cfa6003 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -420,11 +420,12 @@ class BaseSimpleQueueTest: def setUp(self): self.q = self.type2test() - def feed(self, q, seq, rnd): + def feed(self, q, seq, rnd, sentinel): while True: try: val = seq.pop() except IndexError: + q.put(sentinel) return q.put(val) if rnd.random() > 0.5: @@ -463,11 +464,10 @@ class BaseSimpleQueueTest: return results.append(val) - def run_threads(self, n_feeders, n_consumers, q, inputs, - feed_func, consume_func): + def run_threads(self, n_threads, q, inputs, feed_func, consume_func): results = [] sentinel = None - seq = inputs + [sentinel] * n_consumers + seq = inputs.copy() seq.reverse() rnd = random.Random(42) @@ -481,11 +481,11 @@ class BaseSimpleQueueTest: return wrapper feeders = [threading.Thread(target=log_exceptions(feed_func), - args=(q, seq, rnd)) - for i in range(n_feeders)] + args=(q, seq, rnd, sentinel)) + for i in range(n_threads)] consumers = [threading.Thread(target=log_exceptions(consume_func), args=(q, results, sentinel)) - for i in range(n_consumers)] + for i in range(n_threads)] with threading_helper.start_threads(feeders + consumers): pass @@ -543,7 +543,7 @@ class BaseSimpleQueueTest: # Test a pair of concurrent put() and get() q = self.q inputs = list(range(100)) - results = self.run_threads(1, 1, q, inputs, self.feed, self.consume) + results = self.run_threads(1, q, inputs, self.feed, self.consume) # One producer, one consumer => results appended in well-defined order self.assertEqual(results, inputs) @@ -553,7 +553,7 @@ class BaseSimpleQueueTest: N = 50 q = self.q inputs = list(range(10000)) - results = self.run_threads(N, N, q, inputs, self.feed, self.consume) + results = self.run_threads(N, q, inputs, self.feed, self.consume) # Multiple consumers without synchronization append the # results in random order @@ -564,7 +564,7 @@ class BaseSimpleQueueTest: N = 50 q = self.q inputs = list(range(10000)) - results = self.run_threads(N, N, q, inputs, + results = self.run_threads(N, q, inputs, self.feed, self.consume_nonblock) self.assertEqual(sorted(results), inputs) @@ -574,7 +574,7 @@ class BaseSimpleQueueTest: N = 50 q = self.q inputs = list(range(1000)) - results = self.run_threads(N, N, q, inputs, + results = self.run_threads(N, q, inputs, self.feed, self.consume_timeout) self.assertEqual(sorted(results), inputs) diff --git a/Misc/NEWS.d/next/Tests/2021-11-17-14-28-08.bpo-45835.Mgyhjx.rst b/Misc/NEWS.d/next/Tests/2021-11-17-14-28-08.bpo-45835.Mgyhjx.rst new file mode 100644 index 0000000..6a73b01 --- /dev/null +++ b/Misc/NEWS.d/next/Tests/2021-11-17-14-28-08.bpo-45835.Mgyhjx.rst @@ -0,0 +1 @@ +Fix race condition in test_queue tests with multiple "feeder" threads. -- cgit v0.12