futu_cache/qot_cache/
waiters.rs

1use std::sync::Arc;
2use std::sync::atomic::Ordering;
3
4use dashmap::mapref::entry::Entry;
5use tokio::sync::Notify;
6
7use super::QotCache;
8
9/// v1.4.110 codex Phase 3 Slice 6c: cold-cache wait key for BasicQot.
10///
11/// 输入 cache_key (legacy `"market_code"` 或 broker-aware
12/// `"market_code@b1007"`), 输出 `"<cache_key>:basic"` 形式 wait key.
13/// 与 OrderBook wait key 分桶, 避免互相错唤醒.
14#[inline]
15pub fn basic_qot_wait_key(cache_key: &str) -> String {
16    format!("{cache_key}:basic")
17}
18
19/// v1.4.110 codex Phase 3 Slice 6c: cold-cache wait key for OrderBook.
20#[inline]
21pub fn order_book_wait_key(cache_key: &str) -> String {
22    format!("{cache_key}:orderbook")
23}
24
25#[inline]
26pub fn odd_lot_order_book_cache_key(cache_key: &str) -> String {
27    format!("{cache_key}:orderbook_odd")
28}
29
30#[inline]
31pub fn odd_lot_order_book_wait_key(cache_key: &str) -> String {
32    format!("{cache_key}:orderbook_odd")
33}
34
35pub fn ticker_wait_key(cache_key: &str) -> String {
36    format!("{cache_key}:ticker")
37}
38
39impl QotCache {
40    /// v1.4.110 codex Phase 3 Slice 6c: 注册 cold-cache wait waiter.
41    ///
42    /// 返已存在或新建的 `Arc<Notify>`. handler 调:
43    /// 1. `register_cold_cache_waiter("91_BTCUSDT@b1007:basic")` 获 Notify
44    /// 2. 主动发 Pull_SubData CMD6824
45    /// 3. `tokio::time::timeout(Duration::from_secs(3), notify.notified())` 等
46    /// 4. 再 `get_basic_qot_broker(&key)` 读 cache (可能仍 None — 真 timeout)
47    ///
48    /// `wait_kind` 推荐: `"basic"` / `"orderbook"`. 不混 sub_type 数字防误唤.
49    pub fn register_cold_cache_waiter(&self, wait_key: &str) -> Arc<Notify> {
50        self.register_cold_cache_waiter_flight(wait_key).0
51    }
52
53    /// Register a cold-cache waiter and identify the request that created the
54    /// flight.
55    ///
56    /// The returned bool is true only for the caller that inserted the waiter.
57    /// Handlers use it to singleflight the active Pull_SubData request while
58    /// still letting all concurrent callers await the same Notify.
59    pub fn register_cold_cache_waiter_flight(&self, wait_key: &str) -> (Arc<Notify>, bool) {
60        match self.cold_cache_waiters.entry(wait_key.to_string()) {
61            Entry::Occupied(entry) => (entry.get().clone(), false),
62            Entry::Vacant(entry) => {
63                self.cold_cache_waiter_entries
64                    .fetch_add(1, Ordering::Relaxed);
65                (entry.insert(Arc::new(Notify::new())).clone(), true)
66            }
67        }
68    }
69
70    /// v1.4.110 codex Phase 3 Slice 6c: 唤醒指定 cold-cache wait waiter.
71    ///
72    /// push parser update path 调 (`update_basic_qot` / `update_order_book` /
73    /// `_broker` 变种). 没 waiter → no-op. 有 waiter → `notify_waiters()`
74    /// broadcast 给所有 awaiter, 然后 remove (Arc 仍被 awaiter 持有, 自然释放).
75    pub fn notify_cold_cache_waiters(&self, wait_key: &str) {
76        if !self.has_cold_cache_waiters() {
77            return;
78        }
79        if let Some((_, n)) = self.cold_cache_waiters.remove(wait_key) {
80            self.decrement_cold_cache_waiter_entries();
81            n.notify_waiters();
82        }
83    }
84
85    pub fn has_cold_cache_waiters(&self) -> bool {
86        self.cold_cache_waiter_entries.load(Ordering::Relaxed) > 0
87    }
88
89    pub fn notify_basic_qot_cold_cache_waiters(&self, cache_key: &str) {
90        if self.has_cold_cache_waiters() {
91            self.notify_cold_cache_waiters(&basic_qot_wait_key(cache_key));
92        }
93    }
94
95    pub fn notify_order_book_cold_cache_waiters(&self, cache_key: &str) {
96        if self.has_cold_cache_waiters() {
97            self.notify_cold_cache_waiters(&order_book_wait_key(cache_key));
98        }
99    }
100
101    pub fn notify_odd_lot_order_book_cold_cache_waiters(&self, cache_key: &str) {
102        if self.has_cold_cache_waiters() {
103            self.notify_cold_cache_waiters(&odd_lot_order_book_wait_key(cache_key));
104        }
105    }
106
107    pub fn notify_ticker_cold_cache_waiters(&self, cache_key: &str) {
108        if self.has_cold_cache_waiters() {
109            self.notify_cold_cache_waiters(&ticker_wait_key(cache_key));
110        }
111    }
112
113    fn decrement_cold_cache_waiter_entries(&self) {
114        let _ = self.cold_cache_waiter_entries.fetch_update(
115            Ordering::Relaxed,
116            Ordering::Relaxed,
117            |count| count.checked_sub(1),
118        );
119    }
120
121    /// v1.4.110 codex audit Round3 #22: cold-cache wait timeout 后清 idle waiter.
122    ///
123    /// `wait_for_basic_cache` / `wait_for_order_book_cache` 3s timeout 仍 cache
124    /// miss 时调. 若 push 始终没来, `notify_cold_cache_waiters` 不会触发, entry
125    /// 会一直留在 `cold_cache_waiters` map (虽 bounded by distinct wait_key 数,
126    /// 仍是慢速 leak).
127    ///
128    /// **只删 caller 自己注册的那个 entry, 且无其他并发 awaiter 时才删**:
129    /// `remove_if` closure 在 entry lock 下原子检查两条:
130    /// 1. `Arc::ptr_eq(stored, caller_notify)` — stored 必须就是 caller 当初
131    ///    `register_cold_cache_waiter` 拿到的同一 Arc. 防 race: caller timeout
132    ///    后到本调用之间, 若 push 触发 `notify_cold_cache_waiters` 删了旧 entry,
133    ///    另一个 `wait_for_*` 又 register 建了**新** entry (不同 Arc), `ptr_eq`
134    ///    false → 不误删别人的新 entry.
135    /// 2. `Arc::strong_count(stored) <= 2` — DashMap stored Arc 1 + caller
136    ///    持有的 `caller_notify` 1. `> 2` 表示有其他 `wait_for_*` 仍 await 同
137    ///    entry → 保留让它们能被 notify 唤醒.
138    ///
139    /// caller 约定: 必须把 `register_cold_cache_waiter` 返回的 `Arc<Notify>`
140    /// 原样传进来 (caller 全程持有未 drop).
141    pub fn cleanup_cold_cache_waiter_if_idle(&self, wait_key: &str, caller_notify: &Arc<Notify>) {
142        if !self.has_cold_cache_waiters() {
143            return;
144        }
145        if self
146            .cold_cache_waiters
147            .remove_if(wait_key, |_, stored| {
148                Arc::ptr_eq(stored, caller_notify) && Arc::strong_count(stored) <= 2
149            })
150            .is_some()
151        {
152            self.decrement_cold_cache_waiter_entries();
153        }
154    }
155}
futu_cache/qot_cache/waiters.rs

futu_cache/qot_cache/
waiters.rs
