Async in, sync out

Most of the time, you don’t have to think about the line between sync and async code. Async libraries go into async apps, sync libraries into sync ones, and if something doesn’t fit, you replace it.

It gets interesting when you can’t.

Maybe an existing consumer depends on a synchronous API. Maybe you’re moving a component to async without breaking everyone who uses it. Or the library you need is synchronous and there’s no good alternative. Then you build a small adapter, and even a small one can force you to understand what’s going on underneath.

That’s where I ended up. I had to use a synchronous library that streams large files to object storage, from inside an async Starlette app. The obvious solutions worked until they didn’t: buffering changed the behavior, threads brought their own coordination problems, and the question became less about uploads and more about how async and sync code actually talk to each other.

The setup

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: take a chunk, upload it, take the next one. The whole payload never sits 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 problem? request.stream() returns an AsyncIterator[bytes], and upload_streamed wants an Iterator[bytes].

They look alike, but they don’t speak the same language. A sync iterator hands you the next value directly. An async iterator hands you a coroutine, and that coroutine 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 the data is ready.

Async code in asyncio doesn’t block the thread. It suspends the coroutine. A single thread runs an event loop that juggles thousands of coroutines. When a coroutine waits for I/O, the loop registers the file descriptor with epoll¹ and runs other coroutines until the kernel reports that the descriptor is ready.

If you want to see an event loop built up step by step, the talks by David Beazley and Łukasz Langa are great.

Off-loading sync code to a thread

To call blocking sync code from an async handler, the usual move is to off-load it to a worker thread. The textbook example wraps 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, and the thread tells the loop when it’s done. For a single call, the standard library hides how that back-and-forth works. Once one call isn’t enough, it becomes the central question.

Where do the iterators live?

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 that thread until a value is ready.

An async iterator is tied to an event loop. __anext__() returns a coroutine, and a running loop has to drive that coroutine to get a value. Here, that loop is Starlette’s, on the main thread.

So the two iterators live in different worlds:

• The async iterator lives on the event loop’s thread.
• The sync iterator has to work from the worker thread that runs upload_streamed.

Failed attempt 1: just load it all

Can’t we read the whole payload first, then hand it out chunk by chunk from a sync iterator?

That defeats the point. We’d buffer the whole file in memory just to “stream” it.

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

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

The idea is a small wrapper that takes an async iterator and exposes it as a sync one. upload_streamed iterates the wrapper like 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 next() turns into await anext(aiter) under the hood.

That’s different from the usual to_thread example. When you off-load requests.get(), the worker thread runs purely synchronous code and never needs an event loop. Our wrapper only looks synchronous: getting the next chunk still means driving an async iterator. That’s why asyncio.run() shows up. It creates a fresh event loop in the worker thread and runs anext() on it.

The problem is that the async iterator doesn’t belong to that new loop.

By the time the handler runs, Starlette has accepted the TCP connection and registered its socket with the running loop’s selector, so incoming bytes get noticed. request.stream() doesn’t open anything new. It’s a view over that connection, owned by the original loop on the main thread.

The new loop in the worker has no connection. It never registered the socket, so from its point of view there’s nothing to read. Meanwhile the original loop keeps receiving bytes: packets arrive, the socket becomes readable, the stream is ready to produce the next chunk. All of that happens on the original loop.

One loop owns the connection and receives the data, and another one tries to consume it. At best, asyncio notices and raises an error. At worst, everything looks valid and nothing moves: the worker waits for data while the loop that owns the stream waits for someone to drive it.

The rule:

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

Our async iterator was created on Starlette’s loop. Wherever the uploader runs, the iterator still has to be driven there. A second loop is a dead end.

Drive it on the original loop

What we need:

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

asyncio.run_coroutine_threadsafe exists for exactly this. It’s the function in asyncio meant to be called from another thread to run work on a 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:
        # Runs anext(aiter) on the original loop and returns a thread-side Future.
        cf: ThreadFuture[T] = asyncio.run_coroutine_threadsafe(anext(aiter), loop)
        try:
            yield cf.result()  # blocks this worker thread until the loop has 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 smaller problems hide in that question. They’re easier one at a time.

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

The obvious move is for the worker thread to add its callback to the queue the loop drains:

loop._ready.append(callback)

Asyncio doesn’t want you to. _ready is private, and most of the loop’s state, like its heap of timers, isn’t safe to touch from another thread.

Exactly one method on the loop is meant to be called from another thread: loop.call_soon_threadsafe(callback). Everything else, including run_coroutine_threadsafe, is built on it.

2. Why does call_soon_threadsafe need a trick?

An asyncio event loop is a while True on one thread, switching between two things:

1. Drain the ready queue. Callbacks scheduled with loop.call_soon(...) are popped and run, one by one.
2. Poll the selector. When the queue is empty, the loop calls selector.select(timeout), which wraps epoll_wait. The thread blocks in the kernel until a registered file descriptor is ready, or until the next timer is due.

Coroutines, tasks and transports are all built on top of that.

Now picture the loop asleep in step 2 while another thread appends a callback to _ready. Nothing happens. Appending to a queue doesn’t wake a thread that’s sleeping in epoll_wait(). The selector only returns when a file descriptor it watches becomes ready.

So asyncio gives the selector something to notice: a wakeup pipe, created with the loop and registered with its own selector. In CPython it’s actually a socket pair, but it works like a pipe.

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

call_soon_threadsafe does exactly two things:

1. Append the callback to _ready. It’s a deque, and appending to one is thread-safe.
2. Write one byte to the wakeup pipe.

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

Put another way, a sleeping loop wakes up for three reasons: a file descriptor is ready, the next timer is due, or another thread wrote to the wakeup pipe. To the selector, that last one looks like any other socket.

3. How does the value come back?

Waking the loop is only half of it. The worker thread also has to receive the result, and that’s 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 until the value is ready: future.result().

That’s why there are two Future classes:

• asyncio.Future lives on the event loop. It’s awaitable and not thread-safe, so it needs no locks.
• concurrent.futures.Future lives in the thread world. result() blocks on a condition variable, and set_result() wakes every thread waiting on it.

run_coroutine_threadsafe uses both. It returns a concurrent.futures.Future right away, so the worker has something to block on. On the loop, it wraps the coroutine in an asyncio.Task and copies the task’s result onto the worker’s future when the task finishes.

The whole 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. Each next() is a request to the loop, and each chunk is the reply.

Coming back to to_thread

asyncio.to_thread(func, ...) uses the same three pieces as run_coroutine_threadsafe: a concurrent.futures.Future, an asyncio.Future and call_soon_threadsafe. It just puts them together from the other end.

Both directions cross the boundary the same way: a byte on the wakeup pipe and a Future on each side. The only difference is who asks. With to_thread, the loop asks a thread for work. With run_coroutine_threadsafe, a thread asks the loop.

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 uploader runs in a worker thread, and each next(chunks)
    # goes 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, with no buffering and no second event loop.

What did it cost?

Per chunk:

• one round trip from the worker to the loop and back,
• two Future objects,
• one write() syscall to wake the selector.

For a multipart upload with 64 KiB chunks, that’s nothing next to the network. At 16 bytes per chunk, it would dominate. The cost is per chunk, so the chunk size decides whether it matters.

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

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 and sends each next() to a worker thread, like any other sync function. No run_coroutine_threadsafe, and no work started by the thread.

Think about which Future class you still need, which one drops out, and what each round trip looks like. Then compare with Starlette’s iterate_in_threadpool.