Skip to content

Awaiting concurrent.futures Futures in asyncio

Plenty of production libraries predate asyncio and speak concurrent.futures instead: message-queue publishers whose publish() returns a future that resolves when the broker acknowledges, gRPC's future() call style, cloud SDKs with batch uploaders, and internal code built on ThreadPoolExecutor.submit(). Inside an async service, the tempting line is future.result() — which blocks the event loop thread until the other thread finishes, freezing every request in the process. The asyncio Future and the concurrent.futures.Future look alike and are not interchangeable: one is resolved by the event loop on its own thread, the other by any thread at any time, and they have different cancellation semantics. This guide shows how to await the thread-side kind correctly, what cancelling it actually stops, how to handle done-callbacks that fire on foreign threads, and how to combine both kinds in one gather.

Prerequisites

Two futures that look alike 2 columns contrasting asyncio.Future, concurrent.futures.Future. Two futures that look alike asyncio.Future belongs to one event loop resolved on the loop thread await suspends a coroutine cancel() always takes effect callbacks run on the loop concurrent.futures.Future shared between threads resolved by any thread result() blocks the thread cancel() only while queued callbacks run on the worker asyncio.wrap_future links the right-hand kind to the left-hand one.

1. Wrap the future instead of calling result()

asyncio.wrap_future() creates an asyncio future bound to the running loop and links the two: when the thread-side future completes, its result or exception is copied into the asyncio future on the loop thread, using the loop's thread-safe scheduling. Awaiting the wrapper suspends only the current coroutine.

import asyncio
import concurrent.futures
import time

pool = concurrent.futures.ThreadPoolExecutor(max_workers=2)


def legacy_submit(x: int) -> concurrent.futures.Future[int]:
    """Stand-in for a library call that returns a concurrent.futures.Future."""
    return pool.submit(lambda: (time.sleep(0.2), x * 10)[1])


async def main() -> None:
    # legacy_submit(4).result()         # wrong: freezes the event loop for 200 ms
    print(await asyncio.wrap_future(legacy_submit(4)))    # 40, loop stays free

    try:
        await asyncio.wrap_future(pool.submit(lambda: 1 / 0))
    except ZeroDivisionError:
        print("exception crossed the bridge unchanged")


asyncio.run(main())
pool.shutdown()

loop.run_in_executor() already does this wrapping for you — it submits to the executor and returns the asyncio side. Reach for wrap_future() when the concurrent.futures.Future is created by code you do not control.

Verify: run a 10 ms heartbeat task alongside the call. With wrap_future the heartbeat keeps its rhythm; with .result() it stalls for the full 200 ms, which event loop lag measurement would show as a spike.

2. Know what cancellation can and cannot stop

Cancelling the asyncio wrapper propagates to the thread-side future, but concurrent.futures.Future.cancel() only succeeds while the work is still queued. Once a worker thread has started running it, nothing can interrupt it: the wrapper reports cancellation to your coroutine, and the thread carries on to completion.

import asyncio
import concurrent.futures
import time

pool = concurrent.futures.ThreadPoolExecutor(max_workers=2)
finished: list[int] = []


def work(tag: int) -> int:
    time.sleep(0.2)
    finished.append(tag)
    return tag


async def main() -> None:
    running = pool.submit(work, 1)
    await asyncio.sleep(0.05)                     # the worker has started it
    wrapper = asyncio.wrap_future(running)
    wrapper.cancel()
    try:
        await wrapper
    except asyncio.CancelledError:
        print("asyncio side: cancelled")
    print("thread side cancelled:", running.cancelled(), "still running:", running.running())

    await asyncio.sleep(0.3)
    print("work completed anyway:", finished)     # [1]


asyncio.run(main())
pool.shutdown()

The output is asyncio side: cancelled, then thread side cancelled: False still running: True, and finally work completed anyway: [1]. A future that was still waiting in the executor queue, by contrast, is genuinely cancelled and never runs. Design for both outcomes: side effects of "cancelled" work may still happen, so the operation must be safe to complete after the caller has moved on — the same reasoning that idempotency keys formalise for retries.

