第 13 章：犀牛鸟竞赛实战指南

//! 自适应多路径调度器
//!
//! 基于 UCB1 算法在多条路径间进行探索与利用的平衡，
//! 同时考虑 RTT 和带宽两个维度的路径质量。
//!
//! # 算法概述
//! 1. 每条路径维护一个"得分"，综合 RTT、带宽、探索奖励
//! 2. 每次调度选得分最高的路径
//! 3. 定期更新得分，衰减旧测量值
//!
//! # 参考
//! - [UCB1] Auer et al., "Finite-time Analysis of the Multiarmed Bandit Problem"
//! - [BLEST] Ferlin et al., "BLEST: Blocking Estimation-based MPTCP Scheduler"

use std::time::{Duration, Instant};
use crate::multipath::path::PathId;
use crate::multipath::scheduler::MultipathScheduler;
use crate::connection::Connection;

/// 路径统计信息，用于调度决策
#[derive(Debug, Clone)]
struct PathStats {
    /// 平滑 RTT（指数移动平均）
    smoothed_rtt: Duration,
    /// 估计带宽（bytes/sec）
    estimated_bandwidth: u64,
    /// 该路径被选中的总次数
    selection_count: u64,
    /// 该路径的累计"奖励"
    cumulative_reward: f64,
    /// 上次更新时间
    last_updated: Instant,
}

/// 自适应调度器的配置参数
#[derive(Debug, Clone)]
pub struct AdaptiveConfig {
    /// UCB 探索系数 c，越大越倾向探索
    /// 推荐范围：0.5 - 2.0，默认 1.0
    pub exploration_coefficient: f64,
    /// RTT 权重 α
    pub rtt_weight: f64,
    /// 带宽权重 β
    pub bandwidth_weight: f64,
    /// 测量值衰减因子（每秒衰减多少）
    pub decay_factor: f64,
    /// 最小探测间隔
    pub min_probe_interval: Duration,
}

impl Default for AdaptiveConfig {
    fn default() -> Self {
        Self {
            exploration_coefficient: 1.0,
            rtt_weight: 0.6,
            bandwidth_weight: 0.4,
            decay_factor: 0.95,
            min_probe_interval: Duration::from_millis(100),
        }
    }
}

/// 自适应多路径调度器
pub struct AdaptiveScheduler {
    config: AdaptiveConfig,
    path_stats: HashMap<PathId, PathStats>,
    total_selections: u64,
}

impl AdaptiveScheduler {
    pub fn new(config: AdaptiveConfig) -> Self {
        Self {
            config,
            path_stats: HashMap::new(),
            total_selections: 0,
        }
    }

    /// 计算路径的 UCB 得分
    /// 得分 = 利用项 + 探索项
    /// 利用项 = α × RTT得分 + β × 带宽得分
    /// 探索项 = c × sqrt(ln(N) / n_i)
    fn compute_score(&self, path_id: PathId) -> f64 {
        let stats = match self.path_stats.get(&path_id) {
            Some(s) => s,
            None => return f64::MAX, // 未探索的路径优先
        };

        // 利用项：归一化的路径质量
        let rtt_score = 1.0 / stats.smoothed_rtt.as_secs_f64().max(0.001);
        let bw_score = stats.estimated_bandwidth as f64;
        let exploitation = self.config.rtt_weight * rtt_score
            + self.config.bandwidth_weight * bw_score;

        // 探索项：UCB1 公式
        let exploration = if stats.selection_count > 0 && self.total_selections > 0 {
            self.config.exploration_coefficient
                * ((self.total_selections as f64).ln() / stats.selection_count as f64).sqrt()
        } else {
            f64::MAX
        };

        exploitation + exploration
    }

    /// 更新路径统计（在收到 ACK 后调用）
    fn update_path_stats(&mut self, path_id: PathId, rtt: Duration, bytes_acked: u64) {
        let stats = self.path_stats.entry(path_id).or_insert(PathStats {
            smoothed_rtt: rtt,
            estimated_bandwidth: 0,
            selection_count: 0,
            cumulative_reward: 0.0,
            last_updated: Instant::now(),
        });

        // EWMA 更新 RTT
        let alpha = 0.125;
        stats.smoothed_rtt = Duration::from_secs_f64(
            (1.0 - alpha) * stats.smoothed_rtt.as_secs_f64() + alpha * rtt.as_secs_f64(),
        );

        // 更新带宽估计
        if rtt > Duration::ZERO {
            let instant_bw = bytes_acked as f64 / rtt.as_secs_f64();
            stats.estimated_bandwidth =
                ((1.0 - alpha) * stats.estimated_bandwidth as f64 + alpha * instant_bw) as u64;
        }

        stats.last_updated = Instant::now();
    }
}

