Streaming Tokens From a Blocking SDK

Our pet-health assistant streams its answers token by token. Under load,
something odd showed up: response times for completely unrelated endpoints
climbed in lockstep with how many people were mid-conversation. History reads, a
settings fetch, a suggestion request. CPU sat close to idle. Nothing was
throttled, nothing was retrying, no query was slow.

The cause was four lines of entirely reasonable-looking code. The AWS SDK for
Python is synchronous, and we were iterating a blocking response stream directly
inside a coroutine. For the whole duration of a generation, seconds at a time,
that coroutine never handed the event loop back. One user's answer was
serializing every other request the worker had.

What follows is the fix, in detail, because two of the decisions in it run
against instinct: the queue is deliberately unbounded, and the producer
never raises.
PROBLEM — One blocking iterator, one stalled process

There is no async interface to boto3. For a single request/response call that
is a solved problem: wrap the blocking call in a thread and await the result.

# single call, no stream — fine
payload = await asyncio.to_thread(client.invoke_model, modelId=model, body=body)

Streaming is not that shape. The streaming call returns almost immediately, but
the response it hands back contains a blocking iterator: every next() on it
waits on the socket for the next chunk. The natural translation into an async
generator looks like this, and it is wrong.

# WRONG — the version that stalls the loop
async def stream_answer(...):
    resp = client.converse_stream(**kwargs)      # blocking call
    for event in resp["stream"]:                 # blocking iterator
        if text := _delta(event):
            yield text                          # yields to the caller, not the loop

That last comment is the whole bug. yield in an async generator hands a value
to whoever is iterating it. It does not yield control to the event loop. There is
no await anywhere in that loop body, so from the loop's point of view the
coroutine is running one very long uninterruptible step, and every socket read
inside next() happens with the loop pinned.

This is a genuinely nasty failure to diagnose, because it does not look like a
bug. It looks like a slow model. Latency rises with concurrency, CPU stays flat,
and every trace attributes the time to the LLM call, which is technically true.
The tell is that unrelated endpoints slow down by roughly the amount of
generation happening beside them.

Iterate on a thread, deliver over a queue

The blocking iteration has to happen somewhere that is not the event loop, and
the tokens have to arrive somewhere that is. That is a producer/consumer split
with a thread boundary in the middle, and the boundary has exactly one legal
crossing.

An asyncio.Queue is not thread-safe. Calling put_nowait on it from a
worker thread races on its internals and, worse, will not reliably wake a
consumer already suspended in get(). The supported handoff is to schedule the
mutation on the loop instead.

# the only legal crossing
loop = asyncio.get_running_loop()
queue: asyncio.Queue[str | BaseException | None] = asyncio.Queue()

def emit(item: str | BaseException | None) -> None:
    loop.call_soon_threadsafe(queue.put_nowait, item)

That type annotation is the protocol, and it is worth reading as one. Exactly
three things can cross the boundary: a token, a failure, or
end-of-stream. Nothing else, ever.

The consumer is then an ordinary async generator, and await queue.get() is the
entire difference. That is a real suspension point, so between every single token
the loop is free to run everyone else.

# the consumer side
pump = asyncio.create_task(asyncio.to_thread(pump_stream, client, kwargs, emit))
try:
    while True:
        item = await queue.get()          # a real suspension point
        if item is None:
            return
        if isinstance(item, BaseException):
            raise item
        yield item
finally:
    if pump.done():                       # see "a thread cannot be cancelled"
        await pump

PROCESS Three decisions that are easy to get backwards

1. THE QUEUE IS UNBOUNDED, ON PURPOSE

Every instinct says put a maxsize on it. Unbounded queues are how you get
memory blowups; backpressure is a virtue.

Here, backpressure is a deadlock.

A bounded asyncio.Queue's blocking put is a coroutine. You cannot await it
from a worker thread. You would need
asyncio.run_coroutine_threadsafe(queue.put(item), loop).result(), and that
.result() blocks the producer thread until the consumer drains a slot.

Now consider a client that disconnects mid-answer. The consumer generator is
abandoned, nothing calls get() again, the queue stays full, and the producer
thread blocks forever inside .result(). It never returns to the default
executor's thread pool — and that pool is small, shared, and used by everything
in the process that offloads a blocking call. A few dozen abandoned streams and
every blocking call anywhere in the worker hangs. A dropped mobile connection is
not an edge case; it is Tuesday.

The resolution is to notice the backlog already has a bound, just not on the
buffer. Generation is capped by a max-output-tokens setting, so the worst case is
one capped answer's worth of short strings sitting in memory. Bound the work,
not the buffer — and when you cannot bound the work, do not put the producer's
liveness in the consumer's hands.

