Why this 'simple' adapter led me into asyncio's internals

Most of the time, crossing the boundary between sync and async code is not something you have to think about.

You use async libraries in async applications, synchronous libraries in synchronous ones, and if something does not fit, you usually replace it.

The interesting cases are the ones where you cannot.

Maybe an existing consumer depends on a synchronous API. Maybe you are migrating a component to async without breaking everyone using it. Maybe the library you need is synchronous and there is no good alternative.

At that point, you end up building a compatibility layer. And while the layer itself might look simple, it often forces you to understand what is happening underneath.

That was the situation I ran into. I needed to use a synchronous library that streams large files to object storage from an async Starlette application. The obvious solutions worked until they didn’t: buffering changed the behavior, threads introduced new coordination problems, and suddenly the question became less about uploads and more about how async and sync code actually communicate.

It turned into a surprisingly fun exercise in understanding threads, event loops, and the mechanics behind asyncio.

The setup

The setup is small enough to fit in a few lines. On one side, the synchronous uploader I had to keep using:

from collections.abc import Iterator

def upload_streamed(key: str, chunks: Iterator[bytes]) -> None:
    """Uploads `chunks` to `key`, one part at a time. Never buffers the whole payload."""
    for chunk in chunks:
        _upload_part(key, chunk)
    _complete_multipart(key)

The upload has to stay streamed: get a chunk, upload it, get the next one. Nothing about materialising the whole payload in memory.

On the other side, the Starlette handler that has to feed it:

from starlette.requests import Request
from starlette.responses import Response

async def upload(request: Request) -> Response:
    chunks = request.stream()        # AsyncIterator[bytes]
    upload_streamed("my-key", chunks)
    return Response(status_code=204)

Do you see the incompatibility? request.stream() returns an AsyncIterator[bytes]; upload_streamed wants an Iterator[bytes].

They look similar but do not speak the same language. A sync iterator hands you the next value directly. An async iterator hands you a coroutine that has to be awaited inside an event loop.

A step back: async vs sync

Synchronous code blocks the thread it runs on. When it makes an I/O call, the OS suspends the thread until data is ready.

Asynchronous code in Python’s asyncio doesn’t block the thread — it suspends the coroutine. A single thread runs an event loop that multiplexes thousands of coroutines. When a coroutine wants to do I/O, it registers interest in a file descriptor with epoll¹ and yields control back to the loop. The loop runs other coroutines until the kernel reports the descriptor is ready.

For a visual walkthrough of the event loop internals, the talks by David Beazley and Łukasz Langa are highly recommended.

Off-loading sync code to a thread

When you need to call blocking sync code from an async handler, the standard move is to off-load it to a worker thread. The textbook example is wrapping a synchronous HTTP client:

import asyncio
import requests  # synchronous

async def fetch_user(user_id: int) -> dict:
    response = await asyncio.to_thread(requests.get, f"/api/users/{user_id}")
    return response.json()

The loop tells the thread to start; the thread tells the loop when it is done. How that back-and-forth actually works is hidden by the standard library for a single call, and becomes the central question once a single call is no longer enough.

Where do iterators live in our code?

sync_iter:  Iterator[bytes]       = iter(...)        # __next__()
async_iter: AsyncIterator[bytes]  = request.stream() # __anext__() returns a coroutine

A sync iterator lives wherever the thread that calls next(it) lives. The call blocks the calling thread until a value is produced.

An async iterator is tied to an event loop. __anext__() returns a coroutine, and that coroutine has to be driven by a running loop to produce a value. In our case, that loop is Starlette’s, on the main thread.

The two iterators therefore live in different worlds:

• The async iterator lives on the event-loop thread.
• The sync iterator has to be consumable from the worker thread that runs upload_streamed.

Failed attempt #1: just materialise it

Can’t we convert the async iterator by loading the whole payload, then yield chunk by chunk from a sync iterator?

That defeats the entire point. We would be buffering the whole file in memory to “stream” it.

Failed attempt #2: spawn a new event loop in the worker thread

Then just pass the async iterator to the worker thread and call anext() there.

The idea is a small wrapper: take an async iterator, expose it as a sync one. upload_streamed iterates the wrapper as if it were any other Iterator[bytes], and each next() drives one anext() underneath.

import asyncio
from collections.abc import AsyncIterator, Iterator

def async_to_sync(aiter: AsyncIterator[bytes]) -> Iterator[bytes]:
    while True:
        try:
            yield asyncio.run(anext(aiter))
        except StopAsyncIteration:
            return

Wired into the handler:

