手写 mini-sglang 阶段二 · 排得好：同时服务很多请求

调度器骨架与离线接口¶

上一章我们手工扮演了调度器：分配 KV 位置、填 batch 字段、追加 token。这一章把它自动化。mini-sglang 的调度器由四个小管理器组成——请求表、KV 缓存、decode 集合、prefill 队列——主循环只有十几行：收消息、选一个 batch、准备、前向、处理结果。我们先实现最朴素的版本（不复用前缀、不分块、不重叠），并用离线接口 LLM 驱动它。

自测：能答上来就可以跳过本章

连续批处理（continuous batching）和静态批处理的区别是什么？
mini-sglang 一轮只做 prefill 或只做 decode，还是会混在一起？优先做哪个？
decode 集合是一个 set，组 batch 时为什么要按 uid 排序？
离线接口 LLM 没有 ZMQ，它怎样复用调度器的主循环？

自测参考答案（先自己答，再展开对照）

静态批处理要等一整批请求全部结束才换下一批；连续批处理每一轮都重新组 batch，结束的请求立刻离开、新请求随时加入。
一轮只做一种：有能接纳的 prefill 就优先做 prefill，否则做 decode。
张量并行时每个 rank 都运行自己的调度器，必须组出完全相同的 batch（请求顺序决定了 KV 的布局）；set 的遍历顺序不确定，按 uid 排序保证各 rank 一致。
LLM 继承调度器，覆盖"接收消息"和"发送结果"两个方法：直接从内存里的请求列表取输入、把结果收集到本地，主循环完全复用。

本章要写的文件：scheduler/config.py、scheduler/utils.py、scheduler/decode.py、scheduler/prefill.py、scheduler/scheduler.py、llm/llm.py（scheduler/table.py 上一章已写；scheduler/cache.py 下一章写，本章用它的 naive 模式）。

到这一章为止的文件：61 个（新建 9，修改 0，提前引入 7）

minisgl/
├── __init__.py
├── core.py
├── env.py  ← 本章新建
├── attention/  （4 个文件，不变）
├── distributed/  （3 个文件，不变）
├── engine/  （5 个文件，不变）
├── kernel/  （2 个文件，不变）
├── kvcache/  （4 个文件，不变）
├── layers/  （8 个文件，不变）
├── llm/
│   ├── __init__.py  ← 本章新建
│   ├── llm.py  ← 本章新建
├── message/
│   ├── __init__.py  ← 提前引入（第 12 章讲）
│   ├── backend.py  ← 提前引入（第 12 章讲）
│   ├── utils.py  ← 提前引入（第 12 章讲）
│   ├── tokenizer.py  ← 提前引入（第 12 章讲）
├── models/  （7 个文件，不变）
├── moe/  （2 个文件，不变）
├── scheduler/
│   ├── table.py
│   ├── __init__.py  ← 本章新建
│   ├── config.py  ← 本章新建
│   ├── utils.py  ← 本章新建
│   ├── decode.py  ← 本章新建
│   ├── prefill.py  ← 本章新建
│   ├── scheduler.py  ← 本章新建
│   ├── cache.py  ← 提前引入（第 8 章讲）
│   ├── io.py  ← 提前引入（第 14 章讲）
├── utils/
│   ├── __init__.py
│   ├── misc.py
│   ├── hf.py
│   ├── torch_utils.py
│   ├── registry.py
│   ├── logger.py
│   ├── device.py
│   ├── mp.py  ← 提前引入（第 12 章讲）

这一步的 main：examples/ch07_scheduler.py——它只用到上面这些文件；python tools/steps.py check 会逐章搭出这棵树、跑这个 main。

动画：连续批处理与准入控制（约 1.5 分钟，覆盖本章和下一章）

组成¶

minisgl/scheduler/scheduler.py · Scheduler.__init__

