When the request list is larger than service capacity

Imagine checking several thousand orders against an internal service. Overlapping a few requests can reduce waiting, but submitting the entire input at once can retain unnecessary jobs and pressure the destination. Worker count and submission count are different controls. Define how many functions may execute and how many submitted results may remain unconsumed before choosing numbers for a production workload. A fast producer should not silently become an unlimited work queue.

Python 3.14 added buffersize to Executor.map. A configured buffer makes iteration over the input pause when the permitted submitted-but-unyielded results fill up. Results still arrive in input order, so a slow early item can delay delivery of later completed results. This is a useful capacity contract, not a promise that a particular workload becomes faster. Compare latency and memory with realistic service behavior rather than relying on a thread-count benchmark alone.

Queue boundaries, result order and failure behavior

The bounded_map implementation provides a small compatible approach for projects using older Python versions. It submits a window of pending jobs, consumes one ordered result and then reads the next input item. The source can be a generator; the entire data set is not materialized first. Workers limit concurrent execution while pending limits this submission window. Choosing a small window can intentionally leave workers idle when preserving order is more important than aggressive throughput.

A bounded queue does not supply network timeouts. Configure each external call with a deadline and a useful error report. Cancelling a Future does not forcibly stop a function that is already running, and leaving an executor context can wait for that function. The example propagates worker errors and cancels queued jobs where possible. A real client still needs its own timeout, retry policy and shutdown behavior, especially when the destination is slow or unavailable.

The tests inspect input consumption instead of merely checking arithmetic. A recording generator proves that the first yielded result does not require consuming the entire source. Additional cases verify ordering, error propagation and invalid capacity values. These assertions protect the queue contract when the implementation changes later. They also give reviewers something concrete to check before increasing capacity, rather than treating a larger worker count as an automatic performance improvement.

Complete files, execution and tests

Python 3.11+ and the standard library only; the native buffersize file requires Python 3.14+. Save the four complete files below together using the stated filenames; no code is replaced by ellipses.

1. A bounded ordered iterator

File: bounded_workers.py. Save this complete file without removing lines.

from collections import deque
from concurrent.futures import ThreadPoolExecutor
from itertools import islice


def bounded_map(function, values, *, workers=4, pending=8):
    """Yield ordered results while retaining at most pending submitted jobs."""
    if workers < 1 or pending < 1:
        raise ValueError('workers and pending must be positive')
    source = iter(values)
    with ThreadPoolExecutor(max_workers=workers) as pool:
        jobs = deque(pool.submit(function, value)
                     for value in islice(source, pending))
        try:
            while jobs:
                yield jobs.popleft().result()
                try:
                    value = next(source)
                except StopIteration:
                    continue
                jobs.append(pool.submit(function, value))
        finally:
            for job in jobs:
                job.cancel()

This complete module exports bounded_map. Submission stays inside the pending window; executor shutdown is not a mechanism for forcibly terminating a running thread.

2. The Python 3.14+ API

File: python314_map.py. Save this complete file without removing lines.

import inspect
from concurrent.futures import ThreadPoolExecutor


def main():
    if 'buffersize' not in inspect.signature(ThreadPoolExecutor.map).parameters:
        print('SKIP: this example requires Python 3.14 or newer')
        return
    with ThreadPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(lambda n: n * n, range(100), buffersize=8))
    assert results == [n * n for n in range(100)]
    print(sum(results))  # 328350


if __name__ == '__main__':
    main()

Run python python314_map.py. A supported interpreter prints 328350; an older interpreter reports SKIP rather than pretending to exercise the new API.

3. Capacity, ordering and failure tests

File: test_bounded_workers.py. Save this complete file without removing lines.

import unittest
from bounded_workers import bounded_map


class BoundedMapTests(unittest.TestCase):
    def test_order_and_bounded_input_consumption(self):
        consumed = []
        def source():
            for value in range(20):
                consumed.append(value)
                yield value
        stream = bounded_map(lambda n: n * n, source(), workers=2, pending=4)
        self.assertEqual(next(stream), 0)
        self.assertEqual(len(consumed), 4)
        self.assertEqual(next(stream), 1)
        self.assertEqual(len(consumed), 5)
        self.assertEqual(list(stream), [n * n for n in range(2, 20)])

    def test_worker_error_is_not_silently_dropped(self):
        def fail(value):
            if value == 2:
                raise RuntimeError('invalid item')
            return value
        with self.assertRaisesRegex(RuntimeError, 'invalid item'):
            list(bounded_map(fail, range(10), pending=3))

    def test_invalid_capacity(self):
        for options in ({'workers': 0}, {'pending': 0}):
            with self.assertRaises(ValueError):
                list(bounded_map(str, [1], **options))


if __name__ == '__main__':
    unittest.main()

Run python -m unittest test_bounded_workers.py. The three tests inspect consumed input as well as returned values so the capacity limit is directly observable.

4. A configurable command-line run

File: run_workers.py. Save this complete file without removing lines.

import argparse
import json
from bounded_workers import bounded_map


parser = argparse.ArgumentParser()
parser.add_argument('--count', type=int, default=100)
parser.add_argument('--workers', type=int, default=4)
parser.add_argument('--pending', type=int, default=8)
args = parser.parse_args()
if args.count < 0:
    parser.error('--count must not be negative')
values = list(bounded_map(lambda n: n * n, range(args.count),
                          workers=args.workers, pending=args.pending))
print(json.dumps({'count': len(values), 'sum': sum(values)}))

Run python run_workers.py --workers 2 --pending 4. Expect count 100 and sum 328350. Zero capacity and a negative item count are rejected.

Running the examples and choosing real limits

Save all four files in one directory and run python -m unittest test_bounded_workers.py. The default command-line example reports one hundred results whose sum is 328350. The native map example produces the same result on Python 3.14 or newer and explicitly reports SKIP elsewhere. Start with conservative settings, measure memory and tail latency, and coordinate your limits with the receiving service. Keep the executable tests alongside the application configuration.

Implementation checklist

  • Run input-consumption, ordering and exception tests.
  • Give external calls a real timeout.
  • Configure pending jobs separately from worker count.
  • Exercise the native buffersize example on Python 3.14+.

References

Editorial guidance: Liyan Knowledge.

Published: