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;
/// 先验:时间 → 这个时刻丢包的历史概率(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)), 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);
}