Skip to content
mini-sglang from Scratch Stage 3 · Serving: make it a service

The API server and the launcher

The last piece: a FastAPI application takes OpenAI-format requests, gives each a uid, sends it to the tokenizer, and streams the delta text that comes back to the client over SSE; a launcher starts every process in the right order. With this chapter done, python -m minisgl --model Qwen/Qwen3-0.6B is an inference service any OpenAI client can talk to.

Self-test: if you can answer these, skip the chapter

  1. Hundreds or thousands of concurrent HTTP requests share one ZMQ receive queue. How does each one get only its own replies?
  2. What does SSE (Server-Sent Events) look like? What is the last message of OpenAI's streaming interface?
  3. The client disconnects mid-stream. What does the server have to do?
  4. Why does the launcher use spawn rather than fork for its subprocesses?
Answers (try it yourself first, then expand)
  1. Each request has a unique uid: the receiving coroutine puts a reply into the list for that uid and sets that request's asyncio.Event, and each request's handler coroutine waits only on its own event and reads only its own list.
  2. Each message is data: <JSON> followed by a blank line; OpenAI's streaming interface gives each chunk a delta.content, the last chunk a finish_reason, and then sends data: [DONE].
  3. Send an AbortMsg to the scheduler, free the request slot and KV it holds, and stop generating for it.
  4. fork copies the parent's state, including an already initialized CUDA context, threads and the ZMQ context, none of which can be used safely in the child; spawn starts a clean interpreter that initializes its own.

Files you will write: server/args.py, server/api_server.py, server/launch.py, __main__.py, shell.py.

Files so far: 73 (6 new, 0 modified, 0 brought in early)
minisgl/
├── __init__.py
├── core.py
├── env.py
├── __main__.py  ← new in this chapter
├── shell.py  ← new in this chapter
├── attention/  (4 files, unchanged)
├── distributed/  (3 files, unchanged)
├── engine/  (5 files, unchanged)
├── kernel/  (2 files, unchanged)
├── kvcache/  (5 files, unchanged)
├── layers/  (8 files, unchanged)
├── llm/  (2 files, unchanged)
├── message/  (5 files, unchanged)
├── models/  (7 files, unchanged)
├── moe/  (2 files, unchanged)
├── scheduler/  (9 files, unchanged)
├── server/
│   ├── __init__.py  ← new in this chapter
│   ├── args.py  ← new in this chapter
│   ├── api_server.py  ← new in this chapter
│   ├── launch.py  ← new in this chapter
├── tokenizer/  (4 files, unchanged)
├── utils/  (8 files, unchanged)
This step's main: examples/ch15_server.py — it uses only the files above; python tools/steps.py check rebuilds this tree chapter by chapter and runs it.

One shared queue, many waiters

The API server is an asyncio program. Every request's replies come in through the same ZMQ queue and have to be dispatched to the right handler coroutine by uid:

