Rust in Everything
基于 pingora-limits 0.9.0

限流

主线位置：请求主线的 request_filter。pingora-limits 不是中间件，也不会自己拒绝请求：它只是三个线程安全的计数器。用户在钩子里调 observe 拿到计数，自己决定是否写一个 429 并返回 Ok(true) 短路（见 ProxyHttp）。

flowchart TB
  obs["observe"] --> mr["maybe_reset"]
  rt["rate / rate_with"] --> mr
  mr --> due{"过了 interval"}
  due -- 是 --> clr["清空另一格再翻转"]
  due -- 否 --> keep["不动"]
  obs --> cur["当前格 Estimator"]
  rt --> prev["上一格 Estimator"]
  inf["Inflight::incr"] --> est["Estimator"]
  inf --> guard["Guard 析构时减回"]

pingora-limits 的 src/ 一共 633 行：estimator.rs（138 行）、rate.rs（347 行，一半是测试）、inflight.rs（116 行）和 32 行的 lib.rs。唯一的依赖是 ahash。门面 crate pingora 不依赖它，要在 Cargo.toml 里单独写（见 源码地图）。

实验平台：M2 Max、macOS 27、rustc 1.99.0，pingora 0.9.0，tokio 钉在 1.53.2（原因见怎么读这一卷）。程序在 lb-lab/src/bin/limits.rs 和 ratelimit.rs，没有特别说明的都是 release 构建。

Estimator：count-min sketch

Estimator 是一个 hashes × slots 的 AtomicIsize 矩阵，每一行配一个随机种子的 ahash::RandomState。incr：

    pub fn incr<T: Hash>(&self, key: T, value: isize) -> isize {
        self.estimator
            .iter()
            .fold(isize::MAX, |min, (slot, hasher)| {
                let hash = hash(&key, hasher) as usize;
                let counter = &slot[hash % slot.len()];
                // Overflow is allowed for simplicity
                let current = counter.fetch_add(value, Ordering::Relaxed);
                std::cmp::min(min, current + value)
            })
    }

每一行用自己的哈希挑一格，原子地加上 value，返回所有行里最小的那个值。get 同样取最小值。这就是 count-min sketch：不存键，只存计数；别的键哈希到同一格会把计数抬高，所以估计值只会偏大，不会偏小，取多行的最小值是为了让「每一行都撞上」的概率变小。全程没有锁，内存固定为 hashes × slots × 8 字节，和键的个数无关。

偏大多少，取决于格数和这段时间里总共有多少事件。用 1 万个键、计数按 Zipf 分布（第 i 个键 100000 / (i+1) 次，最少 10 次）各配置跑 5 次取平均，误差 = 估计值 − 真实值：

10000 keys, 973855 events, top key 100000 events, 0 keys seen once
hashes  slots    memory   exact%     mean      p99      max     e*N/w
     1   8192      64KiB    29.6%    118.1     1288    75042       323
     2   4096      64KiB    16.6%     50.0      314     2897       646
     4    256       8KiB     0.0%   1374.5     2860     8352     10341
     4   1024      32KiB     0.0%    210.7      573     1654      2585
     8   1024      64KiB     0.0%    153.7      346      608      2585
     4   8192     256KiB    75.3%      4.6       44      198       323
     4  65536    2048KiB   100.0%      0.0        0       13        40
  • 4 × 1024 是 Rate 默认的配置。1 万个键时没有一个键的计数是准的，平均多算 211，比一个键平均的真实次数（97）还大。Rate 的源码注释说 1024「can be lower if interval is short (key cardinality is low)」：格数要按一个窗口里出现的不同键的个数来定，而不是按请求速率。
  • 4 × 8192 是 Inflight 的配置，256 KiB，75% 的键准确。inflight.rs 里的注释写的是「False positive ratio is 1/(n^h)」，按 n = 8192、h = 4 算是 2×10⁻¹⁶；实测有四分之一的键不准。这个式子里没有键的个数，键一多就不成立。
  • 同样 64 KiB，只用 1 行的平均误差最大，最大误差（5 次的平均）75042：某个冷门的键和一个热键落在同一格，只有一行，就没有别的行能把它纠正回来。8 行的最大误差最小，但平均误差比 2 行大：行数多了，每行的格数就少了。
  • 最后一列是 count-min sketch 的理论上界 e·N/w（N 是事件总数，w 是格数），对每个键成立的概率是 1 − e^(−行数)。

