bpo-29595: Expose max_queue_size in ThreadPoolExecutor

Lib/concurrent/futures/thread.py

@@ -81,13 +81,15 @@ def _worker(executor_reference, work_queue):
         _base.LOGGER.critical('Exception in worker', exc_info=True)
 
 class ThreadPoolExecutor(_base.Executor):
-    def __init__(self, max_workers=None, thread_name_prefix=''):
+    def __init__(self, max_workers=None, thread_name_prefix='', max_queue_size=0):
         """Initializes a new ThreadPoolExecutor instance.
 
         Args:
             max_workers: The maximum number of threads that can be used to
                 execute the given calls.
             thread_name_prefix: An optional name prefix to give our threads.
+            max_queue_size: The maximum number of work items to buffer before
+                submit() blocks, defaults to 0 (infinite).
         """
         if max_workers is None:
             # Use this number because ThreadPoolExecutor is often
@@ -96,8 +98,12 @@ def __init__(self, max_workers=None, thread_name_prefix=''):
         if max_workers <= 0:
             raise ValueError("max_workers must be greater than 0")
 
+        # check that max_queue_size is a positive integer
+        if type(max_queue_size) is not int or max_queue_size < 0:
+            raise ValueError("max_queue_size must be equal or greater 0")
+
         self._max_workers = max_workers
-        self._work_queue = queue.Queue()
+        self._work_queue = queue.Queue(maxsize=max_queue_size)
         self._threads = set()
         self._shutdown = False
         self._shutdown_lock = threading.Lock()

Lib/test/test_concurrent_futures.py

@@ -178,6 +178,25 @@ def test_thread_names_default(self):
             self.assertRegex(t.name, r'^.*ThreadPoolExecutor.*_[0-4]$')
             t.join()
 
+    def test_default_max_queue_size(self):
+        executor = futures.ThreadPoolExecutor()
+        self.assertEqual(executor._work_queue.maxsize, 0)
+
+    def test_custom_max_queue_size(self):
+        for i in range(0, 10):
+            executor = futures.ThreadPoolExecutor(max_queue_size=i)
+            self.assertEqual(executor._work_queue.maxsize, i)
+
+    def test_negative_max_queue_size(self):
+        for i in range(-1, -10):
+            with self.assertRaises(ValueError):
+                futures.ThreadPoolExecutor(max_queue_size=i)
+
+    def test_invalid_max_queue_size(self):
+        for bad_value in [None, "5", False, True, 3.14]:
+            with self.assertRaises(ValueError):
+                futures.ThreadPoolExecutor(max_queue_size=bad_value)
+
 
 class ProcessPoolShutdownTest(ProcessPoolMixin, ExecutorShutdownTest, unittest.TestCase):
     def _prime_executor(self):