ThreadPoolExecutor: how to limit the queue maxsize?

from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED

limit = 10

futures = set()

with ThreadPoolExecutor(max_workers=1) as executor:
    for arg in range(10000000):
        if len(futures) >= limit:
            completed, futures = wait(futures, return_when=FIRST_COMPLETED)
        futures.add(executor.submit(some_func, arg))

You should use a semaphore, as demonstrated here https://www.bettercodebytes.com/theadpoolexecutor-with-a-bounded-queue-in-python/

One possible issue with andres.riancho's answer, is that if max_size is reached when trying to shutdown the pool, self._work_queue.put(None) (see excerpt below) may block, effectively making the shutdown synchronous.

    def shutdown(self, wait=True):
        with self._shutdown_lock:
            self._shutdown = True
            self._work_queue.put(None)
        if wait:
            for t in self._threads:
                t.join(sys.maxint)

Python's ThreadPoolExecutor doesn't have the feature you're looking for, but the provided class can be easily sub-classed as follows to provide it:

from concurrent import futures
import queue

class ThreadPoolExecutorWithQueueSizeLimit(futures.ThreadPoolExecutor):
    def __init__(self, maxsize=50, *args, **kwargs):
        super(ThreadPoolExecutorWithQueueSizeLimit, self).__init__(*args, **kwargs)
        self._work_queue = queue.Queue(maxsize=maxsize)

I've been doing this by chunking the range. Here's a working example.

from time import time, strftime, sleep, gmtime
from random import randint
from itertools import islice
from concurrent.futures import ThreadPoolExecutor, as_completed

def nap(id, nap_length):
    sleep(nap_length)
    return nap_length


def chunked_iterable(iterable, chunk_size):
    it = iter(iterable)
    while True:
        chunk = tuple(islice(it, chunk_size))
        if not chunk:
            break
        yield chunk


if __name__ == '__main__':
    startTime = time()

    range_size = 10000000
    chunk_size = 10
    nap_time = 2

    # Iterate in chunks.
    # This consumes less memory and kicks back initial results sooner.
    for chunk in chunked_iterable(range(range_size), chunk_size):

        with ThreadPoolExecutor(max_workers=chunk_size) as pool_executor:
            pool = {}
            for i in chunk:
                function_call = pool_executor.submit(nap, i, nap_time)
                pool[function_call] = i

            for completed_function in as_completed(pool):
                result = completed_function.result()
                i = pool[completed_function]

                print('{} completed @ {} and slept for {}'.format(
                    str(i + 1).zfill(4),
                    strftime("%H:%M:%S", gmtime()),
                    result))

    print('==--- Script took {} seconds. ---=='.format(
        round(time() - startTime)))

The downside to this approach is the chunks are synchronous. All threads in a chunk must complete before the next chunk is added to the pool.