incr 里那句「Overflow is allowed for simplicity」只在 release 下成立。一个格子加到 isize::MAX 再加 1：

== release overflow
9223372036854775807
-9223372036854775808
== debug overflow
9223372036854775807

thread 'main' (5737193) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/pingora-limits-0.9.0/src/estimator.rs:53:36:
attempt to add with overflow

fetch_add 本身是回绕的，panic 的是后面计算返回值的 current + value。0.9.0 之后 main 上的 c1c4a0d（「fix(limits): use wrapping addition for estimator increments」，同步自 PR #1041）改成了 wrapping_add，还没有发版。

Rate：两格轮换的固定窗口

Rate：

pub struct Rate {
    // 2 slots so that we use one to collect the current events and the other to report rate
    red_slot: Estimator,
    blue_slot: Estimator,
    red_or_blue: AtomicBool, // true: the current slot is red, otherwise blue
    start: Instant,
    // Use u64 below instead of Instant because we want atomic operation
    reset_interval_ms: u64, // the time interval to reset `current` and move it to `previous`
    last_reset_time: AtomicU64, // the timestamp in ms since `start`
    interval: Duration,
}

两个 Estimator，一个收当前窗口的事件，另一个保存上一个窗口的结果。Rate::new 用 4 × 1024，一个 Rate 就是两个 32 KiB 的矩阵；new_with_estimator_config 可以换。

  • observe 往当前格加，返回当前窗口到目前为止的计数。上游文档的限流例子拿的就是这个返回值。
  • rate 返回上一个完整窗口的计数除以窗口长度，当前窗口里的事件不算。
  • rate_with 把上一格、当前格的计数和「当前窗口走了几成」交给用户的函数。crate 带一个 PROPORTIONAL_RATE_ESTIMATE_CALC_FN：prev × (1 − 进度) + curr，用上一格按比例「补齐」一个完整的窗口，近似一个滑动窗口。

翻转发生在 maybe_reset 里，observe 和 rate 每次都先调它：

        let now = Instant::now().duration_since(self.start).as_millis() as u64;
        let last_reset = self.last_reset_time.load(Ordering::SeqCst);
        let past_ms = now - last_reset;

        if past_ms < self.reset_interval_ms {
            // no need to reset
            return past_ms;
        }
        let red_or_blue = self.red_or_blue();
        match self.last_reset_time.compare_exchange(
            last_reset,
            now,
            Ordering::SeqCst,
            Ordering::Acquire,
        ) {
            Ok(_) => {
                // first clear the previous slot
                self.previous(red_or_blue).reset();
                // then flip the flag to tell others to use the reset slot
                self.red_or_blue.store(!red_or_blue, Ordering::SeqCst);
                // if current time is beyond 2 intervals, the data stored in the previous slot
                // is also stale, we should clear that too
                if now - last_reset >= self.reset_interval_ms * 2 {
                    // Note that this is the previous one now because we just flipped self.red_or_blue
                    self.current(red_or_blue).reset();
                }
            }
            Err(new) => {
                // another thread beats us to it
                assert!(new >= now - 1000); // double check that the new timestamp looks right
            }
        }

        past_ms
  • 没有定时器。窗口在「过了 interval 之后的第一次调用」时才翻，新窗口从这次调用的时刻开始（last_reset_time 存的是 now，不是对齐到 interval 的整数倍）。所以窗口的边界由请求决定，一个窗口实际可能比 interval 长。
  • 两个线程同时发现该翻了，只有 compare_exchange 成功的那个去清空和翻转。
  • 距上次翻转超过两个 interval，两格都清空，rate 直接返回 0。

interval 为 1 秒，前 450 ms 里 10 个事件，之后在不同时刻读：

