Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 10 additions & 20 deletions src/CanKit.Pro.J1939Tp/J1939TpChannel.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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<TxCompletion, LinkedListNode<PendingTx>> _queuedNodes = new();
private readonly Dictionary<RxSessionKey, RxSession> _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<byte, TxSession> _abortsAwaitingRts = new();

// Sends admitted per destination and not yet completed: posted to the actor and not yet run,
Expand Down Expand Up @@ -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);
Expand Down
5 changes: 5 additions & 0 deletions src/CanKit.Pro.J1939Tp/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
53 changes: 52 additions & 1 deletion tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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<OperationCanceledException>();
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.
Expand Down Expand Up @@ -5229,7 +5280,7 @@ public Task<TxConfirmation> 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;
}

Expand Down
Loading