1. THE PRODUCER NEVER RAISES

The function driving the stream inside the thread swallows everything:

# the pump · never raises
try:
    resp = client.converse_stream(**kwargs)
    for event in resp["stream"]:
        if text := _delta(event):
            emit(text)
except ClientError as exc:
    emit(StreamError(_kind(exc)))     # the failure, as a value
except Exception:
    emit(StreamError("internal"))
finally:
    emit(None)                        # the sentinel, on every path

Two reasons, both about liveness rather than tidiness.

The sentinel must be in finally. The consumer's loop terminates on exactly
one condition: it received None. If the producer can die without emitting it,
the consumer waits in get() forever, and a vendor error has been converted into
a hung request holding an open connection. A sentinel emitted on the happy path
only is a sentinel that is missing precisely when you need it.

A raising producer leaves an unretrieved exception. On early consumer exit we
do not await the pump task. Had its coroutine raised, Python would report
Task exception was never retrieved at some later garbage-collection point:
noise attached to no request, surfacing minutes after the thing that caused it.
Handing the exception across as a value and re-raising it on the consumer side
also preserves the caller-visible contract exactly. Callers still catch the same
error type from async for, with a traceback on the request that actually caused
it, and a throttle still arrives categorised as rate-limited rather than
collapsing into a generic internal error.

1. A THREAD CANNOT BE CANCELLED

asyncio.to_thread returns a future you can cancel. The thread it wrapped keeps
running regardless — there is no way to interrupt a thread parked in a socket
read. So on early consumer exit there is nothing to stop. The pump will finish
its current read, hit finally, emit into a queue nobody is reading, and exit on
its own.

Which is why the generator's finally only awaits the pump if it has already
finished. Awaiting an in-flight pump there would block closing the generator on
the remainder of a generation the client already walked away from, turning a
disconnect into a request that hangs for as long as the model feels like talking.
And not awaiting it is only safe because the pump cannot raise. Those two
decisions hold each other up; change either one alone and you reintroduce a bug.

RESULTS:

• Unrelated endpoint latency stopped tracking the number of concurrent
  generations. Time-to-first-token is unchanged: the win is entirely in what is
  now allowed to happen alongside a generation.
• One rule now covers every synchronous SDK call in the process: single calls go
  through to_thread, streams go through a pump. There is no third pattern, and
  no remaining place where a blocking iterator is touched from a coroutine.
• A mid-stream disconnect costs one thread until the in-flight generation
  finishes, and costs nothing else. Under the bounded-queue design it would have
  cost a thread permanently.
• Failure categories survive the thread boundary intact, so a throttle is still a
  throttle by the time a caller sees it.

The diagnostic lesson is worth as much as the fix. The signature of a blocked
event loop is latency that scales with concurrency while CPU stays flat, and
unrelated handlers slowing down together. If you see that in a Python service
that talks to any synchronous SDK, go looking for a loop body with no await in
it before you go looking at the vendor.

TAKEAWAYS

1. Only await yields. yield in an async generator hands a value to a
   caller, not control to the loop. A loop body containing no await is one long
   uninterruptible step no matter how many yields are in it.
2. A blocking iterator needs a thread and a queue, not just to_thread.
   Wrapping a single call is easy; a stream needs a producer running
   independently of the consumer.
3. call_soon_threadsafe is the only way into an asyncio.Queue from a
   thread. Not put_nowait, however tempting the fact that it does not need
   awaiting.
4. Backpressure that parks a pooled thread on a consumer you do not control is
   a deadlock. Bound the work instead, and treat thread-pool exhaustion as a
   process-wide outage, because that is what it is.
5. Across a thread boundary, send failures as values and always send a
   sentinel. Exceptions do not cross usefully, and a missing sentinel turns an
   error into a hang.

WHERE THIS FITS AT HOOMANELY

Hoomanely is building connected care for pets: hardware, an app, and the health
intelligence between them, so the people who love an animal notice a change
before it becomes an emergency.

A lot of that intelligence arrives as generated text, and it shares a worker with
everything else the app is doing at that moment. Sensor readings still need
ingesting while an answer streams. So the boundary between our async services and
the synchronous SDKs underneath them is not an incidental detail; it decides
whether one user's long answer is allowed to slow down everyone else's data.
Getting this seam right once, in one place, is what lets the rest of the platform
keep treating a streamed answer as just another request.

AUTHOR'S NOTE

We found this by accident, staring at a latency chart that made no sense. Every
metric we had was pointing at the model. If you take one operational habit from
this: chart p95 for a handler that does no I/O of its own, and watch it while
something else in the process streams. A stalled event loop cannot hide from
that.