def __init__(self, config: SchedulerConfig):
    from minisgl.engine import Engine

    self.engine = Engine(config)
    self.config = self.engine.config
    # 调度器自己用一条流准备元数据，引擎在另一条流上计算，两者可以重叠
    self.device = self.engine.device
    self.stream = create_stream(self.device)
    self.engine_stream_ctx = stream_ctx(self.engine.stream)
    set_stream(self.stream)

    self.table_manager = TableManager(config.max_running_req, self.engine.page_table)
    self.cache_manager = CacheManager(
        self.engine.num_pages, config.page_size, self.engine.page_table, config.cache_type)
    self.decode_manager = DecodeManager(config.page_size)
    self.prefill_manager = PrefillManager(
        self.cache_manager, self.table_manager, self.decode_manager)

    self.finished_reqs: Set[Req] = set()
    # 重叠调度下，释放的请求槽要等"可能用到它的 batch"算完才能重新分配（见 _free_req_resources）
    self._defer_slot_free = False
    self._deferred_slots: List[int] = []
    self.tokenizer = load_tokenizer(config.model_path)
    self.eos_token_id = self.tokenizer.eos_token_id
    self.token_pool = self.table_manager.token_pool
    self.prefill_budget = config.max_extend_tokens
    super().__init__(config, self.engine.tp_cpu_group)

调度器的组成与主循环

管理器	职责	章节
`TableManager`	page table / token pool 的行：每个运行中的请求一行	KV 池
`CacheManager`	KV 池的页分配与回收，前缀缓存的门面	CacheManager
`DecodeManager`	正在 decode 的请求集合	本章
`PrefillManager`	等待 prefill 的请求队列，组 prefill batch	本章，分块见第 10 章

调度器还继承了 SchedulerIOMixin，提供 receive_msg 和 send_result，在线服务时它们走 ZMQ（第 14 章），离线时由 LLM 覆盖。

主循环¶

minisgl/scheduler/scheduler.py · Scheduler.normal_loop

def normal_loop(self) -> None:
    """朴素循环：收消息 → 调度 → 前向 → 等结果 → 处理结果。CPU 和 GPU 轮流干活。"""
    self._defer_slot_free = False  # 普通循环里没有在途的 batch，释放立即生效
    blocking = not (self.prefill_manager.runnable or self.decode_manager.runnable)
    for msg in self.receive_msg(blocking=blocking):
        self._process_one_msg(msg)
    forward_input = self._schedule_next_batch()
    ongoing = None
    if forward_input is not None:
        ongoing = (forward_input, self._forward(forward_input))
    self._process_last_data(ongoing)

每一轮四步：

收消息。如果手上没有任何可以做的事（没有等待 prefill 的，也没有正在 decode 的），就阻塞等待；否则只取走已经到达的消息，不等待。
选 batch。有等待的请求就做 prefill，否则做 decode：
minisgl/scheduler/scheduler.py · Scheduler._schedule_next_batch
```
def _schedule_next_batch(self) -> ForwardInput | None:
    # prefill 优先：有等待的请求就先做 prefill，否则做 decode
    batch = (self.prefill_manager.schedule_next_batch(self.prefill_budget)
             or self.decode_manager.schedule_next_batch())
    return self._prepare_batch(batch) if batch else None
```
这是"prefill 优先"策略：新请求尽快得到第一个 token（TTFT 低），代价是正在 decode 的请求会被 prefill 打断一轮。一轮只做一种，不混合——这让每轮的 batch 形状简单（decode batch 可以用 CUDA Graph），也是 mini-sglang 比 SGLang 正式版简单的地方之一。 3. 准备并前向。_prepare_batch 把 batch 的各个字段填好，_forward 调用引擎。 4. 处理结果。把新 token 交给请求，判断结束，回复 detokenizer，释放结束请求的资源。

run_forever 就是反复调用这个循环（重叠调度的版本 overlap_loop 在第 11 章）。

两个管理器¶

PrefillManager 维护一个先来先服务的等待队列，每轮用一个 PrefillAdder 从队首开始尽量多地接纳请求：

minisgl/scheduler/prefill.py · PrefillManager.sc

PrefillAdder 为每个请求做三件事：在前缀缓存里匹配已缓存的部分、检查请求槽和 hedule_next_batch

