Update queue from 3.14.2

Author: ShaharNaveh

Lib/queue.py

@@ -80,9 +80,6 @@ def task_done(self):
         have been processed (meaning that a task_done() call was received
         for every item that had been put() into the queue).
 
-        shutdown(immediate=True) calls task_done() for each remaining item in
-        the queue.
-
         Raises a ValueError if called more times than there were items
         placed in the queue.
         '''
@@ -239,9 +236,11 @@ def shutdown(self, immediate=False):
         By default, gets will only raise once the queue is empty. Set
         'immediate' to True to make gets raise immediately instead.
 
-        All blocked callers of put() and get() will be unblocked. If
-        'immediate', a task is marked as done for each item remaining in
-        the queue, which may unblock callers of join().
+        All blocked callers of put() and get() will be unblocked.
+
+        If 'immediate', the queue is drained and unfinished tasks
+        is reduced by the number of drained tasks.  If unfinished tasks
+        is reduced to zero, callers of Queue.join are unblocked.
         '''
         with self.mutex:
             self.is_shutdown = True

Lib/test/test_queue.py

@@ -2,12 +2,11 @@
 # to ensure the Queue locks remain stable.
 import itertools
 import random
-import sys
 import threading
 import time
 import unittest
 import weakref
-from test.support import gc_collect
+from test.support import gc_collect, bigmemtest
 from test.support import import_helper
 from test.support import threading_helper
 
@@ -964,33 +963,33 @@ def test_order(self):
         # One producer, one consumer => results appended in well-defined order
         self.assertEqual(results, inputs)
 
-    def test_many_threads(self):
+    @bigmemtest(size=50, memuse=100*2**20, dry_run=False)
+    def test_many_threads(self, size):
         # Test multiple concurrent put() and get()
-        N = 50
         q = self.q
         inputs = list(range(10000))
-        results = self.run_threads(N, q, inputs, self.feed, self.consume)
+        results = self.run_threads(size, q, inputs, self.feed, self.consume)
 
         # Multiple consumers without synchronization append the
         # results in random order
         self.assertEqual(sorted(results), inputs)
 
-    def test_many_threads_nonblock(self):
+    @bigmemtest(size=50, memuse=100*2**20, dry_run=False)
+    def test_many_threads_nonblock(self, size):
         # Test multiple concurrent put() and get(block=False)
-        N = 50
         q = self.q
         inputs = list(range(10000))
-        results = self.run_threads(N, q, inputs,
+        results = self.run_threads(size, q, inputs,
                                    self.feed, self.consume_nonblock)
 
         self.assertEqual(sorted(results), inputs)
 
-    def test_many_threads_timeout(self):
+    @bigmemtest(size=50, memuse=100*2**20, dry_run=False)
+    def test_many_threads_timeout(self, size):
         # Test multiple concurrent put() and get(timeout=...)
-        N = 50
         q = self.q
         inputs = list(range(1000))
-        results = self.run_threads(N, q, inputs,
+        results = self.run_threads(size, q, inputs,
                                    self.feed, self.consume_timeout)
 
         self.assertEqual(sorted(results), inputs)