impl MultipathScheduler for AdaptiveScheduler {
    fn on_schedule(&mut self, conn: &Connection) -> Option<PathId> {
        let available_paths = conn.paths().available();

        if available_paths.is_empty() {
            return None;
        }

        // 选择得分最高的路径
        let best_path = available_paths
            .iter()
            .max_by(|a, b| {
                self.compute_score(*a)
                    .partial_cmp(&self.compute_score(*b))
                    .unwrap_or(std::cmp::Ordering::Equal)
            })
            .copied();

        if let Some(path_id) = best_path {
            self.total_selections += 1;
            if let Some(stats) = self.path_stats.get_mut(&path_id) {
                stats.selection_count += 1;
            }
        }

        best_path
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// 未探索的路径应该优先被选中
    #[test]
    fn test_unexplored_path_priority() {
        let mut scheduler = AdaptiveScheduler::new(AdaptiveConfig::default());

        // 模拟 path_0 已经被选了 100 次
        scheduler.path_stats.insert(PathId(0), PathStats {
            smoothed_rtt: Duration::from_millis(50),
            estimated_bandwidth: 1_000_000,
            selection_count: 100,
            cumulative_reward: 80.0,
            last_updated: Instant::now(),
        });

        // path_1 是新路径，从未被探索
        // 它的得分应该是 f64::MAX，所以应该被优先选中

        let score_0 = scheduler.compute_score(PathId(0));
        let score_1 = scheduler.compute_score(PathId(1));

        assert!(score_1 > score_0, "未探索路径应有更高得分");
    }

    /// RTT 更小的路径在利用阶段应该得分更高
    #[test]
    fn test_lower_rtt_higher_exploitation_score() {
        let mut scheduler = AdaptiveScheduler::new(AdaptiveConfig {
            exploration_coefficient: 0.0, // 关闭探索，只看利用
            ..Default::default()
        });

        scheduler.path_stats.insert(PathId(0), PathStats {
            smoothed_rtt: Duration::from_millis(20),
            estimated_bandwidth: 1_000_000,
            selection_count: 50,
            cumulative_reward: 40.0,
            last_updated: Instant::now(),
        });

        scheduler.path_stats.insert(PathId(1), PathStats {
            smoothed_rtt: Duration::from_millis(100),
            estimated_bandwidth: 1_000_000,
            selection_count: 50,
            cumulative_reward: 30.0,
            last_updated: Instant::now(),
        });

        scheduler.total_selections = 100;

        let score_0 = scheduler.compute_score(PathId(0));
        let score_1 = scheduler.compute_score(PathId(1));

        assert!(score_0 > score_1, "RTT 更小的路径应得分更高");
    }

    /// 探索系数增大应该增加低探索路径的得分
    #[test]
    fn test_exploration_coefficient_effect() {
        let mut scheduler_low = AdaptiveScheduler::new(AdaptiveConfig {
            exploration_coefficient: 0.5,
            ..Default::default()
        });
        let mut scheduler_high = AdaptiveScheduler::new(AdaptiveConfig {
            exploration_coefficient: 2.0,
            ..Default::default()
        });

        let stats = PathStats {
            smoothed_rtt: Duration::from_millis(50),
            estimated_bandwidth: 1_000_000,
            selection_count: 5,  // 很少被选
            cumulative_reward: 3.0,
            last_updated: Instant::now(),
        };

        scheduler_low.path_stats.insert(PathId(0), stats.clone());
        scheduler_low.total_selections = 100;

        scheduler_high.path_stats.insert(PathId(0), stats);
        scheduler_high.total_selections = 100;

        let score_low = scheduler_low.compute_score(PathId(0));
        let score_high = scheduler_high.compute_score(PathId(0));

        assert!(
            score_high > score_low,
            "更高的探索系数应给低探索路径更高得分"
        );
    }

    /// RTT 更新应该是指数移动平均
    #[test]
    fn test_rtt_ewma_update() {
        let mut scheduler = AdaptiveScheduler::new(AdaptiveConfig::default());

        // 初始 RTT = 100ms
        scheduler.update_path_stats(PathId(0), Duration::from_millis(100), 1000);
        let rtt_1 = scheduler.path_stats[&PathId(0)].smoothed_rtt;

        // 新测量 RTT = 20ms（大幅下降）
        scheduler.update_path_stats(PathId(0), Duration::from_millis(20), 1000);
        let rtt_2 = scheduler.path_stats[&PathId(0)].smoothed_rtt;

        // EWMA 应该让 rtt_2 在 20 和 100 之间，且比 rtt_1 小
        assert!(rtt_2 < rtt_1, "新的小 RTT 应拉低平滑值");
        assert!(rtt_2 > Duration::from_millis(20), "EWMA 不会一步跳到新值");
    }
}

// 实现最简单但有效的调度策略
use std::collections::HashMap;
use std::time::Duration;

struct PathStats {
    smoothed_rtt: Duration,
    available_cwnd: usize,  // 拥塞窗口剩余空间
    is_active: bool,
}

struct MinRttScheduler {
    paths: HashMap<u64, PathStats>,
}

impl MinRttScheduler {
    fn select_path(&self) -> Option<u64> {
        // 只在有可用窗口的活跃路径中选择
        self.paths.iter()
            .filter(|(_, stats)| stats.is_active && stats.available_cwnd > 0)
            .min_by_key(|(_, stats)| stats.smoothed_rtt)
            .map(|(id, _)| *id)
    }

