streaming notes: async def, threadpools, and bridging a synchronous pipeline into SSE

a question answering service sat on a blank spinner for a minute or more per query, and the fix went through four changes that were all really about one seam between synchronous work and an async framework

1. A synchronous pipeline sat behind an async def handler and held the event loop for its entire run. One query in flight per container, ever.
2. Deleting the word async made the server concurrent, because a plain def handler gets handed to a threadpool instead of the loop.
3. Progress events fire from a callback deep in the call stack. A generator has to yield from the top. A single thread cannot do both.
4. A queue and a background thread bridge the two, and the pipeline never learns it is being streamed.
5. Tokens stream out raw and are then thrown away, because the text cleaner only works on a complete answer.
6. Nothing got faster. The waiting stopped being blank, which turned out to be the part that mattered.

The system behind this post answers questions over a large set of financial regulatory filings. A question goes through entity extraction, embedding, vector retrieval, a sentence expansion step, context assembly, and finally one call to a hosted language model that writes the answer. Simple questions took around ten seconds end to end. Comparison questions spanning several companies and years took a minute, and the worst measured case took close to four.

For all of that time the interface showed a spinner and nothing else. The work was progressing perfectly and the screen had no way to say so.

the seam everything else sits on

The pipeline is ordinary synchronous Python. It calls a cloud SDK, waits, parses, calls again. There is not a single await anywhere in it. The web framework in front of it is asynchronous. Every problem in this post lives at that join.

      synchronous pipeline
      blocking SDK calls, no await points
                  |
    ==============|==============  the seam
                  |
      async web framework
      one event loop, cooperative switching

Four small changes were made across one afternoon, in an order that mattered.

what async actually promises

A server needs to serve many requests without one thread per request. There are two ways to get there.

The first is to give the operating system the problem. Each request runs on its own thread, and the OS interrupts whichever thread it likes whenever it likes. Preemption is involuntary, so a thread that blocks on a network read hurts only itself.

The second is an event loop. One thread runs a list of things that are ready to make progress, picks one, runs it until it says it is waiting on something, and moves to the next. The switching is cooperative, and await is the entire cooperation protocol. It means: this is going to take a while, hand the thread to something else, come back when the result lands.

That protocol has one requirement, which is that every coroutine actually reaches an await. A coroutine containing none is a normal function wearing a keyword. It runs start to finish without offering the loop a single opportunity to do anything else.

Which is what was happening.

async def, one blocking call inside

  event loop  [===== query, 9.6s+ =====]
  /health         blocked ..........
  query 2         blocked ..........

Three consequences followed, and the third is the one that mattered most.

The fix was to delete four characters.

# before
@app.post("/query")
async def query_endpoint(request: QueryRequest):
    return answer_query(request.question)

# after
@app.post("/query")
def query_endpoint(request: QueryRequest):
    return answer_query(request.question)

Frameworks in this family inspect the handler’s declaration and route it accordingly. A coroutine is trusted to yield and is run directly on the loop. A plain function is assumed to block and is handed to a threadpool, where the operating system can preempt it. Declaring the handler synchronous describes what it actually does, which lets the framework schedule it correctly.

plain def, handed to a pool

  event loop  ..free.. ..free.. ..free..
  worker 1    [==== query 1, 9.12s ====]
  worker 2    [==== query 2, 9.66s ====]
  /health       ^2ms    ^9ms    ^1ms

Measured against the running service: while one real query was in flight for 10.14 seconds, three health probes returned 200 in 1.5 to 9.6 milliseconds. Two real queries fired at the same time finished at 9.12 and 9.66 seconds, overlapping almost entirely rather than adding up to eighteen.

concurrency that arrives is concurrency that has to be handled

An earlier change in the same sequence had started caching the component graph. Building it per request cost 1164.7 milliseconds of constructors and table loads that were discarded when the request returned, so it moved to a process-wide cache. Second and third calls measured 0.0 milliseconds.

That cache had a consequence nobody was looking for. The query logger became a single shared instance, and its append is a download, modify, and reupload against object storage with no lock and no conditional write. Two concurrent appends would read the same object, and one would overwrite the other’s row.

