连续批处理：状态、时间轴与可证伪模拟

模拟对象与不被偷换的假设

前三章已经给出一次前向的计算与空间模型，现在引入多个到达请求。本章的项目是离散调度器，不执行 Transformer 前向。每个请求有唯一整数编号 id、到达刻 arrival、所需服务词元数 tokens；每个时刻的长度是一 tick，所有活跃请求每 tick 各完成一个服务词元。并行槽位数为 S。词元服务是已知的有限工作量，没有预填充、网络、采样停止或可变计算时间，因此 tick 是抽象单位，不是毫秒。

这种简化让我们隔离一个机制：同一批次中短请求完成后，空出的槽位何时交给下一个请求？静态 FIFO 在整组请求全部完成前保留组边界，结束的槽位空等；连续 FIFO 在下一 tick 开始时回收完成者，按到达时间与编号顺序补入等待者。两种策略使用完全相同的输入和单位服务假设。若连续策略表现更好，只能归因于该模型中的槽位利用差异，不能据此断言真实设备上连续批处理必有相同比例加速。

可追溯状态和一步转移

状态由尚未接纳的等待队列、活跃请求及剩余服务量、已完成记录和当前时刻组成。每 tick 先处理上一 tick 的完成回收，再接纳到达且有空位的请求，然后给当前运行者各减一，剩余量变成零者记录完成时刻为当前 tick 加一。如果系统没有运行者且下一请求尚未到达，时钟可直接跳到下次到达；这段空闲仍包含在从零到最终完成的观测区间内。排序只用事先约定字段，不能为了漂亮结果偷偷优先短任务。

调度的安全不变量包括：一个编号不能同时存在两份活跃副本；未到达请求不能开始；活跃数不超过槽位；剩余量非负；每个请求恰有一份最终完成记录。工作守恒要求所有 tick 中运行槽位数之和等于全部请求词元服务量之和。此模型没有取消或失败，因而守恒等式应精确成立。若未来加入取消，就必须把已服务、已取消和剩余工作分别记录，不能仍沿用同一个完成量断言。

完整算例一：四个请求、两个槽位

请求零到三都在零时刻到达，服务量依次为一、四、二、三。静态策略先运行零与一：第一 tick 两者都服务，零完成；之后三个 tick 只剩一运行，直到时刻四整组结束。第二组二与三从四开始，二在六完成，三在七完成。总工作量十个服务词元，占用七个 tick、两个槽位，利用率为 $10/(2\times7)=5/7$。短请求零留下的三个空槽位是本模型中可观察的浪费。

连续策略第一 tick 后释放零，在时刻一接入请求二；二服务两 tick 于三完成，再于三接入请求三。请求一在四完成，请求三在六完成。完成时刻依次为一、四、三、六，总跨度六 tick，利用率为 $10/12=5/6$。吞吐从每 tick 十除以七变成十除以六，比例七除以六，而不是按利用率差的百分点来解释。等待时间零、一、二、三号依次为零、零、一、三，均由开始减到达计算。

from lab import schedule
jobs=[{'id':i,'arrival':0,'tokens':n} for i,n in enumerate([1,4,2,3])]
a=schedule(jobs,2,False);b=schedule(jobs,2,True)
assert a['makespan_ticks']==7 and b['makespan_ticks']==6
assert a['busy_slots']==b['busy_slots']==10
assert [j['finish'] for j in b['completed']]==[1,4,3,6]
print(a['utilization'],b['utilization'])

完整算例二：改策略也可能没有收益

若两请求都在时刻三到达、服务量都为四且槽位为二，则两者从三开始并于七完成。静态与连续策略完全相同，因为没有一个提前释放而可被复用的槽位。按从零观测的利用率为八除以十四，若改从首请求到达观测则为一；两个值各有明确分母，不能挑较大者而不标区间。这个负结果可证伪“连续批处理必定更快”的过度结论，并检查时钟跳跃是否错误抹掉了初始空闲。

另一类容易遗漏的边界是请求在 tick 边界恰好到达。当前协议先回收再接纳，到达时刻等于当前 tick 的请求可立即加入本 tick；在上一 tick 开始之后才到达者不能倒流进入。若时间离散化来自真实事件，应事先规定取整方式，否则将到达时间向下取整会人为减少等待。本项目输入直接采用整数 tick，因此没有将真实秒数粗化的误差。测试同时检查延迟到达、相同长度以及所有开始时刻不早于到达。

等待、完成与公平分别是什么

定义请求 i 的等待为 $w_i=s_i-a_i$、流经时间为 $r_i=f_i-a_i$，其中 a、s、f 分别为到达、开始服务和完成；此处 f 不再表示首词元，避免与第一章时间符号混淆。由于每个运行 tick 都服务一个词元且无抢占空档，本模型有 $r_i=w_i+g_i$。完成时间均值不能替代最大等待，长时间排队的某个请求会被均值稀释。FIFO 只定义接纳顺序，不自动给出任意负载下有限等待的保证。