    fn on_packet_sent(&mut self, path_id: u64, size: usize) {
        if let Some(stats) = self.paths.get_mut(&path_id) {
            stats.available_cwnd = stats.available_cwnd.saturating_sub(size);
        }
    }

    fn on_ack_received(&mut self, path_id: u64, size: usize, rtt: Duration) {
        if let Some(stats) = self.paths.get_mut(&path_id) {
            // 释放窗口空间
            stats.available_cwnd += size;
            // 更新 RTT（简化版 EWMA）
            let alpha = 0.125;
            let new_rtt = Duration::from_secs_f64(
                (1.0 - alpha) * stats.smoothed_rtt.as_secs_f64()
                + alpha * rtt.as_secs_f64()
            );
            stats.smoothed_rtt = new_rtt;
        }
    }
}

// 冗余调度：同一数据发到所有路径
struct RedundantScheduler {
    paths: HashMap<u64, PathStats>,
}

impl RedundantScheduler {
    fn select_paths(&self) -> Vec<u64> {
        // 返回所有有窗口的活跃路径
        self.paths.iter()
            .filter(|(_, stats)| stats.is_active && stats.available_cwnd > 0)
            .map(|(id, _)| *id)
            .collect()
    }

    fn schedule_packet(&self, data: &[u8]) -> Vec<(u64, Vec<u8>)> {
        // 同一份数据复制到所有可用路径
        self.select_paths().into_iter()
            .map(|path_id| (path_id, data.to_vec()))
            .collect()
    }
}

// 结合 RTT 和带宽的加权调度
struct AdaptiveScheduler {
    paths: HashMap<u64, PathStats>,
    // 每条路径的"评分"
    scores: HashMap<u64, f64>,
}

struct PathStats {
    smoothed_rtt: Duration,
    estimated_bandwidth: f64,  // bytes/sec
    loss_rate: f64,            // 0.0 - 1.0
    available_cwnd: usize,
    is_active: bool,
}

impl AdaptiveScheduler {
    fn update_scores(&mut self) {
        for (id, stats) in &self.paths {
            if !stats.is_active {
                self.scores.insert(*id, 0.0);
                continue;
            }

            // 综合评分 = 带宽 / (RTT * (1 + loss_rate))
            // 直觉：高带宽、低延迟、低丢包 = 高分
            let rtt_sec = stats.smoothed_rtt.as_secs_f64().max(0.001);
            let score = stats.estimated_bandwidth
                / (rtt_sec * (1.0 + stats.loss_rate * 10.0));

            self.scores.insert(*id, score);
        }
    }