Verify: your logs show the thread-side future's final state alongside the cancellation, and nothing downstream assumes that a cancelled await means the work did not happen.

Cancelling a wrapped thread-side future A decision on Where was the work when you cancelled with 3 outcomes. Cancelling a wrapped thread-side future Where was the work when you cancelled? still queued in the executor really cancelled the function never runs running in a worker thread await is cancelled the work still completes already finished nothing to cancel result was already set Only the first branch stops side effects; design for the second.

3. Apply timeouts on the asyncio side

Because the wrapper is an ordinary awaitable, asyncio.timeout() bounds it like any other call. The timeout cancels the wrapper, which then follows the rules from step 2: queued work is dropped, running work finishes in the background.

import asyncio
import concurrent.futures


async def publish_with_deadline(publisher_future: concurrent.futures.Future[str],
                                seconds: float) -> str | None:
    try:
        async with asyncio.timeout(seconds):
            return await asyncio.wrap_future(publisher_future)
    except TimeoutError:
        # The broker may still acknowledge later; record it rather than retrying blindly.
        publisher_future.add_done_callback(
            lambda f: None if f.cancelled() else print("late ack:", f.exception() or f.result())
        )
        return None

Attaching a done-callback after the timeout turns an unknown outcome into a logged one. Without it, a late failure on the thread side is never observed, and a late success is indistinguishable from a lost message.

Verify: force the publisher to take longer than the deadline. The coroutine returns None on time, and a late ack line appears when the thread-side work completes.

4. Handle done-callbacks on the right thread

concurrent.futures.Future.add_done_callback() runs the callback in whichever thread completes the future — a worker thread, not the loop thread. Touching asyncio objects from there (setting an asyncio.Event, putting on an asyncio.Queue, resolving an asyncio future) is not thread-safe. Hop onto the loop with call_soon_threadsafe() first.

import asyncio
import concurrent.futures
import threading
import time

pool = concurrent.futures.ThreadPoolExecutor(max_workers=4)


async def collect_acks(n: int) -> list[int]:
    loop = asyncio.get_running_loop()
    acks: asyncio.Queue[int] = asyncio.Queue()

    def on_done(fut: concurrent.futures.Future[int]) -> None:
        assert threading.current_thread() is not threading.main_thread()
        if not fut.cancelled() and fut.exception() is None:
            loop.call_soon_threadsafe(acks.put_nowait, fut.result())   # hop to the loop

    for i in range(n):
        pool.submit(lambda i=i: (time.sleep(0.01 * i), i)[1]).add_done_callback(on_done)

    return [await acks.get() for _ in range(n)]


print(asyncio.run(collect_acks(5)))
pool.shutdown()

This is the pattern to use when a library exposes only callbacks or when you want results in completion order without holding every future. If you control the futures, wrapping them and using asyncio.as_completed() is simpler; the callback form earns its place for high-volume publishers where creating a wrapper per message is wasteful.

A done-callback crossing back to the loop 5 stages from worker done to coroutine. A done-callback crossing back to the loop worker done result set done-callback on worker thread thread-safe hop call_soon_threadsafe loop thread queue.put_nowait coroutine await get() Everything left of the hop must not touch asyncio objects.

Verify: the list contains all five values, and running with PYTHONASYNCIODEBUG=1 produces no "non-thread-safe operation invoked on an event loop other than the current one" errors.

5. Mix both kinds in one gather or TaskGroup

asyncio.gather() accepts awaitables, so wrapped futures and coroutines combine directly. Inside a TaskGroup, wrap first and await inside a small coroutine, because create_task() requires a coroutine rather than a future.

import asyncio
import concurrent.futures
import time

pool = concurrent.futures.ThreadPoolExecutor(max_workers=4)


