Request Lifecycle、Streaming、Cancellation 与 Timeout

一个请求是跨组件状态机，不只是一段 model.generate

服务端常见状态可以抽象为：

received → tokenized → queued → admitted → prefill → decode ↔ streaming → terminal

received/tokenized：校验参数，应用 Chat Template 与 Special Tokens，得到 token ids 与长度。
queued/admitted：等待 scheduler，根据 token、KV blocks、batch slots、deadline 和优先级预算接纳。
Scheduler Budget / Admission Control定义这一步如何核算 iteration work 与未来 KV growth；本页负责状态与终态清理。
prefill/decode：执行模型并更新 KV Cache；可能被 chunk、preempt 或跨 iterations 调度。
streaming：把已提交 token ids 增量转换成 bytes/text，处理 special tokens、UTF-8、stop 条件和客户端背压。
terminal：stop、length、cancel、timeout、error 等都必须只终止一次，并回收 KV、slot、stream、metrics/context。

它是服务控制平面，不改变 attention 或 sampling 数学；但错误的状态边界会造成重复 token、泄漏 KV、终止后仍写 socket、collective 次序不一致或错误计费。

INTERACTIVE EXPLAINER

请求从 token ids、queue、模型执行到 stream 与幂等清理

静态图区分控制平面状态、模型执行、bytes/text streaming 与所有终止原因汇合的 cleanup；箭头是逻辑依赖，不承诺线程/进程布局。

图中省略认证、rate limit、prefix lookup、distributed coordinator、scheduler subqueues、retry、SSE/gRPC framing 与多副本路由。Prefill、decode 和 streaming 可以流水重叠；图上的线性顺序只表示一个 token 在发布前的因果关系。

Token 边界、byte 边界与 stop 条件

模型与 cache 的身份建立在 token id 序列上，不是原始字符串。Streaming 输出却必须处理 byte/text 边界：

一个 token 的 byte payload 可能只是一个 UTF-8 code point 的前半段；
byte-fallback tokenizer 可能用多个 token 表示一个字符；
special token 可能参与模型状态，却不应直接显示给用户；
stop token id 可在采样后、写入下一个模型 step 前检查；
stop string 可能跨多个 token/text chunks，需要保留最长可能前缀，不能先把它的一部分发给客户端；
若 stop string 与输出重叠，API 必须定义是否包含 stop text、最长/最早匹配与 normalization 口径。

例如 token bytes 为：

[e4 bd] [a0] [e5] [a5 bd] [E] [N] [D]

前四组逐步组成 UTF-8 “你好”；后三组跨 token 组成 stop string END。Incremental decoder 在 [e4 bd] 或 [e5] 后不能输出 replacement character；streamer 还要暂存 E、EN，确认不是 stop 前缀后才能安全发送。

Cancellation、timeout 与 backpressure

Cancellation 是协作式终止

客户端断开或显式 cancel 后，服务端要标记请求，在安全点把它从 scheduler active set 移除，并回收 KV blocks。正在执行的 GPU kernel 通常不会因为 Python flag 立刻中止；多 rank batch 中也不能让一个 rank 擅自跳过其余 ranks 仍会进入的 collective。常见做法是在本轮执行/collective 保持一致后，下一 scheduler boundary 停止该请求。

取消请求已经写出的 token 是否计费、是否记录 finish_reason=cancelled、prefix blocks 是否仍可进入 cache，应由明确策略决定。Cleanup 必须幂等，因为 client disconnect、deadline 和 backend error 可能并发到达。

Timeout 有多个时钟

queue/admission deadline：未开始模型执行前等待过久；通常没有 KV 可释放。
TTFT deadline：包括 queue、tokenize 与 prefill；超时可能已有 partial KV。
inter-token timeout：decode/stream 长时间无新 token；要区分 GPU stall、backpressure 与网络阻塞。
overall deadline：请求从接收到终态的总预算。

只在 API gateway 超时并断开 socket，却不把 cancellation 传播给 model worker，会形成 orphan request，继续占 GPU/KV。

Backpressure 必须有界

若客户端读取很慢，stream producer 不能无限把 text chunks 堆在内存。可使用有界 queue、高水位暂停、超时后 cancel，或丢弃连接；但阻塞 streaming 线程是否会阻塞 scheduler/其他请求必须隔离。已经生成但尚未成功发送的 token 在 latency/计费指标中也要明确口径。

可运行状态机与增量 UTF-8/stop 实验

实验问题：能否在没有 tokenizer/model 下载的情况下，证明 UTF-8 code point 与 stop string 都可跨 token bytes 正确处理？正常停止、prefill 中取消和 queue 超时是否都进入唯一终态并精确释放 KV ownership？

import codecs,platform
from dataclasses import dataclass,field

TERMINAL={"finished","cancelled","timed_out","failed"}

