alopex_core/storage/format/
backpressure.rs

1//! バックプレッシャ制御用のコンパクション負債トラッカー。
2
3use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
4use std::time::Duration;
5
6/// 書き込みスロットリング設定。
7#[derive(Debug, Clone)]
8pub struct WriteThrottleConfig {
9    /// ペンディングセクション数上限。
10    pub max_pending_sections: usize,
11    /// ペンディングバイト数上限。
12    pub max_pending_bytes: u64,
13    /// スロットリング開始閾値（0.0-1.0）。
14    pub throttle_threshold: f64,
15}
16
17impl Default for WriteThrottleConfig {
18    fn default() -> Self {
19        Self {
20            max_pending_sections: 10,
21            max_pending_bytes: 64 * 1024 * 1024, // 64MB
22            throttle_threshold: 0.7,
23        }
24    }
25}
26
27/// コンパクション負債を追跡する。
28#[derive(Debug)]
29pub struct CompactionDebtTracker {
30    pending_sections: AtomicUsize,
31    pending_bytes: AtomicU64,
32    config: WriteThrottleConfig,
33}
34
35impl CompactionDebtTracker {
36    /// 新規トラッカーを作成する。
37    pub fn new(config: WriteThrottleConfig) -> Self {
38        Self {
39            pending_sections: AtomicUsize::new(0),
40            pending_bytes: AtomicU64::new(0),
41            config,
42        }
43    }
44
45    /// ペンディングセクションを追加する。
46    pub fn add_pending(&self, bytes: u64) {
47        self.pending_sections.fetch_add(1, Ordering::Relaxed);
48        self.pending_bytes.fetch_add(bytes, Ordering::Relaxed);
49    }
50
51    /// ペンディングセクションを完了する。
52    pub fn complete_pending(&self, bytes: u64) {
53        self.pending_sections.fetch_sub(1, Ordering::Relaxed);
54        self.pending_bytes.fetch_sub(bytes, Ordering::Relaxed);
55    }
56
57    /// スロットリングが必要か判定する。
58    pub fn should_throttle(&self) -> bool {
59        self.debt_ratio() >= self.config.throttle_threshold
60    }
61
62    /// 現在の負債率を返す（0.0-1.0超の可能性あり）。
63    pub fn debt_ratio(&self) -> f64 {
64        let sections = self.pending_sections.load(Ordering::Relaxed) as f64;
65        let bytes = self.pending_bytes.load(Ordering::Relaxed) as f64;
66        let section_ratio = sections / self.config.max_pending_sections as f64;
67        let bytes_ratio = bytes / self.config.max_pending_bytes as f64;
68        section_ratio.max(bytes_ratio)
69    }
70
71    /// スロットリング待機時間を返す（指数バックオフ）。
72    pub fn throttle_duration(&self) -> Duration {
73        let ratio = self.debt_ratio();
74        if ratio <= self.config.throttle_threshold {
75            return Duration::from_millis(0);
76        }
77        // 閾値超過分に比例させ、最大2秒にクリップ。
78        let over = ratio - self.config.throttle_threshold;
79        let millis = (over * 1000.0).clamp(10.0, 2000.0);
80        Duration::from_millis(millis as u64)
81    }
82}
alopex_core/storage/format/backpressure.rs

alopex_core/storage/format/
backpressure.rs
