gh-134173: Optimize concurrent.futures→asyncio state transfer with atomic snapshot

[bdraco]

👋 from PyCon @ Pittsburgh

This PR significantly improves performance when transferring future state from concurrent.futures.Future to asyncio.Future, a common operation when dispatching executor jobs in asyncio applications.

The current _copy_future_state implementation requires multiple method calls and lock acquisitions to retrieve the source future's state:

1. done() - acquires lock to check state
2. cancelled() - acquires lock again
3. exception() - acquires lock to get exception
4. result() - acquires lock to get result

Each method call involves thread synchronization overhead, making this operation a bottleneck for high-frequency executor dispatches.

Our use case involves dispatching a large number of small executor jobs from asyncio to a thread pool. These jobs typically involve open or stat on files that are already cached by the OS, so the actual I/O returns almost instantly. However, we still have to offload them to avoid blocking the event loop, since there's no reliable way to determine in advance whether a read will hit the cache.

As a result, the majority of the overhead isn't from the I/O itself, but from the cost of scheduling. Most of the time is spent copying future state, which involves locking. This PR reduces that overhead, which has a meaningful impact at scale.

Add a new _get_snapshot() method to concurrent.futures.Future that atomically retrieves all state information in a single lock acquisition:

• Returns tuple: (done, cancelled, result, exception)
• Uses optimized fast path for already-finished futures (no lock needed)
• Provides atomic state capture for other states

The _copy_future_state function in asyncio now uses this snapshot method to retrieve the state from the concurrent.future.Future

Benchmark results show dramatic improvements for the common case:

• concurrent.futures→asyncio transfer: 4.40x faster
• asyncio→asyncio transfer: while originally tested, this was removed as its always copying a concurrent.futures per gh-134173: Optimize concurrent.futures→asyncio state transfer with atomic snapshot #134174 (comment)

This optimization particularly benefits applications that:

• Dispatch many small executor jobs (e.g., filesystem operations, DNS lookups)

• Use thread pools for I/O-bound operations in asyncio

• Have high frequency of executor task completion

• Adds _get_snapshot() to concurrent.futures.Future for atomic state retrieval

• Updates _copy_future_state() to use the snapshot method

• Maintains full backwards compatibility with existing code

• Minimal code changes with focused optimization

These show consistent 4.4x+ speedup for the critical concurrent.futures→asyncio path.

=== 1. Benchmarking concurrent.futures -> asyncio ===
Writing benchmark scripts...

=== Benchmarking concurrent.futures -> asyncio ===
Running original...
concurrent_to_asyncio: Mean +- std dev: 977 ns +- 13 ns
Running optimized...
concurrent_to_asyncio: Mean +- std dev: 222 ns +- 3 ns

Comparison:
Mean +- std dev: [concurrent_original] 977 ns +- 13 ns -> [concurrent_optimized] 222 ns +- 3 ns: 4.40x faster

Cleaning up...

import pyperf
import concurrent.futures
import asyncio
import subprocess
import os
import sys

