给模型调用加一个并发信号量，能限制同时执行的请求，却未必限制等待它的任务。若入口仍不断创建协程，执行区里只有两条请求，内存里仍可能积着成千上万条等待。

背压要同时约束执行中的请求、排队的请求，以及队列之外等待入队的生产者。本文接续总超时预算与安全重试，用不联网的事件模拟区分准入、拒绝、排队过期和迟到完成。容量与耗时均为教学设定，不是模型服务参数建议。

并发上限与入口容量不是同一个数

Python 的 asyncio.Semaphore用计数控制同时获得许可的任务；它不为整个请求入口规定一个有限的等待任务总数。

asyncio.Queue可以设置正数 maxsize 来限制队内项目，但 await put() 会等待空位；如果调用者无限创建等待 put() 的任务，压力仍留在队列外。完整设计还需要限制生产者数量，或在入口采用明确的有限等待、立即拒绝等策略。本文选择队满立即拒绝。

层级 本文计数 上限与边界
正在执行 running 最多 C 个；结束后才释放执行槽
已经入队 waiting 最多 Q 个，按到达顺序启动
尚未准入 入口请求 不另设无限等待区，满载立即拒绝
教学记录与输入脚本 模拟器自身数据 保留全部轨迹用于验证，不声称整个模拟器内存只有 C+Q 条记录

C+Q 限制的是这个模型内已准入且尚未终止的工作数，不包含输入脚本、日志、连接池和其他进程。多副本服务的本地信号量也不自动等于全局配额。

队列满时为什么要反馈给上游

当输入长期超过处理能力，排队会把即时拒绝变成更晚的失败。AWS Builders Library 的队列积压讨论强调，需要关注任务年龄、积压恢复与上游压力反馈，而不只看成功入队的比例。

拒绝策略取决于业务语义：实时问答可以明确报告忙碌并让上游按契约处理；必须持久化的后台作业则需要持久队列、优先级与恢复机制。不能为了让入口成功率好看，偷偷把实时请求变成无限等待的后台工作。

  1. 有空闲执行槽就启动

    先给旧队列分配空位，再接纳同一时刻的新到达，保持 FIFO。

  2. 没有执行槽但队列有位就排队

    截止时刻从到达开始计，等待不会获得新的总预算。

  3. 执行区与队列都满就拒绝

    不给被拒请求另建一个隐形等待任务；拒绝不等于已完成模型计算。

五条请求的手工演算

设 C=2、Q=2，A 到 E 都在 0 毫秒到达。预算从到达时刻开始，deadline=arrival+budget。以下毫秒数均为合成数据。

请求 合成执行耗时 总预算 入口决定
A 100 300 立即执行
B 100 300 立即执行
C 50 200 排队
D 50 90 排队
E 20 200 执行槽、队列均满，拒绝
两个执行槽先处理A与B，C等待100毫秒后执行，D等待到90毫秒时过期，E在入口立即拒绝
原创合成事件时间线：D 从未占用执行槽，E 从未入队；它们不能混进“已完成调用”的统计。查看原图
时刻 事件 正在执行 排队
0 A、B 启动；C、D 入队；E 拒绝 A、B C、D
90 D 达到截止时刻，从队列移除 A、B C
100 A、B 完成；C 启动 C 空
150 C 完成，距到达共 150 毫秒 空 空

C 的执行只花 50 毫秒，但端到端耗时是 150 毫秒。只报告运行耗时会漏掉前面的排队。

排队过期与执行超时要分开

尚未启动的请求达到截止时刻，可以直接从等待队列移除，不产生一次下游调用。已经启动的请求则不同：客户端不再等待，并不证明下游已经停止执行。

本模拟器不假设服务端取消成功。已经运行的任务即使过期，也继续占用执行槽，直到合成的实际完成事件出现；完成时间达到或超过 deadline 时，结果标为 late，不记作按时成功。这样可以避免“客户端超时就立刻释放槽，再启动另一个后台实际仍重叠的调用”的错误计数。

可运行的纯 Python 事件模拟

依赖 Python 3.10 或更新版本的标准库，不创建协程、不访问网络、不真实睡眠。执行耗时是脚本的未来完成事件，准入决定不根据它预测真实服务性能。

同一时刻的顺序为：等待请求过期、运行请求完成、启动旧队列、处理新到达。同一时刻到达按输入顺序处理。执行耗时必须为正，避免零耗时事件反复在同一批次产生。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
from collections import OrderedDict
from dataclasses import dataclass
import heapq
import json


@dataclass(frozen=True)
class Job:
    id: str
    arrival_ms: int
    service_ms: int
    budget_ms: int