若平均到达工作量超过系统长期可提供的服务，队列会增长；提高槽位利用率不能取消这个守恒约束。还需注意同一 tick 多个槽位并行不表示单个请求的词元可以互相并行，因为自回归后续依赖此前输出。在真实模型中增大 batch 可能改善权重复用，也可能增加每轮计算时间；本模拟假定轮长不随 batch 改变，所以它专门检验调度空洞，而不预测硬件的最优 batch。

已知全部输出长度是离线模拟的便利，真实服务往往只知道上限，实际何时生成 EOS 不确定。FIFO 连续策略不需要提前知道总长来做接纳，它只在完成事件发生后释放槽位；但我们的模拟器用预设整数服务量产生完成事件。若研究长度预测或优先短请求，应把预测误差、最大等待保护与未知完成状态加入模型，并与不使用未来信息的基线比较，不能利用预知真实长度获得不可实现的成绩。

把预填充和空间约束重新接入真实系统

真实连续批处理不仅有解码完成，还要插入新提示的预填充。长提示可能占用一轮大量计算，使既有请求的输出间隔变大；分块预填充把长提示拆开交错执行，但会引入调度选择和缓存分配。要评价这类系统，状态还需记录每请求处于预填充或解码、已处理提示长度、KV 页、模型与位置身份，以及该轮允许的词元预算。本实验没有实现这些机制，连续批处理官方文档用于核对概念，项目成绩只覆盖更小的离散模型。

第三章的字节准入与本章槽位准入应联合满足。两个请求即使槽位空着，若增长到上限所需 KV 无预算，也不应接纳；反之内存足够也不代表计算服务能力足够。真实系统还可能支持抢占重算或转移缓存，此时需要记录额外服务成本和恢复身份。没有这些状态的调度表，只能回答静态与连续回填的理想差别，不能作为生产服务容量承诺。最终报告会把模拟输出与 CPU 前向时间单列，不用 tick 乘某个平均毫秒伪造端到端延迟。

jobs=[(0,0,1),(1,0,4),(2,0,2),(3,0,3)]
# An independent hand-written event ledger for the continuous case.
running=[[0,1],[1,2],[1,2],[1,3],[3],[3]]
work=sum(len(row) for row in running)
counts={i:sum(i in row for row in running) for i,_,_ in jobs}
assert work==10
assert counts=={i:n for i,_,n in jobs}
assert max(map(len,running))==2
print(work/(2*len(running)))

第一类验证是事件重放：逐 tick 检查每个编号实际服务次数、开始时刻和完成时刻，而非只看总跨度。第二类验证是变形与极限：单请求两策略一致；槽位大于请求数且同时到达时无回填收益；全部等长请求的整组边界不产生短请求空洞。若这些性质失败，应先修复状态转移，再讨论优化。当前十项测试包含代表性正例和无收益例，没有给出任意输入的形式化全称证明。

练习与项目增量

本章项目增量是 schedule 及保存下来的逐 tick 运行者列表。它提供调度机制的可重放证据，不伪装成网络服务。第五章转入量化：与缓存等价和调度顺序不同，量化主动改变数值函数，因此要引入误差预算与质量评价，再到第七章在固定条件下测量 CPU 时间。

先验证整数工作域，再讨论终止

schedule 现在先检查 slots 是正整数；每条 job 必须含 id、arrival、tokens，三者 type 恰为 int，排除 Python 的 bool，arrival≥0、tokens≥1，id 唯一。小数、NaN、无穷、字符串、缺字段立即给 ValueError。若让 tokens=1.5 进入原循环，剩余量按 1 递减会经过 0.5、−0.5 而永不等于零，完成条件永远不触发。 合法有限整数输入下，令总剩余工作 W 为所有未完成请求 remaining 的和。运行一个非空 tick 至少完成一个词元，使 W 严格下降至少一；空闲时直接跳到下一个有限 arrival，接纳至少一个任务。W 起始是有限非负整数，故运行 tick 数不超过总词元量，每个任务最终仅完成一次。这个终止论证依赖输入域，不能只证明示例整数而放任其他数值进入。

# schedule
def schedule(jobs,slots):
    raise NotImplementedError('schedule')

# schedule
def schedule(jobs,slots):
 if type(slots)is not int or slots<1:raise ValueError('positive integer slots')
 rows=[];seen=set()
 for job in jobs:
  if set(job)!={'id','arrival','tokens'}or any(type(job[k])is not int for k in job):raise ValueError('exact integer id/arrival/tokens')
  if job['arrival']<0 or job['tokens']<1 or job['id']in seen:raise ValueError('invalid/duplicate job')
  seen.add(job['id']);rows.append(dict(job))
 waiting=sorted(rows,key=lambda j:(j['arrival'],j['id']));active=[];time=0;finished={};trace=[]
 while waiting or active:
  if not active and waiting:time=max(time,waiting[0]['arrival'])
  while waiting and waiting[0]['arrival']<=time and len(active)<slots:active.append(waiting.pop(0))
  served=[j['id']for j in active];trace.append({'tick':time,'served':served})
  for job in active:job['tokens']-=1
  for job in active:
   if job['tokens']==0:finished[job['id']]=time+1
  active=[j for j in active if job_remaining(j)>0];time+=1
 return {'finished':finished,'trace':trace,'makespan':time}