def write_benchmark_scripts():
    """Write individual benchmark scripts for each scenario."""

    # Common helper code
    common_imports = '''
import pyperf
import concurrent.futures
import asyncio

def _convert_future_exc(exc):
    exc_class = type(exc)
    if exc_class is concurrent.futures.CancelledError:
        return asyncio.CancelledError(*exc.args)
    elif exc_class is concurrent.futures.TimeoutError:
        return asyncio.TimeoutError(*exc.args)
    elif exc_class is concurrent.futures.InvalidStateError:
        return asyncio.InvalidStateError(*exc.args)
    else:
        return exc
'''

    # Optimization patch code
    optimization_patch = '''
FINISHED = concurrent.futures._base.FINISHED
CANCELLED = concurrent.futures._base.CANCELLED
CANCELLED_AND_NOTIFIED = concurrent.futures._base.CANCELLED_AND_NOTIFIED

def _get_snapshot_implementation(self):
    """Get a snapshot of the future's current state."""
    # Fast path: check if already finished without lock
    if self._state == FINISHED:
        return True, False, self._result, self._exception

    # Need lock for other states since they can change
    with self._condition:
        if self._state == FINISHED:
            return True, False, self._result, self._exception
        if self._state in {CANCELLED, CANCELLED_AND_NOTIFIED}:
            return True, True, None, None
        return False, False, None, None

concurrent.futures.Future._get_snapshot = _get_snapshot_implementation
'''

    # Original copy implementation (for concurrent.futures.Future)
    original_copy = '''
def copy_future_original(source, dest):
    """Original implementation using individual method calls."""
    if dest.cancelled():
        return

    if hasattr(source, 'done'):
        assert source.done()

    if source.cancelled():
        dest.cancel()
    else:
        exception = source.exception()
        if exception is not None:
            dest.set_exception(_convert_future_exc(exception))
        else:
            result = source.result()
            dest.set_result(result)
'''

    # Optimized copy implementation (matches current code)
    optimized_copy = '''
def copy_future_optimized(source, dest):
    """Optimized implementation using _get_snapshot."""
    if dest.cancelled():
        return
    assert not dest.done()
    done, cancelled, result, exception = source._get_snapshot()
    assert done
    if cancelled:
        dest.cancel()
    elif exception is not None:
        dest.set_exception(_convert_future_exc(exception))
    else:
        dest.set_result(result)
'''

    # 1. concurrent.futures -> asyncio (original)
    with open('bench_concurrent_to_asyncio_original.py', 'w') as f:
        f.write(common_imports + original_copy + '''
source = concurrent.futures.Future()
source.set_result(42)
loop = asyncio.new_event_loop()

def task():
    """Single copy operation benchmark."""
    dest = asyncio.Future(loop=loop)
    copy_future_original(source, dest)
    dest.cancel()

runner = pyperf.Runner()
runner.bench_func('concurrent_to_asyncio', task)
''')

    # 2. concurrent.futures -> asyncio (optimized)
    with open('bench_concurrent_to_asyncio_optimized.py', 'w') as f:
        f.write(common_imports + optimization_patch + optimized_copy + '''
source = concurrent.futures.Future()
source.set_result(42)
loop = asyncio.new_event_loop()

def task():
    """Single copy operation benchmark."""
    dest = asyncio.Future(loop=loop)
    copy_future_optimized(source, dest)
    dest.cancel()

runner = pyperf.Runner()
runner.bench_func('concurrent_to_asyncio', task)
''')

def run_benchmarks():
    """Run all benchmarks and compare results."""
    print("Writing benchmark scripts...")
    write_benchmark_scripts()

    # Clean up old results
    for f in ['concurrent_original.json', 'concurrent_optimized.json']:
        if os.path.exists(f):
            os.remove(f)

    print("\n=== Benchmarking concurrent.futures -> asyncio ===")
    print("Running original...")
    subprocess.run([sys.executable, 'bench_concurrent_to_asyncio_original.py',
                   '-o', 'concurrent_original.json', '--quiet'])

    print("Running optimized...")
    subprocess.run([sys.executable, 'bench_concurrent_to_asyncio_optimized.py',
                   '-o', 'concurrent_optimized.json', '--quiet'])

    print("\nComparison:")
    subprocess.run([sys.executable, '-m', 'pyperf', 'compare_to',
                   'concurrent_original.json', 'concurrent_optimized.json'])

    # Clean up
    print("\nCleaning up...")
    for f in ['bench_concurrent_to_asyncio_original.py',
              'bench_concurrent_to_asyncio_optimized.py']:
        if os.path.exists(f):
            os.remove(f)

    print("\n=== Summary ===")
    print("concurrent.futures -> asyncio: Should show significant speedup with _get_snapshot")

if __name__ == "__main__":
    run_benchmarks()

• Issue: concurrent.futures→asyncio state transfer is a bottleneck #134173

[bdraco]

Thanks!