The race was unreachable while the event loop serialised everything. Freeing the loop made it reachable. A lock went around the five logging call sites before the hazard went live rather than after, and the concurrency test above confirmed it held: two simultaneous queries, two rows, written 161 milliseconds apart, nothing lost.

the callback that cannot yield

Timing instrumentation already existed inside the pipeline, recording durations at six points. Turning those into events needed an optional callback, defaulted to None so that every existing caller kept byte-identical behaviour.

def run_retrieval(query, components, on_progress=None):

    def emit(stage, **detail):
        if on_progress is not None:
            on_progress(stage, detail)

    t0 = now()
    entities = extract_entities(query)
    emit("entities", ms=elapsed(t0), companies=entities.companies)

    t0 = now()
    hits = retrieve(embedding, filters)
    emit("retrieve", ms=elapsed(t0), n_hits=len(hits))

Six added lines and a helper. No control flow changed.

Then the actual problem appears. That callback fires four levels down a call stack, and it fires by calling. A generator produces values by yielding, and it yields from the top. These are opposite directions, and one call stack cannot run in both at once.

  the pipeline pushes:

    retrieve()  --calls-->  on_progress(...)

  the generator pulls:

    caller  --asks-->  yield

  one thread, one stack, cannot do both

Waiting for the pipeline to finish and then replaying the events defeats the purpose, since they exist to be seen while the work is still running.

The bridge is a queue and a second thread. The pipeline runs on the worker, the callback writes into the queue, and the generator sits in the calling thread draining it.

def answer_query_stream(query):
    events = Queue()

    def on_progress(stage, detail):
        try:
            events.put({"type": "stage", "stage": stage, **detail})
        except Exception:
            log.error("progress bridge failed, non-fatal")

    def worker():
        context = build_context(query, on_progress=on_progress)

        for kind, payload in llm.invoke_stream(context):
            if kind == "text":
                events.put({"type": "token", "text": payload})
            elif kind == "final":
                result = package(payload)

        with LOG_LOCK:
            logger.log_query(result)

        events.put({"type": "replace", "text": result["answer"]})
        events.put({"type": "done", "metadata": result["metadata"]})

    thread = Thread(target=worker, daemon=True)
    thread.start()

    try:
        while True:
            item = events.get()
            yield item
            if item["type"] in ("error", "done"):
                break
    finally:
        thread.join(timeout=5.0)

  worker thread          calling thread

    pipeline
       |  on_progress
       v
    queue.put()  ------->  queue.get()
                              |
                              v
                            yield

Two details in that shape are worth pulling out.

The loop has no sentinel value and no liveness check. It stops on the content of an ordinary event, whichever one carries a type of done or error. The terminator travels in band, in the same channel and the same format as the payload, which means the protocol has exactly one kind of thing in it.

The guard around events.put exists because the callback runs inside retrieval. A formatting mistake in the reporting layer would otherwise propagate up and take down the query it was reporting on. The guard was written down as a requirement before any consuming code existed, so that a bug in the reporting could never reach the work being reported.

the four layers, end to end

Once events exist, the rest is transport. Four layers, each one thin.

orchestrator    worker thread, queue, generator
     |          yields plain dicts
     v
streaming route
     |          data: {json}\n\n
     v          media type text/event-stream
http client
     |          stream=True, iterate lines
     v          parse back into dicts
interface       status label, growing text

The route wraps each dict in the server-sent events framing and hands the generator to a streaming response.

@app.post("/query/stream")
def query_stream_endpoint(request):
    def event_source():
        for event in answer_query_stream(request.question):
            yield f"data: {json.dumps(event)}\n\n"

    return StreamingResponse(event_source(),
                             media_type="text/event-stream")

The client reverses it, and stream=True is the load-bearing argument. Without it the HTTP library collects the whole body before returning anything, and the stream becomes a slow way to send one message.

def query_stream(self, question):
    response = requests.post(url, json=payload, stream=True)
    for line in response.iter_lines(decode_unicode=True):
        if not line or not line.startswith("data: "):
            continue
        yield json.loads(line[len("data: "):])

The interface drives two widgets from one loop.

status = st.status("Starting...")
placeholder = st.empty()
text = ""

