推理服务的请求级 QoS 保障基于 Token 桶的优先级调度与延迟 SLO 监控一、没有 QoS 的推理服务一个慢请求拖垮所有请求推理服务上线后很快遇到一个棘手的问题当某个用户发送了一个超长 prompt32K tokenPrefill 阶段的计算耗时长达 2 秒。在这 2 秒内其他所有请求的 Decode 都被阻塞——等待这个慢请求完成 Prefill。后果是其他用户的 TPOTTime Per Output Token从 25ms 飙升至 2000ms。这是典型的无 QoS 共享资源问题。所有请求公平竞争 GPU 资源但不同请求的计算需求差异巨大。没有优先级和流量控制的推理服务就像没有红绿灯的十字路口——早晚会发生拥堵和饿死。请求级 QoS 需要解决三个问题如何分类请求优先级、如何控制每种优先级的资源消耗、如何在 SLO 违约时自动降级。二、Token 桶与优先级队列的双层调度架构graph TD A[请求到达] -- B{优先级分类} B --|P0: 实时| C[P0 Token 桶] B --|P1: 普通| D[P1 Token 桶] B --|P2: 批量| E[P2 Token 桶] C --|Token 充足| F[P0 请求队列] D --|Token 充足| G[P1 请求队列] E --|Token 充足| H[P2 请求队列] C --|Token 不足| I[限流/排队] F -- J[优先级调度器] G -- J H -- J J --|按优先级| K[GPU 执行] K -- L[SLO 监控] L --|延迟超标| M[动态调整 Token 速率] L --|正常| N[继续] style B fill:#1a1a2e,stroke:#e94560,color:#fff style J fill:#16213e,stroke:#0f3460,color:#fff style L fill:#1a1a2e,stroke:#e94560,color:#fffToken 桶算法是最经典的流量控制方案。每个优先级的请求池有一个独立的 Token 桶Token 以恒定速率rate生成桶有最大容量burst每个请求消耗的 Token 数等于其输入 token 数量 × 优先级权重当桶为空时该优先级的请求被限流排队或拒绝优先级调度器在 Token 充足的请求间选择最高优先级的执行。P0 总是优先于 P1但 P1 有最小执行保证不会永远饿死。这套分层设计背后有一个常被忽视的制衡机制Token 桶的速率和 Burst 容量本质上定义了一个预算信封。P0 桶的速率设为 GPU 总吞吐的 60%P1 占 35%P2 占 5%。这种分配并非随意——它源于一个称为降级弹性的概念。当 P0 的 Token 桶满载时Burst 未耗尽P1 的请求可以在 P0 桶消耗完毕的间隙被执行但当 P0 持续满载时P1 的 35% 速率保证了它不会被完全挤压。这 35% 不是一个固定数字而是根据 SLO 监控实时动态调整的——如果 P0 的延迟 SLO 达标率为 99.8%高于目标的 99%系统会自动将 P0 的 Token 生成速率下调 5%释放出的剩余预算会自动回流到 P1 和 P2 桶中。这种闭环调节机制避免了人工调参的滞后性问题运维人员不需要在流量高峰时手动调整 QPS 限制系统自己感知延迟翘动并自适应收紧或放宽。值得注意的是Token 桶的 Burst 容量在这里扮演的是短期缓冲器的角色——一个突然的 P0 流量尖峰可以被 Burst 吸收而不触发限流但如果尖峰持续超过 Burst/rate 的时间窗口通常设定为 0.5 秒就会触发拒绝。这个时间窗口正是从允许突发到保护系统的切换阈值。三、QoS 引擎的核心实现use std::collections::{VecDeque, BinaryHeap}; use std::cmp::Reverse; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::sync::{RwLock, Notify}; use std::sync::atomic::{AtomicU64, Ordering}; /// 请求优先级 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] pub enum Priority { P0 0, // 实时交互 P1 1, // 普通请求 P2 2, // 批量处理 } /// Token 桶实现 /// /// 为什么用 AtomicU64 而非 Mutex /// Token 桶的 acquire 操作在请求路径上是高频调用 /// AtomicU64 的 CAS 操作延迟 ~5nsMutex 延迟 ~50ns pub struct TokenBucket { /// 当前可用 Token 数浮点数的高精度表示tokens × 10^6 tokens: AtomicU64, /// Token 生成速率每秒产生的 Token 数 rate: f64, /// 最大 Token 数burst capacity max_tokens: f64, /// 上次 Token 更新时间 last_refill: RwLockInstant, /// token 精度因子将浮点数转为整数的乘数 precision: u64, } impl TokenBucket { const PRECISION: u64 1_000_000; // 微秒级精度 pub fn new(rate: f64, burst: f64) - Self { Self { tokens: AtomicU64::new((burst * Self::PRECISION as f64) as u64), rate, max_tokens: burst, last_refill: RwLock::new(Instant::now()), precision: Self::PRECISION, } } /// 尝试获取 token /// /// 为什么返回 bool 而非 Result /// Token 不足不是错误——是正常的流量控制行为 /// 调用方根据返回值决定排队或拒绝 pub fn try_acquire(self, requested: f64) - bool { // 先补充 token基于时间流逝 self.refill(); let requested_int (requested * self.precision as f64) as u64; // CAS 循环尝试原子扣减 token // 为什么用 CAS 而非 fetch_sub 检查 // fetch_sub 后再检查需要撤销操作加回来 // CAS 一次操作完成无中间状态 loop { let current self.tokens.load(Ordering::Acquire); if current requested_int { return false; // 不足 } let new_val current - requested_int; if self.tokens.compare_exchange( current, new_val, Ordering::AcqRel, Ordering::Acquire, ).is_ok() { return true; } // CAS 失败其他线程同时修改了值重试 } } /// 根据时间流逝补充 token fn refill(self) { let now Instant::now(); let mut last self.last_refill.blocking_write(); let elapsed now.duration_since(*last).as_secs_f64(); if elapsed 0.001 { return; // 小于 1ms跳过补充避免高频更新 } let new_tokens elapsed * self.rate * self.precision as f64; loop { let current self.tokens.load(Ordering::Acquire); let max_int (self.max_tokens * self.precision as f64) as u64; let new_val (current as f64 new_tokens) as u64; let new_val new_val.min(max_int); if self.tokens.compare_exchange( current, new_val, Ordering::AcqRel, Ordering::Acquire, ).is_ok() { break; } } *last now; } } /// SLO 监控器 /// /// 为什么需要独立于调度器的 SLO 监控 /// 调度器负责分配资源监控器负责验证资源分配是否达到预期效果 /// 两者解耦使得监控逻辑可以独立调整 pub struct SLOMonitor { /// P0 延迟 SLO毫秒 p0_latency_slo: Duration, /// P1 延迟 SLO p1_latency_slo: Duration, /// 滑动窗口的延迟样本 latency_samples: RwLockVecDequeLatencySample, /// 窗口大小保留最近 N 个样本 window_size: usize, } #[derive(Debug, Clone)] struct LatencySample { priority: Priority, latency: Duration, timestamp: Instant, input_tokens: usize, } impl SLOMonitor { pub fn new(p0_slo_ms: u64, p1_slo_ms: u64) - Self { Self { p0_latency_slo: Duration::from_millis(p0_slo_ms), p1_latency_slo: Duration::from_millis(p1_slo_ms), latency_samples: RwLock::new(VecDeque::new()), window_size: 1000, } } /// 记录一次请求的延迟 pub async fn record_latency( self, priority: Priority, latency: Duration, input_tokens: usize, ) { let mut samples self.latency_samples.write().await; samples.push_back(LatencySample { priority, latency, timestamp: Instant::now(), input_tokens, }); // 保持窗口大小 while samples.len() self.window_size { samples.pop_front(); } } /// 计算 SLO 达标率 /// /// 为什么用滑动窗口而非全量统计 /// 全量统计无法反映最近的性能趋势 /// 滑动窗口最近 1000 个请求是实时服务质量的有效代理 pub async fn slo_compliance(self, priority: Priority) - f64 { let samples self.latency_samples.read().await; let slo match priority { Priority::P0 self.p0_latency_slo, Priority::P1 self.p1_latency_slo, Priority::P2 return 1.0, // P2 无 SLO 要求 }; let (total, violations) samples.iter() .filter(|s| s.priority priority) .fold((0usize, 0usize), |(t, v), s| { (t 1, v (s.latency slo) as usize) }); if total 0 { 1.0 } else { (total - violations) as f64 / total as f64 } } /// 根据 SLO 达标率调整 Token 桶速率 /// /// 为什么用 PID 控制思想而非固定阈值 /// 固定阈值在负载波动下会震荡过高→限流过猛→过低→放开太猛 /// 增量调整更平滑 pub async fn adjust_rate(self, bucket: TokenBucket, compliance: f64) - f64 { let target 0.99; // 目标 99% 的请求达标 if compliance target { // 超出目标可以放宽限制增加 token 速率 bucket.rate * 1.05 } else if compliance 0.95 { // 低于 95%收紧限制减少 token 速率 bucket.rate * 0.90 } else { bucket.rate // 在容忍区间内保持不变 } } } /// 优先级感知的请求调度器 pub struct PriorityScheduler { /// 按优先级分组的请求队列 queues: [RwLockVecDequeScheduledRequest; 3], /// 每优先级的 Token 桶 buckets: [TokenBucket; 3], /// SLO 监控器 monitor: SLOMonitor, /// 新请求通知 notifier: Notify, } impl PriorityScheduler { /// 提交请求可能被限流 pub async fn submit(self, req: ScheduledRequest) - Result(), QueueFull { let p req.priority as usize; // Token 桶检查 if !self.buckets[p].try_acquire(req.input_tokens as f64) { return Err(QueueFull); } self.queues[p].write().await.push_back(req); self.notifier.notify_one(); Ok(()) } /// 获取下一个要执行的请求优先级调度 /// /// 为什么 P2 请求也有机会执行 /// 纯粹的优先级调度会让 P2 饿死 /// 使用starvation counterP2 每被跳过 N 次临时提升优先级 pub async fn dequeue(self) - OptionScheduledRequest { let p2_skip_count: usize 0; // 简化实现 loop { // 从高到低检查每个优先级 for p in [Priority::P0, Priority::P1, Priority::P2] { let mut queue self.queues[p as usize].write().await; if let Some(req) queue.pop_front() { return Some(req); } } // 所有队列为空等待通知 self.notifier.notified().await; } } } #[derive(Debug, Clone)] struct ScheduledRequest { id: u64, priority: Priority, input_tokens: usize, arrived_at: Instant, } #[derive(Debug)] struct QueueFull;四、QoS 系统的自稳定性分析Token 桶速率的死循环风险SLO 监控 → 调整速率 → 改变延迟分布 → 重新触发 SLO 调整——如果调整步长过大可能形成震荡。使用较小的调整系数5%~10%和冷却期两次调整间隔 5 秒可以避免。P2 饿死的概率分析在最坏情况下P0 和 P1 持续满载P2 请求的等待时间 当前 P0P1 队列长度 × 平均处理时间。通过设置 P2 请求的最大等待时间超时后自动降级为拒绝并返回错误避免无限等待。不适用优先级调度的场景所有请求的延迟需求相同如纯离线批量推理请求数量极少 10 QPS调度器的复杂性不值得计算资源极其充足无需流量控制五、总结Token 桶通过恒定速率 Burst 容量提供精确的流量控制AtomicU64 的 CAS 保证了高频调用路径下的性能优先级调度 饥饿保护starvation counter在保证高优先级延迟 SLO 的同时防止低优先级完全饿死SLO 监控器的滑动窗口 增量速率调整类比 PID 控制提供自适应的 QoS 调优调整步长不宜过大5%~10%且需要冷却期防止速率调整引起的系统震荡QoS 系统在纯离线批量推理和极低 QPS 场景下为过度设计应评估实际是否需要
推理服务的请求级 QoS 保障：基于 Token 桶的优先级调度与延迟 SLO 监控
推理服务的请求级 QoS 保障基于 Token 桶的优先级调度与延迟 SLO 监控一、没有 QoS 的推理服务一个慢请求拖垮所有请求推理服务上线后很快遇到一个棘手的问题当某个用户发送了一个超长 prompt32K tokenPrefill 阶段的计算耗时长达 2 秒。在这 2 秒内其他所有请求的 Decode 都被阻塞——等待这个慢请求完成 Prefill。后果是其他用户的 TPOTTime Per Output Token从 25ms 飙升至 2000ms。这是典型的无 QoS 共享资源问题。所有请求公平竞争 GPU 资源但不同请求的计算需求差异巨大。没有优先级和流量控制的推理服务就像没有红绿灯的十字路口——早晚会发生拥堵和饿死。请求级 QoS 需要解决三个问题如何分类请求优先级、如何控制每种优先级的资源消耗、如何在 SLO 违约时自动降级。二、Token 桶与优先级队列的双层调度架构graph TD A[请求到达] -- B{优先级分类} B --|P0: 实时| C[P0 Token 桶] B --|P1: 普通| D[P1 Token 桶] B --|P2: 批量| E[P2 Token 桶] C --|Token 充足| F[P0 请求队列] D --|Token 充足| G[P1 请求队列] E --|Token 充足| H[P2 请求队列] C --|Token 不足| I[限流/排队] F -- J[优先级调度器] G -- J H -- J J --|按优先级| K[GPU 执行] K -- L[SLO 监控] L --|延迟超标| M[动态调整 Token 速率] L --|正常| N[继续] style B fill:#1a1a2e,stroke:#e94560,color:#fff style J fill:#16213e,stroke:#0f3460,color:#fff style L fill:#1a1a2e,stroke:#e94560,color:#fffToken 桶算法是最经典的流量控制方案。每个优先级的请求池有一个独立的 Token 桶Token 以恒定速率rate生成桶有最大容量burst每个请求消耗的 Token 数等于其输入 token 数量 × 优先级权重当桶为空时该优先级的请求被限流排队或拒绝优先级调度器在 Token 充足的请求间选择最高优先级的执行。P0 总是优先于 P1但 P1 有最小执行保证不会永远饿死。这套分层设计背后有一个常被忽视的制衡机制Token 桶的速率和 Burst 容量本质上定义了一个预算信封。P0 桶的速率设为 GPU 总吞吐的 60%P1 占 35%P2 占 5%。这种分配并非随意——它源于一个称为降级弹性的概念。当 P0 的 Token 桶满载时Burst 未耗尽P1 的请求可以在 P0 桶消耗完毕的间隙被执行但当 P0 持续满载时P1 的 35% 速率保证了它不会被完全挤压。这 35% 不是一个固定数字而是根据 SLO 监控实时动态调整的——如果 P0 的延迟 SLO 达标率为 99.8%高于目标的 99%系统会自动将 P0 的 Token 生成速率下调 5%释放出的剩余预算会自动回流到 P1 和 P2 桶中。这种闭环调节机制避免了人工调参的滞后性问题运维人员不需要在流量高峰时手动调整 QPS 限制系统自己感知延迟翘动并自适应收紧或放宽。值得注意的是Token 桶的 Burst 容量在这里扮演的是短期缓冲器的角色——一个突然的 P0 流量尖峰可以被 Burst 吸收而不触发限流但如果尖峰持续超过 Burst/rate 的时间窗口通常设定为 0.5 秒就会触发拒绝。这个时间窗口正是从允许突发到保护系统的切换阈值。三、QoS 引擎的核心实现use std::collections::{VecDeque, BinaryHeap}; use std::cmp::Reverse; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::sync::{RwLock, Notify}; use std::sync::atomic::{AtomicU64, Ordering}; /// 请求优先级 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] pub enum Priority { P0 0, // 实时交互 P1 1, // 普通请求 P2 2, // 批量处理 } /// Token 桶实现 /// /// 为什么用 AtomicU64 而非 Mutex /// Token 桶的 acquire 操作在请求路径上是高频调用 /// AtomicU64 的 CAS 操作延迟 ~5nsMutex 延迟 ~50ns pub struct TokenBucket { /// 当前可用 Token 数浮点数的高精度表示tokens × 10^6 tokens: AtomicU64, /// Token 生成速率每秒产生的 Token 数 rate: f64, /// 最大 Token 数burst capacity max_tokens: f64, /// 上次 Token 更新时间 last_refill: RwLockInstant, /// token 精度因子将浮点数转为整数的乘数 precision: u64, } impl TokenBucket { const PRECISION: u64 1_000_000; // 微秒级精度 pub fn new(rate: f64, burst: f64) - Self { Self { tokens: AtomicU64::new((burst * Self::PRECISION as f64) as u64), rate, max_tokens: burst, last_refill: RwLock::new(Instant::now()), precision: Self::PRECISION, } } /// 尝试获取 token /// /// 为什么返回 bool 而非 Result /// Token 不足不是错误——是正常的流量控制行为 /// 调用方根据返回值决定排队或拒绝 pub fn try_acquire(self, requested: f64) - bool { // 先补充 token基于时间流逝 self.refill(); let requested_int (requested * self.precision as f64) as u64; // CAS 循环尝试原子扣减 token // 为什么用 CAS 而非 fetch_sub 检查 // fetch_sub 后再检查需要撤销操作加回来 // CAS 一次操作完成无中间状态 loop { let current self.tokens.load(Ordering::Acquire); if current requested_int { return false; // 不足 } let new_val current - requested_int; if self.tokens.compare_exchange( current, new_val, Ordering::AcqRel, Ordering::Acquire, ).is_ok() { return true; } // CAS 失败其他线程同时修改了值重试 } } /// 根据时间流逝补充 token fn refill(self) { let now Instant::now(); let mut last self.last_refill.blocking_write(); let elapsed now.duration_since(*last).as_secs_f64(); if elapsed 0.001 { return; // 小于 1ms跳过补充避免高频更新 } let new_tokens elapsed * self.rate * self.precision as f64; loop { let current self.tokens.load(Ordering::Acquire); let max_int (self.max_tokens * self.precision as f64) as u64; let new_val (current as f64 new_tokens) as u64; let new_val new_val.min(max_int); if self.tokens.compare_exchange( current, new_val, Ordering::AcqRel, Ordering::Acquire, ).is_ok() { break; } } *last now; } } /// SLO 监控器 /// /// 为什么需要独立于调度器的 SLO 监控 /// 调度器负责分配资源监控器负责验证资源分配是否达到预期效果 /// 两者解耦使得监控逻辑可以独立调整 pub struct SLOMonitor { /// P0 延迟 SLO毫秒 p0_latency_slo: Duration, /// P1 延迟 SLO p1_latency_slo: Duration, /// 滑动窗口的延迟样本 latency_samples: RwLockVecDequeLatencySample, /// 窗口大小保留最近 N 个样本 window_size: usize, } #[derive(Debug, Clone)] struct LatencySample { priority: Priority, latency: Duration, timestamp: Instant, input_tokens: usize, } impl SLOMonitor { pub fn new(p0_slo_ms: u64, p1_slo_ms: u64) - Self { Self { p0_latency_slo: Duration::from_millis(p0_slo_ms), p1_latency_slo: Duration::from_millis(p1_slo_ms), latency_samples: RwLock::new(VecDeque::new()), window_size: 1000, } } /// 记录一次请求的延迟 pub async fn record_latency( self, priority: Priority, latency: Duration, input_tokens: usize, ) { let mut samples self.latency_samples.write().await; samples.push_back(LatencySample { priority, latency, timestamp: Instant::now(), input_tokens, }); // 保持窗口大小 while samples.len() self.window_size { samples.pop_front(); } } /// 计算 SLO 达标率 /// /// 为什么用滑动窗口而非全量统计 /// 全量统计无法反映最近的性能趋势 /// 滑动窗口最近 1000 个请求是实时服务质量的有效代理 pub async fn slo_compliance(self, priority: Priority) - f64 { let samples self.latency_samples.read().await; let slo match priority { Priority::P0 self.p0_latency_slo, Priority::P1 self.p1_latency_slo, Priority::P2 return 1.0, // P2 无 SLO 要求 }; let (total, violations) samples.iter() .filter(|s| s.priority priority) .fold((0usize, 0usize), |(t, v), s| { (t 1, v (s.latency slo) as usize) }); if total 0 { 1.0 } else { (total - violations) as f64 / total as f64 } } /// 根据 SLO 达标率调整 Token 桶速率 /// /// 为什么用 PID 控制思想而非固定阈值 /// 固定阈值在负载波动下会震荡过高→限流过猛→过低→放开太猛 /// 增量调整更平滑 pub async fn adjust_rate(self, bucket: TokenBucket, compliance: f64) - f64 { let target 0.99; // 目标 99% 的请求达标 if compliance target { // 超出目标可以放宽限制增加 token 速率 bucket.rate * 1.05 } else if compliance 0.95 { // 低于 95%收紧限制减少 token 速率 bucket.rate * 0.90 } else { bucket.rate // 在容忍区间内保持不变 } } } /// 优先级感知的请求调度器 pub struct PriorityScheduler { /// 按优先级分组的请求队列 queues: [RwLockVecDequeScheduledRequest; 3], /// 每优先级的 Token 桶 buckets: [TokenBucket; 3], /// SLO 监控器 monitor: SLOMonitor, /// 新请求通知 notifier: Notify, } impl PriorityScheduler { /// 提交请求可能被限流 pub async fn submit(self, req: ScheduledRequest) - Result(), QueueFull { let p req.priority as usize; // Token 桶检查 if !self.buckets[p].try_acquire(req.input_tokens as f64) { return Err(QueueFull); } self.queues[p].write().await.push_back(req); self.notifier.notify_one(); Ok(()) } /// 获取下一个要执行的请求优先级调度 /// /// 为什么 P2 请求也有机会执行 /// 纯粹的优先级调度会让 P2 饿死 /// 使用starvation counterP2 每被跳过 N 次临时提升优先级 pub async fn dequeue(self) - OptionScheduledRequest { let p2_skip_count: usize 0; // 简化实现 loop { // 从高到低检查每个优先级 for p in [Priority::P0, Priority::P1, Priority::P2] { let mut queue self.queues[p as usize].write().await; if let Some(req) queue.pop_front() { return Some(req); } } // 所有队列为空等待通知 self.notifier.notified().await; } } } #[derive(Debug, Clone)] struct ScheduledRequest { id: u64, priority: Priority, input_tokens: usize, arrived_at: Instant, } #[derive(Debug)] struct QueueFull;四、QoS 系统的自稳定性分析Token 桶速率的死循环风险SLO 监控 → 调整速率 → 改变延迟分布 → 重新触发 SLO 调整——如果调整步长过大可能形成震荡。使用较小的调整系数5%~10%和冷却期两次调整间隔 5 秒可以避免。P2 饿死的概率分析在最坏情况下P0 和 P1 持续满载P2 请求的等待时间 当前 P0P1 队列长度 × 平均处理时间。通过设置 P2 请求的最大等待时间超时后自动降级为拒绝并返回错误避免无限等待。不适用优先级调度的场景所有请求的延迟需求相同如纯离线批量推理请求数量极少 10 QPS调度器的复杂性不值得计算资源极其充足无需流量控制五、总结Token 桶通过恒定速率 Burst 容量提供精确的流量控制AtomicU64 的 CAS 保证了高频调用路径下的性能优先级调度 饥饿保护starvation counter在保证高优先级延迟 SLO 的同时防止低优先级完全饿死SLO 监控器的滑动窗口 增量速率调整类比 PID 控制提供自适应的 QoS 调优调整步长不宜过大5%~10%且需要冷却期防止速率调整引起的系统震荡QoS 系统在纯离线批量推理和极低 QPS 场景下为过度设计应评估实际是否需要
