vwemirror/riojoy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

RioSerialLink: stop-and-wait command delivery (supersedes NAK-race resend)

Browse Source 
Bench falsified the v1 retransmit (testlogs/riomash-patched-62500-retx):
resends tracked NAKs 1:1 (321/321) yet lamps still stuck/missed — under
mash bursts the NAK arrives after a newer command is already "latest",
so the wrong packet was resent; and total shreds never NAK at all.

Now commands are stop-and-wait: ONE in flight (_commandGate), resolved
by ACK, NAK, or AckTimeout (50ms); NAK/timeout retransmits THE SAME
packet up to CommandRetransmitLimit (2), then drops (idempotent - the
next state update supersedes). Attribution is exact by construction and
timeouts catch silent shreds. Control-byte replies bypass the gate so
board traffic is never delayed; a request's own reply (analog/version/
check) also resolves the wait, so request/reply exchanges never burn
the timeout even if the board sends no explicit ACK.

7 tests (same-packet resend, timeout retry+drop, ACK completion,
serialization, reply-resolves-request, stray-NAK no-op, disable);
282 green. Mash summary now reports NAK/timeout resends.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Cyd
2026-07-19 13:20:30 -05:00
co-authored by Claude Fable 5
parent 1943b7f8eb
commit 47df14cc61
18 changed files with 279 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/RioJoy.Core/Serial/RioSerialLink.cs
+74 -39
View File
@@ -23,13 +23,14 @@ public sealed class RioSerialLink
// Time since the last accepted AnalogReply, for the recovery watchdog.
private readonly Stopwatch _sinceAnalog = new();
// NAK retransmit state: the last COMMAND packet sent (control-byte replies
// are excluded — the board never NAKs those) and how many resends it has
// consumed. Guarded by _retransmitGate; see RioSerialLinkOptions.NakRetransmitLimit.
private readonly object _retransmitGate = new();
private byte[]? _lastCommand;
private int _lastCommandResends;
private long _nakRetransmits;
// Stop-and-wait command state: one command in flight at a time
// (_commandGate), resolved by the board's ACK/NAK or by AckTimeout.
// Control-byte replies (our ACKs) bypass the gate — the board never
// ACK/NAKs those, and they must not queue behind a pending command.
private readonly SemaphoreSlim _commandGate = new(1, 1);
private readonly object _pendingGate = new();
private TaskCompletionSource<bool>? _pendingAck; // true = ACK, false = NAK
private long _retransmits;
public RioSerialLink(IRioTransport transport, RioSerialLinkOptions? options = null)
{
@@ -58,8 +59,8 @@ public sealed class RioSerialLink
/// <summary>The transport's description, surfaced for status/logging.</summary>
public string Description => _transport.Description;
/// <summary>Total NAK-triggered retransmits since the link was created.</summary>
public long NakRetransmits => Interlocked.Read(ref _nakRetransmits);
/// <summary>Total command retransmits (NAK- or timeout-triggered) since creation.</summary>
public long Retransmits => Interlocked.Read(ref _retransmits);
/// <summary>
/// Run the receive loop and (if enabled) the analog poll loop until
@@ -89,23 +90,56 @@ public sealed class RioSerialLink
}
}
/// <summary>Send a pre-built packet (see <see cref="PacketBuilder"/>) to the RIO.</summary>
/// <summary>
/// Send a pre-built packet (see <see cref="PacketBuilder"/>) to the RIO.
/// Command packets use stop-and-wait: the call completes once the board
/// ACKs, or after <see cref="RioSerialLinkOptions.CommandRetransmitLimit"/>
/// retransmits (NAK- or timeout-triggered) go unacknowledged — the command
/// is then dropped (all commands are idempotent state-setters, and the
/// caller's next update supersedes it). Control bytes bypass the wait.
/// </summary>
public async Task SendAsync(byte[] packet, CancellationToken cancellationToken = default)
{
if (packet is null) throw new ArgumentNullException(nameof(packet));
// Only real command packets participate in NAK retransmit; single
// control bytes (our ACK/NAK replies) are never NAK'd by the board.
if (packet.Length > 1 && _options.NakRetransmitLimit > 0)
// Control-byte replies (len 1) are never ACK/NAK'd by the board, and a
// limit of 0 selects the legacy fire-and-forget behavior.
if (packet.Length <= 1 || _options.CommandRetransmitLimit <= 0)
{
lock (_retransmitGate)
{
_lastCommand = packet;
_lastCommandResends = 0;
}
await WriteAsync(packet, cancellationToken).ConfigureAwait(false);
return;
}
await WriteAsync(packet, cancellationToken).ConfigureAwait(false);
await Compat.TaskCompat.WaitAsync(_commandGate, cancellationToken).ConfigureAwait(false);
try
{
for (int attempt = 0; ; attempt++)
{
var pending = new TaskCompletionSource<bool>();
lock (_pendingGate) _pendingAck = pending;
await WriteAsync(packet, cancellationToken).ConfigureAwait(false);
Task winner = await Compat.TaskCompat.WhenAny(new[]
{
pending.Task,
Compat.TaskCompat.Delay(_options.AckTimeout, cancellationToken),
}).ConfigureAwait(false);
if (winner == pending.Task && pending.Task.Result)
return; // ACK'd
// NAK'd or timed out (a shred so complete the board never NAK'd).
if (attempt >= _options.CommandRetransmitLimit)
return; // budget spent — drop; idempotent, next update supersedes
Interlocked.Increment(ref _retransmits);
}
}
finally
{
lock (_pendingGate) _pendingAck = null;
_commandGate.Release();
}
}
private async Task WriteAsync(byte[] packet, CancellationToken cancellationToken)
@@ -121,24 +155,6 @@ public sealed class RioSerialLink
}
}
// The board NAK'd its most recent inbound packet: resend the last command
// (idempotent, see RioSerialLinkOptions.NakRetransmitLimit) unless its
// retry budget is spent.
private async Task RetransmitOnNakAsync(CancellationToken ct)
{
byte[]? packet;
lock (_retransmitGate)
{
if (_lastCommand is null || _lastCommandResends >= _options.NakRetransmitLimit)
return;
_lastCommandResends++;
packet = _lastCommand;
}
Interlocked.Increment(ref _nakRetransmits);
await WriteAsync(packet!, ct).ConfigureAwait(false);
}
/// <summary>Request an analog update (<see cref="RioCommand.AnalogRequest"/>).</summary>
public Task RequestAnalogAsync(CancellationToken cancellationToken = default) =>
SendAsync(PacketBuilder.AnalogRequest(), cancellationToken);
@@ -188,8 +204,14 @@ public sealed class RioSerialLink
break;
case RioRxEventKind.ControlByte:
if (ev.Byte == (byte)RioControl.Nak)
await RetransmitOnNakAsync(ct).ConfigureAwait(false);
if (ev.Byte is (byte)RioControl.Ack or (byte)RioControl.Nak)
{
// Resolve the in-flight command (stop-and-wait): the board
// only emits ACK/NAK in response to our command packets.
TaskCompletionSource<bool>? pending;
lock (_pendingGate) pending = _pendingAck;
pending?.TrySetResult(ev.Byte == (byte)RioControl.Ack);
}
ControlReceived?.Invoke(ev.Byte);
break;
@@ -201,6 +223,19 @@ public sealed class RioSerialLink
private void DispatchTyped(RioPacket packet)
{
if (packet.Command is RioCommand.AnalogReply or RioCommand.VersionReply or RioCommand.CheckReply)
{
// A reply proves the corresponding request landed, whether or not
// the board also sent an explicit ACK — resolve the in-flight
// command so request/reply exchanges never burn the ACK timeout.
// (One command is in flight at a time, so the only stray case is
// an unsolicited reply resend — harmless: it can only release an
// idempotent command's wait early.)
TaskCompletionSource<bool>? pending;
lock (_pendingGate) pending = _pendingAck;
pending?.TrySetResult(true);
}
switch (packet.Command)
{
case RioCommand.AnalogReply:
src/RioJoy.Core/Serial/RioSerialLinkOptions.cs
+17 -9
View File
@@ -32,14 +32,22 @@ public sealed record RioSerialLinkOptions
/// <summary>
/// How many times a command packet is retransmitted when the board NAKs it
/// (0 = fire-and-forget, the pre-2026-07 behavior). The wire has no sequence
/// numbers, but every PC→RIO command is idempotent (lamp state, analog
/// request, reset), so resending the most recent packet on NAK is safe even
/// in the rare race where the NAK belonged to an earlier one. Bounded so a
/// NAK storm (e.g. a rate the board's RX can't sustain) can't multiply
/// traffic without limit. Bench origin: at 62500 baud ~6.5% of 4-byte lamp
/// commands arrive corrupt (RX-ISR overrun) — one retry cuts the visible
/// lamp-state error rate to ~0.4%, two to ~0.03%.
/// or the ACK never arrives (0 = fire-and-forget, the pre-2026-07 behavior).
/// Commands use stop-and-wait: ONE command in flight, resolved by
/// ACK / NAK / <see cref="AckTimeout"/> before the next is written. The
/// first (NAK-race) design resent "the most recent" packet and misfired
/// under mash bursts — by the time a NAK crossed USB, a newer command was
/// already the latest, so the corrupted write stayed lost (bench
/// 2026-07-19); with one-in-flight the attribution is exact, and the
/// timeout also catches corruptions so complete the board never NAKs.
/// </summary>
public int NakRetransmitLimit { get; init; } = 2;
public int CommandRetransmitLimit { get; init; } = 2;
/// <summary>
/// How long to wait for the board's ACK/NAK of a command before treating it
/// as lost (stop-and-wait; see <see cref="CommandRetransmitLimit"/>). Board
/// turnaround is ~1-4 ms; 50 ms is generous without stalling the 55 ms
/// analog poll cadence on a healthy link.
/// </summary>
public TimeSpan AckTimeout { get; init; } = TimeSpan.FromMilliseconds(50);
}