From 8983ade8649d4f5fbbbc92bc42e91b470b7d3051 Mon Sep 17 00:00:00 2001 From: Dietmar Borgards Date: Sun, 25 Oct 2026 12:41:15 +0100 Subject: [PATCH] fix(j1939tp): order a cancelled CM abort before the next RTS --- src/CanKit.Pro.J1939Tp/J1939TpChannel.cs | 30 ++++------- src/CanKit.Pro.J1939Tp/README.md | 5 ++ .../TestCases/J1939TpTests.cs | 53 ++++++++++++++++++- 3 files changed, 67 insertions(+), 21 deletions(-) diff --git a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs index 91dc0e1a..5a8a6fa8 100644 --- a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs +++ b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs @@ -83,11 +83,10 @@ internal sealed class J1939TpChannel : IJ1939TpChannel // leaves in O(1): a list would shift every remaining entry per send drained (#204). private readonly Dictionary> _queuedNodes = new(); private readonly Dictionary _rxSessions = new(); - // CM sends cancelled while their RTS was still unconfirmed, by destination. Their Connection - // Abort waits for that confirmation, so it cannot overtake the RTS it ends, and the - // destination stays taken until the abort has been handed to the service, so the RTS of a - // send queued behind it cannot be handed over first (#356, T-7; Codex on #390). Actor-loop - // state. + // Cancelled CM sends, by destination. Their Connection Abort waits for the RTS to settle + // if necessary, and the destination stays taken until that abort has been handed to the + // service. This also orders a successor's RTS behind the abort when the cancelled RTS + // was already confirmed or answered by the peer (#356, T-7). Actor-loop state. private readonly Dictionary _abortsAwaitingRts = new(); // Sends admitted per destination and not yet completed: posted to the actor and not yet run, @@ -369,28 +368,19 @@ private void CancelTxOnLoop(TxSessionKey key, TxCompletion tcs, CancellationToke { session.Deadline?.Dispose(); session.Deadline = null; - if (session.IsCm && !session.RtsSettled) + if (session.IsCm) { - // The RTS has not been confirmed yet, so it may not have reached the driver: every - // frame goes out from its own pool work item, and an abort queued now could - // overtake it -- a peer that sees Abort, then RTS, holds a session nobody serves - // for T2 (#356, T-7). The abort goes out from the RTS's confirmation instead - // (OnRtsSettled). The send is cancelled now; the destination stays taken until - // the abort has been handed to the service, so a queued send cannot hand its RTS - // over ahead of it. + // Every control frame goes out from its own pool work item: the Abort must wait + // for an unsettled RTS, and a successor's RTS must wait for this Abort's hand-off. + // Keep the destination occupied through that hand-off even when the original + // RTS was already confirmed or answered. The caller's cancellation completes now. tcs.TrySetCanceled(token); _txSessions.Remove(key); session.AbortAwaitingRts = true; _abortsAwaitingRts[key.DestinationAddress] = session; + if (session.RtsSettled) OnRtsSettled(session); return; } - if (session.IsCm) - { - // Peer already has (or is about to have) an open CM session with us — tell it to - // tear down. Reason 2 ("resources needed for another task") is the closest - // standard code for an application-initiated local cancel. - SendTpCm(CancelAbort(key), destinationAddress: key.DestinationAddress); - } // Complete with the caller's token (do not go through TxSession.Cancel, which would // TrySetCanceled() without the token and win the race against this call). tcs.TrySetCanceled(token); diff --git a/src/CanKit.Pro.J1939Tp/README.md b/src/CanKit.Pro.J1939Tp/README.md index abb2bbee..e4f301c2 100644 --- a/src/CanKit.Pro.J1939Tp/README.md +++ b/src/CanKit.Pro.J1939Tp/README.md @@ -28,6 +28,11 @@ A TP.CM or TP.DT from source address 0xFE (null) or 0xFF (global) is not filtere Sends are serialized per destination: a send whose destination is busy waits behind the one on the wire, and at most `MaxQueuedSendsPerDestination` (default 8, counting sends the channel has accepted but not yet started; every BAM goes to the global address, so this bounds the waiting BAMs too) may wait per destination. One more faults at the call, before anything is queued, with `J1939TpSendRejectedException` instead of growing memory and latency against a peer that answers slowly or never (#204). A queued send has no deadline of its own — a peer may legitimately hold the session ahead of it with CTS(0) — so cancel its token to bound the wait; a cancelled send leaves the queue at once. +Cancelling an active CM send completes its task on the actor without waiting for the Connection +Abort's service hand-off. The destination remains reserved until that hand-off, so the next +send's RTS is handed to the bus service after the Abort. If the original RTS is still +unconfirmed and unanswered, the Abort first waits for it to settle. + The channel ships on nuget.org alongside the other `CanKit.Pro.*` L2/L3 building blocks. It re-uses: - `CanKit.Pro.RawCan` — one `ICanBusService` per channel to demultiplex the TP.CM / TP.DT frames back out of the shared bus stream and to confirm outbound frames. diff --git a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs index d8588d38..8bbfa188 100644 --- a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs +++ b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs @@ -763,6 +763,57 @@ public async Task A_Cm_Send_Cancelled_Before_Its_Rts_Is_Confirmed_Aborts_Only_Af next.IsCompleted.Should().BeFalse("the queued send is on the wire now, waiting for its CTS"); } + // FR-TP-031/034, #356: a peer answer proves the RTS reached the wire even if its + // confirmation is still pending. Cancellation must keep the destination occupied until + // its Abort has crossed the service hand-off, before the successor's RTS is started. + [Fact] + public async Task A_Cm_Send_Cancelled_After_The_Peer_Answered_Hands_Off_Its_Abort_Before_The_Next_Rts() + { + using var clock = new VirtualClock(); + var actor = clock.NewActor(); + using var bus = ControllableBus.EchoCapable(NewSession()); + using var service = new HeldConfirmationBusService(new CanBusService(bus)); + var options = new J1939TpOptions(); + using var sender = new J1939TpChannel(service, sourceAddress: 0x10, options, + ownsService: false, actor); + const byte peerSa = 0x20; + const uint cancelledPgn = 0xFEC6u, nextPgn = 0xFEC7u; + using var cts = new CancellationTokenSource(); + var send = sender.SendCmAsync(cancelledPgn, peerSa, RandomPayload(14, seed: 361), cts.Token); + await service.WaitForHandedAsync(1, ShortTimeout); + + var hold = J1939TpFrames.BuildCts(numPackets: 1, nextPacketSn: 1, dataPgn: cancelledPgn); + hold[1] = 0; // CTS(0): answer the RTS without asking for any DT yet. + bus.RaiseObserved(CanFrame.Classic( + (int)J1939Id.ComposePgn(7, J1939Pgn.TpCm, peerSa, 0x10), hold, isExtendedFrame: true), isEcho: false); + await clock.WaitUntilTimerArmedAsync(actor, options.T4, ShortTimeout); + var next = sender.SendCmAsync(nextPgn, peerSa, RandomPayload(14, seed: 362)); + (await actor.PostAsync(() => sender.QueuedSendCount)).Should().Be(1); + + service.HoldHandOffOf(data => data[0] == J1939TpFrames.ControlAbort); + try + { + cts.Cancel(); + await send.Invoking(t => t.WithTimeout(ShortTimeout)).Should().ThrowAsync(); + var handed = await service.WaitForHandedAsync(2, ShortTimeout); + handed[1][0].Should().Be(J1939TpFrames.ControlAbort); + J1939TpFrames.ReadDataPgn(handed[1]).Should().Be(cancelledPgn); + (await actor.PostAsync(() => sender.QueuedSendCount)).Should().Be(1, + "the next RTS must wait while the cancelled send's Abort is inside the service hand-off"); + + service.ReleaseHandOff(); + handed = await service.WaitForHandedAsync(3, ShortTimeout); + handed[2][0].Should().Be(J1939TpFrames.ControlRts); + J1939TpFrames.ReadDataPgn(handed[2]).Should().Be(nextPgn); + (await actor.PostAsync(() => sender.QueuedSendCount)).Should().Be(0); + next.IsCompleted.Should().BeFalse("its own CTS has not arrived"); + } + finally + { + service.ReleaseHandOff(); + } + } + // The same wait when the RTS is not confirmed: rejected by the driver, or failing inside the // service. Either way it is no longer on its way to the wire, so the abort goes out then -- // a peer that did see the RTS is told -- and the destination is given back. @@ -5229,7 +5280,7 @@ public Task SendConfirmedAsync(CanFrame frame, TimeSpan? timeout lock (_gate) _handed.Add((data, confirmation)); _signal.Release(); if (Volatile.Read(ref _holdHandOff)?.Invoke(data) == true) - _handOffGate.Wait(TimeSpan.FromSeconds(30)); + _handOffGate.Wait(); // released by the test or Dispose, never by a wall-clock deadline return confirmation.Task; }