Skip to content

Resolving Futures from Other Threads Safely

A hardware SDK, a message-bus client or a C extension delivers results by calling your callback on its thread. The natural bridge is an asyncio future: create it, hand future.set_result to the SDK, and await the future. In testing it works. In production, requests occasionally take exactly as long as the next unrelated timer on the loop — two seconds, thirty seconds — although the SDK logged its callback in milliseconds; and under load, InvalidStateError tracebacks appear in the SDK's thread whenever a request timed out just before its reply arrived. Both come from resolving a loop-owned future from a foreign thread. asyncio futures are not thread-safe: setting one from another thread does not wake the loop, and it can race with cancellation on the loop thread. This guide measures the stall, routes every resolution through call_soon_threadsafe, guards against futures that are already done, scales the bridge to many concurrent requests with a correlation table, and handles callbacks that arrive after the loop has shut down.

Prerequisites

Latency of a 100 ms reply, two ways 2 bars comparing direct future.set_result with the others. Latency of a 100 ms reply, two ways direct future.set_result 2.00 s loop.call_soon_threadsafe 0.10 s Measured with the step 1 script on Python 3.14; the direct case waits for the next loop wake-up. Without any other timer, the direct version never resumes at all.

1. Measure the stall from a direct set_result

A thread that calls future.set_result() directly updates the future and schedules its callbacks with the loop's non-thread-safe call_soon(). The loop, asleep in its selector waiting for the next timer or socket event, is not woken, so the awaiting coroutine resumes only when something else wakes the loop.

import asyncio
import threading
import time


def sdk_thread(callback, delay: float) -> None:
    time.sleep(delay)                                  # the device answers after 100 ms
    callback("reading=42")


async def direct() -> None:
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    threading.Thread(target=sdk_thread, args=(future.set_result, 0.1)).start()   # wrong
    loop.call_later(2.0, lambda: None)                 # an unrelated timer, 2 s away
    started = time.monotonic()
    value = await future
    print(f"direct: {value} after {time.monotonic() - started:.2f}s")


async def threadsafe() -> None:
    loop = asyncio.get_running_loop()
    future = loop.create_future()

    def resolve(value: str) -> None:                   # runs on the SDK thread
        loop.call_soon_threadsafe(future.set_result, value)

    threading.Thread(target=sdk_thread, args=(resolve, 0.1)).start()
    started = time.monotonic()
    value = await future
    print(f"threadsafe: {value} after {time.monotonic() - started:.2f}s")


asyncio.run(direct())
asyncio.run(threadsafe())

The direct version returned after 2.00 seconds — the delay of the unrelated timer — although the reply existed after 0.10 seconds. Without that timer it would not return at all. call_soon_threadsafe() schedules the call and writes to the loop's self-pipe, waking the selector immediately, so the threadsafe version returned after 0.10 seconds.

Verify: the direct version's latency equals the delay of the next loop wake-up, not the SDK's delay; the threadsafe version's latency matches the SDK.

2. Guard against futures that are already done

By the time a reply arrives, the awaiting coroutine may have timed out or been cancelled, which cancels the future. Calling set_result() on a cancelled future raises InvalidStateError. If the thread calls it directly, the exception lands in the SDK's thread, often killing its dispatcher; if it is scheduled with call_soon_threadsafe, it lands in the loop's exception handler as noise. Check on the loop thread, at the moment of resolution.

import asyncio


def _settle(future: asyncio.Future, value=None, error: BaseException | None = None) -> None:
    """Runs on the loop thread, so the done() check and the set cannot race."""
    if future.done():                                  # cancelled by a timeout, or already set
        return
    if error is not None:
        future.set_exception(error)
    else:
        future.set_result(value)


def resolver_for(future: asyncio.Future):
    loop = future.get_loop()

    def on_reply(value) -> None:                       # called from any thread
        loop.call_soon_threadsafe(_settle, future, value, None)

    def on_error(exc: BaseException) -> None:
        loop.call_soon_threadsafe(_settle, future, None, exc)

    return on_reply, on_error


async def main() -> None:
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    on_reply, _ = resolver_for(future)
    future.cancel()                                    # the caller gave up first
    on_reply("late reply")                             # no InvalidStateError anywhere
    await asyncio.sleep(0)
    print("cancelled:", future.cancelled())


