phantom_protocol/transport/
scheduler.rs

1//! Phantom Protocol - Multi-Path Scheduler (vestigial)
2//!
3//! A round-robin / low-latency path-selection table over multiple transport legs.
4//!
5//! **Vestigial — not on the live data path.** The project deliberately does
6//! single-path connection migration, not multipath aggregation (bandwidth bonding
7//! was evaluated and rejected). `Scheduler` is still constructed inside `Session`
8//! and reachable via `Session::scheduler()` for diagnostics, but `select_paths` is
9//! never called to steer production traffic; live per-path RTT / loss lives in
10//! `transport::path::PathState` plus the BBR `BandwidthEstimator`. The
11//! `HighThroughput` "multi-path bonding" branch below describes the rejected
12//! aggregation design and never runs against a real socket. Kept intact as the
13//! seam in case multi-path selection is ever rewired.
14
15use crate::transport::types::{LegType, SchedulerMode};
16use parking_lot::RwLock;
17use std::collections::HashMap;
18use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
19
20pub struct PathInfo {
21    pub leg_type: LegType,
22    pub rtt_ms: u32,
23    pub loss_percent: u8,
24    pub active: bool,
25    pub bytes_sent: u64,
26}
27
28impl PathInfo {
29    pub fn new(leg_type: LegType) -> Self {
30        Self {
31            leg_type,
32            rtt_ms: 150, // Default estimate
33            loss_percent: 0,
34            active: true,
35            bytes_sent: 0,
36        }
37    }
38}
39
40pub struct Scheduler {
41    mode: RwLock<SchedulerMode>,
42    paths: RwLock<HashMap<LegType, PathInfo>>,
43    #[allow(dead_code)]
44    rr_counter: AtomicU32,
45    total_bytes: AtomicU64,
46}
47
48impl Scheduler {
49    pub fn new(mode: SchedulerMode) -> Self {
50        Self {
51            mode: RwLock::new(mode),
52            paths: RwLock::new(HashMap::new()),
53            rr_counter: AtomicU32::new(0),
54            total_bytes: AtomicU64::new(0),
55        }
56    }
57
58    pub fn register_path(&self, leg_type: LegType) {
59        let mut paths = self.paths.write();
60        paths
61            .entry(leg_type)
62            .or_insert_with(|| PathInfo::new(leg_type));
63    }
64
65    pub fn set_path_available(&self, leg_type: LegType, available: bool) {
66        let mut paths = self.paths.write();
67        if let Some(path) = paths.get_mut(&leg_type) {
68            path.active = available;
69        }
70    }
71
72    pub fn select_paths(&self, is_priority: bool) -> Vec<LegType> {
73        let paths = self.paths.read();
74        let mode = self.mode.read();
75
76        let mut available: Vec<_> = paths.iter().filter(|(_, p)| p.active).collect();
77
78        if available.is_empty() {
79            return Vec::new();
80        }
81
82        if is_priority {
83            // Pick the single best path (lowest RTT)
84            available.sort_by_key(|(_, p)| p.rtt_ms);
85            return vec![*available[0].0];
86        }
87
88        match *mode {
89            SchedulerMode::LowLatency => {
90                available.sort_by_key(|(_, p)| p.rtt_ms);
91                vec![*available[0].0]
92            }
93            SchedulerMode::HighThroughput => {
94                // Return all active paths for multi-path bonding
95                available.iter().map(|(t, _)| **t).collect()
96            }
97            SchedulerMode::Reliability | SchedulerMode::Stealth => {
98                // Duplicate across all paths? No, just pick two best
99                available.sort_by_key(|(_, p)| p.rtt_ms);
100                available.iter().take(2).map(|(t, _)| **t).collect()
101            }
102        }
103    }
104
105    pub fn update_rtt(&self, leg_type: LegType, rtt: u32) {
106        let mut paths = self.paths.write();
107        if let Some(path) = paths.get_mut(&leg_type) {
108            path.rtt_ms = rtt;
109        }
110    }
111
112    pub fn record_sent(&self, leg_type: LegType, bytes: u64) {
113        let mut paths = self.paths.write();
114        if let Some(path) = paths.get_mut(&leg_type) {
115            path.bytes_sent += bytes;
116        }
117        self.total_bytes.fetch_add(bytes, Ordering::Relaxed);
118    }
119}
phantom_protocol/transport/scheduler.rs

phantom_protocol/transport/
scheduler.rs
