From 0fd38b23671cd184b7f61b07b7491e7f7a8a2810 Mon Sep 17 00:00:00 2001 From: justme Date: Tue, 20 Oct 2026 18:23:31 +0800 Subject: [PATCH] fix: prevent QUIC heartbeat stalls and isolate TCP retransmission counters [CI] [SKIP TESTS] --- backend/probe/src/MoShiro.Probe/ProbeApp.cs | 5 +- .../Net/MsQuicConnection.cs | 80 +++++++++++++++---- .../Net/QuicDatagramTransport.cs | 14 +--- .../shared/MoShiro.Measurement/Net/TcpInfo.cs | 10 ++- .../MoShiro.Protocol/Catalog/Kinds/Kinds.cs | 1 + .../shared/MoShiro.Protocol/Wire/TcpWire.cs | 18 ++++- .../MeasurementTests.cs | 22 ++++- .../MsQuicConnectionTests.cs | 77 ++++++++++++++++++ .../MoShiro.Protocol.Tests/TcpWireTests.cs | 30 +++++++ .../Ops/SessionOps.cs | 32 +++++--- .../MoShiro.Backend.Analysis.Tests/OpTests.cs | 49 ++++++++++++ .../MoShiro.TestSupport/AgentSimulator.cs | 5 ++ .../tests/MoShiro.TestSupport/ChunkFactory.cs | 2 +- docs/design/measurement.md | 4 + 14 files changed, 299 insertions(+), 50 deletions(-) create mode 100644 backend/shared/tests/MoShiro.Measurement.Tests/MsQuicConnectionTests.cs create mode 100644 backend/shared/tests/MoShiro.Protocol.Tests/TcpWireTests.cs diff --git a/backend/probe/src/MoShiro.Probe/ProbeApp.cs b/backend/probe/src/MoShiro.Probe/ProbeApp.cs index 9d85ca8..29ffc4b 100644 --- a/backend/probe/src/MoShiro.Probe/ProbeApp.cs +++ b/backend/probe/src/MoShiro.Probe/ProbeApp.cs @@ -171,10 +171,13 @@ public sealed class ProbeApp : IDisposable private List SnapshotQuicLinks() { + AgentConnection[] connections; lock (_connections) { - return [.. _connections.Values.Select(r => r.Connection.QuicHealth())]; + connections = [.. _connections.Values.Select(r => r.Connection)]; } + + return [.. connections.Select(c => c.QuicHealth())]; } public async Task RunAsync(CancellationToken ct) diff --git a/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs b/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs index e4df037..40a1d01 100644 --- a/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs +++ b/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs @@ -33,6 +33,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection private static readonly MsQuicApi.ConnectionCallback s_callback = OnEvent; private readonly object _sendGate = new(); + private readonly object _nativeGate = new(); private readonly TaskCompletionSource _connected = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource _shutdown = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly GCHandle _self; @@ -48,6 +49,8 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection private int _sendEnabled; private int _disposeStarted; private double? _handshakeDurationMs; + private QuicTransportSnapshot? _transportSnapshot; + private long _statisticsAt; private MsQuicConnection( nint handle, @@ -105,24 +108,33 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection public double? HandshakeDurationMs => _handshakeDurationMs; - public bool TryGetStatistics(out MsQuicApi.QuicStatisticsV2 statistics) + public QuicTransportSnapshot? TransportSnapshot => Volatile.Read(ref _transportSnapshot); + + // Called only from this connection's serialized native callbacks. GetParam is + // inline on the MsQuic worker; calling it from a heartbeat can wait forever + // for a stalled worker and take the entire Fleet heartbeat down with it. + private void UpdateTransportSnapshot() { nint handle = Volatile.Read(ref _handle); - if (handle == 0) + long now = Stopwatch.GetTimestamp(); + if (handle == 0 || (_statisticsAt != 0 && Stopwatch.GetElapsedTime(_statisticsAt, now) < TimeSpan.FromSeconds(1))) { - statistics = default; - return false; + return; } + _statisticsAt = now; try { - statistics = MsQuicApi.GetStatisticsV2(handle); - return true; + MsQuicApi.QuicStatisticsV2 stats = MsQuicApi.GetStatisticsV2(handle); + ulong lost = stats.SendSuspectedLostPackets >= stats.SendSpuriousLostPackets + ? stats.SendSuspectedLostPackets - stats.SendSpuriousLostPackets : 0; + double? loss = stats.SendTotalPackets > 0 ? lost * 100.0 / stats.SendTotalPackets : null; + Volatile.Write(ref _transportSnapshot, new QuicTransportSnapshot(loss, stats.SendTotalPackets, + stats.SendCongestionCount, stats.SendEcnCongestionCount, stats.SendPathMtu)); } catch (InvalidOperationException) { - statistics = default; - return false; + // Some event phases do not yet expose statistics. Retain the last snapshot. } } @@ -214,12 +226,23 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection var connection = new MsQuicConnection(handle, 0, 0, null, datagramHandler, log, closed, peerAddressChanged); MsQuicApi.SetContext(handle, GCHandle.ToIntPtr(connection._self)); MsQuicApi.SetCallbackHandler(handle, s_callback, GCHandle.ToIntPtr(connection._self)); - MsQuicApi.ConnectionSetConfiguration(handle, configuration); MsQuicApi.SetParam(handle, DatagramReceiveEnabled, 1); + // Transport parameters must be set before SetConfiguration starts TLS. + MsQuicApi.ConnectionSetConfiguration(handle, configuration); return connection; } public bool Send(ReadOnlySpan datagram) + { + // ConnectionClose must not free the handle or a pending send while + // its native buffer is still being prepared/submitted. + lock (_nativeGate) + { + return SendCore(datagram); + } + } + + private bool SendCore(ReadOnlySpan datagram) { nint handle = Volatile.Read(ref _handle); if (handle == 0 || !IsReady || datagram.Length > MaxDatagramBytes) @@ -264,9 +287,12 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection public async ValueTask DisposeAsync() { - if (Interlocked.Exchange(ref _disposeStarted, 1) == 0 && Volatile.Read(ref _handle) is nint handle && handle != 0) + lock (_nativeGate) { - MsQuicApi.ConnectionShutdown(handle); + if (Interlocked.Exchange(ref _disposeStarted, 1) == 0 && Volatile.Read(ref _handle) is nint handle && handle != 0) + { + MsQuicApi.ConnectionShutdown(handle); + } } if (Volatile.Read(ref _handle) != 0) @@ -396,6 +422,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection return connection.ValidatePeerCertificate(eventPointer); } + connection.UpdateTransportSnapshot(); return 0; } catch (Exception ex) @@ -418,10 +445,23 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection private void OnDatagramStateChanged(nint eventPointer) { - byte sendEnabled = Marshal.ReadByte(eventPointer, IntPtr.Size); - int maxSendLength = (ushort)Marshal.ReadInt16(eventPointer, IntPtr.Size + 1); + (bool sendEnabled, int maxSendLength) = ReadDatagramState(eventPointer); Volatile.Write(ref _maxDatagramBytes, maxSendLength); - Volatile.Write(ref _sendEnabled, sendEnabled == 0 ? 0 : 1); + Volatile.Write(ref _sendEnabled, sendEnabled ? 1 : 0); + } + + internal static (bool SendEnabled, int MaxSendLength) ReadDatagramState(nint eventPointer) + { + // BOOLEAN is followed by padding before the naturally aligned uint16_t. + var state = Marshal.PtrToStructure(eventPointer + IntPtr.Size); + return (state.SendEnabled != 0, state.MaxSendLength); + } + + [StructLayout(LayoutKind.Sequential)] + private struct DatagramState + { + public byte SendEnabled; + public ushort MaxSendLength; } private void OnDatagramReceived(nint eventPointer) @@ -554,7 +594,17 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection return; } - MsQuicApi.ConnectionClose(handle); + // RegistrationClose waits for native workers/children and is forbidden + // on a MsQuic callback. Return from the callback before waiting for cleanup. + ThreadPool.QueueUserWorkItem(_ => CloseHandles(handle)); + } + + private void CloseHandles(nint handle) + { + lock (_nativeGate) + { + MsQuicApi.ConnectionClose(handle); + } lock (_sendGate) { foreach ((nint context, SendState state) in _sends) diff --git a/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs b/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs index bfe4bc5..59ea32c 100644 --- a/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs +++ b/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs @@ -60,19 +60,7 @@ public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable public double? HandshakeDurationMs => _client?.HandshakeDurationMs; - public QuicTransportSnapshot? GetTransportSnapshot() - { - if (_client is null || !_client.TryGetStatistics(out MsQuicApi.QuicStatisticsV2 stats)) - { - return null; - } - - ulong lost = stats.SendSuspectedLostPackets >= stats.SendSpuriousLostPackets - ? stats.SendSuspectedLostPackets - stats.SendSpuriousLostPackets - : 0; - double? packetLossRate = stats.SendTotalPackets > 0 ? lost * 100.0 / stats.SendTotalPackets : null; - return new QuicTransportSnapshot(packetLossRate, stats.SendTotalPackets, stats.SendCongestionCount, stats.SendEcnCongestionCount, stats.SendPathMtu); - } + public QuicTransportSnapshot? GetTransportSnapshot() => _client?.TransportSnapshot; public void Bind() { diff --git a/backend/shared/MoShiro.Measurement/Net/TcpInfo.cs b/backend/shared/MoShiro.Measurement/Net/TcpInfo.cs index 8a2be0c..c82d61a 100644 --- a/backend/shared/MoShiro.Measurement/Net/TcpInfo.cs +++ b/backend/shared/MoShiro.Measurement/Net/TcpInfo.cs @@ -70,14 +70,15 @@ public static class TcpInfo } /// -/// 一条连接上周期性的内核采样（tcp_info 流）：每个采样一行 (ts, side, srtt, rttvar, cwnd, retrans_total, segs_out, bytes_acked)。 +/// 一条连接上周期性的内核采样（tcp_info 流）：每个采样一行 (ts, side, srtt, rttvar, cwnd, retrans_total, segs_out, bytes_acked, conn)。 /// 两端各采样自己的套接字：Probe 的重传反映上行方向，Agent 的重传反映下行方向（代理场景里大流量走这个方向）。 -/// 累计计数器在 Backend 按 side 分别求差。 +/// 累计计数器在 Backend 按 (side, conn) 分别求差；conn 是当前记录器内分配的连接编号，不包含网络地址。 /// public sealed class TcpInfoRecorder { private readonly MonoClock _clock; private readonly long _side; + private long _nextConnection; public TcpInfoRecorder(StreamSpec spec, Side side, MonoClock clock) { @@ -96,6 +97,9 @@ public sealed class TcpInfoRecorder return; } + // A recorder survives reconnects, and listeners can briefly have more + // than one socket. Kernel counters belong to a socket, not just a side. + long connection = Interlocked.Increment(ref _nextConnection); try { using var timer = new PeriodicTimer(interval, _clock.Time); @@ -107,7 +111,7 @@ public sealed class TcpInfoRecorder } long key = Rows.Append(); - Rows.SetRow(key, [_clock.NowUs(), _side, s.SrttUs, s.RttVarUs, s.Cwnd, s.TotalRetrans, s.SegsOut, s.BytesAcked]); + Rows.SetRow(key, [_clock.NowUs(), _side, s.SrttUs, s.RttVarUs, s.Cwnd, s.TotalRetrans, s.SegsOut, s.BytesAcked, connection]); } } catch (OperationCanceledException) diff --git a/backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs b/backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs index 4cb7f6e..4d1eb77 100644 --- a/backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs +++ b/backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs @@ -22,6 +22,7 @@ public static class Kinds [ Col.TsLocal("ts"), Col.U8("side"), Col.U32("srtt_us"), Col.U32("rttvar_us"), Col.U32("cwnd"), Col.U32("retrans_total"), Col.U32("segs_out"), Col.U64("bytes_acked"), + new ColumnSpec("conn", ColumnType.U64, ColumnEnc.Plain, true), ], Union: true); diff --git a/backend/shared/MoShiro.Protocol/Wire/TcpWire.cs b/backend/shared/MoShiro.Protocol/Wire/TcpWire.cs index 2b18ca7..d680643 100644 --- a/backend/shared/MoShiro.Protocol/Wire/TcpWire.cs +++ b/backend/shared/MoShiro.Protocol/Wire/TcpWire.cs @@ -1,3 +1,4 @@ +using System.Buffers; using System.Buffers.Binary; using System.Security.Cryptography; using Fleet.Core; @@ -124,9 +125,18 @@ public static class TcpWire public static async ValueTask WriteMessageAsync(Stream stream, ReadOnlyMemory body, CancellationToken ct) { - byte[] prefix = new byte[VarInt.MaxBytes]; - int n = VarInt.Write(prefix, (ulong)body.Length); - await stream.WriteAsync(prefix.AsMemory(0, n), ct); - await stream.WriteAsync(body, ct); + byte[] frame = ArrayPool.Shared.Rent(checked(VarInt.MaxBytes + body.Length)); + try + { + int n = VarInt.Write(frame, (ulong)body.Length); + body.CopyTo(frame.AsMemory(n)); + // TCP_NODELAY is enabled: two writes otherwise create a tiny + // length-only segment before every echo body (and two TLS records). + await stream.WriteAsync(frame.AsMemory(0, n + body.Length), ct); + } + finally + { + ArrayPool.Shared.Return(frame); + } } } diff --git a/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs b/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs index 53bd828..de75231 100644 --- a/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs +++ b/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs @@ -203,7 +203,7 @@ public sealed class MeasurementTests var builder = new StreamBuilder(spec); foreach ((long ts, long side, long retrans) in rows) { - builder.Add(ts, side, 40_000, 5_000, 10, retrans, 100, 1_000); + builder.Add(ts, side, 40_000, 5_000, 10, retrans, 100, 1_000, null); } return builder.Build(); @@ -216,8 +216,8 @@ public sealed class MeasurementTests Assert.Equal(info.Columns.Count, info.ColumnsOf(Side.Agent).Count); // each side owns every column var agent = new RowStore(info, Side.Agent, 64); var recorder = new Source { Stores = { ["tcp_info"] = agent } }; - agent.SetRow(agent.Append(), [500, 1, 40_000, 5_000, 10, 7, 100, 1_000]); - agent.SetRow(agent.Append(), [600, 1, 40_000, 5_000, 10, 8, 120, 1_100]); + agent.SetRow(agent.Append(), [500, 1, 40_000, 5_000, 10, 7, 100, 1_000, 1]); + agent.SetRow(agent.Append(), [600, 1, 40_000, 5_000, 10, 8, 120, 1_100, 1]); var msg = EchoFragment((0, 1000, 3000)); var fragment = new RawChunk(msg.Header, [msg.Streams[0], InfoRows(info, (100, 0, 1))]); @@ -227,6 +227,8 @@ public sealed class MeasurementTests Assert.Equal(3, merged.Count); Assert.Equal([0L, 1L, 1L], merged.Find("side")!.Values); Assert.Equal([1L, 7L, 8L], merged.Find("retrans_total")!.Values); + Assert.False(merged.Find("conn")!.IsValid(0)); + Assert.Equal(1, merged.Find("conn")!.Values[1]); // The agent's rows were consumed: the next chunk carries only the probe's own rows IReadOnlyList second = FragmentMerger.Merge(Kinds.TcpEcho, fragment, 1, recorder).Value; @@ -238,7 +240,7 @@ public sealed class MeasurementTests { StreamSpec info = Kinds.TcpEcho.FindStream("tcp_info")!; var agent = new RowStore(info, Side.Agent, 64); - agent.SetRow(agent.Append(), [500, 1, 40_000, 5_000, 10, 7, 100, 1_000]); + agent.SetRow(agent.Append(), [500, 1, 40_000, 5_000, 10, 7, 100, 1_000, 1]); // A probe on a platform without kernel sampling sends the echo rows only RawChunk fragment = EchoFragment((0, 1000, 3000)); @@ -248,6 +250,18 @@ public sealed class MeasurementTests Assert.Equal(7, rows.Find("retrans_total")!.Values[0]); } + [Fact] + public void FragmentMerger_LegacyTcpSamples_KeepConnectionIdentityUnknown() + { + StreamSpec info = Kinds.TcpEcho.FindStream("tcp_info")!; + RawStream sample = InfoRows(info, (100, 0, 1)); + var legacy = new RawStream(sample.Name, sample.Count, sample.Columns.Where(c => c.Name != "conn").ToArray()); + var fragment = new RawChunk(new RawChunkHeader { Kind = "tcp_echo" }, [legacy]); + RawStream merged = Assert.Single(FragmentMerger.Merge(Kinds.TcpEcho, fragment, 1, new Source()).Value); + Assert.False(merged.Find("conn")!.IsValid(0)); + Assert.Equal(1, merged.Find("retrans_total")!.Values[0]); + } + [Fact] public void TcpInfo_OnLinux_ReadsTheKernelCountersOfALiveConnection() { diff --git a/backend/shared/tests/MoShiro.Measurement.Tests/MsQuicConnectionTests.cs b/backend/shared/tests/MoShiro.Measurement.Tests/MsQuicConnectionTests.cs new file mode 100644 index 0000000..8c22e5d --- /dev/null +++ b/backend/shared/tests/MoShiro.Measurement.Tests/MsQuicConnectionTests.cs @@ -0,0 +1,77 @@ +using System.Net; +using System.Runtime.InteropServices; +using System.Security.Cryptography; +using System.Security.Cryptography.X509Certificates; +using Fleet.Core; + +namespace MoShiro.Measurement.Tests; + +public sealed class MsQuicConnectionTests +{ + [Theory] + [InlineData(1200, true)] + [InlineData(1280, true)] + [InlineData(1472, false)] + public void DatagramState_UsesTheNativeUint16Alignment(int maximum, bool enabled) + { + nint nativeEvent = Marshal.AllocHGlobal(IntPtr.Size + 4); + try + { + Marshal.WriteByte(nativeEvent, IntPtr.Size, enabled ? (byte)1 : (byte)0); + Marshal.WriteByte(nativeEvent, IntPtr.Size + 1, 0xA5); // padding is not part of MaxSendLength + Marshal.WriteInt16(nativeEvent, IntPtr.Size + 2, (short)maximum); + Assert.Equal((enabled, maximum), MsQuicConnection.ReadDatagramState(nativeEvent)); + } + finally + { + Marshal.FreeHGlobal(nativeEvent); + } + } + + [Fact] + public async Task DatagramConnection_CanSendCloseAndReconnect_WithoutBlockingNativeWorkers() + { + using ECDsa key = ECDsa.Create(ECCurve.NamedCurves.nistP256); + var request = new CertificateRequest("CN=localhost", key, HashAlgorithmName.SHA256); + using X509Certificate2 certificate = request.CreateSelfSigned(DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch.AddYears(100)); + await using var listener = new MsQuicListener(IPAddress.Loopback, [0], certificate, + (connection, data) => connection.Send(data), NullNodeLog.Instance, null); + listener.Start(); + + for (int attempt = 0; attempt < 3; attempt++) + { + var reply = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + MsQuicConnection client = await MsQuicConnection.ConnectAsync(IPAddress.Loopback, listener.BoundPorts[0], 0, + key.ExportSubjectPublicKeyInfo(), (_, data) => reply.TrySetResult(data.ToArray()), NullNodeLog.Instance, timeout.Token); + try + { + while (!client.IsReady) + { + await Task.Delay(10, timeout.Token); + } + + Assert.InRange(client.MaxDatagramBytes, 64, 1500); + byte[] payload = Enumerable.Range(0, 64).Select(i => (byte)i).ToArray(); + // DATAGRAM is unreliable, and the peer can receive the first + // packet before its handshake-complete notification permits replies. + while (!reply.Task.IsCompleted) + { + Assert.True(client.Send(payload)); + await Task.WhenAny(reply.Task, Task.Delay(20, timeout.Token)); + timeout.Token.ThrowIfCancellationRequested(); + } + Assert.Equal(payload, await reply.Task.WaitAsync(timeout.Token)); + Assert.NotNull(client.TransportSnapshot); + } + finally + { + // The old callback called RegistrationClose, deadlocking its own + // worker. A second connection then hung in SetParam/GetParam. + await client.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(3)); + await client.WaitForShutdownAsync().WaitAsync(TimeSpan.FromSeconds(3)); + } + } + } + +} diff --git a/backend/shared/tests/MoShiro.Protocol.Tests/TcpWireTests.cs b/backend/shared/tests/MoShiro.Protocol.Tests/TcpWireTests.cs new file mode 100644 index 0000000..8d27dd1 --- /dev/null +++ b/backend/shared/tests/MoShiro.Protocol.Tests/TcpWireTests.cs @@ -0,0 +1,30 @@ +namespace MoShiro.Protocol.Tests; + +public sealed class TcpWireTests +{ + [Theory] + [InlineData(0)] + [InlineData(64)] + [InlineData(16384)] + public async Task Message_UsesOneWriteForPrefixAndBody_AndPreservesFraming(int size) + { + using var stream = new CountingStream(); + byte[] body = Enumerable.Range(0, size).Select(i => (byte)i).ToArray(); + await TcpWire.WriteMessageAsync(stream, body, default); + Assert.Equal(1, stream.Writes); + stream.Position = 0; + Assert.Equal(body, await TcpWire.ReadMessageAsync(stream, default)); + Assert.Null(await TcpWire.ReadMessageAsync(stream, default)); + } + + private sealed class CountingStream : MemoryStream + { + public int Writes { get; private set; } + + public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) + { + Writes++; + return base.WriteAsync(buffer, cancellationToken); + } + } +} diff --git a/backend/src/MoShiro.Backend.Analysis/Ops/SessionOps.cs b/backend/src/MoShiro.Backend.Analysis/Ops/SessionOps.cs index 2fd97fb..70e41e4 100644 --- a/backend/src/MoShiro.Backend.Analysis/Ops/SessionOps.cs +++ b/backend/src/MoShiro.Backend.Analysis/Ops/SessionOps.cs @@ -9,7 +9,7 @@ internal static class TcpInfoTime side is not null && side.Values[row] == 0 ? view.ProbeToUtcMs(ts.Values[row]) : view.AgentToUtcMs(ts.Values[row]); } -/// 两次采样之间累计计数的差值（分子）与另一个累计计数的差值（分母），按 side 分别计算。 +/// 两次采样之间累计计数的差值（分子）与另一个累计计数的差值（分母），按 (side, conn) 分别计算；旧数据没有 conn 时按 side 计算。 internal sealed class KernelDeltaOp : MetricOp { private readonly string _field; @@ -21,12 +21,13 @@ internal sealed class KernelDeltaOp : MetricOp _over = over; } - public override string Signature => $"kernel_delta/{_field}/{_over}/1"; + public override string Signature => $"kernel_delta/{_field}/{_over}/2"; public override void Evaluate(ChunkView view, RawStream stream, IObservationSink sink) { RawColumn? ts = stream.Find("ts"); RawColumn? side = stream.Find("side"); + RawColumn? connection = stream.Find("conn"); RawColumn? field = stream.Find(_field); RawColumn? over = stream.Find(_over); if (ts is null || field is null || over is null) @@ -34,21 +35,34 @@ internal sealed class KernelDeltaOp : MetricOp return; } - var last = new Dictionary(); + var last = new Dictionary<(long Side, long Connection), (long Ts, long Field, long Over)>(); for (int i = 0; i < stream.Count; i++) { - long key = side?.Values[i] ?? 0; - if (last.TryGetValue(key, out (long Field, long Over) previous)) + if (!ts.IsValid(i) || !field.IsValid(i) || !over.IsValid(i)) { - long dField = Math.Max(0, field.Values[i] - previous.Field); - long dOver = Math.Max(0, over.Values[i] - previous.Over); - if (dOver > 0) + continue; + } + + var key = (side?.Values[i] ?? 0, connection?.IsValid(i) == true ? connection.Values[i] : 0); + if (last.TryGetValue(key, out (long Ts, long Field, long Over) previous)) + { + // Ignore delayed/duplicate samples without replacing the newer baseline. + if (ts.Values[i] <= previous.Ts) + { + continue; + } + + // Either counter falling means reset/wrap: discard the whole + // interval rather than mixing a new numerator with an old denominator. + long dField = field.Values[i] - previous.Field; + long dOver = over.Values[i] - previous.Over; + if (dField >= 0 && dOver > 0) { sink.Emit(TcpInfoTime.ToUtcMs(view, ts, side, i), dField, dOver); } } - last[key] = (field.Values[i], over.Values[i]); + last[key] = (ts.Values[i], field.Values[i], over.Values[i]); } } } diff --git a/backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs b/backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs index 82d0002..30fa8ab 100644 --- a/backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs +++ b/backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs @@ -211,6 +211,55 @@ public sealed class OpTests Assert.Equal(100UL, retrans.Denominator); } + [Fact] + public void Retrans_InterleavedConnections_DoNotCreateFalseRetransmissions() + { + long ts = ChunkFactory.ProbeUs(0); + var builder = new StreamBuilder(Kinds.TcpEcho.FindStream("tcp_info")!); + builder.Add(ts, 0, 40_000, 0, 10, 10, 1000, 0, 1); + builder.Add(ts + 1_000_000, 0, 40_000, 0, 10, 20, 1100, 0, 2); + builder.Add(ts + 2_000_000, 0, 40_000, 0, 10, 11, 1100, 0, 1); + builder.Add(ts + 3_000_000, 0, 40_000, 0, 10, 21, 1200, 0, 2); + RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 100, + _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [builder.Build()]); + var retrans = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans").Items["tcp.retrans"]; + Assert.Equal(2UL, retrans.Numerator); + Assert.Equal(200UL, retrans.Denominator); + } + + [Fact] + public void Retrans_OutOfOrderSamples_DoNotMoveTheBaselineBackwards() + { + long ts = ChunkFactory.ProbeUs(0); + RawStream info = ChunkFactory.TcpInfo([ + (ts, 0, 40_000u, 0u, 100u), + (ts + 2_000_000, 0, 40_000u, 2u, 300u), + (ts + 1_000_000, 0, 40_000u, 1u, 200u), + (ts + 3_000_000, 0, 40_000u, 3u, 400u), + ]); + RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 100, + _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [info]); + var retrans = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans").Items["tcp.retrans"]; + Assert.Equal(3UL, retrans.Numerator); + Assert.Equal(300UL, retrans.Denominator); + } + + [Fact] + public void Retrans_EqualConnectionIdsOnDifferentSides_AreIndependent() + { + long ts = ChunkFactory.ProbeUs(0); + var builder = new StreamBuilder(Kinds.TcpEcho.FindStream("tcp_info")!); + builder.Add(ts, 0, 40_000, 0, 10, 10, 1000, 0, 1); + builder.Add(ts, 1, 40_000, 0, 10, 100, 2000, 0, 1); + builder.Add(ts + 1_000_000, 0, 40_000, 0, 10, 11, 1100, 0, 1); + builder.Add(ts + 1_000_000, 1, 40_000, 0, 10, 102, 2200, 0, 1); + RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 100, + _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [builder.Build()]); + var retrans = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans").Items["tcp.retrans"]; + Assert.Equal(3UL, retrans.Numerator); + Assert.Equal(300UL, retrans.Denominator); + } + [Fact] public void Connect_Outcomes_AreCountedByClass() { diff --git a/backend/tests/MoShiro.TestSupport/AgentSimulator.cs b/backend/tests/MoShiro.TestSupport/AgentSimulator.cs index 77fac3b..c69fbde 100644 --- a/backend/tests/MoShiro.TestSupport/AgentSimulator.cs +++ b/backend/tests/MoShiro.TestSupport/AgentSimulator.cs @@ -274,6 +274,11 @@ public sealed class AgentSimulator : IAsyncDisposable { // 已经断开 } + catch (OperationCanceledException) when (_stop.IsCancellationRequested) + { + // Canceling the receive loop may abort/dispose the transport + // between the State check and CloseOutputAsync. + } } if (_loop is not null) diff --git a/backend/tests/MoShiro.TestSupport/ChunkFactory.cs b/backend/tests/MoShiro.TestSupport/ChunkFactory.cs index 3f51c50..f093071 100644 --- a/backend/tests/MoShiro.TestSupport/ChunkFactory.cs +++ b/backend/tests/MoShiro.TestSupport/ChunkFactory.cs @@ -79,7 +79,7 @@ public static class ChunkFactory var builder = new StreamBuilder(Kinds.TcpEcho.FindStream("tcp_info")!); foreach ((long ts, byte side, uint srtt, uint retrans, uint segs) in samples) { - builder.Add(ts, side, srtt, srtt / 4, 10, retrans, segs, segs * 100L); + builder.Add(ts, side, srtt, srtt / 4, 10, retrans, segs, segs * 100L, null); } return builder.Build(); diff --git a/docs/design/measurement.md b/docs/design/measurement.md index bdebb6a..c991edf 100644 --- a/docs/design/measurement.md +++ b/docs/design/measurement.md @@ -117,6 +117,10 @@ RawChunk > **两侧各自记录完整行的流**（`StreamSpec.Union`，例如 `tcp_info`）：Probe 和 Agent 各自采样自己的套接字（Linux `TCP_INFO`，通过 `Socket.GetRawSocketOption`，不需要 P/Invoke），每一行带 `side` 标明属于哪一侧；合并时直接拼接，不连接，Agent 一侧的行每行只交出一次。Probe 的重传反映上行方向，Agent 的重传反映下行方向（代理场景里大流量走这个方向）。非 Linux 平台没有这类采样，对应的指标在窗口里就是没有数据。 +`tcp_info` 的可空 `conn` 列标识当前记录器内的一条 TCP 连接，不包含地址。Backend 按 `(side, conn)` 分别求累计计数差，避免把重连前后或同时存活的套接字混为一个序列；时间不递增的样本被忽略，任一计数器回退时重新建立基线。旧数据没有 `conn`，仍按 `side` 分组，无法事后恢复丢失的连接归属。升级时先更新 Backend 和 Agent，再更新 Probe；旧 Agent 合并采样时会丢弃尚不认识的列。 + +TCP 回显的长度头和正文合为一次写入，避免在启用 `TCP_NODELAY` 时为长度头单独发送一个小段。`tcp.retrans` 仍按内核的重传段数与发送段数求比，不由应用消息数估算。 + 采集层只产生**原始列**（`seq`、`size`、`t1…t4`、`t_send`/`t_recv`、`outcome`……），不做任何统计，也不认识上层的方案或协议；丢包率、RTT、抖动由 Backend 的分析层从这些列推算。 每个记录流（`StreamSpec`）声明：