    fn select_path(&mut self) -> Option<u64> {
        self.update_scores();

        self.paths.iter()
            .filter(|(_, stats)| stats.is_active && stats.available_cwnd > 0)
            .max_by(|(id_a, _), (id_b, _)| {
                let score_a = self.scores.get(id_a).unwrap_or(&0.0);
                let score_b = self.scores.get(id_b).unwrap_or(&0.0);
                score_a.partial_cmp(score_b).unwrap()
            })
            .map(|(id, _)| *id)
    }
}

// src/multipath/weighted_rtt.rs
// 加权 RTT 调度器：考虑 RTT 和可用窗口的比例

use super::{MultipathScheduler, SchedulerContext};
use crate::connection::path::PathId;

pub struct WeightedRttScheduler {
    // 配置参数
    rtt_weight: f64,    // RTT 在评分中的权重
    cwnd_weight: f64,   // 可用窗口在评分中的权重
}

impl WeightedRttScheduler {
    pub fn new() -> Self {
        Self {
            rtt_weight: 0.7,
            cwnd_weight: 0.3,
        }
    }

    fn score_path(&self, ctx: &SchedulerContext, path_id: PathId) -> f64 {
        let path = match ctx.get_path(path_id) {
            Some(p) => p,
            None => return 0.0,
        };

        if !path.is_active() || path.available_cwnd() == 0 {
            return 0.0;
        }

        // RTT 分数：越低越好，取倒数
        let rtt_score = 1.0 / path.smoothed_rtt().as_secs_f64().max(0.001);

        // 窗口分数：可用空间占比
        let cwnd_score = path.available_cwnd() as f64
            / path.max_cwnd() as f64;

        // 加权综合
        self.rtt_weight * rtt_score + self.cwnd_weight * cwnd_score
    }
}

impl MultipathScheduler for WeightedRttScheduler {
    fn select_send_path(&self, ctx: &SchedulerContext) -> Option<PathId> {
        ctx.active_paths()
            .iter()
            .filter(|&&pid| {
                ctx.get_path(pid)
                    .map(|p| p.available_cwnd() > 0)
                    .unwrap_or(false)
            })
            .max_by(|&&a, &&b| {
                let score_a = self.score_path(ctx, a);
                let score_b = self.score_path(ctx, b);
                score_a.partial_cmp(&score_b).unwrap_or(std::cmp::Ordering::Equal)
            })
            .copied()
    }

    fn on_packet_sent(&mut self, _ctx: &SchedulerContext, _path_id: PathId, _size: usize) {
        // 基础版本不需要额外状态
        // 高级版本可以在这里更新发送统计
    }

    fn on_packet_acked(&mut self, _ctx: &SchedulerContext, _path_id: PathId, _size: usize) {
        // 基础版本不需要额外状态
        // 高级版本可以在这里更新确认统计
    }

    fn on_packet_lost(&mut self, _ctx: &SchedulerContext, _path_id: PathId, _size: usize) {
        // 基础版本不需要额外状态
        // 高级版本可以在这里降低路径评分
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    // 测试 1：基本选择——应选 RTT 最低的路径
    #[test]
    fn test_selects_lowest_rtt() {
        let mut ctx = MockSchedulerContext::new();
        ctx.add_path(PathId(0), PathStats {
            smoothed_rtt: Duration::from_millis(50),
            available_cwnd: 1000,
            max_cwnd: 10000,
            is_active: true,
        });
        ctx.add_path(PathId(1), PathStats {
            smoothed_rtt: Duration::from_millis(100),
            available_cwnd: 1000,
            max_cwnd: 10000,
            is_active: true,
        });

        let scheduler = WeightedRttScheduler::new();
        assert_eq!(scheduler.select_send_path(&ctx), Some(PathId(0)));
    }

    // 测试 2：窗口耗尽时应跳过
    #[test]
    fn test_skips_exhausted_path() {
        let mut ctx = MockSchedulerContext::new();
        ctx.add_path(PathId(0), PathStats {
            smoothed_rtt: Duration::from_millis(10),  // 最低 RTT
            available_cwnd: 0,                         // 但窗口用完了
            max_cwnd: 10000,
            is_active: true,
        });
        ctx.add_path(PathId(1), PathStats {
            smoothed_rtt: Duration::from_millis(100),
            available_cwnd: 5000,
            max_cwnd: 10000,
            is_active: true,
        });

        let scheduler = WeightedRttScheduler::new();
        // 即使 path 0 RTT 更低，窗口为 0 也不能选
        assert_eq!(scheduler.select_send_path(&ctx), Some(PathId(1)));
    }

    // 测试 3：大窗口可以弥补高 RTT
    #[test]
    fn test_large_cwnd_compensates_high_rtt() {
        let mut ctx = MockSchedulerContext::new();
        // 路径 0：RTT 低但窗口快满了
        ctx.add_path(PathId(0), PathStats {
            smoothed_rtt: Duration::from_millis(20),
            available_cwnd: 100,     // 只剩 1%
            max_cwnd: 10000,
            is_active: true,
        });
        // 路径 1：RTT 高但窗口很空
        ctx.add_path(PathId(1), PathStats {
            smoothed_rtt: Duration::from_millis(80),
            available_cwnd: 9000,    // 90% 空闲
            max_cwnd: 10000,
            is_active: true,
        });

        // 用高窗口权重的调度器
        let scheduler = WeightedRttScheduler {
            rtt_weight: 0.3,
            cwnd_weight: 0.7,
        };
        // 窗口权重高时，应该选窗口更空的路径 1
        assert_eq!(scheduler.select_send_path(&ctx), Some(PathId(1)));
    }

    // 测试 4：所有路径不可用时返回 None
    #[test]
    fn test_no_available_path() {
        let mut ctx = MockSchedulerContext::new();
        ctx.add_path(PathId(0), PathStats {
            smoothed_rtt: Duration::from_millis(50),
            available_cwnd: 0,
            max_cwnd: 10000,
            is_active: true,
        });
        ctx.add_path(PathId(1), PathStats {
            smoothed_rtt: Duration::from_millis(100),
            available_cwnd: 0,
            max_cwnd: 10000,
            is_active: false,
        });

        let scheduler = WeightedRttScheduler::new();
        assert_eq!(scheduler.select_send_path(&ctx), None);
    }
}

// 评估指标结构
struct BenchmarkResults {
    // 吞吐量
    total_bytes: u64,
    duration: Duration,
    goodput_mbps: f64,          // 有效吞吐（排除重传）