minisgl/server/api_server.py · FrontendManager
@dataclass
class FrontendManager:
    config: ServerArgs
    send_tokenizer: ZmqAsyncPushQueue[BaseTokenizerMsg]
    recv_tokenizer: ZmqAsyncPullQueue[BaseFrontendMsg]
    uid_counter: int = 0
    initialized: bool = False
    ack_map: Dict[int, List[UserReply]] = field(default_factory=dict)
    event_map: Dict[int, asyncio.Event] = field(default_factory=dict)

    def new_user(self) -> int:
        uid = self.uid_counter
        self.uid_counter += 1
        self.ack_map[uid] = []
        self.event_map[uid] = asyncio.Event()
        return uid

    async def listen(self) -> None:
        while True:
            for reply in _unwrap(await self.recv_tokenizer.get()):
                if reply.uid not in self.ack_map:  # requests already finished or cancelled
                    continue
                self.ack_map[reply.uid].append(reply)
                self.event_map[reply.uid].set()

    async def send_one(self, msg: BaseTokenizerMsg) -> None:
        if not self.initialized:  # the background task can only be created on the first call, once the event loop is running
            asyncio.create_task(self.listen())
            self.initialized = True
        await self.send_tokenizer.put(msg)

    async def wait_for_ack(self, uid: int) -> AsyncIterator[UserReply]:
        event = self.event_map[uid]
        try:
            while True:
                await event.wait()
                event.clear()
                pending, self.ack_map[uid] = self.ack_map[uid], []
                for ack in pending:
                    yield ack
                    if ack.finished:
                        return
        finally:
            self.ack_map.pop(uid, None)
            self.event_map.pop(uid, None)

    async def stream_generate(self, uid: int) -> AsyncIterator[bytes]:
        async for ack in self.wait_for_ack(uid):
            yield f"data: {ack.incremental_output}\n".encode()
        yield b"data: [DONE]\n"

    async def stream_chat_completions(self, uid: int) -> AsyncIterator[bytes]:
        first = True
        async for ack in self.wait_for_ack(uid):
            delta: Dict[str, str] = {}
            if first:
                delta["role"] = "assistant"
                first = False
            if ack.incremental_output:
                delta["content"] = ack.incremental_output
            chunk = {"id": f"chatcmpl-{uid}", "object": "chat.completion.chunk",
                     "choices": [{"delta": delta, "index": 0, "finish_reason": None}]}
            yield f"data: {json.dumps(chunk)}\n\n".encode()
        end = {"id": f"chatcmpl-{uid}", "object": "chat.completion.chunk",
               "choices": [{"delta": {}, "index": 0, "finish_reason": "stop"}]}
        yield f"data: {json.dumps(end)}\n\n".encode()
        yield b"data: [DONE]\n\n"

    async def stream_with_cancellation(self, gen: AsyncIterator[bytes], request: Request,
                                       uid: int) -> AsyncIterator[bytes]:
        """On a client disconnect, tell the backend to abort this request and free the KV cache it holds."""
        try:
            async for chunk in gen:
                if await request.is_disconnected():
                    raise asyncio.CancelledError
                yield chunk
        except asyncio.CancelledError:
            asyncio.create_task(self.abort_user(uid))
            raise

    async def abort_user(self, uid: int) -> None:
        await asyncio.sleep(0.1)
        self.ack_map.pop(uid, None)
        self.event_map.pop(uid, None)
        logger.warning(f"Aborting request {uid}")
        await self.send_one(AbortMsg(uid=uid))

    def shutdown(self) -> None:
        self.send_tokenizer.stop()
        self.recv_tokenizer.stop()
- new_user() assigns a uid to a new request and creates a reply list and an asyncio.Event for it; - the background coroutine listen() keeps taking UserReplys from ZMQ, appends each to its uid's list, and set()s its event; - each request's handler coroutine awaits event.wait() inside wait_for_ack(), then takes everything in the list and yields it one by one until it sees finished; - listen() is created on the first send, because it needs the event loop to be running already, and there is no loop yet when the FastAPI application starts.

This is the standard asyncio shape for "one producer, many consumers dispatched by key". New replies can arrive and set() the event again after it is clear()ed, so taking the replies swaps the whole list out, and nothing is missed.

The OpenAI-compatible endpoints

minisgl/server/api_server.py · v1_chat_completions
@app.post("/v1/chat/completions")
async def v1_chat_completions(req: OpenAICompletionRequest, request: Request):
    state = get_state()
    if req.messages:
        text: str | List[Dict[str, str]] = [m.model_dump() for m in req.messages]
    else:
        assert req.prompt is not None, "Either 'messages' or 'prompt' must be provided"
        text = req.prompt
    uid = state.new_user()
    await state.send_one(TokenizeMsg(uid=uid, text=text, sampling_params=_sampling_params(req)))
    if req.stream:
        return StreamingResponse(
            state.stream_with_cancellation(state.stream_chat_completions(uid), request, uid),
            media_type="text/event-stream")
    content = ""
    async for ack in state.wait_for_ack(uid):
        content += ack.incremental_output
    return {
        "id": f"chatcmpl-{uid}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": req.model,
        "choices": [{"index": 0, "message": {"role": "assistant", "content": content},
                     "finish_reason": "stop"}],
    }