t=  456ms 10 events      rate()=  0.0 rate_with(proportional)= 10.0
t=  902ms                rate()=  0.0 rate_with(proportional)= 10.0
t= 1108ms                rate()= 10.0 rate_with(proportional)= 10.0
t= 1510ms                rate()= 10.0 rate_with(proportional)=  6.0
t= 2104ms                rate()= 10.0 rate_with(proportional)=  0.0
t= 3208ms                rate()=  0.0 rate_with(proportional)=  0.0
  • 第一个窗口里 rate() 是 0：还没有完整的上一个窗口。
  • 1108 ms 的读取触发翻转，新窗口从 1108 ms 开始，rate() 变成 10。
  • 1510 ms，新窗口走了约 40%，按比例的估计是 10 × 0.6 = 6。
  • 2104 ms 离 1108 ms 还不到 1 秒，没有翻转，rate() 仍报上一个窗口的 10；按比例的估计已经衰减到 10 × 0.004 ≈ 0。两种读法在同一时刻可以差这么多。
  • 3208 ms 离上次翻转超过 2 秒，两格都清空。

tower 的 RateLimit 也是固定窗口，但它是一个包住服务的中间件，一个实例只数一个总数，超额时在 poll_ready 里等（见 tower 卷的 RateLimit）；pingora 的 Rate 按键计数，不等待，也不拒绝。

📝NOTE
冷知识：Rate 的文档和测试都改过一次口径

0.7.0 之前，Rate 的文档注释是「A stable rate estimator that reports the rate of events in the past interval time. It returns the average rate between interval * 2 and interval while collecting the events happening between interval and now.」，还有一句「This estimator ignores events that happen less than once per interval time.」；测试里那个引用 Cloudflare 博客的按比例估计函数写的是 prev * (1. - interval_fraction) + curr * interval_fraction，当前格也乘了进度。

56d8f7b（「pingora-limits Rate: correct some comments, test/example」，首次发布于 0.7.0）把 Rate 的注释改成现在的「reports the rate of events per period of interval time ... returns the average rate of the latest completed period」，给 rate() 补上「This is the average rate of the latest completed period」；按比例的函数改成 prev * (1. - interval_fraction) + curr，从测试里挪出来，作为公开的 PROPORTIONAL_RATE_ESTIMATE_CALC_FN 导出。测试里第 0.5 秒的期望值从 5. * 0.5 改成了 5.。rate_with 本身是 0.4.0 的 0021d41（「Add function for custom rate measurements」）加的。

第一秒里的 panic

上面 Err(new) 分支里的 assert!(new >= now - 1000) 用的是 u64。now 是从 Rate 创建起的毫秒数，在第一秒内小于 1000，now - 1000 下溢。这个分支只在两个线程抢同一次翻转、自己没抢到时才走到；interval 不小于 1 秒时，第一次翻转最早也在 1000 ms，碰不到。interval 小于 1 秒就不一样了：Rate::new(10 ms)，8 个线程同时 observe 900 ms：

== release run 1
   7 
   7 assertion failed: new >= now - 1000
   1 note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
   1 thread '<unnamed>' (5738500) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/pingora-limits-0.9.0/src/rate.rs:174:17:
...
   1 thread done, N observes
   7 thread panicked
== debug run 1
   7 attempt to subtract with overflow
   7 thread '<unnamed>' (tid) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/pingora-limits-0.9.0/src/rate.rs:174:32:
   1 thread done, N observes
   7 thread panicked

（输出经过 sort | uniq -c，每个线程 observe 的次数换成了 N；debug 那组先用 grep 只留下 panic 相关的行，线程号换成了 tid。）

release 三次、debug 两次，每次都是 8 个线程里 7 个 panic。release 下 now - 1000 回绕成一个接近 u64::MAX 的数，断言失败；debug 下减法本身先 panic。放到代理里，对应的是进程刚启动、并发请求一起进来的那一秒：request_filter 里的 observe 会让请求任务 panic（这一步按源码推断，没有在代理上复现）。这个断言从 0bca116（pingora-limits 的第一个提交）起就在，0.9.0 之后的 main 上也没改。

文档里的例子

docs/user_guide/rate_limiter.md 的例子：一个全局的 Rate::new(1 秒)，按请求头 appid 计数，observe 返回值大于 1 就回 429，带三个 X-Rate-Limit-* 头并关掉 keep-alive。原样拷出来只改四处：上游从 1.1.1.1:443 换成本机的 127.0.0.1:18701（明文，SNI 为空，Host 改成 127.0.0.1），监听 127.0.0.1:18710。不改别的就能在 0.9.0 上编译通过。

