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¶
- Python 3.11+, standard library only.
- Future fundamentals from Future Objects & Callbacks, and the reverse bridge — callback APIs into awaitables — in bridging callback APIs to async with Futures.
- Thread offloading from Hybrid Concurrency Models: a
concurrent.futures.Futurealmost always means another thread or process is doing the work.
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.
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.
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.Futuretouches asyncio objects only throughcall_soon_threadsafe(). - Structured waiting still applies: wrapped futures participate in
gather()andTaskGroupwith 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
ProcessPoolExecutorcarry 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 raiseCancelledErrorin 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 raisesTimeoutErrorconstantly. 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.
Related¶
- Future Objects & Callbacks — up to the topic overview for futures, callbacks and their lifecycles.
- Calling async code from synchronous code safely — the opposite bridge, from threads into a running loop.
- Asyncio Fundamentals & Event Loop Architecture — the section overview for the loop, tasks and futures.