fxcp_core/
governor.rs

1// SPDX-License-Identifier: GPL-2.0-or-later
2// Copyright (C) 2025 Joel Wirāmu Pauling <aenertia@aenertia.net>
3//
4//! PSI-based system stress management with QoS floor.
5#![allow(clippy::unwrap_used)]
6use std::sync::atomic::{AtomicU64, Ordering};
7use std::time::Duration;
8use std::sync::Arc;
9use sysinfo::System;
10use tracing::{warn, debug, info};
11use crate::metrics;
12use std::fs;
13use crate::constants;
14use std::thread;
15
16
17/// Store f64 in AtomicU64 via bit transmutation (lock-free reads)
18fn f64_to_u64(v: f64) -> u64 { v.to_bits() }
19fn u64_to_f64(v: u64) -> f64 { f64::from_bits(v) }
20
21/// Detect if running inside a hypervisor (KVM, VMware, Xen, Hyper-V, etc.)
22pub fn detect_hypervisor() -> Option<String> {
23    // Method 1: systemd-detect-virt (most reliable)
24    if let Ok(output) = std::process::Command::new("systemd-detect-virt")
25        .output()
26    {
27        let virt = String::from_utf8_lossy(&output.stdout).trim().to_string();
28        if output.status.success() && virt != "none" && !virt.is_empty() {
29            return Some(virt);
30        }
31    }
32    // Method 2: DMI vendor string
33    if let Ok(vendor) = fs::read_to_string("/sys/class/dmi/id/sys_vendor") {
34        let v = vendor.trim().to_lowercase();
35        if v.contains("qemu") || v.contains("vmware") || v.contains("xen")
36            || v.contains("microsoft") || v.contains("amazon")
37            || v.contains("google") || v.contains("digitalocean") {
38            return Some(v);
39        }
40    }
41    // Method 3: hypervisor CPUID flag (Linux exposes this)
42    if std::path::Path::new("/sys/hypervisor/type").exists()
43        && let Ok(t) = fs::read_to_string("/sys/hypervisor/type") {
44            return Some(t.trim().to_string());
45        }
46    None
47}
48
49/// Detect whether cgroup-scoped PSI pressure files are available.
50///
51/// Returns `true` if `/sys/fs/cgroup/io.pressure` exists and is readable,
52/// indicating cgroup2 with PSI enabled. The cgroup PSI files use the same
53/// format as `/proc/pressure/*` but are scoped to the process's cgroup.
54pub fn detect_cgroup_psi() -> bool {
55    let probe = "/sys/fs/cgroup/io.pressure";
56    std::path::Path::new(probe).exists() && fs::read_to_string(probe).is_ok()
57}
58
59/// PSI-based system stress governor with background monitoring thread.
60pub struct Governor {
61    /// System stress score  --  updated by background thread, read lock-free by workers
62    stress_score: Arc<AtomicU64>,
63    /// Memory usage percentage  --  updated by background thread, read lock-free
64    memory_usage_pct: Arc<AtomicU64>,
65    min_hydration_interval: Duration,
66    #[allow(dead_code)]
67    min_throughput_bytes_sec: AtomicU64,
68    copy_success_count: AtomicU64,
69    copy_failure_count: AtomicU64,
70    /// Epoch millis of failure window start  --  atomic, no mutex
71    failure_window_start_ms: AtomicU64,
72}
73
74impl Governor {
75    /// Create a new governor with the given stress thresholds.
76    pub fn new(max_load: f64, hydration_delay_ms: u64, psi_io_limit: f64, psi_cpu_limit: f64) -> Self {
77        let is_one_shot = constants::ONE_SHOT_MODE.load(Ordering::Relaxed);
78        let in_hypervisor = detect_hypervisor();
79
80        let (effective_io_limit, effective_cpu_limit) = if is_one_shot {
81            debug!("Governor: Applying RELAXED thresholds for One-Shot mode.");
82            (constants::GOVERNOR_PSI_IO_THRESHOLD_RELAXED, constants::GOVERNOR_PSI_CPU_THRESHOLD_RELAXED)
83        } else if let Some(ref virt_type) = in_hypervisor {
84            // Hypervisor environments have inflated PSI metrics due to
85            // virtio-blk -> qcow2 -> host storage indirection. Apply 5x relaxation.
86            let relaxed_io = (psi_io_limit * constants::GOVERNOR_HYPERVISOR_PSI_MULTIPLIER).max(50.0);
87            let relaxed_cpu = (psi_cpu_limit * constants::GOVERNOR_HYPERVISOR_PSI_MULTIPLIER).max(50.0);
88            warn!("Governor: Hypervisor detected ({}). Relaxing PSI thresholds: IO>{:.1}, CPU>{:.1}",
89                  virt_type, relaxed_io, relaxed_cpu);
90            (relaxed_io, relaxed_cpu)
91        } else {
92            (psi_io_limit, psi_cpu_limit)
93        };
94
95        let use_cgroup_psi = detect_cgroup_psi();
96        let psi_available = use_cgroup_psi || std::path::Path::new("/proc/pressure/io").exists();
97        if psi_available {
98            if use_cgroup_psi {
99                info!("PSI: using cgroup-scoped pressure");
100            } else {
101                info!("PSI: using system-wide pressure (cgroup not available)");
102            }
103            debug!("Governor: PSI active. Limits: IO>{:.1}, CPU>{:.1}", effective_io_limit, effective_cpu_limit);
104        }
105
106        // Lock-free f64 storage via AtomicU64 bit transmutation
107        // Safe: f64 and u64 are both 8 bytes, AtomicU64 guarantees no torn reads on x86_64
108        let stress_score = Arc::new(AtomicU64::new(f64_to_u64(0.0)));
109        let memory_usage_pct = Arc::new(AtomicU64::new(f64_to_u64(0.0)));
110
111        let stress_score_thread = stress_score.clone();
112        let memory_usage_thread = memory_usage_pct.clone();
113
114        #[allow(clippy::expect_used)]
115        thread::Builder::new()
116            .name("foxing-governor".into())
117            .spawn(move || {
118                let mut system = System::new();
119                system.refresh_all();
120                loop {
121                    let mut max_score = 0.0;
122                    let mut reason = "None";
123
124                    system.refresh_memory();
125                    let used = system.used_memory();
126                    let total = system.total_memory();
127                    let mem_pct = if total > 0 { used as f64 / total as f64 } else { 0.0 };
128                    
129                    memory_usage_thread.store(f64_to_u64(mem_pct), Ordering::Relaxed);
130
131                    if mem_pct > 0.90 {
132                        let mem_score = (mem_pct - 0.90) * 10.0;
133                        if mem_score > max_score {
134                            max_score = mem_score;
135                            reason = "Memory";
136                        }
137                    }
138
139                    if psi_available {
140                        if let Some(io_psi) = Self::read_psi("io", use_cgroup_psi) {
141                            let score = io_psi.avg10 / effective_io_limit;
142                            if score > max_score {
143                                max_score = score;
144                                reason = "PSI_IO";
145                            }
146                        }
147                        if let Some(cpu_psi) = Self::read_psi("cpu", use_cgroup_psi) {
148                            let score = cpu_psi.avg10 / effective_cpu_limit;
149                            if score > max_score {
150                                max_score = score;
151                                reason = "PSI_CPU";
152                            }
153                        }
154                    } else {
155                        system.refresh_cpu_all();
156                        let load = sysinfo::System::load_average();
157                        metrics::GOVERNOR_LOAD_AVERAGE.with_label_values(&["1m"]).set(load.one);
158                        let score = load.one / max_load;
159                        if score > max_score {
160                            max_score = score;
161                            reason = "LoadAvg";
162                        }
163                    }
164
165                    stress_score_thread.store(f64_to_u64(max_score), Ordering::Relaxed);
166
167                    metrics::GOVERNOR_STRESSED.set(if max_score >= 1.0 { 1.0 } else { 0.0 });
168                    metrics::GOVERNOR_STRESS_SCORE.set(max_score);
169
170                    if max_score > 2.0 {
171                         warn!("System Critical! Score: {:.2} (Reason: {}). Throttling hard.", max_score, reason);
172                    }
173
174                    thread::sleep(Duration::from_millis(constants::GOVERNOR_CHECK_INTERVAL_MS));
175                }
176            }).expect("Failed to spawn governor thread");
177
178        Self {
179            stress_score,
180            memory_usage_pct,
181            min_hydration_interval: Duration::from_millis(hydration_delay_ms),
182            min_throughput_bytes_sec: AtomicU64::new(0),
183            copy_success_count: AtomicU64::new(0),
184            copy_failure_count: AtomicU64::new(0),
185            failure_window_start_ms: AtomicU64::new(0),
186        }
187    }
188
189    #[allow(dead_code)]
190    pub(crate) fn set_min_throughput(&self, bytes_per_sec: u64) {
191        self.min_throughput_bytes_sec.store(bytes_per_sec, Ordering::Relaxed);
192    }
193
194    fn read_psi(resource: &str, use_cgroup: bool) -> Option<PsiMetrics> {
195        let path = if use_cgroup {
196            format!("/sys/fs/cgroup/{}.pressure", resource)
197        } else {
198            format!("/proc/pressure/{}", resource)
199        };
200        let content = fs::read_to_string(path).ok()?;
201        for line in content.lines() {
202            if line.starts_with("some ") {
203                return parse_psi_line(line);
204            }
205        }
206        None
207    }
208
209    #[allow(dead_code)]
210    pub(crate) fn signal_copy_result(&self, success: bool) {
211        if success {
212            self.copy_success_count.fetch_add(1, Ordering::Relaxed);
213        } else {
214            self.copy_failure_count.fetch_add(1, Ordering::Relaxed);
215        }
216    }
217
218    /// Current failure rate as a fraction [0.0, 1.0].
219    pub(crate) fn failure_rate(&self) -> f64 {
220        let success = self.copy_success_count.load(Ordering::Relaxed);
221        let failure = self.copy_failure_count.load(Ordering::Relaxed);
222        let total = success + failure;
223        if total == 0 { return 0.0; }
224        failure as f64 / total as f64
225    }
226
227    /// Reset failure counters (e.g. at the start of a new failure window).
228    pub(crate) fn reset_failure_window(&self) {
229        self.copy_success_count.store(0, Ordering::Relaxed);
230        self.copy_failure_count.store(0, Ordering::Relaxed);
231        self.failure_window_start_ms.store(0, Ordering::Relaxed);
232    }
233
234    /// Lock-free stress score read with failure-rate boost.
235    /// Called ~320 times/second across all workers  --  must be contention-free.
236    pub fn current_stress_score(&self) -> f64 {
237        let mut score = u64_to_f64(self.stress_score.load(Ordering::Relaxed));
238
239        // Boost stress when copy failure rate exceeds threshold
240        let failure_rate = self.failure_rate();
241        if failure_rate > constants::GOVERNOR_FAILURE_RATE_THRESHOLD {
242            score += 0.3;
243        }
244
245        // Auto-reset failure window  --  lock-free, reset after enough samples
246        let total_ops = self.copy_success_count.load(Ordering::Relaxed)
247            + self.copy_failure_count.load(Ordering::Relaxed);
248        if total_ops > constants::GOVERNOR_FAILURE_WINDOW_RESET_OPS {
249            self.reset_failure_window();
250        }
251
252        score
253    }
254
255    /// Return true if the system is under enough stress to throttle.
256    pub fn is_system_stressed(&self) -> bool {
257        self.current_stress_score() >= 1.0
258    }
259
260    /// Lock-free memory usage read.
261    pub fn current_memory_usage_pct(&self) -> f64 {
262        u64_to_f64(self.memory_usage_pct.load(Ordering::Relaxed))
263    }
264
265    /// Sleep proportionally to system stress before hydration work.
266    pub fn pace_hydration(&self) {
267        let score = self.current_stress_score();
268
269        // Skip pacing entirely when system is not stressed  --  during initial
270        // hydration on NVMe source there's no contention worth throttling for.
271        if score < 0.1 { return; }
272
273        let is_one_shot = constants::ONE_SHOT_MODE.load(Ordering::Relaxed);
274        let throttle_threshold = if is_one_shot { 1.5 } else { 0.8 };
275
276        if score < throttle_threshold {
277            if !self.min_hydration_interval.is_zero() {
278                std::thread::sleep(self.min_hydration_interval);
279            }
280            return;
281        }
282
283        let severity = (score - throttle_threshold).max(0.0);
284        let backoff_ms = if is_one_shot {
285             (severity * 50.0).powf(1.1).clamp(0.0, 250.0) as u64
286        } else {
287             (severity * 100.0).powf(1.2).clamp(0.0, 500.0) as u64
288        };
289
290        if backoff_ms > 0 {
291            metrics::GOVERNOR_PACING_DURATION_MS.inc_by(backoff_ms as f64);
292            metrics::GOVERNOR_THROTTLED_EVENTS.inc();
293            std::thread::sleep(Duration::from_millis(backoff_ms));
294        } else if !self.min_hydration_interval.is_zero() {
295            std::thread::sleep(self.min_hydration_interval);
296        }
297    }
298
299    /// Async variant of [`pace_hydration`] for callers running on the tokio executor.
300    ///
301    /// Uses `tokio::time::sleep` instead of `std::thread::sleep` to avoid blocking
302    /// a tokio worker thread. Use this from `async fn` contexts (e.g., `process_hydration_job`).
303    /// For sync callers on dedicated `std::thread`s (e.g., `full_scan`, `execute_frontier_scan`),
304    /// use the sync [`pace_hydration`] instead.
305    pub async fn pace_hydration_async(&self) {
306        let score = self.current_stress_score();
307
308        if score < 0.1 { return; }
309
310        let is_one_shot = constants::ONE_SHOT_MODE.load(Ordering::Relaxed);
311        let throttle_threshold = if is_one_shot { 1.5 } else { 0.8 };
312
313        if score < throttle_threshold {
314            if !self.min_hydration_interval.is_zero() {
315                tokio::time::sleep(self.min_hydration_interval).await;
316            }
317            return;
318        }
319
320        let severity = (score - throttle_threshold).max(0.0);
321        let backoff_ms = if is_one_shot {
322             (severity * 50.0).powf(1.1).clamp(0.0, 250.0) as u64
323        } else {
324             (severity * 100.0).powf(1.2).clamp(0.0, 500.0) as u64
325        };
326
327        if backoff_ms > 0 {
328            metrics::GOVERNOR_PACING_DURATION_MS.inc_by(backoff_ms as f64);
329            metrics::GOVERNOR_THROTTLED_EVENTS.inc();
330            tokio::time::sleep(Duration::from_millis(backoff_ms)).await;
331        } else if !self.min_hydration_interval.is_zero() {
332            tokio::time::sleep(self.min_hydration_interval).await;
333        }
334    }
335}
336
337/// Pressure Stall Information metrics from /proc/pressure/*.
338#[derive(Debug, Clone, Default)]
339pub struct PsiMetrics {
340    /// 10-second average stall percentage.
341    pub avg10: f64,
342    /// 60-second average stall percentage.
343    pub avg60: f64,
344    /// 300-second average stall percentage.
345    pub avg300: f64,
346    /// Total stall time in microseconds.
347    pub total_us: u64,
348}
349
350/// Parse a single PSI line into [`PsiMetrics`].
351///
352/// Accepts lines starting with `"some"` or `"full"`.
353/// Returns `None` if the line does not match the expected PSI format.
354pub fn parse_psi_line(line: &str) -> Option<PsiMetrics> {
355    let parts: Vec<&str> = line.split_whitespace().collect();
356    if parts.len() < 5 {
357        return None;
358    }
359
360    let kind = parts[0];
361    if kind != "some" && kind != "full" {
362        return None;
363    }
364
365    let mut metrics = PsiMetrics::default();
366    let mut parsed_any = false;
367
368    for part in &parts[1..] {
369        if let Some((key, value_str)) = part.split_once('=') {
370            match key {
371                "avg10" => {
372                    metrics.avg10 = value_str.parse().ok()?;
373                    parsed_any = true;
374                }
375                "avg60" => {
376                    metrics.avg60 = value_str.parse().ok()?;
377                }
378                "avg300" => {
379                    metrics.avg300 = value_str.parse().ok()?;
380                }
381                "total" => {
382                    metrics.total_us = value_str.parse().ok()?;
383                }
384                _ => {}
385            }
386        }
387    }
388
389    if parsed_any { Some(metrics) } else { None }
390}
391
392#[cfg(test)]
393#[allow(clippy::unwrap_used, clippy::expect_used)]
394mod governor_tests {
395    use super::*;
396
397    fn make_governor() -> Governor {
398        Governor::new(4.0, 0, 10.0, 25.0)
399    }
400
401    #[test]
402    fn test_governor_default_not_stressed() {
403        let gov = make_governor();
404        assert!(!gov.is_system_stressed());
405    }
406
407    #[test]
408    fn test_governor_current_stress_score_near_zero() {
409        let gov = make_governor();
410        let score = gov.current_stress_score();
411        // Background thread may have updated score, but it should be low on a healthy system
412        assert!(score < 1.0, "expected score < 1.0 on healthy system, got {}", score);
413    }
414
415    #[test]
416    fn test_governor_stressed_when_score_high() {
417        let gov = make_governor();
418        gov.stress_score.store(f64_to_u64(2.0), Ordering::SeqCst);
419        assert!(gov.is_system_stressed());
420        assert!(gov.current_stress_score() >= 2.0);
421    }
422
423    #[test]
424    fn test_pace_hydration_fast_path() {
425        let gov = make_governor();
426        gov.stress_score.store(f64_to_u64(0.0), Ordering::SeqCst);
427        let start = std::time::Instant::now();
428        gov.pace_hydration();
429        let elapsed = start.elapsed();
430        assert!(elapsed.as_millis() < 10, "pace_hydration should return instantly when score < 0.1, took {}ms", elapsed.as_millis());
431    }
432
433    #[test]
434    fn test_stress_score_roundtrip() {
435        let val = 1.2345_f64;
436        let bits = f64_to_u64(val);
437        let recovered = u64_to_f64(bits);
438        assert_eq!(val, recovered);
439
440        let atomic = AtomicU64::new(f64_to_u64(1.2345_f64));
441        let read_back = u64_to_f64(atomic.load(Ordering::Relaxed));
442        assert!((read_back - 1.2345_f64).abs() < f64::EPSILON);
443    }
444
445    #[test]
446    fn test_failure_rate_tracking() {
447        let gov = make_governor();
448        assert_eq!(gov.failure_rate(), 0.0);
449
450        gov.signal_copy_result(true);
451        gov.signal_copy_result(true);
452        gov.signal_copy_result(false);
453        // 1 failure out of 3 total = 0.333...
454        let rate = gov.failure_rate();
455        assert!((rate - 1.0 / 3.0).abs() < 0.01);
456
457        gov.reset_failure_window();
458        assert_eq!(gov.failure_rate(), 0.0);
459    }
460
461    #[test]
462    fn test_pace_hydration_throttled_path() {
463        let gov = make_governor();
464        gov.stress_score.store(f64_to_u64(1.5), Ordering::SeqCst);
465        gov.pace_hydration();
466        assert!(gov.current_stress_score() >= 1.0);
467    }
468
469    #[test]
470    fn test_current_memory_usage_pct() {
471        let gov = make_governor();
472        std::thread::sleep(Duration::from_millis(200));
473        let pct = gov.current_memory_usage_pct();
474        assert!(pct >= 0.0, "memory usage should be >= 0.0, got {}", pct);
475        assert!(pct <= 1.0, "memory usage should be <= 1.0 (fraction), got {}", pct);
476    }
477
478    #[test]
479    fn test_failure_rate_auto_reset() {
480        let gov = make_governor();
481        for _ in 0..1001 {
482            gov.signal_copy_result(true);
483        }
484        let _score = gov.current_stress_score();
485        assert_eq!(gov.failure_rate(), 0.0, "failure rate should be 0.0 after auto-reset");
486        gov.signal_copy_result(false);
487        let rate = gov.failure_rate();
488        assert!((rate - 1.0).abs() < f64::EPSILON, "1 failure out of 1 total = 1.0");
489    }
490}
fxcp_core/governor.rs

fxcp_core/
governor.rs