文档里 request_filter 的核心三行：

        // retrieve the current window requests
        let curr_window_requests = RATE_LIMITER.observe(&appid, 1);
        if curr_window_requests > MAX_REQ_PER_SEC {

文档对测试结果的描述是：「The first request should work and any later requests that arrive within 1s of a previous request should fail」。

❗IMPORTANT
动脑筋：按下面的时间表发请求（时间从第一个请求算起），第 5 个请求，appid=2 在 1050 ms 的那个，得到什么状态码？
时刻01002009001050110011501200
appid1112211无

提示：所有 appid 共用一个 Rate，窗口什么时候翻、从哪一刻开始算？答案在文末。

Inflight：析构时减回去

Inflight 数的是「正在进行的」事件，比如一个客户端同时开着几个请求。incr 先用自己的 RandomState 把键哈希成一个 u64，Guard 里只存这个 u64 和 value（注释说「store the hash instead of the actual key to save space」），然后加一次、返回 (Guard, 估计值)；Guard 被 drop 时 decr 同样的 value。所以把 Guard 放进 CTX，请求结束时 CTX 一释放，计数就减回去，请求中途出错返回也一样。

Guard 还有一个公开的 incr，文档是「Increment the key's value that the Guard was created from」。它每调一次就再加一次 value，析构时却只减一次：

after Inflight::incr: 1
after Guard::incr: 2
after drop, a new incr sees: 2

Guard 已经释放，键 a 的计数还停在 1 上，加上新的一次是 2。这个计数会一直漏下去，除非手动 decr。debug 和 release 结果相同。

动脑筋答案

t=     0ms appid=1 -> 200 Connection: keep-alive
t=   101ms appid=1 -> 429 X-Rate-Limit-Reset: 1 Connection: close
t=   201ms appid=1 -> 429 X-Rate-Limit-Reset: 1 Connection: close
t=   900ms appid=2 -> 200 Connection: keep-alive
t=  1050ms appid=2 -> 200 Connection: keep-alive
t=  1101ms appid=1 -> 200 Connection: keep-alive
t=  1150ms appid=1 -> 429 X-Rate-Limit-Reset: 1 Connection: close
t=  1201ms appid=None -> 200 Connection: keep-alive

200。appid=2 在 900 ms 和 1050 ms 两次都通过了，中间只隔 150 ms：

  • RATE_LIMITER 是 Lazy，第一个请求进来时才创建，start 就是 0 ms，第一个窗口是 [0, 1000)。
  • 900 ms 是 appid=2 在这个窗口里的第一次，计数 1，通过。
  • 1050 ms 的请求是 1000 ms 之后的第一次调用，它触发了翻转，新窗口从 1050 ms 开始，appid=2 在新窗口里又是第一次。
  • 1101 ms 的 appid=1 同理是新窗口里的第一次，通过；1150 ms 的第二次被拒。

所以文档那句「within 1s of a previous request should fail」只对单个 appid、且没有别的键推动翻转时成立：窗口是所有键共用的，翻转时刻由任何一个键的请求决定。另外三点：X-Rate-Limit-Reset 写死是 1，和真实的窗口剩余时间无关；429 响应带 Connection: close，因为例子里调了 session.set_keepalive(None)；没有 appid 头的请求不计数，永远放行。

小结

  • Estimator 是无锁的 count-min sketch，只会多算；误差由格数和窗口内的不同键数决定，实测 1 万个键时 Rate 默认的 4 × 1024 没有一个键是准的，Inflight 的 4 × 8192 有 75% 准确；溢出在 debug 下 panic，修复在 main 上未发版。
  • Rate 是红蓝两个估计器轮换的固定窗口：observe 返回当前窗口的计数，rate 只报上一个完整窗口，rate_with 加 PROPORTIONAL_RATE_ESTIMATE_CALC_FN 才近似滑动窗口；窗口由调用触发翻转，从触发的那一刻开始。
  • interval 小于 1 秒的 Rate 在创建后的第一秒内被并发调用，maybe_reset 里的 now - 1000 下溢，release 断言失败、debug 减法溢出，实测 8 个线程里 7 个 panic。
  • 上游文档的限流例子在 0.9.0 上原样可编；所有键共用一个窗口，实测同一个 appid 两次通过只隔 150 ms。
  • Inflight 的 Guard 析构时减回计数，但手动调 Guard::incr 之后只减一次，计数会漏。

💬 关联讨论

发起讨论
加载中...