class StreamingText:
    def __init__(self,stop_strings):
        self.decoder=codecs.getincrementaldecoder("utf-8")()
        self.stop_strings=tuple(stop_strings); self.pending=""; self.emitted=""; self.stopped=False
    def push(self,token_bytes):
        if self.stopped: return ""
        self.pending+=self.decoder.decode(token_bytes,final=False)
        positions=[self.pending.find(stop) for stop in self.stop_strings]
        positions=[position for position in positions if position>=0]
        if positions:
            cut=min(positions); chunk=self.pending[:cut]
            self.emitted+=chunk; self.pending=""; self.stopped=True
            return chunk
        hold=0
        for stop in self.stop_strings:
            for length in range(1,min(len(stop),len(self.pending))+1):
                if self.pending.endswith(stop[:length]): hold=max(hold,length)
        safe=len(self.pending)-hold; chunk=self.pending[:safe]
        self.pending=self.pending[safe:]; self.emitted+=chunk
        return chunk
    def finish(self):
        if self.stopped: return ""
        self.pending+=self.decoder.decode(b"",final=True)
        chunk=self.pending; self.emitted+=chunk; self.pending=""
        return chunk

@dataclass
class Request:
    request_id:str; deadline:int; state:str="received"; kv_blocks:int=0
    finish_reason:str|None=None; trace:list[str]=field(default_factory=lambda:["received"])
    def transition(self,state):
        assert self.state not in TERMINAL; self.state=state; self.trace.append(state)
    def allocate_kv(self,blocks):
        assert self.kv_blocks==0 and blocks>0; self.kv_blocks=blocks
    def terminate(self,state,reason):
        assert state in TERMINAL and self.state not in TERMINAL
        released=self.kv_blocks; self.kv_blocks=0; self.state=state
        self.finish_reason=reason; self.trace.append(state); return released

def main():
    complete=Request("complete",deadline=20)
    for state in ("tokenized","queued","admitted"): complete.transition(state)
    complete.allocate_kv(3); complete.transition("prefill")
    complete.transition("decode"); complete.transition("streaming")
    stream=StreamingText(("END",))
    token_bytes=[b"\xe4\xbd",b"\xa0",b"\xe5",b"\xa5\xbd",b"E",b"N",b"D",b"ignored"]
    chunks=[stream.push(piece) for piece in token_bytes]
    assert stream.emitted=="你好" and stream.stopped
    assert "END" not in stream.emitted and "ignored" not in stream.emitted
    released_complete=complete.terminate("finished","stop")

    cancelled=Request("cancelled",deadline=20)
    for state in ("tokenized","queued","admitted"): cancelled.transition(state)
    cancelled.allocate_kv(2); cancelled.transition("prefill")
    released_cancelled=cancelled.terminate("cancelled","client_cancel")

    timed_out=Request("timed_out",deadline=3)
    timed_out.transition("tokenized"); timed_out.transition("queued")
    now=4; assert now>timed_out.deadline
    released_timeout=timed_out.terminate("timed_out","queue_deadline")
    assert (released_complete,released_cancelled,released_timeout)==(3,2,0)
    assert all(r.kv_blocks==0 for r in (complete,cancelled,timed_out))
    print(f"python={platform.python_version()} implementation={platform.python_implementation()}")
    print(f"stream_token_bytes={token_bytes}")
    print(f"stream_chunks={chunks} emitted={stream.emitted!r} stopped={stream.stopped}")
    print(f"complete trace={complete.trace} released_kv_blocks={released_complete} finish_reason={complete.finish_reason}")
    print(f"cancelled trace={cancelled.trace} released_kv_blocks={released_cancelled} finish_reason={cancelled.finish_reason}")
    print(f"timed_out trace={timed_out.trace} released_kv_blocks={released_timeout} finish_reason={timed_out.finish_reason}")

if __name__=="__main__": main()

完整文件位于 examples/request_lifecycle.py。实际输出 chunks 为 ['', '你', '', '好', '', '', '', '']，客户端只看到“你好”，END 与之后 bytes 均未发送。正常/取消/排队超时 traces 分别释放 3/2/0 个 KV blocks，最后所有请求 kv_blocks==0。

这只是单线程教学状态机：没有证明 socket/SSE、异步 task、GPU worker 或多 rank cancellation 的生产正确性。真实系统还要用 lock/actor/message ownership 避免 double-finalize，并将 request id、trace span、finish reason、已生成/已发送 token 数写入可核对指标。

服务可靠性、Backpressure 与 Observability进一步用同一慢客户端 trace 验证有界输出队列、幂等终态竞态和 KV/slot 单次回收，并区分 metrics、request traces 与 logs 的基数和成本边界。

与 scheduler、prefix cache 和投机解码的边界

Continuous Batching每轮选择 admitted/active 请求；本页定义请求何时可加入、移除与终止。
Chunked Prefill 与 Preemption是资源不足/公平性下的非终态暂停策略；cancel/timeout 是请求不再继续的终态。
Prefix Cache中的共享 blocks 有独立 refcount；取消一个请求只能释放它的引用，不能破坏其他请求仍使用的 prefix。
投机解码的 rejected draft tokens 不应进入 committed token ids、KV 或外部 stream；只发布 accepted prefix/target correction。
Benchmark 方法应区分 generated、committed、sent tokens，以及 cancelled/timed-out 请求是否计入 throughput/goodput。