- either messages (a conversation) or prompt (plain text), with the former going through the chat template in the tokenizer; - the sampling parameters pass straight to the scheduler, and ignore_eos is an extension field, often used for load testing; - a streaming response returns text/event-stream: each reply becomes a data: {...}\n\n whose choices[0].delta.content is the delta text, followed by a finish_reason: "stop" and data: [DONE]; - a non-streaming response collects everything into one complete JSON.

minisgl/server/api_server.py · FrontendManager.stream_chat_completions
async def stream_chat_completions(self, uid: int) -> AsyncIterator[bytes]:
    first = True
    async for ack in self.wait_for_ack(uid):
        delta: Dict[str, str] = {}
        if first:
            delta["role"] = "assistant"
            first = False
        if ack.incremental_output:
            delta["content"] = ack.incremental_output
        chunk = {"id": f"chatcmpl-{uid}", "object": "chat.completion.chunk",
                 "choices": [{"delta": delta, "index": 0, "finish_reason": None}]}
        yield f"data: {json.dumps(chunk)}\n\n".encode()
    end = {"id": f"chatcmpl-{uid}", "object": "chat.completion.chunk",
           "choices": [{"delta": {}, "index": 0, "finish_reason": "stop"}]}
    yield f"data: {json.dumps(end)}\n\n".encode()
    yield b"data: [DONE]\n\n"

Aborting on disconnect

If the server keeps generating after a client disconnects, that request holds a request slot and its KV cache until max_tokens. So the connection is checked before every chunk:

minisgl/server/api_server.py · FrontendManager.stream_with_cancellation
async def stream_with_cancellation(self, gen: AsyncIterator[bytes], request: Request,
                                   uid: int) -> AsyncIterator[bytes]:
    """On a client disconnect, tell the backend to abort this request and free the KV cache it holds."""
    try:
        async for chunk in gen:
            if await request.is_disconnected():
                raise asyncio.CancelledError
            yield chunk
    except asyncio.CancelledError:
        asyncio.create_task(self.abort_user(uid))
        raise
On a disconnect it deletes that uid's waiting state and sends an AbortMsg. That becomes an AbortBackendMsg through the tokenizer, and the scheduler finds the request in the waiting queue or the decode set and frees its resources (chapter 11 covered the details of how this interleaves with overlap scheduling).

The launcher

minisgl/server/launch.py · start_backend
def start_backend(server_args: ServerArgs) -> None:
    """Start every backend process: one scheduler per TP rank, one detokenizer, and a few tokenizers."""
    from minisgl.tokenizer import tokenize_worker

    logger = init_logger(__name__, "initializer")
    mp.set_start_method("spawn", force=True)  # CUDA does not work in a forked child process
    ack_queue: mp.Queue = mp.Queue()
    world_size = server_args.tp_info.size
    for i in range(world_size):
        mp.Process(target=_run_scheduler,
                   args=(replace(server_args, tp_info=DistributedInfo(i, world_size)), ack_queue),
                   name=f"minisgl-TP{i}-scheduler").start()
    common = dict(tokenizer_path=server_args.model_path, backend_addr=server_args.zmq_backend_addr,
                  frontend_addr=server_args.zmq_frontend_addr, local_bs=1, ack_queue=ack_queue)
    n = server_args.num_tokenizer
    mp.Process(target=tokenize_worker, name="minisgl-detokenizer",
               kwargs=dict(common, addr=server_args.zmq_detokenizer_addr,
                           create=server_args.tokenizer_create_addr, tokenizer_id=n)).start()
    for i in range(n):
        mp.Process(target=tokenize_worker, name=f"minisgl-tokenizer-{i}",
                   kwargs=dict(common, addr=server_args.zmq_tokenizer_addr,
                               create=server_args.tokenizer_create_addr, tokenizer_id=i)).start()
    for _ in range(n + 2):  # the rank 0 scheduler + n tokenizers + 1 detokenizer
        logger.info(ack_queue.get())