asyncio.run(main())

Checking future.done() from the SDK thread before scheduling would not be enough: the future could be cancelled on the loop between the check and the set. Doing the check inside the scheduled callback makes the check and the action one uninterruptible step on the loop thread.

Verify: the script prints cancelled: True with no traceback; removing the done() check produces an InvalidStateError report from the loop's exception handler.

3. Correlate many in-flight requests

Real bridges carry many concurrent requests over one callback: the SDK calls on_message(correlation_id, payload) for every reply. Keep a table from correlation ID to future, owned by the loop, register before sending, and remove entries when the future completes for any reason — including timeouts — so the table cannot grow without bound.

import asyncio
import itertools
import threading
import time


class FakeDeviceBus:
    """A thread-based transport that replies on its own thread."""

    def __init__(self, on_message) -> None:
        self._on_message = on_message

    def send(self, correlation_id: int, command: str) -> None:
        def reply() -> None:
            time.sleep(0.01 if command != "slow" else 0.5)
            self._on_message(correlation_id, f"{command}: ok")
        threading.Thread(target=reply, daemon=True).start()


class AsyncBridge:
    def __init__(self) -> None:
        self._loop = asyncio.get_running_loop()
        self._pending: dict[int, asyncio.Future] = {}
        self._ids = itertools.count(1)
        self._bus = FakeDeviceBus(self._on_message)

    def _on_message(self, correlation_id: int, payload: str) -> None:        # SDK thread
        self._loop.call_soon_threadsafe(self._deliver, correlation_id, payload)

    def _deliver(self, correlation_id: int, payload: str) -> None:           # loop thread
        future = self._pending.get(correlation_id)
        if future is not None and not future.done():
            future.set_result(payload)

    async def request(self, command: str, timeout: float) -> str:
        cid = next(self._ids)
        future = self._loop.create_future()
        self._pending[cid] = future
        future.add_done_callback(lambda _f: self._pending.pop(cid, None))    # always cleaned
        self._bus.send(cid, command)
        async with asyncio.timeout(timeout):
            return await future


async def main() -> None:
    bridge = AsyncBridge()
    replies = await asyncio.gather(*(bridge.request(f"read-{i}", 1.0) for i in range(50)))
    print(len(replies), "replies;", "pending entries:", len(bridge._pending))
    try:
        await bridge.request("slow", timeout=0.05)
    except TimeoutError:
        pass
    await asyncio.sleep(0.6)                              # the late reply arrives and is ignored
    print("after a timed-out request, pending entries:", len(bridge._pending))


asyncio.run(main())

Registering the future before send() closes the race in which a very fast reply arrives before the table knows about the request. The done-callback removes the entry on success, failure and cancellation alike; a late reply for a removed ID finds nothing and is dropped. The same pattern underlies request/response protocols over any shared connection, as in implementing a length-prefixed framing protocol.

Verify: fifty concurrent requests all complete, and the pending table is empty both after the batch and after the timed-out request's late reply.

A correlation table on the loop thread 5 stages from register to cleanup. A correlation table on the loop thread register id to future send SDK call reply thread threadsafe hop deliver if not done cleanup done-callback pops Registration happens before sending; removal happens on every outcome.

4. Survive callbacks after the loop has closed

SDK threads outlive event loops: a reply can arrive while the service is shutting down, after asyncio.run() has closed the loop. call_soon_threadsafe() on a closed loop raises RuntimeError: Event loop is closed in the SDK's thread. Treat that as "nobody is listening" and drop the reply, and stop the SDK before closing the loop where the SDK allows it.

import asyncio
import logging

log = logging.getLogger("bridge")


def post_to_loop(loop: asyncio.AbstractEventLoop, callback, *args) -> bool:
    """Schedule on the loop from any thread; False if the loop is gone."""
    try:
        loop.call_soon_threadsafe(callback, *args)
        return True
    except RuntimeError:                                  # "Event loop is closed"
        log.debug("dropped late callback %s: loop closed", getattr(callback, "__name__", callback))
        return False


def main() -> None:
    loop = asyncio.new_event_loop()
    loop.close()
    delivered = post_to_loop(loop, print, "late reply")
    print("delivered:", delivered)                        # False, and no crash in the SDK thread


