From 9450c751cc2053b1c2e03ec92ed822a41223b142 Mon Sep 17 00:00:00 2001 From: "Miss Islington (bot)" <31488909+miss-islington@users.noreply.github.com> Date: Thu, 18 Nov 2021 08:16:06 -0800 Subject: bpo-45835: Fix race condition in test_queue (GH-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. (cherry picked from commit df3e53d86b2ad67da9ac2b5a3f56257d1f394982) Co-authored-by: Sam Gross --- 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 dd7f0b1..ee58c62 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -418,11 +418,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: @@ -461,11 +462,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) @@ -479,11 +479,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 support.start_threads(feeders + consumers): pass @@ -541,7 +541,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) @@ -551,7 +551,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 @@ -562,7 +562,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) @@ -572,7 +572,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