The startup order:

  1. the main process parses the arguments and creates the FrontendManager (which binds the frontend's ZMQ address);
  2. spawn starts a scheduler process per TP rank, one detokenizer process and N tokenizer processes. CUDA cannot be reinitialized in a forked child, so spawn is required;
  3. each process puts a "ready" message into a multiprocessing.Queue once it is set up (for the schedulers only rank 0 does, and only after every rank has completed sync_all_ranks); the main process waits for num_tokenizer + 2 of them before continuing;
  4. uvicorn starts (or the interactive shell).

minisgl/server/launch.py · _run_scheduler
def _run_scheduler(args: ServerArgs, ack_queue: mp.Queue) -> None:
    import torch
    from minisgl.scheduler import Scheduler

    with torch.inference_mode():
        scheduler = Scheduler(args)
        scheduler.sync_all_ranks()  # start only once every rank has initialized, ZMQ subscriptions included
        if args.tp_info.is_primary():
            ack_queue.put("Scheduler is ready")
        if args.silent_output:
            logging.disable(logging.INFO)
        try:
            scheduler.run_forever()
        except KeyboardInterrupt:
            scheduler.shutdown()
Interactive mode (--shell) runs no HTTP service and chats right in the terminal; it sets max_running_req to 1 and captures only a batch-size-1 CUDA Graph, since there is only one user.

Running it

examples/ch15_server.py
"""Chapter 15: start the complete service and talk to it with an OpenAI client."""

import os
import signal
import socket
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor

import httpx
import psutil
from openai import OpenAI

with socket.socket() as s:
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
proc = subprocess.Popen(
    [sys.executable, "-m", "minisgl", "--model", "models/Qwen3-0.6B", "--dtype", "float32",
     "--port", str(port), "--num-pages", "4096", "--max-seq-len-override", "1024"],
    stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True)
try:
    t = time.time()
    while True:
        try:
            if httpx.get(f"http://127.0.0.1:{port}/v1/models", timeout=1).status_code == 200:
                break
        except httpx.HTTPError:
            time.sleep(0.5)
    children = psutil.Process(proc.pid).children(recursive=True)
    workers = [c for c in children if "multiprocessing-fork" in " ".join(c.cmdline())]
    print(f"service ready (about {time.time() - t:.0f} s): the main process (API server) + {len(workers)} workers"
          f"(schedulers, tokenizers)")
    client = OpenAI(base_url=f"http://127.0.0.1:{port}/v1", api_key="none")

    r = client.chat.completions.create(
        model="qwen3", messages=[{"role": "user", "content": "Describe Beijing in one sentence. /no_think"}],
        max_tokens=48, temperature=0)
    print("non-streaming:", repr(r.choices[0].message.content))

    stream = client.chat.completions.create(
        model="qwen3", messages=[{"role": "user", "content": "Count from 1 to 5. /no_think"}],
        max_tokens=24, temperature=0, stream=True)
    chunks = [c.choices[0].delta.content for c in stream if c.choices[0].delta.content]
    print(f"streaming: {len(chunks)} chunks received:", chunks[:12], "...")

    def ask(i: int) -> float:
        t0 = time.perf_counter()
        client.chat.completions.create(model="qwen3", messages=[{"role": "user", "content": f"{i}+{i}=?"}],
                                       max_tokens=16, temperature=0, extra_body={"ignore_eos": True})
        return time.perf_counter() - t0

    t0 = time.perf_counter()
    with ThreadPoolExecutor(8) as ex:
        lat = list(ex.map(ask, range(8)))
    print(f"8 concurrent requests (16 tokens each) took {time.perf_counter() - t0:.2f}s in total, {min(lat):.2f}-{max(lat):.2f}s each")
finally:
    os.killpg(proc.pid, signal.SIGKILL)
What it prints
service ready (about 9 s): the main process (API server) + 2 workers(schedulers, tokenizers)
non-streaming: '<think>\n\n</think>\n\nBeijing is a major city in China, known for its historical landmarks, cultural significance, and modern infrastructure.'
streaming: 24 chunks received: ['<think>', '\n\n', '</think>', '\n\n', 'Count', ' from', ' ', '1', ' to', ' ', '5', ':'] ...
8 concurrent requests (16 tokens each) took 4.82s in total, 4.81-4.82s each
- Processes: the API server's main process, plus one scheduler process and one tokenizer process (tokenizing and detokenizing share it by default). - Non-streaming: Qwen3's answer starts with an empty <think></think> (/no_think turned thinking off) and then the answer. - Streaming: one chunk per token, with the special token <think> emitted as a complete chunk of its own. - The 8 concurrent requests are merged into one batch by the scheduler and take almost as long in total as a single request, which is the throughput continuous batching buys.

The official implementation

The usage field in upstream's non-streaming /v1/chat/completions response is always 0; we leave it out. Upstream's interactive mode kills every child process with psutil on exit, which we kept.

Tests

tests/test_ch15_server.py · test_client_disconnect_aborts_request
def test_client_disconnect_aborts_request(server):
    """The client disconnects mid-stream: the server aborts the request and frees its resources, and later requests work as usual."""
    port, log_path = server
    with httpx.stream("POST", f"http://127.0.0.1:{port}/v1/chat/completions",
                      json={"model": "m", "prompt": "Tell me a long story.", "max_tokens": 500,
                            "stream": True, "ignore_eos": True}, timeout=60) as r:
        for i, _ in enumerate(r.iter_lines()):
            if i >= 3:
                break  # disconnect after a few chunks
    time.sleep(1.0)
    resp = httpx.post(f"http://127.0.0.1:{port}/v1/chat/completions",
                      json={"model": "m", "prompt": "1 + 1 =", "max_tokens": 4}, timeout=60)
    assert resp.status_code == 200
    assert "Aborting request" in log_path.read_text()
The other two tests in tests/test_ch15_server.py: a non-streaming response matches Hugging Face's greedy output word for word, and both streaming and concurrent requests return correctly.

How to explain it

On the API server: hundreds or thousands of HTTP requests share one ZMQ receive queue through "a reply list dispatched by uid plus one asyncio.Event per request", where a background task puts each reply into its request's list and wakes it. OpenAI's streaming interface is SSE: one data: {...} line per chunk with the content in delta.content, a finish_reason on the last one, and then data: [DONE]. A mid-stream disconnect has to send an abort so the request slot and KV are freed, or the engine keeps generating for nobody. The launcher uses spawn rather than fork because forking copies the parent's already-initialized CUDA, threads and locks, which is not safe.

Exercises

  1. Add a correct usage (prompt tokens, completion tokens) to the non-streaming response. Which process knows those numbers? Which messages need new fields?
  2. Implement /v1/completions (the non-chat completion endpoint). How do its request and response formats differ from /v1/chat/completions?
  3. uids are currently assigned by a counter in the API server. If several API server processes served together later, how would you assign uids without collisions?
Answers
  1. The prompt length is known in the tokenizer and the generated count in the scheduler. Add prompt_tokens and completion_tokens to UserReply, filled by the detokenizer at the end (it can count the tokens it received; the prompt length has to be recorded by the tokenizer at encoding time, or carried back by the scheduler on the last DetokenizeMsg).
  2. The request uses prompt rather than messages and skips the chat template; the response has choices[0].text rather than message.content, each streaming chunk is a choices[0].text, and object is text_completion.
  3. Give each API server a different prefix or stride, for instance uid = process index + process count × local counter, or use UUIDs. The detokenizer's replies also have to route back to the right API server (one receive address per server, with the origin carried in the message).

Summary

  • The API server shares one ZMQ receive queue across all requests through "a reply list dispatched by uid plus an asyncio.Event".
  • OpenAI's streaming interface is SSE: one delta.content per chunk, then a finish_reason and [DONE].
  • A client disconnect sends an AbortMsg so the request's slot and KV are freed.
  • The launcher spawns the scheduler and tokenizer processes and waits for all of them before starting the HTTP service.

Exercises for this chapter (1)

Write the code and run the tests in your browser, or judge locally with python practice/judge.py. The exercises are in Chinese for now.

Found a mistake or a better way to say it? Edit this page on GitHubReport an issue