把请求转发出去收到响应头再把并发计数减一这段代码看起来很顺。但对流式推理而言响应头后面可能还有很长一段正文。计数先归还新的请求就能进来正在生成的请求却没有结束。这篇把目标收窄到一个问题一份本地并发许可应当由谁持有到什么时候不讨论智能路由也不拿客户端断开推断远端已停止计算。让响应体成为 owner请求的建立阶段先持有许可。拿到响应体后把许可一起移动进响应体包装器。只要消费者还可能继续读这份许可就留在里面。图中三条出口不是三次释放而是同一个资源的不同结束路径。正常读到 EOF可以立即释放读取报错后如果包装器决定不再允许继续读也可以释放消费者不读了直接丢弃包装器则由字段的析构归还。Tokio 的计数信号量提供 owned permit适合随响应体移动。它约束的是本进程的许可数量不是远端计算状态。文档和具体行为应对齐使用版本。Tokio Semaphore一个可以单独运行的例子下面用 channel 充当正文帧来源。它没有实现 HTTP 客户端、SSE 解析器或推理后端mock 帧只是把资源生命周期从网络条件里拆出来。真实适配时HTTP 建连失败也要使建立阶段的许可随错误返回而析构。新建目录保存以下两个文件。依赖固定为本次验证时已缓存的 Tokio 1.52.3不需要宏[package] name permit-body-demo version 0.1.0 edition 2024 [dependencies] tokio { version 1.52.3, features [rt, sync] }将下面内容保存为src/main.rsuse std::{future::pending, sync::Arc}; use tokio::sync::{mpsc, OwnedSemaphorePermit, Semaphore}; struct Body { rx: mpsc::UnboundedReceiverResultstatic str, static str, permit: OptionOwnedSemaphorePermit, ended: bool, } impl Body { async fn next(mut self) - OptionResultstatic str, static str { if self.ended { return None; } let item self.rx.recv().await; if matches!(item, None | Some(Err(_))) { self.ended true; self.rx.close(); self.permit.take(); // drop the only owned permit } item } } async fn response( sem: ArcSemaphore, frames: VecResultstatic str, static str, ) - Body { let permit sem.acquire_owned().await.unwrap(); let (tx, rx) mpsc::unbounded_channel(); for frame in frames { tx.send(frame).unwrap(); } drop(tx); Body { rx, permit: Some(permit), ended: false } } fn main() { tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(demo()); } async fn demo() { let sem Arc::new(Semaphore::new(1)); let mut body response(sem.clone(), vec![Ok(data: hello\n\n)]).await; assert_eq!(sem.available_permits(), 0); // headers are not EOF assert!(body.next().await.unwrap().is_ok()); assert_eq!(sem.available_permits(), 0); assert!(body.next().await.is_none()); assert_eq!(sem.available_permits(), 1); drop(body); assert_eq!(sem.available_permits(), 1); // no double release println!(EOF: 1); let mut body response(sem.clone(), vec![Err(mock read error)]).await; assert!(body.next().await.unwrap().is_err()); assert_eq!(sem.available_permits(), 1); assert!(body.next().await.is_none()); drop(body); assert_eq!(sem.available_permits(), 1); println!(Error: 1); let body response(sem.clone(), vec![Ok(partial)]).await; drop(body); assert_eq!(sem.available_permits(), 1); println!(Drop: 1); let body response(sem.clone(), vec![]).await; let task tokio::spawn(async move { let _held body; pending::()().await; }); tokio::task::yield_now().await; assert_eq!(sem.available_permits(), 0); task.abort(); assert!(task.await.unwrap_err().is_cancelled()); assert_eq!(sem.available_permits(), 1); println!(Abort: 1); }执行cargo run --offline。本地验证环境为 macOS aarch64、rustc 1.98.0、Tokio 1.52.3输出为EOF: 1 Error: 1 Drop: 1 Abort: 1每个 1 都是结束后的可用许可数。例子还检查了读到第一帧时许可仍为 0以及结束后再次 drop 不会多释放一次。任务取消的用例会等待被 abort 的任务真正结束再读计数只调用 abort 而不等待容易把调度尚未完成误认为泄漏。这四个断言说明了什么OptionOwnedSemaphorePermit表达了“包装器至多还持有一份许可”。终态用take()取走并立即析构以后再次遇到清理路径取到的只是 None。普通的消费者 Drop 不需要手工减原子计数字段本身会析构。错误路径要定义清楚。本例把任意 mock 读取错误当成终态关闭接收端并禁止下一次读取。如果真实流的某类错误可恢复就不能照搬这个策略在仍允许继续读时先释放。建立连接、响应体和读取任务之间的 owner 转移同样值得画出来。把 guard 留在创建函数的局部变量里即使代码能编译也会在函数返回时过早归还。本地资源释放之后远端是否仍生成 token需要后端协议和日志单独回答。llama-server的监控与接口应对照选定提交的说明核查不能用本例的 channel 关闭代替。llama.cpp Server README本轮没有接真实模型没有测试网络取消传播也没有吞吐、费用或可用性结论。这里拿到的结果很小四条本地结束路径都把同一份许可归还了一次。10 月仓库标签也不等于已经验证过的生产版本例如 b11829 发布页标有预发布属性。实际发稿前应再核所用版本与接口状态。