for event in client.query_stream(question):
    kind = event["type"]

    if kind == "stage":
        label = f"{event['stage']} ({event['ms']:.0f} ms)"
        status.update(label=label)

    elif kind == "token":
        text += event["text"]
        placeholder.markdown(text + " |")

    elif kind == "replace":
        text = event["text"]

    elif kind == "done":
        metadata = event["metadata"]
        status.update(label="Done", state="complete")

Five stage names reach the screen in practice. Measured on one real question:

The instrumented call ran 3613.9 milliseconds against 3379.3 for the same work with no callback attached, which is the same order of magnitude and no measurable overhead.

One piece of correctness in the framing happened by accident. Each event is serialised to JSON and only then wrapped in the line-oriented envelope. A token containing a newline is therefore escaped inside the JSON string and cannot break the parser at the far end. Encoding before framing rather than framing before encoding is the only reason a naive line reader works here at all.

And one measurement that had to be taken rather than assumed. Application servers are allowed to buffer a response until it completes, which would produce something that looks like streaming code and behaves like the old blocking endpoint. Against a rebuilt container, time to first byte was 0.004343 seconds against a total of 8.959013 seconds. A buffering server would have made those two numbers equal.

the streamed text gets discarded at the end

The finished text passes through a cleaner before anyone sees it. The cleaner is one substitution that escapes literal dollar signs, so that an answer quoting several figures does not get read as mathematical notation by the markdown renderer and collapsed into symbols.

That cleaner runs on the complete text. It cannot run on a fragment, because a chunk boundary can land between a backslash and the character it escapes.

Three options were weighed before any streaming code existed.

The first option shipped. Tokens arrive raw and get painted immediately, then a single replace event carries the cleaned full answer and the accumulated string is discarded.

A second requirement points the same way. The final answer ends with a citation block that the interface parses into source chips. That block is the last thing the model emits, so any attempt to parse it mid-stream finds nothing. Calling the renderer once on complete text produces the same result as calling it repeatedly, without the flicker.

The predicted cost turned out cheaper than the design expected. Answers about company finances contain dollar amounts essentially always, so the cleaned string is almost never byte-equal to the raw one, and replace genuinely replaces something on nearly every query. An escaped dollar sign and a bare one render identically, so the swap changes the string and leaves the rendered page alone.

Worth stating plainly, since it was never measured: whether that swap is perceptible to someone watching was not tested. The event fires exactly once, and that is the whole of what is known about it.

add a sibling, never modify

There is one pattern underneath all four changes, and it is visible five times.

Nothing on the working path was edited. A command line entry point, a batch evaluation harness, and a set of notebooks all call the original functions and were incapable of noticing that any of this happened. The callback default carries the same guarantee in a smaller space: absent an argument, the code path is identical.

The streaming function reuses the cached component getters and the logging lock, and its final event is deliberately shaped like the non-streaming return value so that response packaging and cost accounting did not need a second implementation.

The discipline has a real cost. The two orchestrator functions share no body, so roughly a hundred lines of error handling exist twice, and they have already drifted. The streaming copy wraps its error-path logging in a guard that the original still lacks, so the fork is more careful than its parent. In the other direction the streaming path quietly drops a field of file paths that the original attaches to its result. A duplicated fork drifts. That is the price, and it bought the ability to add a feature to a finished system without a regression story.

what the four changes add up to

Read downward, each row is a precondition for the one below it. Streaming had been waiting on a free event loop, and the loop was held by a keyword promising a cooperation that the code underneath never performed.

what actually changed, and what did not

Total time did not move. A query that took nine seconds still takes nine seconds, and the comparison questions still take a minute. Nothing in the pipeline was made faster and no work was removed from it.

What changed is that the minute stopped being blank. The stage labels appear within a few milliseconds of submission and update as the pipeline advances, and the answer builds on screen a token at a time instead of appearing in one piece. The system was deployed to Fargate in this state and demonstrated there, which is the test that matters for a change whose whole purpose is how the waiting feels.

Three limits are worth carrying away. Time to first token was never measured on a full-length answer, and the only recorded figure came from a probe capped at twenty tokens where the whole response arrived in a single chunk, making it useless for the purpose. The hosted model’s own stream carries a server-side first-byte latency metric in its final event, which was noticed and left unused as out of scope, so the one number that would settle it was in hand the whole time. And a client that disconnects mid-answer stops the generator but not the worker, which runs the pipeline to completion and pays for the model call that nobody will read.