namespace MoShiro.Measurement; /// 探测节奏的状态。 是线路几乎全丢:不再按正常频率白白发包,指数退避着探测,等到有回包再恢复。 public enum ProbeState : byte { Quiet = 0, Watch = 1, Active = 2, Dead = 3, } /// 自适应探测频率的参数。频率范围、精度目标和各状态的保持时间都在这里。 public sealed record AdaptiveRateOptions { /// 平稳时的最低频率(Hz)。 public double MinHz { get; init; } = 2; /// /// 最高频率(Hz),也是权重的参考网格:每个探测包的权重是它代表的时间除以 1/MaxHz, /// 分析时丢包率和 RTT 分位数按权重累计,所以频率怎么变,得到的仍是按时间平均的值,不会偏向探测更密的时段。 /// public double MaxHz { get; init; } = 10; /// 平稳时 RTT 分位数(p99)的 95% 置信区间半宽容差:取 max(下限毫秒, 百分比 × p99)。样本多到超出这个精度才会降频。 public double RttTolPct { get; init; } = 2; public double RttTolFloorMs { get; init; } = 0.3; /// 平稳线路要能证明的丢包上界(百分比,95% 置信)。三法则:零丢包需要 3 / τ 个样本。 public double TauPct { get; init; } = 0.2; /// 证明窗口(秒):平稳频率 = 3 / (τ × 窗口)。 public double CertifySeconds { get; init; } = 300; /// 历史上一直平稳的时段可以拉长证明窗口,同样的精度用更少的包。 public double QuietCertifySeconds { get; init; } = 900; /// 回包等多久算丢(秒)。 public double GraceSeconds { get; init; } = 3; /// 刚启动没有历史:先按 Watch 频率跑这么久(秒)。 public double WarmupSeconds { get; init; } = 300; /// 一次丢包事件之后,Watch 至少保持多久(秒)。 public double WatchHoldSeconds { get; init; } = 1800; /// 进入 Active 之后,没有新的丢包事件要保持多久才降到 Watch(秒)。 public double ActiveHoldSeconds { get; init; } = 900; /// 证明窗口内有这么多个不同的秒出现丢包,就算持续丢包,升到 Active。 public int ActiveLossCount { get; init; } = 3; /// 判断线路全丢的窗口(秒)和最少样本数、丢包比例。 public double DeadWindowSeconds { get; init; } = 10; public int DeadMinSent { get; init; } = 10; public double DeadLossRatio { get; init; } = 0.95; /// Dead 状态的探测间隔:从 1 秒起翻倍,最长这么久(秒),带 ±20% 抖动。 public double DeadMaxIntervalSeconds { get; init; } = 30; /// RTT 比长期水平高出这个倍数(再加下面的绝对值)持续几秒,就当作链路在变差,至少升到 Watch。 public double RttFactor { get; init; } = 1.5; public double RttFloorMs { get; init; } = 5; /// 用 RTT p99 的置信区间限制降频。样本稀少的任务(每分钟几次)估不出区间,关掉后平稳频率只由丢包上界决定。 public bool RttGuard { get; init; } = true; /// 先验:时间 → 这个时刻丢包的历史概率(0 到 1);null 表示没有历史。 public Func? Prior { get; init; } /// 先验里的"丢包高发"阈值:达到它在到来之前 提前预热。 public double PriorWatchAt { get; init; } = 0.15; public double PriorActiveAt { get; init; } = 0.5; public double PriorLeadSeconds { get; init; } = 1800; } /// 状态或频率变了: 是给日志和审计看的固定短语。 public sealed record ProbeTransition(ProbeState From, ProbeState To, double Hz, string Reason, double LossPct, int Sent, int Lost); /// /// 自适应探测频率:按"要证明的精度"决定频率,丢包多了自动升频,线路全丢就指数退避。 /// 平稳线路只需要能证明丢包低于 τ 的样本数(三法则 N = 3/τ),有丢包的线路用满频率保精度和丢包的细节(连续丢包、乱序)。 /// 状态靠丢包事件的频率升降,带滞回,不会在阈值附近来回跳;时间画像可以让高发时段提前预热。 /// 纯逻辑:时间由调用方传入,不依赖时钟,方便测试。 /// public sealed class AdaptiveRate { private const long UsPerSecond = 1_000_000; private const int Buckets = 1200; // 20 分钟,够覆盖最长的统计窗口 private readonly AdaptiveRateOptions _options; private readonly Random _random; private readonly object _gate = new(); private readonly long[] _second = new long[Buckets]; private readonly int[] _sent = new int[Buckets]; private readonly int[] _replied = new int[Buckets]; private long _startUs = -1; private long _lastLossUs = long.MinValue; private long _activeUntilUs = long.MinValue; private long _lastReplyUs = long.MinValue; private long _deadSinceUs = long.MinValue; private int _deadProbes; private readonly Queue<(long Second, double Ms)> _rtts = new(); private long _lastGuardSecond = long.MinValue; private double _guardHz = double.NaN; private TimeSpan _lastDelay = TimeSpan.Zero; private double _rttFast = double.NaN; private double _rttSlow = double.NaN; private int _rttHighSeconds; private long _lastRttSecond = long.MinValue; public AdaptiveRate(AdaptiveRateOptions options, Random? random = null) { _options = options; _random = random ?? Random.Shared; State = ProbeState.Watch; Hz = _options.MaxHz; } public ProbeState State { get; private set; } /// 当前的探测频率(Hz)。Dead 时是退避后的近似频率。 public double Hz { get; private set; } /// 发出一个探测包。 public void OnSend(long nowUs) { lock (_gate) { if (_startUs < 0) { _startUs = nowUs; } long second = nowUs / UsPerSecond; int slot = Slot(second); if (_second[slot] != second) { _second[slot] = second; _sent[slot] = 0; _replied[slot] = 0; } _sent[slot]++; if (State == ProbeState.Dead) { _deadProbes++; } } } /// 收到回包: 是它发出的时间, 是往返时延。 public void OnReply(long sentUs, long nowUs, double rttMs) { lock (_gate) { long second = sentUs / UsPerSecond; int slot = Slot(second); if (_second[slot] == second && _replied[slot] < _sent[slot]) { _replied[slot]++; } if (double.IsNaN(_rttSlow)) { _rttSlow = _rttFast = rttMs; } else { // 时间常数:快的 5 秒,慢的 15 分钟(按时间而不是按包数,频率变了也一样) double dt = Math.Max(0, (nowUs - Math.Max(_lastReplyUs, 0)) / (double)UsPerSecond); _rttFast += (1 - Math.Exp(-dt / 5)) * (rttMs - _rttFast); _rttSlow += (1 - Math.Exp(-dt / 900)) * (rttMs - _rttSlow); } _lastReplyUs = nowUs; _rtts.Enqueue((nowUs / UsPerSecond, rttMs)); while (_rtts.Count > 6000) { _rtts.Dequeue(); } } } /// 连接(重新)建立了:之前线路不通退避着的话,马上恢复探测,不用等退避间隔。 public void Reconnected(long nowUs) { lock (_gate) { if (State == ProbeState.Dead) { _lastReplyUs = nowUs; } } } /// 距离下一个探测包还有多久(带抖动,平均值等于频率的倒数)。 public TimeSpan NextDelay() { lock (_gate) { if (State == ProbeState.Dead) { double seconds = Math.Min(_options.DeadMaxIntervalSeconds, Math.Pow(2, Math.Min(_deadProbes, 10))); return _lastDelay = TimeSpan.FromSeconds(seconds * (0.8 + 0.4 * _random.NextDouble())); } // 平稳时间隔精确(不加抖动):发包时刻规整,列块里的时间戳差分更小;抖动只用在退避里避免多条线路同步重试 return _lastDelay = TimeSpan.FromSeconds(1.0 / Hz); } } /// 刚刚等过的那个间隔所代表的权重(以 1/MaxHz 为单位):下一个发出的包用它。平稳时是 MaxHz/Hz,退避时是几十到几百。 public int Weight { get { lock (_gate) { return (int)Math.Clamp(Math.Round(_lastDelay.TotalSeconds * _options.MaxHz), 1, 65535); } } } /// 重新评估状态,大约每秒调用一次;状态或频率变了就返回变化。 public ProbeTransition? Evaluate(long nowUs) { lock (_gate) { if (_startUs < 0) { return null; } long finalSecond = nowUs / UsPerSecond - (long)Math.Ceiling(_options.GraceSeconds); (int sent, int replied, int _) shortWindow = Window(finalSecond, (int)_options.DeadWindowSeconds); (int sent, int replied, int lossSeconds) certify = Window(finalSecond, (int)_options.CertifySeconds); int lostCertify = certify.sent - certify.replied; double lossPct = certify.sent == 0 ? 0 : 100.0 * lostCertify / certify.sent; // 最近一次有丢包的时间:用最近 CertifySeconds 里最晚一个有丢包的整秒 for (long second = finalSecond; second > finalSecond - (long)_options.CertifySeconds; second--) { int slot = Slot(second); if (_second[slot] == second && _sent[slot] > _replied[slot]) { _lastLossUs = Math.Max(_lastLossUs, second * UsPerSecond); break; } } UpdateRttTrend(nowUs); ProbeState next; string reason; if (State == ProbeState.Dead) { if (_lastReplyUs > _deadSinceUs) { next = ProbeState.Watch; reason = "reply received"; _deadProbes = 0; _lastLossUs = nowUs; } else { return null; } } else if (shortWindow.sent >= _options.DeadMinSent && shortWindow.replied <= shortWindow.sent * (1 - _options.DeadLossRatio)) { next = ProbeState.Dead; reason = "line unreachable"; _deadSinceUs = nowUs; _deadProbes = 0; } else { (next, reason) = Decide(nowUs, certify.lossSeconds, lostCertify, lossPct); } UpdateGuard(nowUs); double hz = next switch { ProbeState.Quiet => Quantize(Math.Max(QuietHz(QuietWindowSeconds(nowUs)), !_options.RttGuard ? _options.MinHz : double.IsNaN(_guardHz) ? _options.MaxHz : _guardHz)), ProbeState.Watch or ProbeState.Active => _options.MaxHz, _ => Math.Max(_options.MinHz / 4, 1.0 / _options.DeadMaxIntervalSeconds), }; hz = Math.Min(_options.MaxHz, hz); if (next == State && Math.Abs(hz - Hz) / Hz < 0.05) { return null; } ProbeTransition result = new(State, next, hz, reason, lossPct, certify.sent, lostCertify); State = next; Hz = hz; return result; } } private (ProbeState State, string Reason) Decide(long nowUs, int lossSeconds, int lostCertify, double lossPct) { ProbeState state = ProbeState.Quiet; string reason = "quiet"; double sinceLoss = _lastLossUs == long.MinValue ? double.PositiveInfinity : (nowUs - _lastLossUs) / (double)UsPerSecond; if (lossSeconds >= _options.ActiveLossCount || (lossPct >= _options.TauPct * 2 && lossSeconds >= 2)) { _activeUntilUs = nowUs + (long)(_options.ActiveHoldSeconds * UsPerSecond); } if (nowUs < _activeUntilUs) { return (ProbeState.Active, "sustained loss"); } if (sinceLoss <= _options.WatchHoldSeconds) { state = ProbeState.Watch; reason = "recent loss"; } if (_rttHighSeconds >= 3) { state = ProbeState.Watch; reason = "rtt above baseline"; } double? prior = _options.Prior?.Invoke(nowUs); double? ahead = _options.Prior?.Invoke(nowUs + (long)(_options.PriorLeadSeconds * UsPerSecond)); double expected = Math.Max(prior ?? 0, ahead ?? 0); if (expected >= _options.PriorActiveAt) { return (ProbeState.Active, "expected peak"); } if (expected >= _options.PriorWatchAt && state == ProbeState.Quiet) { state = ProbeState.Watch; reason = "expected loss window"; } if (state == ProbeState.Quiet && _options.Prior is null && nowUs - _startUs < (long)(_options.WarmupSeconds * UsPerSecond)) { state = ProbeState.Watch; reason = "warm-up"; } return (state, reason); } /// 历史上这个时段(现在和将要到来的一段时间)都平稳时,证明窗口可以更长,频率更低。 private double QuietWindowSeconds(long nowUs) { if (_options.Prior is null) { return _options.CertifySeconds; } double now = _options.Prior(nowUs) ?? 1; double ahead = _options.Prior(nowUs + (long)(_options.PriorLeadSeconds * UsPerSecond)) ?? 1; return Math.Max(now, ahead) <= 0.02 ? _options.QuietCertifySeconds : _options.CertifySeconds; } /// 频率取 MaxHz/k(k 为整数):每个包的权重正好是整数 k,分析时没有舍入误差。 private double Quantize(double hz) { double k = Math.Max(1, Math.Floor(_options.MaxHz / Math.Max(hz, 1e-9))); return _options.MaxHz / k; } /// /// 分位数精度保护:用最近一个证明窗口里的 RTT 样本估计 p99 的 95% 置信区间(次序统计量), /// 半宽超过容差就升频,远小于容差才降频(单步最多减半),使平稳线路的 RTT 分位数始终保持在设定的精度内。 /// private void UpdateGuard(long nowUs) { long second = nowUs / UsPerSecond; if (_lastGuardSecond != long.MinValue && second - _lastGuardSecond < 10) { return; } _lastGuardSecond = second; long from = second - (long)_options.CertifySeconds; double[] window = [.. _rtts.Where(r => r.Second >= from).Select(r => r.Ms).Order()]; if (window.Length < 200) { _guardHz = double.NaN; // 样本不够估计精度:按最高频率 return; } int n = window.Length; double rank = n * 0.99; double half = 1.96 * Math.Sqrt(n * 0.99 * 0.01); int lo = Math.Clamp((int)Math.Floor(rank - half), 0, n - 1); int hi = Math.Clamp((int)Math.Ceiling(rank + half), 0, n - 1); double halfWidth = (window[hi] - window[lo]) / 2; double tolerance = Math.Max(_options.RttTolFloorMs, _options.RttTolPct / 100.0 * window[Math.Min(n - 1, (int)rank)]); double ratio = Math.Clamp(halfWidth / tolerance, 0.7, 2.0); // 单步变化有限,避免来回跳 double current = Hz; _guardHz = Math.Clamp(current * ratio * ratio, _options.MinHz, _options.MaxHz); } private double QuietHz(double windowSeconds) => Math.Min(_options.MaxHz, Math.Max(_options.MinHz, 3.0 / (_options.TauPct / 100.0 * windowSeconds))); /// 窗口内发出、回来的包数,以及有丢包的整秒数(一次连续丢包只算一个"事件",不会因为一次短暂中断就判成持续丢包)。 private (int Sent, int Replied, int LossSeconds) Window(long finalSecond, int seconds) { int sent = 0; int replied = 0; int lossSeconds = 0; for (long second = finalSecond; second > finalSecond - seconds; second--) { int slot = Slot(second); if (_second[slot] == second) { sent += _sent[slot]; replied += _replied[slot]; if (_sent[slot] > _replied[slot]) { lossSeconds++; } } } return (sent, replied, lossSeconds); } private void UpdateRttTrend(long nowUs) { long second = nowUs / UsPerSecond; if (second == _lastRttSecond || double.IsNaN(_rttFast) || double.IsNaN(_rttSlow)) { return; } _lastRttSecond = second; bool high = _rttFast > _rttSlow * _options.RttFactor + _options.RttFloorMs; _rttHighSeconds = high ? _rttHighSeconds + 1 : 0; } private static int Slot(long second) => (int)(((second % Buckets) + Buckets) % Buckets); }