def schedule_next_batch(self, prefill_budget: int) -> Batch | None: if not self.pending_list: return None adder = PrefillAdder( token_budget=prefill_budget, reserved_size=self.decode_manager.inflight_tokens,  # 正在 decode 的请求已经承诺的空间 cache_manager=self.cache_manager, table_manager=self.table_manager, ) reqs: List[Req] = [] chunked_list: List[PendingReq] = [] for pending_req in self.pending_list:  # 先来先服务：遇到第一个放不下的就停 req = adder.try_add_one(pending_req) if req is None: break pending_req.chunked_req = None if isinstance(req, ChunkedReq): pending_req.chunked_req = req chunked_list.append(pending_req) reqs.append(req) if not reqs: return None # 没做完的分块请求排到队首，下一轮优先继续 self.pending_list = chunked_list + self.pending_list[len(reqs):] return Batch(reqs=reqs, phase="prefill") KV 空间够不够（准入控制，下一章详讲）、在 token pool 里写入本轮要算的 token。遇到第一个放不下的请求就停止——不跳过它去接纳后面更小的请求，这样保证公平，不会有请求一直被插队。
DecodeManager 更简单，就是一个请求集合：
minisgl/scheduler/decode.py · DecodeManager@dataclass
class DecodeManager:
    page_size: int
    running_reqs: Set[Req] = field(default_factory=set)

    def filter_reqs(self, reqs: Iterable[Req]) -> None:
        """一轮前向之后：把本轮的请求并入运行集合，去掉已经不能再生成的。"""
        self.running_reqs = {req for req in self.running_reqs.union(reqs) if req.can_decode}

    def remove_req(self, req: Req) -> None:
        self.running_reqs.discard(req)

    def abort_req(self, uid: int) -> Req | None:
        for req in self.running_reqs:
            if req.uid == uid:
                self.running_reqs.remove(req)
                return req
        return None

    @property
    def inflight_tokens(self) -> int:
        """运行中的请求未来还可能占用多少 token 的 KV（每个请求再预留一页的余量）。"""
        reserved = (self.page_size - 1) * len(self.running_reqs)
        return sum(req.remain_len for req in self.running_reqs) + reserved

    def schedule_next_batch(self) -> Batch | None:
        if not self.runnable:
            return None
        # 按 uid 排序：TP 的各个 rank 上集合的遍历顺序可能不同，排序保证大家组出同一个 batch
        return Batch(reqs=sorted(self.running_reqs, key=lambda r: r.uid), phase="decode")

    @property
    def runnable(self) -> bool:
        return len(self.running_reqs) > 0

每轮前向之后，filter_reqs 把本轮的请求并进来，并去掉已经不能再生成的（can_decode 为假）。刚做完 prefill 的请求就这样自然地加入了 decode 集合。组 batch 时按 uid 排序：张量并行下每个 rank 都有一个 DecodeManager，Python 的 set 遍历顺序取决于对象的内存地址，不同进程里可能不同；而各 rank 的 batch 必须完全一致（同样的请求、同样的顺序），否则 all-reduce 就把不同请求的数据加在了一起。官方仓库最新的一次提交修的正是这个问题。
准备 batch¶
minisgl/scheduler/scheduler.py · Scheduler._prepare_batchdef _prepare_batch(self, batch: Batch) -> ForwardInput:
    self.engine.graph_runner.pad_batch(batch)
    self.cache_manager.allocate_paged(batch.reqs)
    batch.positions = _make_positions(batch, self.device)
    input_mapping = _make_input_tuple(batch, self.device)
    write_mapping = _make_write_tuple(batch, self.device)
    batch.out_loc = self.engine.page_table[input_mapping]
    self.engine.attn_backend.prepare_metadata(batch)
    return ForwardInput(batch, self.engine.sampler.prepare(batch), input_mapping, write_mapping)

按顺序：

补齐（第 18 章，不用 CUDA Graph 时什么也不做）；
分配 KV 页：为本轮要算的 token 分配位置并写进 page table（第 8 章）；
位置：每个请求 [cached_len, device_len)，拼成一维；
两组二维下标：input_tuple = (行号, 位置) 用来从 token pool 取输入、从 page table 取 out_loc；write_tuple = (行号, device_len) 用来把采样结果写回 token pool 的下一个位置；
注意力元数据和采样参数。

minisgl/scheduler/scheduler.py · _make_write_tupledef _make_write_tuple(batch: Batch, device: torch.device) -> Indice2D:
    """采样出的 token 写到 token_pool[table_idx, device_len]，也就是下一轮的输入位置。

    列号为 -1 表示不写：分块中的请求和已经不能再生成的请求（写 -1 相当于写到最后一列，无害）。
    注意 device_len 在 forward_batch 里会加 1，这里读取的是加 1 之前的值，正好是新 token 的位置。
    """
    rows = torch.tensor([r.table_idx for r in batch.reqs], dtype=torch.int64, pin_memory=pin(device))
    cols = torch.tensor([r.device_len if r.can_decode else -1 for r in batch.reqs],
                        dtype=torch.int64, pin_memory=pin(device))
    return rows.to(device, non_blocking=True), cols.to(device, non_blocking=True)

然后前向：
minisgl/scheduler/scheduler.py · Scheduler._forwarddef _forward(self, forward_input: ForwardInput) -> ForwardOutput:
    batch, sample_args, input_mapping, write_mapping = forward_input
    batch.input_ids = self.token_pool[input_mapping]  # 在设备上取输入，不经过 CPU
    output = self.engine.forward_batch(batch, sample_args)
    self.token_pool[write_mapping] = output.next_tokens_gpu  # 采样结果直接写回，作为下一轮输入
    self.decode_manager.filter_reqs(batch.reqs)
    return output

输入 token_pool[input_tuple] 和写回 token_pool[write_tuple] = next_tokens 都是设备上的索引操作，CPU 不需要知道 token 的值。
处理结果¶
minisgl/scheduler/scheduler.py · Scheduler._process_last_datadef _process_last_data(self, last_data: ForwardData | None) -> None:
    if last_data is None:
        return
    batch, (_, next_tokens_cpu, copy_done) = last_data[0].batch, last_data[1]
    copy_done.synchronize()  # 等采样结果拷回 CPU；这也意味着这个 batch 以及之前的 batch 都算完了
    self._release_deferred_slots()
    reply: List[DetokenizeMsg] = []
    new_finished: Set[Req] = set()
    with self.cache_manager.lazy_free_region():
        for i, req in enumerate(batch.reqs):
            if isinstance(req, ChunkedReq):  # 分块中的请求这一轮没有输出
                continue
            if req in self.finished_reqs:
                # 重叠调度下，请求在上一轮已经结束（遇到 EOS），但结束的消息还没处理，这一轮就已经
                # 把它带上了。这一轮的结果是多余的，直接丢弃（官方实现会多发一条消息，前端恰好会忽略）。
                continue
            next_token = next_tokens_cpu[i]
            req.append_host(next_token.unsqueeze(0))
            token = int(next_token.item())
            # 判断是否达到 max_tokens 不能用 req.can_decode：重叠调度下，处理第 N 轮结果时
            # 第 N+1 轮已经发射，req 的 device_len 已经被推进了一步，会提前一个 token 判定结束。
            # 用 CPU 侧已经收到的 token 数判断，与调度节奏无关。
            finished = len(req.input_ids) >= req.max_device_len
            if not req.sampling_params.ignore_eos:
                finished |= token == self.eos_token_id
            reply.append(DetokenizeMsg(uid=req.uid, next_token=token, finished=finished))
            if finished:
                self.decode_manager.remove_req(req)
                self._free_req_resources(req)
                new_finished.add(req)
            elif batch.is_prefill:  # prefill 刚结束：立刻把提示词的 KV 交给前缀缓存
                self.cache_manager.cache_req(req, finished=False)
    self.finished_reqs = new_finished
    self.send_result(reply)

对 batch 中的每个请求：把新 token 追加到 CPU 上的 input_ids，判断是否结束（达到 max_tokens，或者遇到 EOS 且没有设置 ignore_eos），生成一条 DetokenizeMsg。结束的请求释放资源：归还 page table 的行，把 KV 交给缓存（naive 缓存会全部释放）。刚做完 prefill 的请求，把提示词的 KV 交给前缀缓存（第 9 章）。
这段代码里有两处是为第 11 章的重叠调度准备的——finished_reqs 和"用 len(req.input_ids) 而不是 req.can_decode 判断长度"——那一章再解释。
新请求的处理：
minisgl/scheduler/scheduler.py · Scheduler._process_one_msgdef _process_one_msg(self, msg: BaseBackendMsg) -> None:
    if isinstance(msg, BatchBackendMsg):
        for m in msg.data:
            self._process_one_msg(m)
    elif isinstance(msg, ExitMsg):
        raise KeyboardInterrupt
    elif isinstance(msg, UserMsg):
        input_len, max_seq_len = len(msg.input_ids), self.engine.max_seq_len
        max_output_len = max_seq_len - input_len
        if max_output_len <= 0:
            logger.warning_rank0(f"Input length {input_len} exceeds {max_seq_len}, "
                                 f"request {msg.uid} dropped.")
            return
        if msg.sampling_params.max_tokens > max_output_len:
            msg.sampling_params.max_tokens = max_output_len
        self.prefill_manager.add_one_req(msg)
    elif isinstance(msg, AbortBackendMsg):
        req = self.prefill_manager.abort_req(msg.uid) or self.decode_manager.abort_req(msg.uid)
        if req is not None:
            self._free_req_resources(req)
            # 重叠调度下它可能还在在途的 batch 里：记为已结束，那一轮的结果会被丢弃，
            # 不会再对一个已经释放的请求调用 cache_req
            self.finished_reqs.add(req)
    else:
        raise NotImplementedError(f"Unknown message type: {type(msg)}")

提示词加上 max_tokens 超过模型最大长度时，max_tokens 被截短；提示词本身就超长的请求直接丢弃（官方同样只打一条警告，前端不会收到任何回复——这是一个可以改进的地方）。
离线接口 LLM¶
minisgl/llm/llm.py · LLMclass LLM(Scheduler):
    """离线推理接口：不起任何进程，直接复用调度器的主循环。

    它覆盖了调度器的两个 IO 方法：receive_msg 从待处理列表里取请求（按 prefill 预算分批放入），
    send_result 把生成的 token 记下来。所有请求处理完、调度器又要阻塞等待新消息时，
    抛出 RequestAllFinished 跳出 run_forever。
    """

    def __init__(self, model_path: str, dtype: torch.dtype = torch.bfloat16, **kwargs):
        config = SchedulerConfig(model_path=model_path, tp_info=DistributedInfo(0, 1),
                                 dtype=dtype, offline_mode=True, **kwargs)
        super().__init__(config)
        self.pending_requests: List[Tuple[List[int] | str, SamplingParams]] = []
        self.status_map: Dict[int, RequestStatus] = {}
        self.counter = 0

    def _tokenize_one(self, prompt: List[int] | str) -> torch.Tensor:
        if isinstance(prompt, str):
            return self.tokenizer.encode(prompt, return_tensors="pt").view(-1).to(torch.int32)
        return torch.tensor(prompt, dtype=torch.int32)

    def offline_receive_msg(self, blocking: bool = False) -> List[BaseBackendMsg]:
        if blocking and not self.pending_requests:
            raise RequestAllFinished()
        results: List[BaseBackendMsg] = []
        added = total_len = 0
        for prompt, sp in self.pending_requests:
            if total_len >= self.prefill_budget:
                break
            input_ids = self._tokenize_one(prompt)
            total_len += len(input_ids)
            uid = self.counter + added
            added += 1
            results.append(UserMsg(uid=uid, input_ids=input_ids, sampling_params=sp))
            self.status_map[uid] = RequestStatus(uid, input_ids.tolist(), [])
        self.counter += added
        self.pending_requests = self.pending_requests[added:]
        return results

    def offline_send_result(self, reply: List[DetokenizeMsg]) -> None:
        for msg in reply:
            status = self.status_map[msg.uid]
            if not (msg.finished and msg.next_token == self.eos_token_id):
                status.output_ids.append(msg.next_token)

    def generate(self, prompts: List[str] | List[List[int]],
                 sampling_params: List[SamplingParams] | SamplingParams
                 ) -> List[Dict[str, str | List[int]]]:
        self.pending_requests, self.status_map, self.counter = [], {}, 0
        if isinstance(sampling_params, SamplingParams):
            sampling_params = [sampling_params] * len(prompts)
        # 每次都复制一份：调度器可能会修改 max_tokens
        self.pending_requests = [(p, SamplingParams(**vars(sp)))
                                 for p, sp in zip(prompts, sampling_params)]
        try:
            self.run_forever()
        except RequestAllFinished:
            pass
        return [{"text": self.tokenizer.decode(self.status_map[i].output_ids),
                 "token_ids": self.status_map[i].output_ids} for i in range(len(prompts))]

LLM 继承 Scheduler，用 offline_mode=True 让 IO 层不创建任何 ZMQ 队列，然后覆盖两个方法：

offline_receive_msg：从待处理列表里取请求，按 prefill 预算（max_extend_tokens）分批放入。当调度器"没事可做、要阻塞等待新消息"而列表也空了时，说明全部完成，抛出 RequestAllFinished 跳出 run_forever；
offline_send_result：把每条回复的 token 记到对应请求上（EOS 不记）。

这个小技巧让离线接口与在线服务共用同一个主循环，测过一个就等于测过另一个的调度逻辑。
运行¶
examples/ch07_scheduler.py"""第 7 章：离线接口跑 5 个请求，打印每一轮调度出的 batch。

为了只看"骨架"，关掉重叠调度、用 naive 缓存（不复用前缀）。
"""

import torch
from minisgl.core import SamplingParams
from minisgl.env import ENV
from minisgl.llm import LLM
from transformers import AutoModelForCausalLM, AutoTokenizer

ENV.DISABLE_OVERLAP_SCHEDULING.value = True
path = "models/Qwen3-0.6B"
llm = LLM(path, dtype=torch.float32, max_running_req=8, num_page_override=1024,
          max_seq_len_override=512, cache_type="naive")
orig = llm._forward


def traced(fi):
    b = fi.batch
    detail = ", ".join(f"uid{r.uid}:{r.extend_len}" for r in b.reqs)
    print(f"  {b.phase:<7} bs={b.size}  token 数={len(b.positions):<3} [{detail}]")
    return orig(fi)


llm._forward = traced
prompts = ["The capital of France is", "List three prime numbers:", "def fibonacci(n):",
           "1 + 1 =", "Once upon a time,"]
lengths = [6, 2, 5, 1, 4]  # 每个请求的 max_tokens 不同：短的先结束离开
print("调度轨迹：")
outputs = llm.generate(prompts, [SamplingParams(max_tokens=n, ignore_eos=True) for n in lengths])
tok = AutoTokenizer.from_pretrained(path)
hf = AutoModelForCausalLM.from_pretrained(path, dtype=torch.float32)
for p, n, o in zip(prompts, lengths, outputs):
    ids = tok(p, return_tensors="pt").input_ids
    ref = hf.generate(ids, max_new_tokens=n, min_new_tokens=n, do_sample=False)[0, ids.shape[1]:].tolist()
    print(f"{p!r:<30} -> {o['text']!r:<32} 与 HF 一致: {o['token_ids'] == ref}")
llm.shutdown()

运行结果调度轨迹：
  prefill bs=5  token 数=24  [uid0:5, uid1:5, uid2:4, uid3:5, uid4:5]
  decode  bs=4  token 数=4   [uid0:1, uid1:1, uid2:1, uid4:1]
  decode  bs=3  token 数=3   [uid0:1, uid2:1, uid4:1]
  decode  bs=3  token 数=3   [uid0:1, uid2:1, uid4:1]
  decode  bs=2  token 数=2   [uid0:1, uid2:1]
  decode  bs=1  token 数=1   [uid0:1]
'The capital of France is'     -> ' Paris. The capital of Italy'   与 HF 一致: True
'List three prime numbers:'    -> ' 2'                             与 HF 一致: True
'def fibonacci(n):'            -> '\n    if n =='                  与 HF 一致: True
'1 + 1 ='                      -> ' '                              与 HF 一致: True
'Once upon a time,'            -> ' there were 3'                  与 HF 一致: True

第一轮 5 个请求一起 prefill（24 个 token 拼成一维）。max_tokens=1 的 uid3 在 prefill 之后就结束了，没有进入 decode；uid1 在一轮 decode 后离开；其余请求继续，batch 逐渐变小。这就是连续批处理：请求按自己的节奏加入和离开，不用等整个 batch 结束。5 个请求的输出都与 Hugging Face 单独生成的结果相同。

官方实现

主循环：scheduler/scheduler.py · Scheduler.normal_loop（第 108～118 行）
准备与前向：scheduler/scheduler.py · Scheduler._prepare_batch（第 204～217 行）、scheduler/scheduler.py · Scheduler._forward（第 227～233 行）
prefill 队列：scheduler/prefill.py · PrefillManager.schedule_next_batch（第 126～151 行）
decode 集合：scheduler/decode.py · DecodeManager（第 9～39 行）
离线接口：llm/llm.py · LLM（第 28～98 行）


测试¶
tests/test_ch07_scheduler.py · test_requests_with_different_lengths_leave_and_joindef test_requests_with_different_lengths_leave_and_join():
    """max_tokens 各不相同：短的先结束离开 batch，其余继续；新一轮 generate 的请求随后加入。"""
    ENV.DISABLE_OVERLAP_SCHEDULING.value = True
    llm = build_llm(QWEN3, cache_type="naive")
    lengths = [3, 12, 7, 1, 9]
    out = llm.generate(PROMPTS, [greedy(n) for n in lengths])
    ref = hf_greedy(QWEN3, PROMPTS, 12)
    assert [o["token_ids"] for o in out] == [r[:n] for r, n in zip(ref, lengths)]
    assert llm.table_manager.available_size == 8  # 所有行都已归还
    llm.shutdown()


怎么讲清楚
讲调度器：连续批处理按"一轮一轮"组 batch，请求随时加入、做完立即离开，不用等整批结束（静态批处理的短请求要陪长请求跑完）。mini-sglang 的主循环是收消息 → 选 batch → 准备 → 前向 → 处理结果，一轮只做一种：prefill 优先（先来先服务，放不下就停），否则做 decode；decode 集合在组 batch 时按 uid 排序，保证每个 TP rank 组出完全相同的 batch。和 vLLM V1 的区别：vLLM 用统一的 token 预算把 prefill 和 decode 混在同一步里，mini-sglang 分开做，实现简单但长 prefill 会让 decode 停一轮。离线接口 LLM 只替换收发两个方法，复用同一个主循环。

练习¶

把调度策略改成"decode 优先"：有正在 decode 的请求就先做 decode，没有时才做 prefill。会有什么问题？（提示：考虑一个长期有请求在 decode 的服务里，新请求什么时候能被处理。）
PrefillManager 遇到第一个放不下的请求就停止。如果改成"跳过它、继续尝试后面的请求"，吞吐会怎样？有什么风险？
在 _process_one_msg 里，超长的提示词被直接丢弃，前端永远收不到回复。怎样改成给前端返回一个错误？需要改哪些消息类型？


参考答案

只要 decode 集合不空，新请求就永远得不到 prefill，TTFT 无上限（饥饿）。实用的折中是限制 decode 连续执行的轮数，或者像 SGLang 正式版那样把 prefill 和 decode 混在一个 batch 里。
吞吐可能提高（小请求填满了预算），但大请求可能一直被后来的小请求插队，出现饥饿。需要"老化"机制：等待越久优先级越高。
调度器直接回复一条带错误标记的 DetokenizeMsg（例如新增 error 字段，finished=True），detokenizer 转成带错误信息的 UserReply，API Server 据此返回 400。


小结¶

 调度器 = 四个管理器 + 一个主循环：收消息 → 选 batch（prefill 优先，一轮只做一种）→ 准备 → 前向 → 处理结果。
 prefill 队列先来先服务、遇到放不下的就停；decode 集合在每轮前向后更新，组 batch 时按 uid 排序保证各 rank 一致。
 输入从 token pool 取、采样结果写回 token pool，都在设备上完成。
 离线接口 LLM 覆盖收发两个方法，复用同一个主循环。


本章练习题（1 道）
在浏览器里直接写代码、跑测试；也可以在本地用 python practice/judge.py 判题。

191. PrefillAdder：准入控制与分块 prefill（中等）









  
  






                
  
  
    
    发现错误或有更好的写法？在 GitHub 上编辑此页 · 提交问题