Finding and fixing asyncio event-loop stalls in a streaming gateway: a stack-capturing watchdog, per-process shared resources, and the benchmark behind the numbers.

A developer built a toy LLM streaming gateway that reproduces asyncio event-loop stalls under many concurrent token streams, pairing it with a stack-capturing watchdog that samples the loop thread's frames when heartbeats run late. The fix moves per-request setup work—tokenizer loading, TLS context creation, and retrieval scoring—out of the request path and into once-per-process initialization, and the accompanying benchmark quantifies the resulting stall reduction.

| | """A toy LLM streaming gateway that reproduces event-loop stalls. | | | | | | Many clients hold open token streams on one asyncio loop. New requests keep | | | arriving, and each one does some setup before its own stream starts. In | | | "naive" mode the setup does the usual per-request things on the loop; in | | | "fixed" mode the same work is built once per process before traffic. | | | | | | Run: python gateway.py naive \| python gateway.py fixed \| python gateway.py both | | | """ | | | | | | from future import annotations | | | | | | import asyncio | | | import gc | | | import json | | | import os | | | import random | | | import ssl | | | import statistics | | | import sys | | | import tempfile | | | import threading | | | import time | | | import traceback | | | import zlib | | | from collections import Counter | | | | | | import certifi | | | import numpy as np | | | | | | STREAMS = 40 long-lived client streams | | | CHUNK EVERY = 0.020 each stream sends a token chunk every 20 ms | | | ARRIVAL MEAN = 0.25 a new request every 250 ms on average | | | DURATION = 20.0 seconds of traffic per run | | | VOCAB SIZE = 200 000 | | | DOCS, DIMS = 400, 384 retrieval candidates scored per request | | | PAYLOAD = 1 000 000 bytes of already-compressed data per request | | | STARTUP OBJECTS = 1 500 000 | | | | | | rng = random.Random 7 | | | | | | | | | ---------- watchdog same design as the article ---------- | | | | | | class LoopWatchdog: | | | def init self, loop, tick=0.010, margin=0.040, poll=0.005, report=None : | | | self.loop, self.report = loop, report | | | self.tick, self.limit, self.poll = tick, tick + margin, poll | | | self.loop thread = threading.get ident | | | self.heartbeat = self.seen = time.monotonic | | | self.stacks = | | | self. stop = False | | | threading.Thread target=self. watch, daemon=True .start | | | | | | async def beat self : | | | while True: | | | self.heartbeat = time.monotonic | | | await asyncio.sleep self.tick | | | | | | def stop self : | | | self. stop = True | | | | | | def watch self : | | | while not self. stop: | | | time.sleep self.poll | | | beat = self.heartbeat | | | if beat = self.seen: | | | gap, self.seen = beat - self.seen, beat | | | stacks, self.stacks = self.stacks, | | | if gap self.limit: | | | blocked ms = gap - self.tick 1000 | | | self.loop.call soon threadsafe self.report, blocked ms, stacks | | | continue | | | overdue = time.monotonic - beat self.limit | | | if overdue and len self.stacks < 2: | | | frame = sys. current frames .get self.loop thread | | | if self.heartbeat == beat: | | | self.stacks.append traceback.extract stack frame, limit=15 | | | | | | | | | ---------- the expensive things a request start does ---------- | | | | | | def make vocab file - str: | | | path = os.path.join tempfile.gettempdir , "toy vocab.json" | | | if not os.path.exists path : | | | vocab = {f"tok {i:06d} {i 7919 % 100003}": i for i in range VOCAB SIZE } | | | with open path, "w" as f: | | | json.dump vocab, f | | | return path | | | | | | | | | def load tokenizer path: str - dict: | | | with open path as f: | | | vocab = json.load f | | | return {v: k for k, v in vocab.items } | | | | | | | | | def make tls context - ssl.SSLContext: | | | return ssl.create default context cafile=certifi.where | | | | | | | | | def score python query: list float , docs: list list float - list float : | | | out = | | | for doc in docs: | | | dot = qn = dn = 0.0 | | | for q, d in zip query, doc : | | | dot += q d | | | qn += q q | | | dn += d d | | | out.append dot / qn 0.5 dn 0.5 | | | return out | | | | | | | | | def score numpy query: np.ndarray, docs: np.ndarray - np.ndarray: | | | return docs @ query / np.linalg.norm docs, axis=1 np.linalg.norm query | | | | | | | | | def compress payload data: bytes - bytes: | | | return zlib.compress data, 6 | | | | | | | | | def make garbage : | | | Per-request bookkeeping: lots of small containers, some cyclic. | | | items = | | | for i in range 20 000 : | | | d = {"i": i, "meta": i, str i } | | | d "self" = d | | | items.append d | | | return len items | | | | | | | | | ---------- the gateway ---------- | | | | | | class Gateway: | | | def init self, mode: str : | | | self.mode = mode | | | self.vocab path = make vocab file | | | self.docs list = rng.random for in range DIMS for in range DOCS | | | self.docs np = np.array self.docs list | | | self.payload = os.urandom PAYLOAD | | | A big long-lived heap, like a prompt/template catalog loaded at boot. | | | self.catalog = i, str i , i for i in range STARTUP OBJECTS | | | self.setup ms: list float = | | | self.step ms: dict str, list float = {} | | | if mode == "fixed": | | | self.tls = make tls context | | | self.tokenizer = load tokenizer self.vocab path | | | gc.collect | | | gc.freeze | | | | | | def step self, name, fn : | | | t = time.perf counter | | | out = fn | | | self.step ms.setdefault name, .append time.perf counter - t 1000 | | | return out | | | | | | def start request self : | | | t0 = time.perf counter | | | step = self. step | | | if self.mode == "naive": | | | clients = step "TLS context", lambda: make tls context for in range 3 | | | tokenizer = step "tokenizer load", lambda: load tokenizer self.vocab path | | | q = rng.random for in range DIMS | | | step "pure-Python scoring", lambda: score python q, self.docs list | | | step "compressing incompressible data", lambda: compress payload self.payload | | | else: | | | clients = self.tls 3 | | | tokenizer = self.tokenizer | | | q = np.array rng.random for in range DIMS | | | step "numpy scoring", lambda: score numpy q, self.docs np | | | payload is already compressed: send as is | | | step "request bookkeeping", make garbage | | | self.setup ms.append time.perf counter - t0 1000 | | | return clients, tokenizer | | | | | | | | | TIMELINE: list tuple float, float = seconds since start, lateness for --dump | | | | | | | | | async def stream lateness: list float , stop at: float : | | | next due = time.monotonic | | | while True: | | | next due += CHUNK EVERY | | | await asyncio.sleep max 0.0, next due - time.monotonic | | | now = time.monotonic | | | if now = stop at: | | | return | | | lateness.append now - next due | | | TIMELINE.append now - stop at - DURATION , now - next due | | | if now - next due CHUNK EVERY: fell behind: resync like a real sender | | | next due = now | | | | | | | | | def arrival offsets - list float : | | | r, t, out = random.Random 11 , 0.0, | | | while True: | | | t += r.expovariate 1 / ARRIVAL MEAN | | | if t = DURATION - 0.5: | | | return out | | | out.append t | | | | | | | | | async def arrivals gw: Gateway, start: float : | | | Same arrival times in both modes, even if the loop falls behind. | | | for offset in arrival offsets : | | | await asyncio.sleep max 0.0, start + offset - time.monotonic | | | gw.start request | | | | | | | | | def bucket worst - list float : | | | buckets = 0.0 int DURATION 10 | | | for t, late in TIMELINE: | | | i = min int t 10 , len buckets - 1 | | | buckets i = max buckets i , late 1000 | | | TIMELINE.clear | | | return round b, 1 for b in buckets | | | | | | | | | CATEGORIES = | | | "make tls context", "TLS context" , | | | "load tokenizer", "tokenizer load" , | | | "score python", "pure-Python scoring" , | | | "compress payload", "compressing incompressible data" , | | | | | | | | | | | | def categorize stack - str: | | | names = {f.name for f in stack} | | | for fn, label in CATEGORIES: | | | if fn in names: | | | return label | | | return "other" | | | | | | | | | async def run mode: str - dict: | | | gw = Gateway mode | | | gc pauses: list float = | | | gc start = 0.0 | | | | | | def on gc phase, info : | | | if phase == "start": | | | gc start 0 = time.perf counter | | | else: | | | gc pauses.append time.perf counter - gc start 0 1000 | | | | | | gc.callbacks.append on gc | | | stalls = | | | | | | def report blocked ms, stacks : | | | stalls.append blocked ms, categorize stacks 0 if stacks else "no stack GIL held " | | | | | | wd = LoopWatchdog asyncio.get running loop , report=report | | | beat = asyncio.create task wd.beat | | | stop at = time.monotonic + DURATION | | | lateness: list list float = for in range STREAMS | | | await asyncio.gather arrivals gw, stop at - DURATION , stream l, stop at for l in lateness | | | wd.stop | | | beat.cancel | | | gc.callbacks.remove on gc | | | gc.unfreeze | | | | | | all late = x 1000 for l in lateness for x in l | | | | | | def share late threshold ms : | | | return 100 sum x threshold ms for x in all late / len all late | | | | | | causes = Counter c for , c in stalls | | | return { | | | "mode": mode, | | | "requests": len gw.setup ms , | | | "setup p50 ms": statistics.median gw.setup ms , | | | "stalls per request": len stalls / len gw.setup ms , | | | "stall p50 ms": statistics.median s for s, in stalls if stalls else 0.0, | | | "stall max ms": max s for s, in stalls , default=0.0 , | | | "loop blocked pct": 100 sum s for s, in stalls / 1000 / DURATION, | | | "chunk late p99 ms": float np.percentile all late, 99 , | | | "chunk late max ms": max all late , | | | "chunks late 50ms pct": share late 50 , | | | "chunks late 100ms pct": share late 100 , | | | "gc collections": len gc pauses , | | | "gc pause max ms": max gc pauses, default=0.0 , | | | "gc pause total ms": sum gc pauses , | | | "setup step p50 ms": {k: round statistics.median v , 2 for k, v in gw.step ms.items }, | | | "timeline 100ms worst late ms": bucket worst , | | | "stall causes": {k: round 100 v / len stalls , 1 for k, v in causes.most common } if stalls else {}, | | | } | | | | | | | | | def main : | | | args = sys.argv 1: | | | dump = None | | | if "--dump" in args: | | | i = args.index "--dump" | | | dump = args i + 1 | | | del args i:i + 2 | | | modes = args or "both" | | | if modes == "both" : | | | modes = "naive", "fixed" | | | results = | | | for mode in modes: | | | result = asyncio.run run mode | | | timeline = result.pop "timeline 100ms worst late ms" | | | print json.dumps result, indent=2 | | | result "timeline 100ms worst late ms" = timeline | | | results.append result | | | gc.collect | | | if dump: | | | with open dump, "w" as f: | | | json.dump results, f | | | | | | | | | if name == " main ": | | | main |