Python Basics: Collecting Thread-Pool Results with concurrent.futures

Several independent file reads can share a small thread pool instead of creating one thread per file. concurrent.futures provides an executor that schedules functions and Future objects that hold their eventual results or exceptions.

The examples cover ordered collection, per-job failure handling, and cancelling a job that has not started. They demonstrate coordination rather than claim a speedup: the workloads are intentionally tiny and repeatable.

Prepare a scratch directory

These three standalone examples were tested on Linux with Python 3.12.14. The examples require Python 3.6 or newer because they use f-strings. They use ThreadPoolExecutor and Future APIs available before that version, without newer map buffersize or shutdown cancel_futures options. Use Bash, a text editor, and the standard mktemp, rm, and rmdir utilities. No third-party packages are required. Keep assertions enabled by running without Python’s -O option.

Create a fresh directory and keep using the same terminal:

start_dir="$PWD"
work_dir="$(mktemp -d)" || exit 1
cd -- "$work_dir" || exit 1

Save each named source file here as UTF-8.

Avoid source filenames such as concurrent.py, pathlib.py, tempfile.py, or threading.py. The first example creates and automatically removes its own temporary text files. The other two use memory only.

1. Read local files while preserving input order

Save as count_files_in_pool.py:

from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from tempfile import TemporaryDirectory


def count_lines(path):
    return path.name, len(path.read_text(encoding="utf-8").splitlines())


with TemporaryDirectory() as directory:
    root = Path(directory)
    paths = []
    for name, content in [("first.txt", "one\ntwo\n"), ("empty.txt", "")]:
        path = root / name
        path.write_text(content, encoding="utf-8")
        paths.append(path)
    with ThreadPoolExecutor(max_workers=2) as pool:
        results = list(pool.map(count_lines, paths))
        assert list(pool.map(count_lines, [])) == []
    assert results == [("first.txt", 2), ("empty.txt", 0)]
    for name, total in results:
        print(f"{name}: {total} lines")

assert not root.exists()
print("Temporary files removed:", not root.exists())

Run it:

python3 count_files_in_pool.py

Output:

first.txt: 2 lines
empty.txt: 0 lines
Temporary files removed: True

map returns results in input order, even if a later read finishes first. The empty file has zero lines, and mapping over an empty list produces an empty result. All readers finish before the temporary directory is removed.

The inner with block closes the executor and waits for its work. The outer block owns the files. This nesting prevents cleanup from racing unfinished readers. For this tiny batch, a regular loop would also be reasonable; use measurements on realistic work before choosing a pool for performance.

2. Collect a failure without losing other results

Save as collect_job_failures.py:

from concurrent.futures import ThreadPoolExecutor, as_completed


def parse_quantity(text):
    if not text.isdecimal():
        raise ValueError("quantity must contain decimal digits")
    return int(text)


inputs = ["12", "bad", "0"]
outcomes = {}
with ThreadPoolExecutor(max_workers=2) as pool:
    futures = {pool.submit(parse_quantity, text): index
               for index, text in enumerate(inputs)}
    for future in as_completed(futures):
        index = futures[future]
        try:
            quantity = future.result()
        except ValueError as error:
            outcomes[index] = f"rejected: {error}"
        else:
            outcomes[index] = f"accepted: {quantity}"

assert outcomes == {
    0: "accepted: 12",
    1: "rejected: quantity must contain decimal digits",
    2: "accepted: 0",
}
for index, text in enumerate(inputs):
    print(f"{text}: {outcomes[index]}")

Run it:

python3 collect_job_failures.py

Output:

12: accepted: 12
bad: rejected: quantity must contain decimal digits
0: accepted: 0

submit returns one Future per call. as_completed exposes whichever futures finish next; result returns a value or raises the function’s exception. This example catches the expected validation error for each job and continues collecting the remaining outcomes.

The final print loop restores input order using the saved indexes, rather than depending on thread timing. Zero is accepted, while bad is rejected with an intentional application message. Unexpected exception types are not silently converted to ordinary validation failures.

3. Cancel queued work without stopping a running thread

Save as cancel_waiting_job.py:

from concurrent.futures import CancelledError, ThreadPoolExecutor
from threading import Event


started = Event()
release = Event()


def hold_worker():
    started.set()
    release.wait()
    return "active job finished"


with ThreadPoolExecutor(max_workers=1) as pool:
    active = pool.submit(hold_worker)
    try:
        assert started.wait(timeout=10), "Worker did not start"
        waiting = pool.submit(str.upper, "queued job")
        assert active.cancel() is False
        cancelled = waiting.cancel()
        assert cancelled is True
        print("Queued job cancelled:", cancelled)
        try:
            waiting.result()
        except CancelledError:
            print("Queued result: CancelledError")
        else:
            raise AssertionError("A cancelled job produced a result")
    finally:
        release.set()
    assert active.result(timeout=10) == "active job finished"
    print(active.result())

assert waiting.cancelled()

Run it:

python3 cancel_waiting_job.py

Output:

Queued job cancelled: True
Queued result: CancelledError
active job finished

One worker is deliberately held by an Event, so the second job cannot start. Cancelling that queued Future succeeds and retrieving its result raises CancelledError. The running Future refuses cancellation.

The finally block always releases the active worker, including if a check fails. This avoids hanging executor shutdown while a thread waits for an event nobody will set. A result timeout limits how long the caller waits; it does not terminate the underlying function. Long-running tasks need their own cooperative stopping mechanism.

Do not have a worker wait for another job in the same exhausted pool: that can deadlock. Threads are useful for overlapping blocking I/O. In the tested CPython 3.12 runtime, the GIL means ordinary CPU-bound Python code does not gain multi-core execution merely by using a thread pool.

Clean up

In the same terminal, remove only these source files, return to the original directory, and remove the now-empty scratch directory:

rm -- count_files_in_pool.py collect_job_failures.py cancel_waiting_job.py
cd -- "$start_dir" || exit 1
rmdir -- "$work_dir"
unset start_dir work_dir

If rmdir reports that anything remains, inspect the directory rather than deleting unknown files.

References

• Python 3.12 concurrent.futures documentation
• Future cancellation
• CPython 3.12 threading and the GIL

Acknowledgements

This post was created with assistance from AI.