Using concurrent.futures as_completed in Python

Python's concurrent.futures.as_completed() function is an iterator that yields Future instances as they finish executing, rather than waiting for them to complete in the order they were submitted. This article explains how as_completed() operates under the hood, how to implement it using ThreadPoolExecutor and ProcessPoolExecutor, and why it provides superior throughput and responsiveness when handling asynchronous or concurrent workloads.

What is as_completed()?

When working with concurrent.futures, submitting tasks via an executor returns Future objects representing pending operations. Standard batch processing methods, such as executor.map(), yield results strictly in the order tasks were initiated, which can cause execution bottlenecks if early tasks take longer than later ones.

as_completed() solves this issue. It accepts an iterable of Future objects and returns an iterator that yields each future the moment its state changes to finished or cancelled.

How as_completed() Works Internally

1. State Tracking with Waiters: When you pass a collection of futures to as_completed(), an internal _Waiter object is registered with each future. This waiter utilizes threading locks and condition variables (threading.Condition).
2. Notification on Completion: When a worker thread or process finishes its assigned callable, it updates the future's state to FINISHED (or CANCELLED) and invokes the callback attached to the waiter.
3. Queue-Based Yielding: The completed future is placed into an internal queue monitored by the waiter. The as_completed() generator polls this queue, yielding completed futures immediately back to the caller loop without waiting for the remaining tasks.
4. Timeout Handling: You can supply an optional timeout argument. If the timeout expires before the next future finishes, as_completed() raises a concurrent.futures.TimeoutError.

Basic Implementation Example

The following example demonstrates how tasks with variable completion times are yielded out of submission order:

import time
import random
from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_data(task_id, delay):
    time.sleep(delay)
    return f"Task {task_id} completed in {delay}s"

delays = [3, 1, 4, 2]

with ThreadPoolExecutor(max_workers=4) as executor:
    # Submit tasks and map each Future to its task_id
    future_to_id = {
        executor.submit(fetch_data, i, d): i 
        for i, d in enumerate(delays)
    }

    # Process results as they become available
    for future in as_completed(future_to_id):
        task_id = future_to_id[future]
        try:
            result = future.result()
            print(result)
        except Exception as exc:
            print(f"Task {task_id} generated an exception: {exc}")

Output Order: Even though Task 0 was submitted first, Task 1 (1 second) and Task 3 (2 seconds) will be yielded first because they finish earlier.

Error Handling with as_completed()

Unlike standard sequential loops, exceptions raised within worker threads or processes are captured by the Future object rather than raised immediately.

When iterating with as_completed():

• The iterator itself does not raise the task's exception.
• Calling future.result() inside the loop retrieves the return value or raises the original exception caught during execution.
• Wrapping future.result() in a standard try/except block allows you to handle failures on a per-task basis without interrupting the processing of other running tasks.

Key Differences: as_completed() vs executor.map()

When to Use as_completed()

Use as_completed() when:

• Tasks have unpredictable durations (e.g., HTTP requests, database queries, file downloads).
• You want to process, stream, or store results incrementally as soon as they become ready.
• You need precise control over task exceptions and metadata mapping for individual tasks.