async def upload(request: Request) -> Response:
    sync_chunks = async_to_sync(request.stream())
    await asyncio.to_thread(upload_streamed, "my-key", sync_chunks)
    return Response(status_code=204)

The wrapper looks synchronous, but it isn’t. Every call to next() turns into await anext(aiter) under the hood.

This is a departure from the usual asyncio.to_thread() example. When you off-load requests.get(), the worker thread runs entirely synchronous code. It doesn’t know about event loops and doesn’t need to. Our wrapper only appears synchronous — producing the next chunk still requires driving an async iterator, and that is why asyncio.run() shows up in the implementation. It creates a fresh event loop in the worker thread and uses it to execute anext().

The problem is that the async iterator does not belong to that new loop.

By the time the handler runs, Starlette is already inside the request lifecycle: a TCP connection has been accepted, associated with the running event loop, and the socket registered with the loop’s selector so incoming bytes get noticed. request.stream() doesn’t open anything new — it is a view over that already-registered connection, owned by the original event loop on the main thread.

We moved consumption of that stream into a worker thread and started a new event loop there. The new loop has no connection. It never registered the socket and isn’t part of the machinery receiving the request data. From its perspective, there is nothing to read from. Meanwhile, the original loop is still receiving bytes — packets arrive, the socket becomes readable, the stream is scheduled to produce the next chunk — all of it on the original loop.

One loop owns the connection and receives the data. Another loop is trying to consume it. In the best case, asyncio notices and raises an error; in the worst case, everything looks valid and nothing makes progress. The worker thread waits for data while the loop that owns the stream waits for someone to drive it.

The rule is:

Asyncio objects are generally expected to be used from the event loop that created them.

Our async iterator was created on Starlette’s event loop. Wherever the synchronous uploader runs, the iterator still has to be driven there. Spawning a loop in the worker is a dead end.

Drive it on the original loop

What we need:

• the async iterator keeps being driven on the original event-loop thread,
• the worker thread can ask for the next chunk and block until it arrives,
• there is a thread-safe channel between the two.

asyncio.run_coroutine_threadsafe exists for exactly this. It is the one function in asyncio designed to be called from a worker thread to drive work on the loop.

import asyncio
from collections.abc import AsyncIterator, Iterator
from concurrent.futures import Future as ThreadFuture
from typing import TypeVar

T = TypeVar("T")

def async_to_sync_iter(
    aiter: AsyncIterator[T],
    loop: asyncio.AbstractEventLoop,
) -> Iterator[T]:
    """Consume `aiter` (driven on `loop`) from a worker thread, one item at a time."""
    while True:
        # Schedule `anext(aiter)` on the original loop, from this worker thread.
        # Returns a concurrent.futures.Future — the *thread-world* flavor.
        cf: ThreadFuture[T] = asyncio.run_coroutine_threadsafe(anext(aiter), loop)
        try:
            yield cf.result()         # blocks this worker thread until the loop produces a value
        except StopAsyncIteration:
            return

Those few lines rely on a fair amount of machinery. The rest of this section is about what run_coroutine_threadsafe actually does.

How does a worker thread talk to an event loop?

Three sub-problems are hiding in that question. It’s easier to work through them one at a time.

1. Why can’t the thread just poke the loop directly?

The obvious move is for the worker thread to append the callback to whatever queue the loop drains from:

loop._ready.append(callback)

Asyncio deliberately does not let you do this. Its internal state — the ready queue, task list, timers — is owned by a single thread. Two threads mutating it at the same time would race with the loop itself.

Only one thing in the whole asyncio API is safe to call from another thread: loop.call_soon_threadsafe(callback). Everything else, including run_coroutine_threadsafe, is built on top of it.

2. Why does call_soon_threadsafe need a trick?

An asyncio event loop is a while True on a single thread, alternating between two things:

1. Drain a ready queue. Callbacks scheduled by loop.call_soon(...) get popped and run, one by one.
2. Poll a selector. When the queue empties, the loop calls selector.select(timeout) — the syscall that wraps epoll. The thread blocks in the kernel until a registered file descriptor is ready.

Everything else — coroutines, tasks, transports — is sugar on top of that.

Now imagine the loop is asleep in step 2, and another thread appends a callback to _ready. Nothing happens. Appending to a Python list does not wake a thread sleeping inside epoll_wait(); the selector will not return until a file descriptor it is watching becomes ready.

