diff --git a/backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs b/backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs
index d0f28d1..cabd503 100644
--- a/backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs
+++ b/backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs
@@ -217,7 +217,7 @@ public sealed class StreamInfo
KindSpec spec = Kinds.Find(kind)!;
DeclaredPps = kind switch
{
- "udp_echo" or "tcp_echo" => parameters.GetInt("rate_hz"),
+ "udp_echo" or "tcp_echo" => Kinds.PeakRateHz(kind, parameters),
"udp_train" => parameters.GetNumber("mbps") * 1_000_000 / 8 / Math.Max(1, parameters.GetInt("size")),
_ => 20,
};
diff --git a/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs b/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs
index afa65cf..60e28e0 100644
--- a/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs
+++ b/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs
@@ -1,4 +1,5 @@
using System.Net;
+using Fleet.Core;
using MoShiro.Protocol;
namespace MoShiro.Measurement;
@@ -41,6 +42,76 @@ public abstract class BlockBase : IBlock
Env.Fragments!.Ship(Spec.KindId, Spec.Scope, streams, parameters ?? Spec.Params.Values, Env.Clock.UtcMsAt(first), (int)((last - first) / 1000), Spec.Port);
}
+ ///
+ /// 按自适应的节奏发包：下一个包的时刻按绝对时间排（不累积误差），每秒评估一次状态，状态变化记进日志（供审计）。
+ /// 落后了就从现在重新排，不补发。
+ ///
+ protected async Task PaceAsync(AdaptiveRate rate, Func sendOne, CancellationToken ct)
+ {
+ long due = Env.Clock.NowUs();
+ long lastEvaluation = due;
+ while (!ct.IsCancellationRequested)
+ {
+ due += (long)(rate.NextDelay().TotalMilliseconds * 1000);
+ long wait = due - Env.Clock.NowUs();
+ if (wait > 0)
+ {
+ await Task.Delay(TimeSpan.FromMicroseconds(wait), Env.Time, ct);
+ }
+ else
+ {
+ due = Env.Clock.NowUs();
+ }
+
+ await sendOne();
+ long now = Env.Clock.NowUs();
+ if (now - lastEvaluation >= 1_000_000 || rate.State == ProbeState.Dead)
+ {
+ lastEvaluation = now;
+ if (rate.Evaluate(now) is { } change)
+ {
+ LogTransition(change);
+ }
+ }
+ }
+ }
+
+ /// 读这个任务的自适应频率参数（udp_echo 和 tcp_echo 共用）。
+ protected AdaptiveRate? CreateAdaptiveRate()
+ {
+ if (!Spec.Bool("adaptive"))
+ {
+ return null;
+ }
+
+ double max = Math.Max(Spec.Int("rate_max_hz"), Spec.Int("rate_min_hz"));
+ return new AdaptiveRate(new AdaptiveRateOptions
+ {
+ MinHz = Spec.Int("rate_min_hz"),
+ MaxHz = max,
+ TauPct = Spec.Number("tau_pct"),
+ RttTolPct = Spec.Number("rtt_tol_pct"),
+ Prior = Env.LossPrior,
+ });
+ }
+
+ private void LogTransition(ProbeTransition change)
+ {
+ var fields = new (string, object?)[]
+ {
+ ("stream", Spec.StreamId), ("kind", Spec.KindId), ("from", change.From.ToString().ToLowerInvariant()), ("to", change.To.ToString().ToLowerInvariant()), ("hz", Math.Round(change.Hz, 1)),
+ ("reason", change.Reason), ("loss_pct", Math.Round(change.LossPct, 3)), ("sent", change.Sent), ("lost", change.Lost),
+ };
+ if (change.To == ProbeState.Dead)
+ {
+ Env.Log.Warn("probe rate changed", fields);
+ }
+ else
+ {
+ Env.Log.Info("probe rate changed", fields);
+ }
+ }
+
/// 什么也不做、一直等到取消（监听方里不需要自己运行逻辑的积木）。
protected async Task WaitUntilCancelledAsync(CancellationToken ct)
{
diff --git a/backend/shared/MoShiro.Measurement/Blocks/Echo.cs b/backend/shared/MoShiro.Measurement/Blocks/Echo.cs
index 3627608..f8fa36a 100644
--- a/backend/shared/MoShiro.Measurement/Blocks/Echo.cs
+++ b/backend/shared/MoShiro.Measurement/Blocks/Echo.cs
@@ -23,6 +23,7 @@ internal sealed class EchoRows
private readonly DriverLog _log;
private readonly int _size;
+ private readonly int _w;
private readonly int _t1;
private readonly int _t4;
private readonly long _graceUs;
@@ -36,26 +37,36 @@ internal sealed class EchoRows
StreamSpec spec = kind.FindStream("msg")!;
_log = new DriverLog(spec, Side.Dialer, capacity);
_size = _log.Store.ColumnIndex("size");
+ _w = _log.Store.ColumnIndex("w");
_t1 = _log.Store.ColumnIndex("t1");
_t4 = _log.Store.ColumnIndex("t4");
}
public long Count => _log.Count - Interlocked.Read(ref _discarded);
- public uint Add(int size, long t1Us)
+ /// ：这个包代表多少个基础网格间隔（频率变化时分析按它累计，丢包率和分位数仍是按时间平均的）。
+ public uint Add(int size, long t1Us, int weight = 1)
{
long key = _log.Store.Append();
_log.Store.Set(key, _size, size);
+ _log.Store.Set(key, _w, weight);
_log.Store.Set(key, _t1, t1Us);
return (uint)key;
}
- public void Reply(uint seq, long t4Us)
+ public bool Reply(uint seq, long t4Us) => Reply(seq, t4Us, out _);
+
+ /// 回填 t4；第一次回复才算， 是这个包的发出时间。
+ public bool Reply(uint seq, long t4Us, out long t1Us)
{
- if (_log.Store.Contains(seq) && !_log.Store.IsSet(seq, _t4))
+ t1Us = 0;
+ if (_log.Store.Contains(seq) && !_log.Store.IsSet(seq, _t4) && _log.Store.TryGet(seq, _t1, out t1Us))
{
_log.Store.Set(seq, _t4, t4Us);
+ return true;
}
+
+ return false;
}
public bool Discard(uint seq)
@@ -83,6 +94,7 @@ public sealed class DatagramEchoDialer : BlockBase
private readonly EchoRows _rows;
private readonly object _gate = new();
private long _nextSendFailureWarningUs;
+ private AdaptiveRate? _rate;
public DatagramEchoDialer(BlockEnv env, BlockSpec spec)
: base(env, spec)
@@ -100,29 +112,50 @@ public sealed class DatagramEchoDialer : BlockBase
IPEndPoint target = PeerEndpoint(Spec.Port);
int size = (int)Spec.Int("size");
byte[] packet = new byte[size];
+ _rate = CreateAdaptiveRate();
+
+ void SendOne()
+ {
+ long t1 = Env.Clock.NowUs();
+ lock (_gate)
+ {
+ uint seq = _rows.Add(size, t1, _rate?.Weight ?? 1);
+ MeasurementDatagram.Write(packet, new MeasurementDatagramHeader(MeasurementDatagramType.EchoRequest, session.SessionId, Spec.StreamId, seq, t1), session.Key);
+ if (datagrams.Send(session, target, packet))
+ {
+ _rate?.OnSend(t1);
+ return;
+ }
+
+ _rows.Discard(seq);
+ if (t1 >= _nextSendFailureWarningUs)
+ {
+ Env.Log.Warn("QUIC echo datagram was not queued; omitted from loss samples",
+ ("stream", Spec.StreamId), ("seq", seq), ("bytes", size));
+ _nextSendFailureWarningUs = t1 + 5_000_000;
+ }
+ }
+ }
+
datagrams.Register(Spec.StreamId, OnPacket);
try
{
- using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1.0 / Spec.Int("rate_hz")), Env.Time);
- while (await timer.WaitForNextTickAsync(ct))
+ if (_rate is null)
{
- long t1 = Env.Clock.NowUs();
- lock (_gate)
+ using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1.0 / Spec.Int("rate_hz")), Env.Time);
+ while (await timer.WaitForNextTickAsync(ct))
{
- uint seq = _rows.Add(size, t1);
- MeasurementDatagram.Write(packet, new MeasurementDatagramHeader(MeasurementDatagramType.EchoRequest, session.SessionId, Spec.StreamId, seq, t1), session.Key);
- if (!datagrams.Send(session, target, packet))
- {
- _rows.Discard(seq);
- if (t1 >= _nextSendFailureWarningUs)
- {
- Env.Log.Warn("QUIC echo datagram was not queued; omitted from loss samples",
- ("stream", Spec.StreamId), ("seq", seq), ("bytes", size));
- _nextSendFailureWarningUs = t1 + 5_000_000;
- }
- }
+ SendOne();
}
}
+ else
+ {
+ await PaceAsync(_rate, () =>
+ {
+ SendOne();
+ return ValueTask.CompletedTask;
+ }, ct);
+ }
}
catch (OperationCanceledException)
{
@@ -157,7 +190,10 @@ public sealed class DatagramEchoDialer : BlockBase
{
lock (_gate)
{
- _rows.Reply(delivery.Header.Seq, delivery.ReceiveUs);
+ if (_rows.Reply(delivery.Header.Seq, delivery.ReceiveUs, out long t1))
+ {
+ _rate?.OnReply(t1, delivery.ReceiveUs, (delivery.ReceiveUs - t1) / 1000.0);
+ }
}
}
}
@@ -211,6 +247,7 @@ public sealed class TcpEchoDialer : BlockBase
{
private readonly EchoRows _rows;
private readonly TcpInfoRecorder _info;
+ private AdaptiveRate? _rate;
public TcpEchoDialer(BlockEnv env, BlockSpec spec)
: base(env, spec)
@@ -224,12 +261,14 @@ public sealed class TcpEchoDialer : BlockBase
public override async Task RunAsync(CancellationToken ct)
{
var backoff = new Fleet.Core.Backoff(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(30));
+ _rate = CreateAdaptiveRate();
while (!ct.IsCancellationRequested)
{
try
{
using TcpLink link = await TcpLink.OpenAsync(Env, Spec.Port, Spec.Bool("tls"), Spec.StreamId, sendAuth: true, ct);
backoff.Reset();
+ _rate?.Reconnected(Env.Clock.NowUs());
await ExchangeAsync(link, ct);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
@@ -300,16 +339,29 @@ public sealed class TcpEchoDialer : BlockBase
int min = (int)Spec.Int("min");
int max = (int)Spec.Int("max");
byte[] body = new byte[Math.Max(max, min)];
- using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1.0 / Spec.Int("rate_hz")), Env.Time);
- while (await timer.WaitForNextTickAsync(ct))
+
+ async ValueTask SendOneAsync()
{
int size = Math.Max(TcpWire.MinEchoBody, max > min ? Random.Shared.Next(min, max + 1) : min);
long t1 = Env.Clock.NowUs();
- uint seq = _rows.Add(size, t1);
+ uint seq = _rows.Add(size, t1, _rate?.Weight ?? 1);
body.AsSpan(0, size).Clear();
TcpWire.WriteEchoBody(body, seq, t1);
await TcpWire.WriteMessageAsync(stream, body.AsMemory(0, size), ct);
await stream.FlushAsync(ct);
+ _rate?.OnSend(t1);
+ }
+
+ if (_rate is not null)
+ {
+ await PaceAsync(_rate, SendOneAsync, ct);
+ return;
+ }
+
+ using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1.0 / Spec.Int("rate_hz")), Env.Time);
+ while (await timer.WaitForNextTickAsync(ct))
+ {
+ await SendOneAsync();
}
}
@@ -324,9 +376,9 @@ public sealed class TcpEchoDialer : BlockBase
}
long now = Env.Clock.NowUs();
- if (TcpWire.TryReadEchoBody(reply, out uint seq, out _))
+ if (TcpWire.TryReadEchoBody(reply, out uint seq, out _) && _rows.Reply(seq, now, out long t1))
{
- _rows.Reply(seq, now);
+ _rate?.OnReply(t1, now, (now - t1) / 1000.0);
}
}
}
diff --git a/backend/shared/MoShiro.Measurement/Core/AdaptiveRate.cs b/backend/shared/MoShiro.Measurement/Core/AdaptiveRate.cs
new file mode 100644
index 0000000..b7eb0bc
--- /dev/null
+++ b/backend/shared/MoShiro.Measurement/Core/AdaptiveRate.cs
@@ -0,0 +1,449 @@
+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);
+}
diff --git a/backend/shared/MoShiro.Measurement/Core/Contracts.cs b/backend/shared/MoShiro.Measurement/Core/Contracts.cs
index 177d503..dd1fd9d 100644
--- a/backend/shared/MoShiro.Measurement/Core/Contracts.cs
+++ b/backend/shared/MoShiro.Measurement/Core/Contracts.cs
@@ -186,6 +186,8 @@ public sealed class BlockSpec
public long Int(string name) => Params.GetInt(name);
+ public double Number(string name) => Params.GetNumber(name);
+
public bool Bool(string name) => Params.GetBool(name);
/// 参数或端口变化时调和器重启这个积木（不含流编号：Agent 的流编号每次重建都会重新分配）。
@@ -221,6 +223,9 @@ public sealed class BlockEnv
public QuicDatagramTransport? Datagrams { get; init; }
+ /// 拨号方：这条线路在每个时刻（单调时钟，微秒）丢包的历史概率（0 到 1），自适应探测用它在高发时段提前升频；没有历史为 null。
+ public Func? LossPrior { get; init; }
+
/// 监听方的准入策略（限速等）；拨号方为 null。
public IAdmission? Admission { get; init; }
@@ -257,6 +262,7 @@ public sealed class BlockEnv
Log = Log,
Session = () => session,
Datagrams = Datagrams,
+ LossPrior = LossPrior,
Admission = Admission,
Counters = Counters,
TcpListen = TcpListen,
diff --git a/backend/shared/MoShiro.Measurement/Execution/BudgetGuard.cs b/backend/shared/MoShiro.Measurement/Execution/BudgetGuard.cs
index 934e6a9..06bd324 100644
--- a/backend/shared/MoShiro.Measurement/Execution/BudgetGuard.cs
+++ b/backend/shared/MoShiro.Measurement/Execution/BudgetGuard.cs
@@ -148,7 +148,7 @@ public sealed class BudgetGuard
private static long EstimatePps(string kindId, ParamValues values) => kindId switch
{
- "udp_echo" or "tcp_echo" => values.GetInt("rate_hz") * 2,
+ "udp_echo" or "tcp_echo" => Kinds.PeakRateHz(kindId, values) * 2,
"icmp_echo" => values.GetInt("rate_hz") * 2,
"udp_train" => (long)(values.GetNumber("mbps") * 1_000_000 / 8 / Math.Max(1, values.GetInt("size"))),
"connect" => values.GetInt("per_minute") * 6 / 60,
diff --git a/backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs b/backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs
index c0984f1..f20d420 100644
--- a/backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs
+++ b/backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs
@@ -8,10 +8,14 @@ public static class Kinds
{
private static readonly string[] s_directions = ["fwd", "rev", "bidir"];
+ /// 这个任务可能达到的最高探测频率（Hz）：自适应时是 rate_max_hz，否则是固定的 rate_hz。预算、限速和准入都按它算。
+ public static long PeakRateHz(string kindId, ParamValues values) =>
+ (kindId is "udp_echo" or "tcp_echo") && values.GetBool("adaptive") ? Math.Max(values.GetInt("rate_max_hz"), values.GetInt("rate_min_hz")) : values.GetInt("rate_hz");
+
private static readonly StreamSpec s_echoMsg = new(
"msg",
[
- Col.Seq("seq"), Col.U16("size"),
+ Col.Seq("seq"), Col.U16("size"), Col.U16("w"),
Col.TsDialer("t1"), Col.TsListener("t2", nullable: true), Col.TsListener("t3", nullable: true), Col.TsDialer("t4", nullable: true),
],
Side.Dialer,
@@ -33,12 +37,17 @@ public static class Kinds
Scope.Line,
new BlockSet("quic_datagram", "fixed", "fixed", "echo", []),
[
- P.Int("rate_hz", 10, 1, 200, "Hz"),
+ P.Bool("adaptive", true, "Adapt the probe rate to the loss rate: low when the line is quiet, up to rate_max_hz when it loses packets, exponential back-off when it is unreachable."),
+ P.Int("rate_hz", 10, 1, 200, "Hz", "Fixed rate, used when adaptive is false."),
+ P.Int("rate_min_hz", 2, 1, 200, "Hz"),
+ P.Int("rate_max_hz", 10, 1, 200, "Hz", "Highest rate, and the reference grid for sample weights."),
+ P.Number("tau_pct", 0.2, 0.01, 5, "%", "Loss rate a quiet line has to be able to prove below, at 95% confidence."),
+ P.Number("rtt_tol_pct", 2, 0.1, 20, "%", "Half-width of the 95% confidence interval of the RTT p99 a quiet line has to keep, as a share of p99 (never below 0.3 ms)."),
P.Int("size", 64, 44, 1400, "B"),
P.Int("port", 7443, 1, 65535),
],
[s_echoMsg],
- p => Bandwidth.UdpEcho(p.GetInt("rate_hz"), p.GetInt("size")));
+ p => Bandwidth.UdpEcho(PeakRateHz("udp_echo", p), p.GetInt("size")));
public static readonly KindSpec UdpTrain = new(
"udp_train",
@@ -66,15 +75,20 @@ public static class Kinds
Scope.Line,
new BlockSet("tcp", "fixed", "uniform", "echo", ["tcp_info"]),
[
- P.Int("rate_hz", 20, 1, 200, "Hz"),
+ P.Bool("adaptive", true, "Adapt the echo rate: as low as the loss bound and the RTT percentile precision allow when the line is quiet, the maximum when it is not, back-off when it stalls."),
+ P.Int("rate_hz", 20, 1, 200, "Hz", "Fixed rate, used when adaptive is false."),
+ P.Int("rate_min_hz", 2, 1, 200, "Hz"),
+ P.Int("rate_max_hz", 20, 1, 200, "Hz", "Highest rate, and the reference grid for sample weights."),
+ P.Number("tau_pct", 0.2, 0.01, 5, "%", "Share of retransmitted segments a quiet line has to be able to prove below, at 95% confidence."),
+ P.Number("rtt_tol_pct", 2, 0.1, 20, "%", "Half-width of the 95% confidence interval of the RTT p99 a quiet line has to keep, as a share of p99 (never below 0.3 ms)."),
P.Int("min", 32, 16, 1400, "B"),
P.Int("max", 64, 16, 1400, "B"),
- P.Int("tcp_info_ms", 100, 10, 1000, "ms"),
+ P.Int("tcp_info_ms", 1000, 10, 1000, "ms"),
P.Int("port", 7444, 1, 65535),
P.Bool("tls", false),
],
[s_echoMsg, s_tcpInfo],
- p => Bandwidth.TcpEcho(p.GetInt("rate_hz"), p.GetInt("max")));
+ p => Bandwidth.TcpEcho(PeakRateHz("tcp_echo", p), p.GetInt("max")));
public static readonly KindSpec TcpBulk = new(
"tcp_bulk",
diff --git a/backend/shared/tests/MoShiro.Measurement.Tests/AdaptiveRateTests.cs b/backend/shared/tests/MoShiro.Measurement.Tests/AdaptiveRateTests.cs
new file mode 100644
index 0000000..c908403
--- /dev/null
+++ b/backend/shared/tests/MoShiro.Measurement.Tests/AdaptiveRateTests.cs
@@ -0,0 +1,175 @@
+using MoShiro.Measurement;
+
+namespace MoShiro.Measurement.Tests;
+
+public sealed class AdaptiveRateTests
+{
+ private const long Second = 1_000_000;
+
+ /// 按当前频率发包、按 决定回不回，每秒评估一次；返回结束时的秒数。
+ private static long Run(AdaptiveRate rate, long fromSecond, int seconds, Func lost, double rttMs = 20, List? log = null, bool stopAtDead = false)
+ {
+ long second = fromSecond;
+ for (int i = 0; i < seconds; i++, second++)
+ {
+ int probes = Math.Max(1, (int)Math.Round(rate.Hz));
+ if (rate.State == ProbeState.Dead)
+ {
+ probes = i % 4 == 0 ? 1 : 0; // 退避：几秒才发一个
+ }
+
+ for (int k = 0; k < probes; k++)
+ {
+ long sent = second * Second + k * Second / probes;
+ rate.OnSend(sent);
+ if (!lost(second))
+ {
+ rate.OnReply(sent, sent + (long)(rttMs * 1000), rttMs);
+ }
+ }
+
+ if (rate.Evaluate((second + 1) * Second) is { } change)
+ {
+ log?.Add(change);
+ if (stopAtDead && change.To == ProbeState.Dead)
+ {
+ return second + 1;
+ }
+ }
+ }
+
+ return second;
+ }
+
+ [Fact]
+ public void QuietLine_AfterWarmUp_ProbesJustEnoughToProveTheLossBound()
+ {
+ var rate = new AdaptiveRate(new AdaptiveRateOptions());
+ long at = Run(rate, 0, 600, _ => false);
+ Assert.Equal(ProbeState.Quiet, rate.State);
+ Assert.Equal(5, rate.Hz, 0.01); // 3 / (0.2% × 300 s)
+ Assert.True(at > 0);
+ }
+
+ [Fact]
+ public void SustainedLoss_RaisesToTheMaximum_AndComesBackOnlyAfterTheHoldTimes()
+ {
+ var rate = new AdaptiveRate(new AdaptiveRateOptions());
+ long at = Run(rate, 0, 600, _ => false);
+ at = Run(rate, at, 60, second => second % 10 == 0); // 每 10 秒丢一整秒的包
+ Assert.Equal(ProbeState.Active, rate.State);
+ Assert.Equal(10, rate.Hz);
+
+ at = Run(rate, at, 800, _ => false); // 不到 15 分钟：还在 Active
+ Assert.Equal(ProbeState.Active, rate.State);
+ at = Run(rate, at, 400, _ => false); // 超过 15 分钟：降到 Watch
+ Assert.Equal(ProbeState.Watch, rate.State);
+ Run(rate, at, 1900, _ => false); // 最后一次丢包 30 分钟之后：回到 Quiet
+ Assert.Equal(ProbeState.Quiet, rate.State);
+ }
+
+ [Fact]
+ public void AnIsolatedLoss_OnlyRaisesToWatch()
+ {
+ var rate = new AdaptiveRate(new AdaptiveRateOptions());
+ long at = Run(rate, 0, 600, _ => false);
+ var log = new List();
+ Run(rate, at, 120, second => second == at + 10, log: log);
+ Assert.Equal(ProbeState.Watch, rate.State);
+ Assert.DoesNotContain(log, change => change.To == ProbeState.Active);
+ }
+
+ [Fact]
+ public void Unreachable_BacksOffExponentially_AndRecoversOnTheFirstReply()
+ {
+ var rate = new AdaptiveRate(new AdaptiveRateOptions(), new Random(1));
+ long at = Run(rate, 0, 600, _ => false);
+ at = Run(rate, at, 60, _ => true, stopAtDead: true);
+ Assert.Equal(ProbeState.Dead, rate.State);
+
+ var delays = new List();
+ long now = at * Second;
+ for (int i = 0; i < 8; i++)
+ {
+ double delay = rate.NextDelay().TotalSeconds;
+ delays.Add(delay);
+ now += (long)(delay * Second);
+ rate.OnSend(now);
+ }
+
+ Assert.InRange(delays[0], 0.8, 1.2); // 从 1 秒开始
+ Assert.True(delays[3] > delays[0] * 4, string.Join(",", delays)); // 翻倍增长（有抖动）
+ Assert.All(delays, delay => Assert.InRange(delay, 0.8, 36.0)); // 最长 30 秒，抖动 ±20%
+
+ rate.OnReply(now, now + 30_000, 30);
+ Assert.Equal(ProbeState.Watch, rate.Evaluate(now + 2 * Second)!.To);
+ }
+
+ [Fact]
+ public void QuietLineWithTightRtt_MayDropBelowTheGrid_ButAJitteryOneStaysAtTheMaximum()
+ {
+ var tight = new AdaptiveRate(new AdaptiveRateOptions());
+ Run(tight, 0, 1200, _ => false, rttMs: 20); // RTT 恒定：p99 的置信区间几乎为零，降到丢包下界 5 Hz
+ Assert.Equal(ProbeState.Quiet, tight.State);
+ Assert.Equal(5, tight.Hz, 0.01);
+
+ var jittery = new AdaptiveRate(new AdaptiveRateOptions());
+ var random = new Random(7);
+ long second = 0;
+ for (int i = 0; i < 1200; i++, second++)
+ {
+ int probes = Math.Max(1, (int)Math.Round(jittery.Hz));
+ for (int k = 0; k < probes; k++)
+ {
+ long sent = second * Second + k * Second / probes;
+ double rtt = 20 + random.NextDouble() * random.NextDouble() * 60; // 长尾 20 到 80 ms
+ jittery.OnSend(sent);
+ jittery.OnReply(sent, sent + (long)(rtt * 1000), rtt);
+ }
+
+ jittery.Evaluate((second + 1) * Second);
+ }
+
+ Assert.Equal(10, jittery.Hz, 0.01); // 分位数精度保护：抖动大，样本不能少
+ }
+
+ [Fact]
+ public void Weight_IsTheTimeAProbeStandsFor_InUnitsOfTheGrid()
+ {
+ var rate = new AdaptiveRate(new AdaptiveRateOptions());
+ Run(rate, 0, 1200, _ => false);
+ Assert.Equal(5, rate.Hz, 0.01);
+ rate.NextDelay();
+ Assert.Equal(2, rate.Weight); // 5 Hz 的一个包代表 2 个 10 Hz 网格间隔
+ }
+
+ [Fact]
+ public void RttWellAboveTheBaseline_RaisesToWatch()
+ {
+ var rate = new AdaptiveRate(new AdaptiveRateOptions());
+ long at = Run(rate, 0, 900, _ => false, rttMs: 20);
+ Assert.Equal(ProbeState.Quiet, rate.State);
+ Run(rate, at, 30, _ => false, rttMs: 120);
+ Assert.Equal(ProbeState.Watch, rate.State);
+ }
+
+ [Fact]
+ public void Prior_WarmsUpBeforeTheEveningPeak_AndLetsHistoricallyQuietHoursUseTheLowestRate()
+ {
+ const long peakStart = 1_000 * 3600; // 秒
+ double? Prior(long us)
+ {
+ long second = us / Second;
+ return second >= peakStart && second < peakStart + 4 * 3600 ? 0.8 : 0.0;
+ }
+
+ var rate = new AdaptiveRate(new AdaptiveRateOptions { Prior = Prior });
+ long at = Run(rate, 100_000, 1200, _ => false); // 远离高峰，历史平稳：证明窗口拉长，2 Hz
+ Assert.Equal(ProbeState.Quiet, rate.State);
+ Assert.Equal(2, rate.Hz, 0.01);
+
+ Run(rate, peakStart - 1500, 120, _ => false); // 距离高峰不到半小时：提前升到 Active
+ Assert.Equal(ProbeState.Active, rate.State);
+ Assert.True(at > 0);
+ }
+}
diff --git a/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs b/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs
index e7c654a..64b3e46 100644
--- a/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs
+++ b/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs
@@ -65,7 +65,7 @@ public sealed class MeasurementTests
{
StreamSpec msg = Kinds.UdpEcho.FindStream("msg")!;
var dialer = new RowStore(msg, Side.Dialer, 64);
- Assert.Equal(["size", "t1", "t4"], msg.ColumnsOf(Side.Dialer).Select(c => c.Name));
+ Assert.Equal(["size", "w", "t1", "t4"], msg.ColumnsOf(Side.Dialer).Select(c => c.Name));
Assert.Equal(["t2", "t3"], msg.ColumnsOf(Side.Listener).Select(c => c.Name));
for (int i = 0; i < 4; i++)
@@ -78,7 +78,7 @@ public sealed class MeasurementTests
dialer.Set(2, dialer.ColumnIndex("t4"), 5000);
RawStream cut = dialer.Cut(1, 4, includeKey: true);
Assert.Equal(3, cut.Count);
- Assert.Equal(["seq", "size", "t1", "t4"], cut.Columns.Select(c => c.Name));
+ Assert.Equal(["seq", "size", "w", "t1", "t4"], cut.Columns.Select(c => c.Name));
Assert.Equal([1L, 2L, 3L], cut.Find("seq")!.Values);
Assert.Equal([false, true, false], cut.Find("t4")!.Valid!);
@@ -111,10 +111,10 @@ public sealed class MeasurementTests
private static RawChunk EchoFragment(params (uint Seq, long T1, long? T4)[] rows)
{
- var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.TsDialer("t1"), Col.TsDialer("t4", nullable: true)]));
+ var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.U16("w"), Col.TsDialer("t1"), Col.TsDialer("t4", nullable: true)]));
foreach ((uint seq, long t1, long? t4) in rows)
{
- msg.Add(seq, 64, t1, t4);
+ msg.Add(seq, 64, 1, t1, t4);
}
return new RawChunk(new RawChunkHeader { Kind = "udp_echo", Clock = new ChunkClock { DialerOffsetUs = 1000 } }, [msg.Build()]);
@@ -131,7 +131,7 @@ public sealed class MeasurementTests
Result> merged = FragmentMerger.Merge(Kinds.UdpEcho, fragment, 1, source);
RawStream msg = Assert.Single(merged.Value);
- Assert.Equal(["seq", "size", "t1", "t2", "t3", "t4"], msg.Columns.Select(c => c.Name));
+ Assert.Equal(["seq", "size", "w", "t1", "t2", "t3", "t4"], msg.Columns.Select(c => c.Name));
Assert.Equal([true, false, true], Enumerable.Range(0, 3).Select(i => msg.Find("t2")!.IsValid(i)));
Assert.Equal(2050, msg.Find("t3")!.Values[0]);
Assert.False(msg.Find("t4")!.IsValid(1));
diff --git a/backend/shared/tests/MoShiro.Protocol.Tests/RawChunkCodecTests.cs b/backend/shared/tests/MoShiro.Protocol.Tests/RawChunkCodecTests.cs
index f7bb2d7..54015b3 100644
--- a/backend/shared/tests/MoShiro.Protocol.Tests/RawChunkCodecTests.cs
+++ b/backend/shared/tests/MoShiro.Protocol.Tests/RawChunkCodecTests.cs
@@ -92,6 +92,7 @@ public sealed class RawChunkCodecTests
public void GoldenBody_LayoutIsStable()
{
// 布局变化（列编码、头部字段、字节序）会让这个测试失败：必须同时升级 container_version 并更新黄金字节。
+ // 只是回显流多了一列权重 w（块头自描述新增列，不用升级 container_version），黄金字节里多出六个 01。
RawChunk chunk = SampleChunk();
chunk.Header.Clock = new ChunkClock { ListenerMonoUs = 1000, UtcMs = 1_700_000_000_000, DialerOffsetUs = 500 };
byte[] body = RawChunkCodec.EncodeBody(chunk);
@@ -104,7 +105,7 @@ public sealed class RawChunkCodecTests
Assert.Equal(GoldenColumns, columns);
}
- private const string GoldenColumns = "00020202020240404040404080fbf3b407c09a0cc09a0cc09a0cc09a0cc09a0c3bc0e0d8b907889c0c90b818889c0c889c0c3ba4e1d8b907889c0c90b818889c0c889c0c2bb4fcf8b407889c0c90b81890b818";
+ private const string GoldenColumns = "00020202020240404040404001010101010180fbf3b407c09a0cc09a0cc09a0cc09a0cc09a0c3bc0e0d8b907889c0c90b818889c0c889c0c3ba4e1d8b907889c0c90b818889c0c889c0c2bb4fcf8b407889c0c90b81890b818";
[Fact]
public void Decode_RejectsBadMagicVersionCrcAndTruncation()
diff --git a/backend/src/MoShiro.Backend.Analysis/Engine/ChunkView.cs b/backend/src/MoShiro.Backend.Analysis/Engine/ChunkView.cs
index 0d3cd55..fa466d1 100644
--- a/backend/src/MoShiro.Backend.Analysis/Engine/ChunkView.cs
+++ b/backend/src/MoShiro.Backend.Analysis/Engine/ChunkView.cs
@@ -33,5 +33,8 @@ public sealed class ChunkView
public double ParamNumber(string name, double fallback) =>
Header.Params.TryGetValue(name, out JsonElement v) && v.ValueKind == JsonValueKind.Number && v.TryGetDouble(out double d) ? d : fallback;
+ public bool ParamBool(string name, bool fallback) =>
+ Header.Params.TryGetValue(name, out JsonElement v) && v.ValueKind is JsonValueKind.True or JsonValueKind.False ? v.GetBoolean() : fallback;
+
private static long FloorDiv(long a, long b) => (a / b) - ((a % b != 0 && (a < 0) != (b < 0)) ? 1 : 0);
}
diff --git a/backend/src/MoShiro.Backend.Analysis/Ops/EchoOps.cs b/backend/src/MoShiro.Backend.Analysis/Ops/EchoOps.cs
index 29747ec..0ddfc9d 100644
--- a/backend/src/MoShiro.Backend.Analysis/Ops/EchoOps.cs
+++ b/backend/src/MoShiro.Backend.Analysis/Ops/EchoOps.cs
@@ -26,6 +26,7 @@ internal readonly struct EchoColumns
T2 = stream.Find("t2");
T3 = stream.Find("t3");
T4 = stream.Find("t4");
+ W = stream.Find("w");
}
public int Count { get; }
@@ -40,6 +41,11 @@ internal readonly struct EchoColumns
public RawColumn? T4 { get; }
+ /// 每个包的权重（它代表多少个基础网格间隔）；旧数据和固定频率的任务没有这一列，按 1 算。
+ public RawColumn? W { get; }
+
+ public int Weight(int i) => W is null || !W.IsValid(i) ? 1 : (int)Math.Clamp(W.Values[i], 1, 65535);
+
public bool HasBasics => T1 is not null && T4 is not null;
/// 行 i 的发送时刻（UTC 毫秒）。
@@ -75,7 +81,7 @@ internal readonly struct EchoColumns
internal sealed class RttOp : MetricOp
{
- public override string Signature => "rtt/1";
+ public override string Signature => "rtt/2";
public override void Evaluate(ChunkView view, RawStream stream, IObservationSink sink)
{
@@ -100,7 +106,11 @@ internal sealed class RttOp : MetricOp
if (rttUs >= 0)
{
- sink.Emit(c.SentUtcMs(view, i), rttUs / 1000.0);
+ // 权重 k：这个样本代表 k 个网格间隔，按 k 次累计，分位数就是按时间平均的
+ for (int k = c.Weight(i); k > 0; k--)
+ {
+ sink.Emit(c.SentUtcMs(view, i), rttUs / 1000.0);
+ }
}
}
}
@@ -160,7 +170,7 @@ internal sealed class SlowShareOp : MetricOp
double threshold = Math.Max(baseline * 1.8, baseline + Math.Max(15, baseline * 0.35));
foreach ((int row, double ms) in rtts)
{
- sink.Emit(c.SentUtcMs(view, row), ms >= threshold ? 1 : 0, 1);
+ sink.Emit(c.SentUtcMs(view, row), ms >= threshold ? c.Weight(row) : 0, c.Weight(row));
}
}
}
@@ -174,7 +184,7 @@ internal sealed class LossOp : MetricOp
_dir = dir;
}
- public override string Signature => "loss/" + _dir + "/1";
+ public override string Signature => "loss/" + _dir + "/2";
public override void Evaluate(ChunkView view, RawStream stream, IObservationSink sink)
{
@@ -194,18 +204,18 @@ internal sealed class LossOp : MetricOp
switch (_dir)
{
case Dir.Up:
- sink.Emit(c.SentUtcMs(view, i), c.IsUpLost(i) ? 1 : 0, 1);
+ sink.Emit(c.SentUtcMs(view, i), c.IsUpLost(i) ? c.Weight(i) : 0, c.Weight(i));
break;
case Dir.Down:
// 只统计上行已经到达 Agent 的包：回复没回来就是下行丢失。
if (!c.IsUpLost(i))
{
- sink.Emit(c.SentUtcMs(view, i), c.IsReplyLost(i) ? 1 : 0, 1);
+ sink.Emit(c.SentUtcMs(view, i), c.IsReplyLost(i) ? c.Weight(i) : 0, c.Weight(i));
}
break;
default:
- sink.Emit(c.SentUtcMs(view, i), c.IsReplyLost(i) ? 1 : 0, 1);
+ sink.Emit(c.SentUtcMs(view, i), c.IsReplyLost(i) ? c.Weight(i) : 0, c.Weight(i));
break;
}
}
@@ -452,7 +462,7 @@ internal sealed class GapOp : MetricOp
_thresholdMs = thresholdMs;
}
- public override string Signature => $"gap/{_output}/{_rateParam}/{_thresholdMs}/1";
+ public override string Signature => $"gap/{_output}/{_rateParam}/{_thresholdMs}/2";
public override void Evaluate(ChunkView view, RawStream stream, IObservationSink sink)
{
@@ -462,11 +472,13 @@ internal sealed class GapOp : MetricOp
return;
}
- double rate = Math.Max(0.1, view.ParamNumber(_rateParam, 1));
- double expectedMs = 1000.0 / rate;
+ // 自适应频率时，每个包的权重就是它和上一个包的间隔（以 1/rate_max_hz 为单位），期望间隔按它算
+ bool adaptive = view.ParamBool("adaptive", false);
+ double rate = Math.Max(0.1, view.ParamNumber(adaptive ? "rate_max_hz" : _rateParam, 1));
int[] order = c.ArrivalOrder();
for (int i = 1; i < order.Length; i++)
{
+ double expectedMs = 1000.0 * (adaptive ? c.Weight(order[i]) : 1) / rate;
double gapMs = (c.T4!.Values[order[i]] - c.T4.Values[order[i - 1]]) / 1000.0;
if (gapMs - expectedMs <= _thresholdMs)
{
diff --git a/backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs b/backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs
index 9da06ce..dfbe7e4 100644
--- a/backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs
+++ b/backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs
@@ -31,6 +31,25 @@ public sealed class OpTests
Assert.InRange(rtt.Quantile(0.5)!.Value, 50 * 0.955, 50 * 1.045);
}
+ [Fact]
+ public void SampleWeights_KeepRttQuantilesAndLossTimeAveraged_WhenTheProbeRateChanges()
+ {
+ static EchoPacket At(int i) => new(LostUp: i % 20 < 2, LostDown: false, i % 10 < 2 ? 100_000 : 20_000, 20_000);
+
+ // 每 100 ms 一个包，每个权重 1；对照：只取偶数行，每 200 ms 一个包，每个权重 2——代表同样长的时间
+ RawChunk dense = ChunkFactory.Echo(s_agent, "udp_echo", T0, 100, 100, At);
+ RawChunk sparse = ChunkFactory.Echo(s_agent, "udp_echo", T0, 50, 200, i => At(i * 2), weight: _ => 2);
+
+ MetricSet d = EvaluateOneBucket(dense, "rtt", "loss.up", "loss.total");
+ MetricSet s = EvaluateOneBucket(sparse, "rtt", "loss.up", "loss.total");
+ var dr = (HistogramAcc)d.Items["rtt"];
+ var sr = (HistogramAcc)s.Items["rtt"];
+ Assert.Equal(dr.Count, sr.Count); // 稀疏的 45 个回复各按权重 2 累计，和密集的 90 个一样
+ Assert.Equal(dr.Quantile(0.5), sr.Quantile(0.5));
+ Assert.Equal(dr.Quantile(0.95), sr.Quantile(0.95));
+ Assert.Equal(((RatioAcc)d.Items["loss.up"]).Percent, ((RatioAcc)s.Items["loss.up"]).Percent);
+ }
+
[Fact]
public void Loss_SeparatesUpAndDown()
{
diff --git a/backend/tests/MoShiro.TestSupport/ChunkFactory.cs b/backend/tests/MoShiro.TestSupport/ChunkFactory.cs
index cf41653..9011337 100644
--- a/backend/tests/MoShiro.TestSupport/ChunkFactory.cs
+++ b/backend/tests/MoShiro.TestSupport/ChunkFactory.cs
@@ -48,7 +48,8 @@ public static class ChunkFactory
Func packet,
int durationMs = 10_000,
IReadOnlyDictionary? parameters = null,
- IEnumerable? extraStreams = null)
+ IEnumerable? extraStreams = null,
+ Func? weight = null)
{
var msg = new StreamBuilder(Kinds.Find(kind)!.FindStream("msg")!);
for (int i = 0; i < count; i++)
@@ -58,7 +59,7 @@ public static class ChunkFactory
long t2 = t1 + DialerOffsetUs + p.OwdUpUs;
long t3 = t2 + p.ProcessUs;
long t4 = t3 - DialerOffsetUs + p.OwdDownUs;
- msg.Add(i, 64, t1, p.LostUp ? null : t2, p.LostUp ? null : t3, p.LostUp || p.LostDown ? null : t4);
+ msg.Add(i, 64, weight?.Invoke(i) ?? 1, t1, p.LostUp ? null : t2, p.LostUp ? null : t3, p.LostUp || p.LostDown ? null : t4);
}
var streams = new List { msg.Build() };
diff --git a/backend/tests/MoShiro.TestSupport/DialerSimulator.cs b/backend/tests/MoShiro.TestSupport/DialerSimulator.cs
index e45dcb0..64ae5f0 100644
--- a/backend/tests/MoShiro.TestSupport/DialerSimulator.cs
+++ b/backend/tests/MoShiro.TestSupport/DialerSimulator.cs
@@ -160,10 +160,10 @@ public sealed class DialerSimulator : IAsyncDisposable
_echoes.TryRemove(task.StreamId, out _);
_quic.Unregister(task.StreamId);
- var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.TsDialer("t1"), Col.TsDialer("t4", nullable: true)]));
+ var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.U16("w"), Col.TsDialer("t1"), Col.TsDialer("t4", nullable: true)]));
foreach ((uint seq, long t1) in collector.Sent.OrderBy(p => p.Key))
{
- msg.Add(seq, size, t1, collector.Received.TryGetValue(seq, out long t4) ? t4 : null);
+ msg.Add(seq, size, 1, t1, collector.Received.TryGetValue(seq, out long t4) ? t4 : null);
}
RawChunkHeader header = Header("udp_echo", task.Id, null, null, startUtc, (int)Math.Max(1, count * interval.TotalMilliseconds), task.Params);
@@ -195,7 +195,7 @@ public sealed class DialerSimulator : IAsyncDisposable
await stream.WriteAsync(auth, ct);
long startUtc = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds();
- var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.TsDialer("t1"), Col.TsDialer("t4", nullable: true)]));
+ var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.U16("w"), Col.TsDialer("t1"), Col.TsDialer("t4", nullable: true)]));
int received = 0;
for (uint seq = 0; seq < count; seq++)
{
@@ -207,7 +207,7 @@ public sealed class DialerSimulator : IAsyncDisposable
byte[]? reply = await TcpWire.ReadMessageAsync(stream, ct);
long? t4 = reply is not null && TcpWire.TryReadEchoBody(reply, out uint echoed, out _) && echoed == seq ? NowUs() : null;
received += t4 is null ? 0 : 1;
- msg.Add(seq, body.Length, t1, t4);
+ msg.Add(seq, body.Length, 1, t1, t4);
await Task.Delay(interval, ct);
}