async def fetch_from_cache(key: str) -> str:
    await asyncio.sleep(0.01)
    return f"cached:{key}"


async def await_cf(fut: concurrent.futures.Future):
    return await asyncio.wrap_future(fut)


async def main() -> None:
    legacy = pool.submit(lambda: (time.sleep(0.05), "legacy:report")[1])
    print(await asyncio.gather(asyncio.wrap_future(legacy), fetch_from_cache("user:1")))

    async with asyncio.TaskGroup() as tg:
        a = tg.create_task(await_cf(pool.submit(lambda: "from-thread")))
        b = tg.create_task(fetch_from_cache("user:2"))
    print(a.result(), b.result())


asyncio.run(main())
pool.shutdown()

When the TaskGroup cancels its children because one failed, each await_cf task cancels its wrapper, which propagates to the thread-side futures under the step 2 rules. Queued work is dropped; running work completes unobserved.

Verify: both prints show the thread result alongside the coroutine result; raising inside fetch_from_cache cancels the group and the legacy futures that had not started report cancelled() == True.

Verification

The bridge is correct when:

  • No blocking waits on the loop: a search for .result( in async code returns only calls on futures already known to be done.
  • Cancellation outcomes are explicit: code and logs distinguish queued work that was dropped from running work that completed after the caller gave up.
  • Timeouts observe late outcomes: thread-side futures abandoned by a timeout still have their result or exception logged.
  • Callbacks hop threads: every callback registered on a concurrent.futures.Future touches asyncio objects only through call_soon_threadsafe().
  • Structured waiting still applies: wrapped futures participate in gather() and TaskGroup with the same failure semantics as coroutines.

Pitfalls & edge cases

  • Wrapping on the wrong loop. wrap_future() binds to the running loop at the time of the call. Wrapping in one loop and awaiting in another raises or hangs; wrap where you await.
  • Process pool futures and large results. Futures from a ProcessPoolExecutor carry pickled results back through a pipe. Awaiting them is non-blocking, but a 500 MB result still costs its transfer; send references as described in reducing pickle overhead.
  • Unobserved exceptions. A thread-side future whose exception nobody retrieves fails silently. Every future you create or receive should be awaited, wrapped, or given a done-callback that logs failures.
  • Executor shutdown during await. Calling executor.shutdown(cancel_futures=True) cancels queued futures, and their wrappers raise CancelledError in the awaiting coroutines. Shut executors down after the tasks that await them.
  • Calling .result(timeout=0) as a poll. Polling a thread-side future from a loop wastes iterations and still raises TimeoutError constantly. Wrap it and await, or use a done-callback.

Frequently Asked Questions

How do I await a concurrent.futures.Future in asyncio?

Call asyncio.wrap_future(future) inside a coroutine and await the result. It returns an asyncio future bound to the running loop that receives the result or exception when the thread-side future completes, so only the awaiting coroutine is suspended. Never call future.result() on the event loop thread, because it blocks the whole loop.

Does cancelling the asyncio wrapper cancel the thread's work?

Only if the work has not started. Cancelling the wrapper calls cancel() on the concurrent.futures.Future, which succeeds while the item is still queued in the executor. Once a worker thread is running the function it cannot be interrupted, so the await raises CancelledError while the work completes in the background.

What is the difference between asyncio.Future and concurrent.futures.Future?

An asyncio.Future belongs to one event loop and must be resolved on that loop's thread; awaiting it suspends a coroutine. A concurrent.futures.Future is thread-safe, can be resolved from any thread, and its result method blocks the calling thread. asyncio.wrap_future and run_coroutine_threadsafe convert between the two.

Is it safe to set an asyncio.Event from a concurrent future callback?

Not directly. Done-callbacks on a concurrent.futures.Future run in the thread that completed it, and asyncio primitives are not thread-safe. Schedule the operation onto the loop with loop.call_soon_threadsafe(event.set), capturing the loop with asyncio.get_running_loop() before registering the callback.