def simulate(jobs, *, workers=2, queue_capacity=2):
    def integer(value, minimum):
        if type(value) is not int or value < minimum:
            raise ValueError("invalid integer setting")

    integer(workers, 1)
    integer(queue_capacity, 0)
    jobs = list(jobs)
    seen = set()
    for job in jobs:
        if not isinstance(job, Job) or type(job.id) is not str or not job.id:
            raise ValueError("invalid job")
        if job.id in seen:
            raise ValueError("duplicate job id")
        seen.add(job.id)
        integer(job.arrival_ms, 0)
        integer(job.service_ms, 1)
        integer(job.budget_ms, 1)
    rows = [{"id": j.id, "arrival_ms": j.arrival_ms,
             "deadline_ms": j.arrival_ms + j.budget_ms,
             "status": "not_arrived", "start_ms": None, "end_ms": None}
            for j in jobs]
    # Event kinds: 0 waiting expiry, 1 actual completion, 2 arrival.
    events = [(j.arrival_ms, 2, i) for i, j in enumerate(jobs)]
    heapq.heapify(events)
    waiting = OrderedDict()
    running, peak_running, peak_waiting, now = 0, 0, 0, 0
    trace = []

    def record(event, i):
        nonlocal peak_running, peak_waiting
        peak_running = max(peak_running, running)
        peak_waiting = max(peak_waiting, len(waiting))
        trace.append({"time_ms": now, "event": event, "id": jobs[i].id,
                      "running": running, "waiting": len(waiting)})

    def start(i):
        nonlocal running
        rows[i]["status"] = "running"
        rows[i]["start_ms"] = now
        running += 1
        heapq.heappush(events, (now + jobs[i].service_ms, 1, i))
        record("start", i)

    while events:
        now = events[0][0]
        batch = []
        while events and events[0][0] == now:
            batch.append(heapq.heappop(events))
        for _, kind, i in batch:
            if kind == 0 and rows[i]["status"] == "queued":
                del waiting[i]
                rows[i].update(status="expired", end_ms=now)
                record("expired", i)
        for _, kind, i in batch:
            if kind == 1:
                running -= 1
                status = "completed" if now < rows[i]["deadline_ms"] else "late"
                rows[i].update(status=status, end_ms=now)
                record(status, i)
        while waiting and running < workers:
            i, _ = waiting.popitem(last=False)
            start(i)
        for _, kind, i in batch:
            if kind != 2:
                continue
            if running < workers:
                start(i)
            elif len(waiting) < queue_capacity:
                waiting[i] = None
                rows[i]["status"] = "queued"
                heapq.heappush(events, (rows[i]["deadline_ms"], 0, i))
                record("enqueue", i)
            else:
                rows[i].update(status="overload", end_ms=now)
                record("overload", i)
    return {"jobs": rows, "peak_running": peak_running,
            "peak_waiting": peak_waiting, "trace": trace}


if __name__ == "__main__":
    jobs = [Job("A", 0, 100, 300), Job("B", 0, 100, 300),
            Job("C", 0, 50, 200), Job("D", 0, 50, 90), Job("E", 0, 20, 200)]
    print(json.dumps(simulate(jobs), ensure_ascii=False, indent=2))

本文 queue_capacity=0 表示没有等待区，不同于 asyncio.Queue(maxsize=0) 的无限队列语义；不能把这个参数直接照搬给实际 Queue。过期事件若发现请求已经启动，则不释放执行槽，等实际完成事件处理。

对于 q 条输入记录，事件数 O(q)，堆维护时间 O(q log q)，轨迹、脚本、结果与事件占 O(q) 空间。OrderedDict 负责 FIFO 与队中任意请求的过期移除，操作平均 O(1)。模拟器刻意保留完整记录供对拍，不是可直接上线的请求服务器。

生产系统还要保护哪些边界

边界 必须明确
单机与全局 多进程、多副本的本地并发数会相加；供应商全局配额需要额外协调
并发与速率 C 限制同时工作数，不等于每秒请求数或 token 额度
重试 SDK 重试与退避是否一直占用本地执行槽，要遵循自己的资源定义
流式输出 首 token 到达不代表完整流结束，不能提前释放仍在工作的槽
租户公平性 单一 FIFO 可能让一个租户占满队列，需要另行设计配额或调度
持久后台任务 不能照搬本例的过期丢弃；需匹配持久化与恢复契约
性能调参 服务时间随批处理、模型、输入长度和设备变化，C 与吞吐不必线性对应
自检：队列最多 2 条，但入口无限创建等待 put() 的任务，内存压力已经有界了吗？

没有。队内数量有界，队外阻塞的任务仍可能无界。必须再限制入口生产者或等待数量，并明确定义拒绝、超时与取消策略。

必测情况 期望结果
C=2、Q=2、同刻到达五条 两条执行、两条排队、一条拒绝
队列中途过期 无下游 start 事件，及时腾出队列位置
完成与新到达同刻 旧队列先启动，新请求只使用剩余容量
完成恰好达到 deadline 标 late，不算按时成功
运行已过期但还未完成 继续占用执行槽，不产生假空位
请求 ID 重复或布尔值冒充整数 拒绝无效输入

项目 tests/verify-bounded-admission-article.cjs 提取并运行本文代码，以独立逐毫秒模型核验随机事件轨迹，检查 running≤C、waiting≤Q 与每个请求唯一终态。测试不会发出真实模型请求。

观测指标应分别报告准入、入口拒绝、排队过期、按时完成与迟到完成，记录排队时长和端到端时长。继续接上RAG 回归评测：质量门槛与容量门槛需要分别通过，不能用一个平均成功率掩盖不同失败。