main()

Catching RuntimeError narrowly around call_soon_threadsafe keeps SDK dispatcher threads alive during shutdown. It is still better to unsubscribe or stop the SDK first, inside the shutdown sequence described in graceful shutdown and signal handling, so replies stop arriving before the loop goes away.

Verify: posting to a closed loop returns False without raising, and a shutdown test that closes the loop while replies are in flight produces no thread tracebacks.

5. Prefer the built-in bridges when they fit

Hand-written future plumbing is justified for callback-based SDKs. When the other side is a plain blocking call or a concurrent.futures.Future, the standard library already does all of the above correctly.

import asyncio
import concurrent.futures
import time


def blocking_read(sensor: str) -> str:
    time.sleep(0.05)
    return f"{sensor}=21.5"


async def main() -> None:
    # A blocking function: to_thread handles the thread, the future and context.
    print(await asyncio.to_thread(blocking_read, "temp"))

    # A library that returns concurrent.futures.Future: wrap it.
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
        cf = pool.submit(blocking_read, "humidity")
        print(await asyncio.wrap_future(cf))


asyncio.run(main())

asyncio.to_thread() and asyncio.wrap_future() both resolve the loop-side future with thread-safe scheduling internally. Reach for call_soon_threadsafe and a correlation table only when the SDK insists on calling you back.

Which bridge for this dependency? A decision on What does the dependency give you with 3 outcomes. Which bridge for this dependency? What does the dependency give you? a blocking function asyncio.to_thread no future handling a concurrent future asyncio.wrap_future await it directly callbacks on its thread call_soon_threadsafe plus a pending table Hand-written plumbing is for the third branch only.

Verify: both values print without any manual future handling, and neither path stalls when the loop has no other timers.

Verification

Cross-thread resolution is correct when:

  • Every resolution from a foreign thread goes through call_soon_threadsafe, and reply latency matches the SDK, not the loop's next timer.
  • Done futures are skipped on the loop thread, so timeouts racing with replies produce no InvalidStateError.
  • Correlation tables stay bounded, with entries removed by a done-callback on every outcome.
  • Late callbacks after shutdown are dropped quietly, without crashing SDK threads.
  • Built-in bridges are used where they apply: to_thread for blocking calls, wrap_future for concurrent futures.

Pitfalls & edge cases

  • Capturing the loop at import time. Resolvers must reference the loop that owns the future — use future.get_loop() rather than a global loop variable.
  • Calling back into the SDK from _deliver. Loop-thread code that calls a blocking SDK method stalls the loop; offload it.
  • Exceptions as results. Pass SDK errors through set_exception with a real exception instance, not a string payload, so await raises as expected.
  • Registering after sending. A reply that beats the registration is dropped and the request times out. Register first.
  • Unbounded fan-in. An SDK that floods the loop with call_soon_threadsafe calls can starve other work; batch deliveries or bound the SDK's in-flight requests.

Frequently Asked Questions

Can I call future.set_result from another thread in asyncio?

You should not. asyncio futures are not thread-safe, and setting one from another thread neither wakes the event loop nor coordinates with cancellation on the loop thread, so the awaiting coroutine may resume late or never. Use loop.call_soon_threadsafe to schedule the set_result on the loop thread instead.

Why does my coroutine wake up late when a thread resolves its future?

The thread resolved the future directly, which scheduled the wake-up with the loop's non-thread-safe call_soon. The loop was blocked in its selector and was only woken by the next unrelated timer or socket event. call_soon_threadsafe writes to the loop's wake-up pipe so the coroutine resumes immediately.

How do I avoid InvalidStateError when a reply arrives after a timeout?

Schedule the resolution on the loop thread and check future.done() inside that scheduled callback before setting a result or exception. A future cancelled by a timeout or already resolved is then skipped. Checking from the foreign thread is not enough, because the state can change between the check and the set.

What happens if a thread calls call_soon_threadsafe after the loop is closed?

It raises RuntimeError with the message Event loop is closed in the calling thread. Catch that error narrowly around the call and drop the callback, since nothing is waiting for it any more, and stop or unsubscribe the thread-based SDK before closing the loop during shutdown.