Asyncio’s fix is to give the selector something it can notice: an internal pipe, opened at loop construction and registered with the loop’s own selector.

                 event-loop thread
                 ┌───────────────────────────────┐
                 │  selector watches:            │
                 │    • sockets, timers, ...     │
                 │    • wakeup pipe (read end) ──┼──┐
                 │                               │  │
                 │  parked in epoll_wait()       │  │
                 └───────────────────────────────┘  │
                                                    │
worker thread                                       │
──────────────                                      │
call_soon_threadsafe(cb):                           │
  1. append cb to _ready  (under lock)              │
  2. write 1 byte ────────────────────────► pipe ───┘
                                                (readable → selector returns)

call_soon_threadsafe therefore does exactly two things:

1. Lock and append the callback to self._ready.
2. Write one byte to the pipe.

The byte makes the read end readable, the selector returns, the loop wakes up, drains _ready, and runs the callback.

Put another way, an event loop only ever wakes up for two kinds of events. Either the kernel has news (a socket became readable, a timer fired), or another thread has news (through the wakeup pipe). Both arrive as file descriptors becoming ready on the same selector.

3. How does the value come back?

Waking the loop is only half the problem. The worker thread also needs to receive the result the loop produces, and this is where Future comes in — twice.

The two sides want different things from a Future:

• The event loop wants to suspend a coroutine until the value is ready: await future.
• The worker thread wants to block its own thread until the value is ready: future.result().

Asyncio has two Future implementations for exactly this reason:

• asyncio.Future lives entirely on the event loop. It is awaitable, not thread-safe, with lock-free internals.
• concurrent.futures.Future lives entirely in thread-land. Its result() blocks on a condition variable; its set_result() wakes any thread waiting on it.

run_coroutine_threadsafe uses both. It returns a concurrent.futures.Future immediately, so the worker thread has something to block on. On the loop side, it schedules the coroutine as an asyncio.Task and arranges for the task’s result to be copied onto the worker’s future when the task completes.

The full timeline of a single next():

Worker Thread                        Event Loop Thread
─────────────                        ─────────────────

next()
  │
  ▼
run_coroutine_threadsafe()
  │
  ├── append callback to _ready ───►
  ├── write byte to wakeup pipe ───►
  │                                   selector.select() returns
  ▼                                          │
cf.result()                                  ▼
  ⋮                                   drain _ready → run callback
  ⋮                                          │
  ⋮  (worker blocked)                        ▼
  ⋮                                   create Task from coroutine
  ⋮                                          │
  ⋮                                          ▼
  ⋮                                   drive task to completion
  ⋮                                          │
  ⋮                                          ▼
  ⋮                          ◄──────  cf.set_result(chunk)
  │
  ▼
yield chunk

The worker thread never touches the async iterator directly. Each next() is a request to the event loop; each yielded chunk is a reply.

Coming back to to_thread

asyncio.to_thread(func, ...) uses the same three ingredients as run_coroutine_threadsafe — a concurrent.futures.Future, an asyncio.Future, and call_soon_threadsafe — just assembled from the other end.

Both directions cross the boundary the same way: a byte on the wakeup pipe, a Future on each side. The only real difference is who plays the asker. run_coroutine_threadsafe is the only place in asyncio where that asker is the thread.

Putting it together

import asyncio
from starlette.requests import Request
from starlette.responses import Response

from bridge import async_to_sync_iter
from sync_uploader import upload_streamed

async def upload(request: Request) -> Response:
    loop = asyncio.get_running_loop()
    chunks = async_to_sync_iter(request.stream(), loop)

    # The sync uploader runs in a worker thread; each `next(chunks)`
    # round-trips back to `loop` for the next async chunk.
    await asyncio.to_thread(upload_streamed, "my-key", chunks)

    return Response(status_code=204)

The async iterator is now consumed as a sync one without buffering and without spawning a second event loop.

What did it cost?

Per chunk:

• one context switch worker → loop → worker,
• two Future allocations,
• one self-pipe write() syscall to wake the selector.

For a 64 KiB-per-chunk multipart upload, these costs are negligible next to network I/O. At 16 bytes per chunk, they would dominate the transfer. The overhead is per-chunk, so the chunk size is what determines whether it matters.

In a greenfield project this alone would be a reasonable argument for going fully async. Here, the sync library was the constraint, and this is what it costs to keep it in place.

The other direction (an exercise)

Consuming a sync iterator as an async one is easier than what we just did. The loop is in charge, dispatching each next() to a worker thread the way it would any other sync function. No run_coroutine_threadsafe, no thread-initiated work.

Think about which Future class you still need and which one drops out, and what the round-trip per call ends up looking like. Then compare with Starlette’s iterate_in_threadpool.