    // 延迟
    latency_p50: Duration,      // 中位数延迟
    latency_p95: Duration,      // 95 分位
    latency_p99: Duration,      // 99 分位
    jitter: Duration,           // 延迟抖动

    // 效率
    retransmit_rate: f64,       // 重传率
    path_utilization: Vec<f64>, // 各路径利用率

    // 公平性
    jain_fairness_index: f64,   // Jain 公平性指数
}

impl BenchmarkResults {
    fn goodput(&self) -> f64 {
        (self.total_bytes as f64 * 8.0) / self.duration.as_secs_f64() / 1_000_000.0
    }

    fn path_balance(&self) -> f64 {
        // 路径利用率的标准差——越小越均衡
        let mean: f64 = self.path_utilization.iter().sum::<f64>()
            / self.path_utilization.len() as f64;
        let variance: f64 = self.path_utilization.iter()
            .map(|x| (x - mean).powi(2))
            .sum::<f64>() / self.path_utilization.len() as f64;
        variance.sqrt()
    }
}

// 思路伪代码
struct AdaptiveWeightScheduler {
    weights: [f64; 3],  // [rtt_weight, bw_weight, loss_weight]
    history: VecDeque<Decision>,
    reward_signal: f64,
}

impl AdaptiveWeightScheduler {
    fn adapt_weights(&mut self) {
        // 根据最近的调度效果调整权重
        // 如果选了某路径后吞吐提升 → 增加当时权重高的因素
        // 如果选了某路径后延迟恶化 → 降低当时权重高的因素
    }
}

// 思路伪代码
struct PredictiveScheduler {
    // 每条路径维护 RTT 趋势
    rtt_history: HashMap<PathId, VecDeque<(Instant, Duration)>>,
}

impl PredictiveScheduler {
    fn predict_rtt(&self, path_id: PathId, horizon: Duration) -> Duration {
        // 用线性回归或指数平滑预测未来 RTT
        // 如果趋势在恶化，预测值 > 当前值
        // 如果趋势在改善，预测值 < 当前值
    }

    fn select_path(&self) -> Option<PathId> {
        // 用预测的 RTT 而不是当前 RTT 做决策
    }
}

// 思路伪代码（UCB 策略）
struct UCBScheduler {
    path_rewards: HashMap<PathId, f64>,     // 平均收益
    path_counts: HashMap<PathId, u64>,      // 选择次数
    total_count: u64,
}

impl UCBScheduler {
    fn ucb_score(&self, path_id: PathId) -> f64 {
        let avg_reward = self.path_rewards[&path_id];
        let count = self.path_counts[&path_id] as f64;
        let exploration_bonus = (2.0 * (self.total_count as f64).ln() / count).sqrt();

        avg_reward + exploration_bonus  // 选择次数少的路径获得更高探索奖励
    }
}