From b782aec468bbbf431c1b4a77a09e4ab25d6ea849 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Sat, 1 Aug 2026 12:07:11 -0600 Subject: [PATCH 1/5] feat(device): serialize operations per device and let consumers declare indivisible sequences (closes #342) Core serialized text exchanges but nothing else, so a Send() from another thread could be written onto the wire while a text query owned the stream and have its reply collected as part of that query's answer. Every concurrent consumer had to build its own gate to avoid it. Core now owns it: - RunExclusiveAsync wraps a sequence of commands so nothing splits it. Reentrant on the same flow, so a body can call the SD/diagnostic methods that open a text exchange of their own, and can Disconnect. - Send() defers rather than blocks while another flow owns the device. It still returns immediately; the message goes out, in order, afterwards. - One lock, not two. RunExclusiveAsync and the text exchange share the existing per-device semaphore, so there is no ordering to get wrong. - The text exchange lets the outbound queue drain before it takes the stream, so replies to earlier commands go to the protobuf consumer. - The producer-less direct-write path is serialized; it was the one place SCPI bytes could genuinely interleave mid-command. The MCP server's hand-rolled gate shrinks to the connection registry, so two devices now run genuinely in parallel instead of behind one process-wide semaphore. Co-Authored-By: Claude Opus 5 --- docs/DEVICE_INTERFACES.md | 83 ++- ...DaqifiDeviceOperationSerializationTests.cs | 649 ++++++++++++++++++ src/Daqifi.Core/Device/DaqifiDevice.cs | 447 +++++++++++- src/Daqifi.Mcp/DaqifiAgent.cs | 190 +++-- 4 files changed, 1222 insertions(+), 147 deletions(-) create mode 100644 src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs diff --git a/docs/DEVICE_INTERFACES.md b/docs/DEVICE_INTERFACES.md index 5582f047..0038ad10 100644 --- a/docs/DEVICE_INTERFACES.md +++ b/docs/DEVICE_INTERFACES.md @@ -937,17 +937,86 @@ Notes: ## Thread Safety -The `DaqifiDevice` message producer uses a background thread with a concurrent queue, making `Send()` calls thread-safe. Multiple threads can safely send commands: +`DaqifiDevice` and `DaqifiStreamingDevice` are safe to use from multiple threads. The contract has +three parts, and the third one is the part people get wrong. + +### 1. Any single operation is safe to call from any thread + +Enable a channel on one thread, drive a DIO pin on another, run a text query on a third — nothing +you can do with individual calls will produce corrupt SCPI on the wire, and a query's reply always +belongs to that query. Core enforces this itself: + +- Every string command goes onto one background queue drained by one writer thread, so two commands + can never be spliced together mid-command. +- Text queries (SD card listings, diagnostics, LAN chip info, the capability document — anything + that reads a reply) run one at a time per device. They take the device's operation lock, and a + `Send()` from another thread while one is running is **held back and delivered afterwards** rather + than written into the middle of somebody else's answer. + +```csharp +// Safe. Nothing here needs coordinating. +Parallel.For(0, 10, i => device.Send(ScpiMessageProducer.GetDeviceInfo)); +``` + +### 2. A sequence that must not be split goes in `RunExclusiveAsync` + +The one thing a single call cannot express is "these commands belong together". Two threads each +doing set-direction-then-set-value can have their commands interleaved, and the pin ends up in a +state neither thread asked for. Wrap the sequence: ```csharp -// Safe to call from multiple threads -Parallel.For(0, 10, i => +await device.RunExclusiveAsync(_ => +{ + streaming.SetDioDirection(channel, ChannelDirection.Output); + streaming.SetDioValue(channel, true); + return Task.CompletedTask; +}); + +// Bodies can be async and can return a value. Text queries nest freely inside — +// the lock is reentrant on the same logical flow. +var files = await device.RunExclusiveAsync(async ct => { - device.Send(ScpiMessageProducer.GetDeviceInfo); + await sd.StartSdCardLoggingAsync(cancellationToken: ct); + return await sd.GetSdCardFilesAsync(ct); }); ``` -However, for connection state changes (Connect/Disconnect), coordinate access from a single thread or use proper synchronization. +While the body runs, other threads' `Send()` calls are deferred (they still return immediately) and +other threads' text queries wait. Keep bodies short, and do not start background work inside one: +a `Task.Run` launched from the body inherits the flow's ownership of the lock, so its commands would +*not* be deferred and could still interleave. + +`RunExclusiveAsync` is per device. Two devices always run in parallel — there is no global lock. + +### 3. Connect / Disconnect / Dispose serialize themselves, but do not wait forever + +The device never drives its transport from two threads at once, so you do not have to funnel +lifecycle calls through one thread. What you should know: + +- `Connect()` throws `TimeoutException` if another connect or disconnect is still in flight after + 10 seconds. Nothing was opened, so retrying is safe. +- `Disconnect()` / `Dispose()` give an in-flight text query or `RunExclusiveAsync` block a bounded + courtesy wait (10 seconds) and then tear down regardless. A teardown that waited forever would + hang on a wedged serial port, which is worse. Calling `Disconnect()` from *inside* your own + `RunExclusiveAsync` body is fine and does not wait at all. +- Deferred sends belonging to an operation that was still running when the device went away are + logged and dropped, consistent with `Send()` never having guaranteed delivery. + +### What is *not* covered + +- Streaming callbacks (`StreamSamplesAsync`, sample events, decode) never take the operation lock + and are never blocked by it. A live stream keeps flowing while control operations run. +- `IChannel` objects are mutable and shared. Reading `channel.IsEnabled` while another thread + reconfigures it can give you a torn view; use `GetChannelsSnapshot()` for a consistent one. +- Two `DaqifiDevice` instances pointed at the *same* physical unit over two transports are two + independent locks and will fight. Use `DaqifiDeviceRegistry`, which detects that duplicate. + +### Reference implementation + +`src/Daqifi.Mcp/DaqifiAgent.cs` is the worked example of a fully concurrent consumer: the MCP +transport dispatches tool calls in parallel, and every multi-command tool goes through +`RunExclusiveAsync`. Its remaining `SemaphoreSlim` guards only its own connection registry, not the +devices. ## Delivery Failures @@ -967,6 +1036,10 @@ device.SendFailed += (_, e) => A single failed write does not stop the queue: the producer keeps draining the remaining messages regardless of whether anything observes the failure. +`Send()` also never blocks — including when another thread holds the device (see Thread Safety +above), where the message is held back and queued as soon as that finishes. So "returned" has always +meant "accepted", never "delivered", and now it can also mean "delivered a little later". + ## Error Surface Reading from the device and decoding its frames both happen on background threads, where an diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs new file mode 100644 index 00000000..e4e6c084 --- /dev/null +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs @@ -0,0 +1,649 @@ +using Daqifi.Core.Communication.Producers; +using Daqifi.Core.Communication.Transport; +using Daqifi.Core.Device; +using System.Diagnostics; +using System.Reflection; +using System.Text; + +namespace Daqifi.Core.Tests.Device; + +/// +/// Coverage for per-device operation serialization (#342). +/// +/// +/// The contract under test: individual calls are safe from any thread; a sequence that must not be +/// split goes in ; a +/// from another thread is deferred rather than blocked while one runs; and nothing here can +/// deadlock against the text-exchange or lifecycle locks that were already in place. +/// +public class DaqifiDeviceOperationSerializationTests +{ + private static readonly TimeSpan DeadlockBudget = TimeSpan.FromSeconds(15); + + // ── Mutual exclusion ──────────────────────────────────────────────────────────────────── + + [Fact] + public async Task RunExclusiveAsync_NeverRunsTwoOperationsAtOnce() + { + using var transport = new RecordingTransport(); + using var device = new DaqifiDevice("Exclusive Device", transport); + device.Connect(); + + var overlapped = false; + var inFlight = 0; + + var operations = Enumerable.Range(0, 8).Select(_ => Task.Run(() => + device.RunExclusiveAsync(async _ => + { + if (Interlocked.Increment(ref inFlight) > 1) + { + Volatile.Write(ref overlapped, true); + } + + await Task.Delay(25); + Interlocked.Decrement(ref inFlight); + }))); + + await Task.WhenAll(operations).WaitAsync(DeadlockBudget); + + Assert.False(Volatile.Read(ref overlapped), "Two exclusive operations ran at the same time."); + + device.Disconnect(); + } + + [Fact] + public async Task RunExclusiveAsync_ReleasesTheLockWhenTheBodyThrows() + { + using var transport = new RecordingTransport(); + using var device = new DaqifiDevice("Throwing Device", transport); + device.Connect(); + + await Assert.ThrowsAsync( + () => device.RunExclusiveAsync(_ => throw new InvalidOperationException("boom"))); + + // A leaked lock would hang here instead of completing. + await device.RunExclusiveAsync(_ => Task.CompletedTask).WaitAsync(DeadlockBudget); + + device.Disconnect(); + } + + [Fact] + public async Task RunExclusiveAsync_WhenDisposed_ThrowsDeviceNotConnected() + { + var transport = new RecordingTransport(); + var device = new DaqifiDevice("Disposed Device", transport); + device.Connect(); + device.Dispose(); + + var ex = await Assert.ThrowsAsync( + () => device.RunExclusiveAsync(_ => Task.CompletedTask)); + + Assert.True(ex.IsShuttingDown); + } + + // ── Reentrancy / deadlock guards ──────────────────────────────────────────────────────── + + [Fact] + public async Task RunExclusiveAsync_IsReentrantOnTheSameFlow() + { + using var transport = new RecordingTransport(); + using var device = new DaqifiDevice("Reentrant Device", transport); + device.Connect(); + + var reached = false; + + await device.RunExclusiveAsync(async ct => + { + await device.RunExclusiveAsync(_ => + { + reached = true; + return Task.CompletedTask; + }, ct); + }).WaitAsync(DeadlockBudget); + + Assert.True(reached); + + device.Disconnect(); + } + + [Fact] + public async Task RunExclusiveAsync_AllowsANestedTextExchange() + { + // The deadlock this guards: text queries (SD listings, diagnostics, the capability + // document) take the same lock RunExclusiveAsync holds. A non-reentrant acquisition here + // would hang forever on a lock this very flow is holding. + using var transport = new RecordingTransport(); + using var device = new TextExchangeDevice("Nesting Device", transport); + device.Connect(); + + var lines = await device.RunExclusiveAsync( + _ => device.RunTextExchangeAsync(() => device.Send(ScpiMessageProducer.GetDeviceInfo))) + .WaitAsync(DeadlockBudget); + + Assert.NotNull(lines); + Assert.Contains(transport.Writes, w => w.Contains("SYSInfoPB", StringComparison.Ordinal)); + + device.Disconnect(); + } + + [Fact] + public async Task Disconnect_FromInsideAnExclusiveOperation_DoesNotStall() + { + // Teardown waits for the operation lock before ripping the transport away. From inside an + // exclusive operation that is the caller's own lock, so it must run nested instead of + // burning the whole 10s courtesy budget waiting on itself. + using var transport = new RecordingTransport(); + using var device = new DaqifiDevice("Self-disconnecting Device", transport); + device.Connect(); + + var sw = Stopwatch.StartNew(); + await device.RunExclusiveAsync(_ => + { + device.Disconnect(); + return Task.CompletedTask; + }).WaitAsync(DeadlockBudget); + sw.Stop(); + + Assert.False(device.IsConnected); + Assert.True( + sw.Elapsed < TimeSpan.FromSeconds(5), + $"Disconnect from inside an exclusive operation took {sw.Elapsed.TotalSeconds:0.#}s; it waited on its own lock."); + } + + [Fact] + public async Task Disconnect_FromAnotherFlow_StillTearsDownWhileAnOperationIsInFlight() + { + // Teardown must never be blocked indefinitely by an operation. The cancellation token + // shortens the courtesy wait, which is the same exit the 10s timeout takes. + using var transport = new RecordingTransport(); + using var device = new DaqifiDevice("Torn-down Device", transport); + device.Connect(); + + var entered = new TaskCompletionSource(); + var release = new TaskCompletionSource(); + + var operation = Task.Run(() => device.RunExclusiveAsync(async _ => + { + entered.SetResult(); + await release.Task; + })); + + await entered.Task.WaitAsync(DeadlockBudget); + + using var shortWait = new CancellationTokenSource(); + await shortWait.CancelAsync(); + await device.DisconnectAsync(shortWait.Token).WaitAsync(DeadlockBudget); + + Assert.False(device.IsConnected); + + release.SetResult(); + await operation.WaitAsync(DeadlockBudget); + } + + // ── Send deferral ─────────────────────────────────────────────────────────────────────── + + [Fact] + public async Task Send_FromAnotherFlow_IsHeldBackUntilTheOperationFinishes() + { + using var transport = new RecordingTransport(); + using var device = new DaqifiDevice("Deferring Device", transport); + device.Connect(); + + var entered = new TaskCompletionSource(); + var release = new TaskCompletionSource(); + + var operation = Task.Run(() => device.RunExclusiveAsync(async _ => + { + entered.SetResult(); + await release.Task; + })); + + await entered.Task.WaitAsync(DeadlockBudget); + + await Task.Run(() => device.Send(ScpiMessageProducer.SetDioPortState(4, 1))); + + // Long enough that the producer thread would have written it had it been queued. + await Task.Delay(250); + Assert.DoesNotContain(transport.Writes, w => w.Contains("DIO:PORt:STATe", StringComparison.Ordinal)); + + release.SetResult(); + await operation.WaitAsync(DeadlockBudget); + + await WaitForWriteAsync(transport, "DIO:PORt:STATe"); + } + + [Fact] + public async Task Send_FromAnotherFlow_DoesNotBlockWhileAnOperationIsInFlight() + { + // Deferred, not blocked: Send has always been fire-and-forget and must keep returning + // immediately even when the device is owned by someone else. + using var transport = new RecordingTransport(); + using var device = new DaqifiDevice("Non-blocking Device", transport); + device.Connect(); + + var entered = new TaskCompletionSource(); + var release = new TaskCompletionSource(); + + var operation = Task.Run(() => device.RunExclusiveAsync(async _ => + { + entered.SetResult(); + await release.Task; + })); + + await entered.Task.WaitAsync(DeadlockBudget); + + var sw = Stopwatch.StartNew(); + await Task.Run(() => device.Send(ScpiMessageProducer.SetDioPortState(4, 1))); + sw.Stop(); + + Assert.True( + sw.Elapsed < TimeSpan.FromSeconds(2), + $"Send blocked for {sw.Elapsed.TotalSeconds:0.##}s while another flow owned the device."); + + release.SetResult(); + await operation.WaitAsync(DeadlockBudget); + } + + [Fact] + public async Task Send_FromTheOwningFlow_GoesStraightOut() + { + // The operation's own commands must not be parked — the operation would be waiting on + // itself to finish before its own commands could be sent. + using var transport = new RecordingTransport(); + using var device = new DaqifiDevice("Owning Device", transport); + device.Connect(); + + await device.RunExclusiveAsync(async _ => + { + device.Send(ScpiMessageProducer.SetDioPortState(4, 1)); + await WaitForWriteAsync(transport, "DIO:PORt:STATe"); + }).WaitAsync(DeadlockBudget); + + device.Disconnect(); + } + + [Fact] + public async Task Send_DeferredMessagesAreDeliveredInOrder() + { + using var transport = new RecordingTransport(); + using var device = new DaqifiDevice("Ordering Device", transport); + device.Connect(); + + var entered = new TaskCompletionSource(); + var release = new TaskCompletionSource(); + + var operation = Task.Run(() => device.RunExclusiveAsync(async _ => + { + entered.SetResult(); + await release.Task; + })); + + await entered.Task.WaitAsync(DeadlockBudget); + + await Task.Run(() => + { + for (var channel = 1; channel <= 5; channel++) + { + device.Send(ScpiMessageProducer.SetDioPortState(channel, 1)); + } + }); + + release.SetResult(); + await operation.WaitAsync(DeadlockBudget); + + await WaitForWriteAsync(transport, "DIO:PORt:STATe 5"); + + var states = transport.Writes + .Where(w => w.Contains("DIO:PORt:STATe", StringComparison.Ordinal)) + .ToList(); + + Assert.Equal(5, states.Count); + for (var channel = 1; channel <= 5; channel++) + { + Assert.Contains($"STATe {channel}", states[channel - 1], StringComparison.Ordinal); + } + + device.Disconnect(); + } + + [Fact] + public async Task Send_FromAnotherFlow_IsHeldBackDuringAPlainTextExchange() + { + // The hazard the whole feature exists for: a text query owns the stream with the protobuf + // consumer stopped, so a command written by another thread has its reply collected as part + // of that query's answer. + using var transport = new RecordingTransport(); + using var device = new TextExchangeDevice("Querying Device", transport); + device.Connect(); + + // The sender is a real thread started before the exchange opens, so it carries none of the + // exchange's execution context. That matters: work started from *inside* the exchange + // inherits its ownership of the lock and is deliberately not deferred. + using var sendNow = new ManualResetEventSlim(false); + using var sent = new ManualResetEventSlim(false); + + var sender = new Thread(() => + { + sendNow.Wait(DeadlockBudget); + device.Send(ScpiMessageProducer.SetDioPortState(4, 1)); + sent.Set(); + }) + { + IsBackground = true, + }; + sender.Start(); + + var exchange = device.RunTextExchangeAsync(() => sendNow.Set()); + + Assert.True(sent.Wait(DeadlockBudget), "The sending thread never ran."); + Assert.DoesNotContain(transport.Writes, w => w.Contains("DIO:PORt:STATe", StringComparison.Ordinal)); + + await exchange.WaitAsync(DeadlockBudget); + await WaitForWriteAsync(transport, "DIO:PORt:STATe"); + + Assert.True(sender.Join(TimeSpan.FromSeconds(5))); + + device.Disconnect(); + } + + [Fact] + public async Task TextExchange_CancelledWhileTheOutboundQueueDrains_DoesNotResubscribeTheConsumer() + { + // The drain added for #342 is the one step before the consumer swap that can throw. If it + // threw from inside the swap's try/finally, that finally would "restart" a consumer that + // was never stopped — Start() early-returns, but the inbound handler is subscribed again, + // and every frame from then on is dispatched twice. + using var transport = new BlockedWriteTransport(); + using var device = new TextExchangeDevice("Draining Device", transport); + device.Connect(); + + // Queue more than the blocked writer can drain, so the exchange is still draining when the + // token fires. + for (var i = 0; i < 5; i++) + { + device.Send(ScpiMessageProducer.SetDioPortState(i, 1)); + } + + var before = InboundSubscriberCount(device); + + using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(30)); + await Assert.ThrowsAnyAsync( + () => device.RunTextExchangeAsync(() => { }, cts.Token)); + + // A restart that never should have run adds a subscriber; the count must be untouched. + Assert.Equal(before, InboundSubscriberCount(device)); + + transport.ReleaseWrites(); + device.Disconnect(); + } + + private static int InboundSubscriberCount(DaqifiDevice device) + { + var consumer = typeof(DaqifiDevice) + .GetField("_messageConsumer", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(device)!; + + var handler = (Delegate?)consumer.GetType() + .GetField("MessageReceived", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(consumer); + + return handler?.GetInvocationList().Length ?? 0; + } + + // ── The inbound path stays clear ──────────────────────────────────────────────────────── + + [Fact] + public async Task RunExclusiveAsync_DoesNotBlockInboundChannelWork() + { + // Streaming callbacks, the reader loop and frame decode must never wait on the operation + // lock — a control operation must not stall a live stream. Channel snapshotting is the + // device-level state those paths touch. + using var transport = new RecordingTransport(); + using var device = new DaqifiDevice("Streaming-through Device", transport); + device.Connect(); + + await device.RunExclusiveAsync(async ct => + { + var inbound = Task.Run(() => + { + var seen = 0; + for (var i = 0; i < 200; i++) + { + seen += device.GetChannelsSnapshot().Count; + } + + return seen; + }, ct); + + await inbound.WaitAsync(TimeSpan.FromSeconds(5)); + }).WaitAsync(DeadlockBudget); + + device.Disconnect(); + } + + // ── Helpers ───────────────────────────────────────────────────────────────────────────── + + private static async Task WaitForWriteAsync(RecordingTransport transport, string fragment) + { + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5); + while (DateTime.UtcNow < deadline) + { + if (transport.Writes.Any(w => w.Contains(fragment, StringComparison.Ordinal))) + { + return; + } + + await Task.Delay(20); + } + + Assert.Fail($"'{fragment}' never reached the wire. Writes: {string.Join(" | ", transport.Writes)}"); + } + + /// Exposes the protected text-exchange entry point. + private sealed class TextExchangeDevice : DaqifiDevice + { + public TextExchangeDevice(string name, IStreamTransport transport) + : base(name, transport) + { + } + + public Task> RunTextExchangeAsync( + Action setupAction, + CancellationToken cancellationToken = default) => + ExecuteTextCommandAsync( + setupAction, + responseTimeoutMs: 300, + completionTimeoutMs: 100, + cancellationToken: cancellationToken); + } + + /// + /// Transport whose writes block until released, so the producer's queue stays non-empty and a + /// text exchange is guaranteed to still be draining it when a cancellation lands. + /// + private sealed class BlockedWriteTransport : IStreamTransport + { + private readonly BlockingStream _stream = new(); + private bool _isConnected; + private bool _disposed; + + public Stream Stream => _disposed + ? throw new ObjectDisposedException(nameof(BlockedWriteTransport)) + : _stream; + + public bool IsConnected => _isConnected && !_disposed; + + public string ConnectionInfo => _isConnected ? "Blocked: Connected" : "Blocked: Disconnected"; + + public event EventHandler? StatusChanged; + + public void ReleaseWrites() => _stream.Release(); + + public Task ConnectAsync() => ConnectAsync(null); + + public Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + if (_disposed) throw new ObjectDisposedException(nameof(BlockedWriteTransport)); + _isConnected = true; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(true, ConnectionInfo)); + return Task.CompletedTask; + } + + public Task DisconnectAsync() + { + _stream.Release(); + _isConnected = false; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(false, ConnectionInfo)); + return Task.CompletedTask; + } + + public void Connect() => ConnectAsync().Wait(); + + public void Disconnect() => DisconnectAsync().Wait(); + + public void Dispose() + { + if (_disposed) return; + _stream.Release(); + _isConnected = false; + _disposed = true; + } + + private sealed class BlockingStream : Stream + { + private readonly ManualResetEventSlim _released = new(false); + + public void Release() => _released.Set(); + + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => true; + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public override void Flush() + { + } + + public override int Read(byte[] buffer, int offset, int count) + { + Thread.Sleep(5); + return 0; + } + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + + public override void Write(byte[] buffer, int offset, int count) => + _released.Wait(TimeSpan.FromSeconds(10)); + } + } + + /// + /// Transport over a stream that records every write and never has anything to read, so tests + /// can assert on exactly what reached the wire and when. + /// + private sealed class RecordingTransport : IStreamTransport + { + private readonly RecordingStream _stream = new(); + private bool _isConnected; + private bool _disposed; + + public IReadOnlyList Writes => _stream.Writes; + + public Stream Stream => _disposed + ? throw new ObjectDisposedException(nameof(RecordingTransport)) + : _stream; + + public bool IsConnected => _isConnected && !_disposed; + + public string ConnectionInfo => _isConnected ? "Recording: Connected" : "Recording: Disconnected"; + + public event EventHandler? StatusChanged; + + public Task ConnectAsync() => ConnectAsync(null); + + public Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + if (_disposed) throw new ObjectDisposedException(nameof(RecordingTransport)); + _isConnected = true; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(true, ConnectionInfo)); + return Task.CompletedTask; + } + + public Task DisconnectAsync() + { + _isConnected = false; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(false, ConnectionInfo)); + return Task.CompletedTask; + } + + public void Connect() => ConnectAsync().Wait(); + + public void Disconnect() => DisconnectAsync().Wait(); + + public void Dispose() + { + if (_disposed) return; + _isConnected = false; + _disposed = true; + } + + private sealed class RecordingStream : Stream + { + private readonly List _writes = new(); + private readonly object _gate = new(); + + public IReadOnlyList Writes + { + get + { + lock (_gate) + { + return _writes.ToList(); + } + } + } + + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => true; + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public override void Flush() + { + } + + public override int Read(byte[] buffer, int offset, int count) + { + // Nothing to read; back off so the consumer's reader loop doesn't spin. + Thread.Sleep(5); + return 0; + } + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + + public override void Write(byte[] buffer, int offset, int count) + { + lock (_gate) + { + _writes.Add(Encoding.UTF8.GetString(buffer, offset, count)); + } + } + } + } +} diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 00f25f6e..48d542e8 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -422,24 +422,35 @@ protected void WithChannelsLock(Action action) /// private const int InitScpiErrorRetryDelayMs = 150; - // Serializes ExecuteTextCommandAsync calls device-wide (closes #186). - // Multiple callers — e.g. concurrent GetSdCardFilesAsync / - // DrainErrorQueueAsync / GetSystemInfoAsync — would otherwise race the - // protobuf-consumer pause/swap/restart sequence on the same stream and - // either intermix SCPI bytes on the wire or interleave reply lines - // between callers' returned lists. SemaphoreSlim chosen over Lock - // because the method is async; counter is (1, 1) for mutual exclusion. + // THE device operation lock. Originally introduced to serialize ExecuteTextCommandAsync + // calls device-wide (closes #186): multiple callers — e.g. concurrent GetSdCardFilesAsync / + // DrainErrorQueueAsync / GetSystemInfoAsync — would otherwise race the protobuf-consumer + // pause/swap/restart sequence on the same stream and either intermix SCPI bytes on the wire + // or interleave reply lines between callers' returned lists. + // + // It is now also what RunExclusiveAsync takes (closes #342), so a caller can declare a + // multi-command sequence indivisible and have text exchanges, deferred Send()s and teardown + // all coordinate against the same one lock. Deliberately ONE lock rather than an operation + // lock layered over the text-exchange lock: two locks would need an ordering, and the code + // that would have to respect it (Disconnect, Dispose, the reconnect loop, every SD + // operation) is exactly the code that must never deadlock. The field name is unchanged + // because the text exchange is still its busiest user. + // + // SemaphoreSlim chosen over Lock because the holders are async; counter is (1, 1) for + // mutual exclusion. Not reentrant, so re-entry is tracked by _ownsOperationLock below. private readonly SemaphoreSlim _textExchangeLock = new(1, 1); // Async-context flag that tracks whether the current logical flow - // already holds _textExchangeLock. AsyncLocal flows across await + // is inside the consumer swap of ExecuteTextCommandAsync. AsyncLocal flows across await // resumptions on different threads, so a setupAction that re-enters // ExecuteTextCommandAsync after a ConfigureAwait(false) hop is still // detected and surfaced as InvalidOperationException — instead of - // wedging on _textExchangeLock.WaitAsync() (the re-entrant call - // would corrupt the consumer swap mid-flight). Plain + // corrupting the consumer swap mid-flight. Plain // Environment.CurrentManagedThreadId capture wouldn't work — the // value seen before await may not match the value seen after. + // + // Distinct from _ownsOperationLock: this one says "a consumer swap is in progress on this + // flow" (nesting is a bug), that one says "this flow holds the lock" (nesting is fine). private readonly AsyncLocal _isInsideTextExchange = new(); /// @@ -904,6 +915,288 @@ private void ReleaseLifecycleLock() #endregion + #region Operation serialization (issue #342) + + /// + /// True while the current logical flow owns _textExchangeLock. + /// + /// + /// Set by and by the text exchange, and read by + /// everything that would otherwise wait on a lock it already holds. + /// rather than a thread id so it survives an await resuming on another thread — the + /// same technique _isInsideLifecycleOperation and _isInsideTextExchange use. + /// + private readonly AsyncLocal _ownsOperationLock = new(); + + /// + /// Guards and as one unit. + /// + /// + /// They have to move together or the deferral leaks: checking the flag and parking the + /// message in two steps lets an operation finish in between, leaving a message in a list + /// nobody will ever flush. + /// + private readonly object _deferralGate = new(); + + /// True while some flow owns the operation lock. + private bool _operationInFlight; + + /// + /// Sends parked by because another flow held the operation lock. + /// Flushed, in order, by that flow on its way out. + /// + private List? _deferredSends; + + /// + /// Serializes writes on the producer-less path, where writes to the + /// stream on the caller's own thread. Two threads writing a stream concurrently is the one + /// place SCPI bytes really can interleave mid-command; the queued path is already safe + /// because a single producer thread does every write. + /// + private readonly object _directWriteGate = new(); + + /// + /// Runs with exclusive use of the device: no other operation, + /// text query or command send from another thread runs alongside it. + /// + /// + /// + /// Individual calls are already safe to make from any thread. This exists for the case a + /// single call cannot express — a sequence that must not be split, such as "set the + /// direction, then drive the pin" or "set duty, then frequency, then enable". Without it, + /// two threads each doing that can have their commands interleaved and leave the device in + /// a state neither asked for. + /// + /// + /// While the operation runs, a from another thread is deferred, + /// not blocked: it returns immediately, as fire-and-forget always has, and the message goes + /// out in order once this operation finishes. Text queries from other threads wait, exactly + /// as they already waited for each other. + /// + /// + /// Reentrant on the same logical flow, so the body is free to call anything on the device, + /// including the SD card and diagnostic methods that open a text exchange of their own, and + /// including . is the one exception worth + /// knowing about: it takes the lifecycle lock, which a concurrent + /// takes before this one, so reconnecting from inside an exclusive block can cost + /// both sides their bounded wait. Reconnect outside the block. + /// + /// + /// Keep the body short and do not fan out inside it: work started with + /// Task.Run/_ = SomethingAsync() inherits the flow's ownership of the lock, so + /// it would not be deferred and could still interleave. Teardown does not wait forever + /// either — gives an in-flight operation a bounded courtesy wait + /// and then tears down regardless. + /// + /// + /// The sequence to run exclusively. + /// Observed while waiting for the lock, then handed to the operation. + /// A task that completes when the operation has finished. + /// Thrown when the device has been disposed. + /// Thrown when cancelled while waiting for the lock. + public Task RunExclusiveAsync( + Func operation, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(operation); + + return RunExclusiveAsync( + async ct => + { + await operation(ct).ConfigureAwait(false); + return null; + }, + cancellationToken); + } + + /// + /// The type the operation produces. + /// The operation's result. + public async Task RunExclusiveAsync( + Func> operation, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(operation); + + // Already ours: run nested, exactly as a reentrant monitor would. This is what lets an + // exclusive block call GetSdCardFilesAsync — which opens a text exchange on this same + // lock — instead of deadlocking against a non-reentrant semaphore. + if (_ownsOperationLock.Value) + { + return await operation(cancellationToken).ConfigureAwait(false); + } + + await AcquireOperationLockAsync(cancellationToken).ConfigureAwait(false); + + // Set HERE rather than inside the helper above: an async method's AsyncLocal writes do + // not flow back to its caller, only forward to its callees. Assigning it in this frame + // is what makes the body — and everything the body awaits — see the ownership. + _ownsOperationLock.Value = true; + MarkOperationInFlight(); + + try + { + return await operation(cancellationToken).ConfigureAwait(false); + } + finally + { + _ownsOperationLock.Value = false; + FlushDeferredSends(); + ReleaseOperationLock(); + } + } + + /// + /// Waits for the operation lock, translating a disposed semaphore into the same clean + /// failure every other caller of this lock reports. + /// + private async Task AcquireOperationLockAsync(CancellationToken cancellationToken) + { + try + { + await _textExchangeLock.WaitAsync(cancellationToken).ConfigureAwait(false); + } + catch (ObjectDisposedException ex) + { + throw new DeviceNotConnectedException( + "No operation can run exclusively on this device because it is disposed.", + ex, + isShuttingDown: true); + } + } + + /// + /// Publishes "an operation owns the device" so starts deferring. + /// + private void MarkOperationInFlight() + { + lock (_deferralGate) + { + _operationInFlight = true; + } + } + + /// + /// Stops deferring and sends everything parked while the operation ran, in order. + /// + /// + /// Called before the semaphore is released, so the parked messages are queued ahead of + /// whatever the next operation does. Clearing the flag and taking the list happen together + /// under , which is what guarantees no message is parked into a + /// list that has already been drained. + /// + /// A parked send that fails is logged and dropped rather than thrown: the caller was told + /// the message was accepted before this point, and has never + /// guaranteed delivery. The usual case is a device that disconnected while the operation + /// ran, where throwing here would surface a teardown as the operation's failure. + /// + /// + private void FlushDeferredSends() + { + List? parked; + lock (_deferralGate) + { + _operationInFlight = false; + parked = _deferredSends; + _deferredSends = null; + } + + if (parked == null) + { + return; + } + + foreach (var send in parked) + { + try + { + send(); + } + catch (Exception ex) + { + SafeLog(() => _logger.LogWarning( + ex, + "A message deferred while an exclusive operation was running could not be " + + "sent afterwards; it was dropped.")); + } + } + } + + private void ReleaseOperationLock() + { + try + { + _textExchangeLock.Release(); + } + catch (ObjectDisposedException) + { + // Raced a Dispose that already tore the semaphore down. + } + } + + /// + /// Parks a send if another flow currently owns the device, and reports whether it did. + /// + /// + /// The flow that owns the lock is never deferred — those are the operation's own commands, + /// and parking them would leave the operation waiting for itself. + /// + private bool TryDeferSend(IOutboundMessage message) + { + if (_ownsOperationLock.Value) + { + return false; + } + + lock (_deferralGate) + { + if (!_operationInFlight) + { + return false; + } + + (_deferredSends ??= new List()).Add(() => SendNow(message)); + return true; + } + } + + /// + /// How long the text exchange lets the outbound queue drain before it takes the stream. + /// + /// + /// Short on purpose: it is only covering messages queued microseconds before the exchange + /// opened, and the exchange has its own stale-line boundary as a backstop. + /// + private static readonly TimeSpan OutboundDrainWait = TimeSpan.FromMilliseconds(250); + + /// + /// Waits, briefly, for messages queued before this exchange to reach the wire. + /// + /// + /// New sends from other threads are already parked by the time this runs, but anything + /// queued just before the exchange opened is still in the producer's queue. Written after + /// the consumer swap, its reply would land in this exchange's lines instead of the protobuf + /// consumer's — a reply matched to the wrong request. Letting the queue empty first puts + /// those replies back where they belong. Bounded, because a device that is not draining its + /// receive buffer must not stall the exchange: the stale-line boundary still covers it. + /// + private async Task DrainOutboundQueueAsync(CancellationToken cancellationToken) + { + var producer = _messageProducer; + if (producer == null) + { + return; + } + + var deadline = DateTime.UtcNow + OutboundDrainWait; + while (producer.QueuedMessageCount > 0 && DateTime.UtcNow < deadline) + { + await Task.Delay(10, cancellationToken).ConfigureAwait(false); + } + } + + #endregion + /// /// Connects to the device. /// @@ -1372,6 +1665,16 @@ private async Task DisconnectCoreUnsynchronizedAsync( /// true when the lock was acquired and must be released after teardown. private bool AcquireTextExchangeLockForTeardown() { + // This flow already owns the lock — a Disconnect() from inside RunExclusiveAsync, or + // from a StatusChanged handler raised within one. Waiting would burn the whole teardown + // budget on a lock we are holding ourselves and then tear down anyway; run nested + // instead, and leave the release to the owner. Reported as "not acquired" precisely so + // FinishDisconnect does not release a lock this teardown never took. + if (_ownsOperationLock.Value) + { + return false; + } + try { return _textExchangeLock.Wait(TextExchangeTeardownWait); @@ -1391,6 +1694,13 @@ private bool AcquireTextExchangeLockForTeardown() /// private async Task AcquireTextExchangeLockForTeardownAsync(CancellationToken cancellationToken) { + // See AcquireTextExchangeLockForTeardown: re-entry from a flow that already owns the + // lock runs nested rather than waiting on itself. + if (_ownsOperationLock.Value) + { + return false; + } + try { return await _textExchangeLock.WaitAsync(TextExchangeTeardownWait, cancellationToken) @@ -1475,6 +1785,21 @@ private void FinishDisconnect(bool lockAcquired, ConnectionStatus finalStatus) /// /// Sends a message to the device. /// + /// + /// + /// Fire-and-forget, and safe to call from any thread: the message is handed to a background + /// queue and this returns before the write happens, so delivery is not guaranteed. See + /// for the only signal that a specific message was not delivered. + /// + /// + /// If another thread is inside or a text query when + /// this is called, the message is held back and sent, in order, as soon as that finishes + /// (issue #342). This call still does not block — it returns as immediately as it always + /// has — but the message can reach the device later than it used to. That is the point: a + /// command written while a text query owns the stream gets its reply mixed into that + /// query's answer. + /// + /// /// The type of the message data payload. /// The message to send to the device. /// Thrown when the device is not connected. @@ -1489,6 +1814,26 @@ public virtual void Send(IOutboundMessage message) throw new DeviceNotConnectedException(); } + // Checked before the message can be parked. A null used to fail loudly at the producer; + // deferred, it would instead fail on a background flush where the exception is logged + // and dropped, turning a caller's bug into a silently missing command. + ArgumentNullException.ThrowIfNull(message); + + if (TryDeferSend(message)) + { + return; + } + + SendNow(message); + } + + /// + /// Puts a message on its way to the device immediately — the body runs + /// once it knows nothing else owns the device, and the body a deferred send is replayed + /// through afterwards. + /// + private void SendNow(IOutboundMessage message) + { // Use the queued message producer when available and the message is string-based; // this is the common path (SCPI text commands). if (_messageProducer != null && message is IOutboundMessage stringMessage) @@ -1508,7 +1853,15 @@ public virtual void Send(IOutboundMessage message) } var bytes = message.GetBytes(); - stream.Write(bytes, 0, bytes.Length); + + // The queued path gets its mutual exclusion from having exactly one writer thread; this + // path has none, and two callers writing at once is the one case where SCPI bytes can + // genuinely interleave mid-command. The lock is a leaf — nothing is acquired underneath + // it — so it cannot participate in a cycle. + lock (_directWriteGate) + { + stream.Write(bytes, 0, bytes.Length); + } } /// @@ -1762,21 +2115,33 @@ private async Task> ExecuteTextCommandCoreAsync( + "do not call it from inside a setupAction callback."); } - try + // The exchange runs under the device's operation lock. A flow that already owns it — + // one inside RunExclusiveAsync, typically — runs nested rather than waiting on a + // semaphore it is itself holding, and leaves the release to the owner. + var ownsLock = !_ownsOperationLock.Value; + if (ownsLock) { - await _textExchangeLock.WaitAsync(cancellationToken).ConfigureAwait(false); - } - catch (ObjectDisposedException ex) - { - // Dispose() raced ahead of us and disposed the semaphore. - // Surface the same clean failure as the post-acquisition - // _disposed check below, instead of leaking a low-level - // teardown exception to callers. The original is kept as - // InnerException so this rare race stays diagnosable. - throw new DeviceNotConnectedException( - "ExecuteTextCommandAsync cannot run because the device is disposed.", - ex, - isShuttingDown: true); + try + { + await _textExchangeLock.WaitAsync(cancellationToken).ConfigureAwait(false); + } + catch (ObjectDisposedException ex) + { + // Dispose() raced ahead of us and disposed the semaphore. + // Surface the same clean failure as the post-acquisition + // _disposed check below, instead of leaking a low-level + // teardown exception to callers. The original is kept as + // InnerException so this rare race stays diagnosable. + throw new DeviceNotConnectedException( + "ExecuteTextCommandAsync cannot run because the device is disposed.", + ex, + isShuttingDown: true); + } + + // Assigned in this frame, not in a helper: an async method's AsyncLocal writes flow + // forward to its callees but never back to its caller. + _ownsOperationLock.Value = true; + MarkOperationInFlight(); } _isInsideTextExchange.Value = true; @@ -1842,6 +2207,17 @@ private async Task> ExecuteTextCommandCoreAsync( SafeLog(() => _logger.LogDebug("[ExecuteTextCommandAsync] Prepare phase completed at {ElapsedMs}ms", sw.ElapsedMilliseconds)); } + // Let anything queued before this exchange opened reach the wire while the protobuf + // consumer is still the one reading, so its reply is not mistaken for an answer to + // a command this exchange is about to send (issue #342). New sends from other + // threads are already parked by this point, so the queue can only shrink. + // + // Deliberately OUTSIDE the swap's try/finally below: this is the one step here that + // can throw (a cancelled token) before the consumer has been stopped, and that + // finally restarts the consumer — which on a consumer that was never stopped means + // subscribing the inbound handler a second time and dispatching every frame twice. + await DrainOutboundQueueAsync(cancellationToken).ConfigureAwait(false); + var collectedLines = new List(); var stream = _transport.Stream; int? originalReadTimeout = null; @@ -2036,18 +2412,17 @@ private async Task> ExecuteTextCommandCoreAsync( } _isInsideTextExchange.Value = false; - // Release can race with Dispose() — Dispose acquires the lock - // before disposing it, but if that acquisition timed out and - // Dispose proceeded anyway, our SemaphoreSlim handle is now - // gone. Treat that as a benign teardown signal rather than - // surfacing it from the finally and masking the original - // exception (if any) from the try body. - try - { - _textExchangeLock.Release(); - } - catch (ObjectDisposedException) + + // Only the flow that took the lock releases it; a nested exchange leaves it to the + // RunExclusiveAsync block that owns it. Parked sends are flushed before the release + // so they are queued ahead of whatever runs next, and ReleaseOperationLock absorbs + // a Dispose that already tore the semaphore down (Dispose acquires the lock first, + // but proceeds anyway if that acquisition times out). + if (ownsLock) { + _ownsOperationLock.Value = false; + FlushDeferredSends(); + ReleaseOperationLock(); } if (finalizeFailure != null) diff --git a/src/Daqifi.Mcp/DaqifiAgent.cs b/src/Daqifi.Mcp/DaqifiAgent.cs index 79a1bad6..601d09fb 100644 --- a/src/Daqifi.Mcp/DaqifiAgent.cs +++ b/src/Daqifi.Mcp/DaqifiAgent.cs @@ -14,10 +14,24 @@ namespace Daqifi.Mcp; /// ). One instance is shared by all tool calls. /// /// -/// The MCP transport may dispatch tool calls concurrently, so every operation that connects, -/// disconnects, or mutates device state is serialized behind . Read-only -/// introspection snapshots the channel collection instead, so it never blocks and never folds the -/// live Channels view while the device's consumer thread repopulates it. +/// The MCP transport may dispatch tool calls concurrently. Serialization is split by what is +/// actually being protected: +/// +/// +/// Per device — a tool call that sends more than one command is wrapped in +/// , so Core holds that device's operation +/// lock for the whole sequence and nothing from another tool call splits it (issue #342). This +/// used to be , which had the side effect of serializing every device against +/// every other one; two devices now run genuinely in parallel. +/// +/// +/// Across the registry — is now only what it needs to be: the lock +/// around adding to and removing from the connection registry, where the thing being protected is +/// this agent's own state rather than any one device. +/// +/// +/// Read-only introspection takes neither: it snapshots the channel collection, so it never blocks +/// and never folds the live Channels view while the device's consumer thread repopulates it. /// The live set of connections is owned by a keyed by our own /// device_id, which also supplies stale-handle pruning, disposal, and cross-transport /// duplicate detection (the same unit reached over both USB and WiFi). @@ -28,6 +42,12 @@ public sealed class DaqifiAgent private readonly ILogger _logger; private readonly ConcurrentDictionary _discovered = new(StringComparer.Ordinal); private readonly DaqifiDeviceRegistry _registry = new(); + + /// + /// Serializes changes to the connection registry — connect, disconnect, shutdown. Device-level + /// serialization is Core's job now (see the class remarks), so this no longer stands between + /// two tool calls that touch different devices. + /// private readonly SemaphoreSlim _gate = new(1, 1); public DaqifiAgent(ServerOptions options, ILogger? logger = null) @@ -143,12 +163,11 @@ public IReadOnlyList ListChannels(string deviceId) => /// public async Task ConfigureAnalogChannelsAsync(string deviceId, int[] enabledChannels) { - await _gate.WaitAsync().ConfigureAwait(false); - try - { - RequireControl(); - var (device, streaming) = RequireStreaming(deviceId); + RequireControl(); + var (device, streaming) = RequireStreaming(deviceId); + return await device.RunExclusiveAsync(async _ => + { var analog = Snapshot(device).Where(c => c.Type == ChannelType.Analog).ToList(); var validNumbers = analog.Select(c => c.ChannelNumber).ToHashSet(); @@ -177,11 +196,7 @@ public async Task ConfigureAnalogChannelsAsync(string deviceId, await RefreshCapabilityDocumentAsync(device, streaming).ConfigureAwait(false); return new ConfigureResult(deviceId, EnabledAnalog(device), streaming.StreamingFrequency); - } - finally - { - _gate.Release(); - } + }).ConfigureAwait(false); } /// @@ -191,12 +206,11 @@ public async Task ConfigureAnalogChannelsAsync(string deviceId, /// public async Task ConfigureDigitalChannelsAsync(string deviceId, int[] enabledChannels) { - await _gate.WaitAsync().ConfigureAwait(false); - try - { - RequireControl(); - var (device, streaming) = RequireStreaming(deviceId); + RequireControl(); + var (device, streaming) = RequireStreaming(deviceId); + return await device.RunExclusiveAsync(async _ => + { var digital = Snapshot(device).Where(c => c.Type == ChannelType.Digital).ToList(); var validNumbers = digital.Select(c => c.ChannelNumber).ToHashSet(); @@ -224,11 +238,7 @@ public async Task ConfigureDigitalChannelsAsync(string d await RefreshCapabilityDocumentAsync(device, streaming).ConfigureAwait(false); return new ConfigureDigitalResult(deviceId, EnabledDigital(device)); - } - finally - { - _gate.Release(); - } + }).ConfigureAwait(false); } /// @@ -239,21 +249,16 @@ public async Task SetDigitalDirectionAsync(string deviceId, in { var parsed = ParseDirection(direction); - await _gate.WaitAsync().ConfigureAwait(false); - try - { - RequireControl(); - var (device, streaming) = RequireStreaming(deviceId); - var ch = RequireDigitalChannel(device, channel); + RequireControl(); + var (device, streaming) = RequireStreaming(deviceId); + var ch = RequireDigitalChannel(device, channel); + return await device.RunExclusiveAsync(_ => + { streaming.SetDioDirection(ch, parsed); - return DigitalPinResult.From(deviceId, ch); - } - finally - { - _gate.Release(); - } + return Task.FromResult(DigitalPinResult.From(deviceId, ch)); + }).ConfigureAwait(false); } /// @@ -262,13 +267,14 @@ public async Task SetDigitalDirectionAsync(string deviceId, in /// public async Task SetDigitalOutputAsync(string deviceId, int channel, bool high) { - await _gate.WaitAsync().ConfigureAwait(false); - try - { - RequireControl(); - var (device, streaming) = RequireStreaming(deviceId); - var ch = RequireDigitalChannel(device, channel); + RequireControl(); + var (device, streaming) = RequireStreaming(deviceId); + var ch = RequireDigitalChannel(device, channel); + // Direction-then-value is the sequence that must not be split: another tool call landing + // between them could flip the pin back to input before the value is driven. + return await device.RunExclusiveAsync(_ => + { if (ch.Direction != ChannelDirection.Output) { streaming.SetDioDirection(ch, ChannelDirection.Output); @@ -276,12 +282,8 @@ public async Task SetDigitalOutputAsync(string deviceId, int c streaming.SetDioValue(ch, high); - return DigitalPinResult.From(deviceId, ch); - } - finally - { - _gate.Release(); - } + return Task.FromResult(DigitalPinResult.From(deviceId, ch)); + }).ConfigureAwait(false); } /// @@ -291,13 +293,12 @@ public async Task SetDigitalOutputAsync(string deviceId, int c /// public async Task SetPwmOutputAsync(string deviceId, int channel, int dutyCyclePercent, int frequencyHz) { - await _gate.WaitAsync().ConfigureAwait(false); - try - { - RequireControl(); - var (device, streaming) = RequireStreaming(deviceId); - var ch = RequireDigitalChannel(device, channel); + RequireControl(); + var (device, streaming) = RequireStreaming(deviceId); + var ch = RequireDigitalChannel(device, channel); + return await device.RunExclusiveAsync(_ => + { // Duty before frequency before enable: the firmware applies a stored duty when the // frequency is (re)programmed, so this order never leaves a stale compare value. // Core.PwmFrequencyHz always holds a commandable value (a session default when @@ -314,12 +315,8 @@ public async Task SetPwmOutputAsync(string deviceId, int channel, int streaming.SetPwmEnabled(ch, true); - return PwmResult.From(deviceId, streaming, ch); - } - finally - { - _gate.Release(); - } + return Task.FromResult(PwmResult.From(deviceId, streaming, ch)); + }).ConfigureAwait(false); } /// @@ -328,21 +325,16 @@ public async Task SetPwmOutputAsync(string deviceId, int channel, int /// public async Task DisablePwmAsync(string deviceId, int channel) { - await _gate.WaitAsync().ConfigureAwait(false); - try - { - RequireControl(); - var (device, streaming) = RequireStreaming(deviceId); - var ch = RequireDigitalChannel(device, channel); + RequireControl(); + var (device, streaming) = RequireStreaming(deviceId); + var ch = RequireDigitalChannel(device, channel); + return await device.RunExclusiveAsync(_ => + { streaming.SetPwmEnabled(ch, false); - return PwmResult.From(deviceId, streaming, ch); - } - finally - { - _gate.Release(); - } + return Task.FromResult(PwmResult.From(deviceId, streaming, ch)); + }).ConfigureAwait(false); } public async Task SetSampleRateAsync(string deviceId, int rateHz) @@ -352,12 +344,11 @@ public async Task SetSampleRateAsync(string deviceId, int rate throw new InvalidOperationException("rate_hz must be >= 1."); } - await _gate.WaitAsync().ConfigureAwait(false); - try - { - RequireControl(); - var (device, streaming) = RequireStreaming(deviceId); + RequireControl(); + var (device, streaming) = RequireStreaming(deviceId); + return await device.RunExclusiveAsync(_ => + { // MaxSamplingRate is the absolute sampling-ISR ceiling, not what the device will // actually accept for the channels enabled right now — that is // CapabilityStreaming.CurrentMaximumRateHz, refreshed after every channel- @@ -383,12 +374,8 @@ public async Task SetSampleRateAsync(string deviceId, int rate } streaming.StreamingFrequency = rateHz; - return new SampleRateResult(deviceId, rateHz); - } - finally - { - _gate.Release(); - } + return Task.FromResult(new SampleRateResult(deviceId, rateHz)); + }).ConfigureAwait(false); } // --------------------------------------------------------- SD card logging @@ -398,42 +385,33 @@ public async Task StartLoggingAsync( { var fmt = ParseFormat(format); - await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); - try - { - RequireControl(); - var (device, streaming) = RequireStreaming(deviceId); - var sd = RequireSdCard(device); + RequireControl(); + var (device, streaming) = RequireStreaming(deviceId); + var sd = RequireSdCard(device); + return await device.RunExclusiveAsync(async ct => + { // Core owns the naming convention and reports the effective on-card filename back to // us, so we no longer duplicate the log_{timestamp} generation here. - var session = await sd.StartSdCardLoggingSessionAsync(fileName, channelMask: null, format: fmt, cancellationToken) + var session = await sd.StartSdCardLoggingSessionAsync(fileName, channelMask: null, format: fmt, ct) .ConfigureAwait(false); return new StartLoggingResult( deviceId, session.FileName, session.Format.ToString(), streaming.StreamingFrequency, EnabledAnalog(device)); - } - finally - { - _gate.Release(); - } + }, cancellationToken).ConfigureAwait(false); } public async Task StopLoggingAsync(string deviceId, CancellationToken cancellationToken) { - await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); - try + RequireControl(); + var (device, _) = RequireStreaming(deviceId); + var sd = RequireSdCard(device); + + return await device.RunExclusiveAsync(async ct => { - RequireControl(); - var (device, _) = RequireStreaming(deviceId); - var sd = RequireSdCard(device); - await sd.StopSdCardLoggingAsync(cancellationToken).ConfigureAwait(false); + await sd.StopSdCardLoggingAsync(ct).ConfigureAwait(false); return $"Stopped SD-card logging on '{deviceId}'."; - } - finally - { - _gate.Release(); - } + }, cancellationToken).ConfigureAwait(false); } // ------------------------------------------------------------------ shutdown From cf16f66bd226001edf02fd96a27963c7db28ae82 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Sat, 1 Aug 2026 13:04:53 -0600 Subject: [PATCH 2/5] fix(device): close two holes in the new operation serialization (Qodo round 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both findings were real and both attacked the point of the change. The outbound drain was not a barrier. It waited for QueuedMessageCount to reach zero, but MessageProducer takes a message off the queue BEFORE it writes it, so the count reads zero while the write is still going out. A text exchange could still take the stream mid-write and collect that command's reply as its own. MessageProducer now tracks the drain batch itself and exposes IsIdle ("nothing queued and nothing being written"), which is what the barrier waits on. IsIdle is a default interface member carrying the old queue-only answer, so existing implementations compile and behave unchanged. Deferred sends could be overtaken. FlushDeferredSends cleared the "deferring" flag before replaying, so a send arriving mid-replay saw nothing in flight and went straight out ahead of messages parked before it. Deferral now stays on for the whole replay, which still runs outside the gate so Send() keeps not blocking; the flush drains in rounds and only stops deferring in the same locked moment it observes an empty list. A sender fast enough to refill every round is bounded by finishing the last round under the gate. Both are covered by tests confirmed to fail without their fix — the ordering one reproduces the exact inversion (A1, B, A2, A3). Co-Authored-By: Claude Opus 5 --- ...DaqifiDeviceOperationSerializationTests.cs | 312 +++++++++++++++++- .../Producers/IMessageProducer.cs | 21 ++ .../Producers/MessageProducer.cs | 99 ++++-- src/Daqifi.Core/Device/DaqifiDevice.cs | 92 +++++- 4 files changed, 467 insertions(+), 57 deletions(-) diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs index e4e6c084..b7b6aba4 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs @@ -1,3 +1,5 @@ +using Daqifi.Core.Communication.Consumers; +using Daqifi.Core.Communication.Messages; using Daqifi.Core.Communication.Producers; using Daqifi.Core.Communication.Transport; using Daqifi.Core.Device; @@ -390,6 +392,156 @@ private static int InboundSubscriberCount(DaqifiDevice device) return handler?.GetInvocationList().Length ?? 0; } + // ── Qodo round 1, finding 1: the drain must cover in-flight writes ────────────────────── + + [Fact] + public void Producer_WithAWriteInFlight_ReportsQueueEmptyButNotIdle() + { + // The trap, stated as an assertion. MessageProducer dequeues BEFORE it writes, so the + // queue reads empty while the write is still going out. Anything using + // QueuedMessageCount == 0 as "the wire is quiet" is wrong; IsIdle is the real signal. + using var stream = new GatedWriteStream(); + using var producer = new MessageProducer(stream); + producer.Start(); + + producer.Send(ScpiMessageProducer.SetDioPortState(4, 1)); + + Assert.True(stream.WaitForWriteToStart(DeadlockBudget), "The producer never started writing."); + + Assert.Equal(0, producer.QueuedMessageCount); + Assert.False(producer.IsIdle, "IsIdle reported true while a write was still in flight."); + + stream.ReleaseWrites(); + } + + [Fact] + public async Task TextExchange_DoesNotTakeTheStreamWhileAWriteIsStillInFlight() + { + // The device-level consequence: swapping the stream's reader mid-write means that + // command's reply is collected into the exchange's answer. The exchange must still be + // waiting — protobuf consumer running, stream not taken — while the write is blocked. + using var transport = new GatedWriteTransport(); + using var device = new SlowDrainTextExchangeDevice("Draining Device", transport); + device.Connect(); + + device.Send(ScpiMessageProducer.SetDioPortState(4, 1)); + Assert.True(transport.WaitForWriteToStart(DeadlockBudget), "The producer never started writing."); + + var exchange = Task.Run(() => device.RunTextExchangeAsync(() => { })); + + // Sampled across the whole window rather than once at the end: a drain that returns early + // swaps within a few tens of milliseconds and then restarts the consumer when the exchange + // finishes, so a single late sample would see it running again and prove nothing. The + // write stays blocked throughout, so with the barrier working the consumer is never + // stopped at any point here. + for (var i = 0; i < 12; i++) + { + await Task.Delay(25); + Assert.True( + ConsumerIsRunning(device), + $"The text exchange took the stream while a command was still being written (sample {i})."); + } + + transport.ReleaseWrites(); + await exchange.WaitAsync(DeadlockBudget); + + device.Disconnect(); + } + + // ── Qodo round 1, finding 2: deferred sends must not be overtaken ─────────────────────── + + [Fact] + public async Task Send_ArrivingDuringTheFlush_DoesNotOvertakeAlreadyDeferredMessages() + { + // Deferral has to stay on while the parked messages are replayed. If it is switched off + // first, a send arriving mid-replay goes straight out and lands ahead of messages that + // were queued before it. + // + // Binary payloads on purpose: they take the direct-write path, so each replayed send is a + // real blocking write and the replay window is wide enough to aim at deterministically. + using var transport = new GatedWriteTransport(writeDelay: TimeSpan.FromMilliseconds(120)); + using var device = new DaqifiDevice("Ordering Device", transport); + device.Connect(); + + var entered = new TaskCompletionSource(); + var release = new TaskCompletionSource(); + + var operation = Task.Run(() => device.RunExclusiveAsync(async _ => + { + entered.SetResult(); + await release.Task; + })); + + await entered.Task.WaitAsync(DeadlockBudget); + + // Parked while the operation holds the device. + foreach (var tag in new[] { "A1", "A2", "A3" }) + { + await Task.Run(() => device.Send(new TaggedBinaryMessage(tag))); + } + + // A competitor that fires once the replay is under way. Started before the flush and + // gated, so it carries none of the flushing flow's context. + using var replayStarted = new ManualResetEventSlim(false); + transport.OnWriteStarted = () => replayStarted.Set(); + + var competitor = new Thread(() => + { + replayStarted.Wait(DeadlockBudget); + device.Send(new TaggedBinaryMessage("B")); + }) + { + IsBackground = true, + }; + competitor.Start(); + + transport.ReleaseWrites(); + release.SetResult(); + + await operation.WaitAsync(DeadlockBudget); + Assert.True(competitor.Join(TimeSpan.FromSeconds(10))); + + await WaitForWriteAsync(transport, "B"); + + var order = transport.Writes.Where(w => w.Length <= 2).ToList(); + Assert.Equal(new[] { "A1", "A2", "A3", "B" }, order); + + device.Disconnect(); + } + + private static bool ConsumerIsRunning(DaqifiDevice device) + { + var consumer = typeof(DaqifiDevice) + .GetField("_messageConsumer", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(device); + + return consumer is IMessageConsumer { IsRunning: true }; + } + + /// A message whose payload is a short ASCII tag, so write order is readable. + private sealed class TaggedBinaryMessage : IOutboundMessage + { + public TaggedBinaryMessage(string tag) => Data = Encoding.UTF8.GetBytes(tag); + + public byte[] Data { get; set; } + + public byte[] GetBytes() => Data; + } + + /// Raises the drain budget so the wait is observable rather than a race. + private sealed class SlowDrainTextExchangeDevice : DaqifiDevice + { + public SlowDrainTextExchangeDevice(string name, IStreamTransport transport) + : base(name, transport) + { + } + + internal override TimeSpan OutboundDrainWait => TimeSpan.FromSeconds(5); + + public Task> RunTextExchangeAsync(Action setupAction) => + ExecuteTextCommandAsync(setupAction, responseTimeoutMs: 300, completionTimeoutMs: 100); + } + // ── The inbound path stays clear ──────────────────────────────────────────────────────── [Fact] @@ -423,12 +575,18 @@ await device.RunExclusiveAsync(async ct => // ── Helpers ───────────────────────────────────────────────────────────────────────────── - private static async Task WaitForWriteAsync(RecordingTransport transport, string fragment) + private static Task WaitForWriteAsync(RecordingTransport transport, string fragment) => + WaitForWriteAsync(() => transport.Writes, fragment); + + private static Task WaitForWriteAsync(GatedWriteTransport transport, string fragment) => + WaitForWriteAsync(() => transport.Writes, fragment); + + private static async Task WaitForWriteAsync(Func> writes, string fragment) { var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5); while (DateTime.UtcNow < deadline) { - if (transport.Writes.Any(w => w.Contains(fragment, StringComparison.Ordinal))) + if (writes().Any(w => w.Contains(fragment, StringComparison.Ordinal))) { return; } @@ -436,7 +594,7 @@ private static async Task WaitForWriteAsync(RecordingTransport transport, string await Task.Delay(20); } - Assert.Fail($"'{fragment}' never reached the wire. Writes: {string.Join(" | ", transport.Writes)}"); + Assert.Fail($"'{fragment}' never reached the wire. Writes: {string.Join(" | ", writes())}"); } /// Exposes the protected text-exchange entry point. @@ -457,6 +615,154 @@ public Task> RunTextExchangeAsync( cancellationToken: cancellationToken); } + /// + /// A stream that parks inside until released, and reports when a write has + /// actually begun — the state where the queue is empty but the wire is not yet quiet. + /// + private sealed class GatedWriteStream : Stream + { + private readonly ManualResetEventSlim _released = new(false); + private readonly ManualResetEventSlim _writeStarted = new(false); + private readonly List _writes = new(); + private readonly object _gate = new(); + private readonly TimeSpan _writeDelay; + + public GatedWriteStream(TimeSpan? writeDelay = null) => _writeDelay = writeDelay ?? TimeSpan.Zero; + + /// Invoked on the writing thread each time a write begins. + public Action? OnWriteStarted { get; set; } + + public IReadOnlyList Writes + { + get + { + lock (_gate) + { + return _writes.ToList(); + } + } + } + + public bool WaitForWriteToStart(TimeSpan timeout) => _writeStarted.Wait(timeout); + + public void ReleaseWrites() => _released.Set(); + + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => true; + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public override void Flush() + { + } + + public override int Read(byte[] buffer, int offset, int count) + { + Thread.Sleep(5); + return 0; + } + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + + public override void Write(byte[] buffer, int offset, int count) + { + _writeStarted.Set(); + OnWriteStarted?.Invoke(); + + _released.Wait(TimeSpan.FromSeconds(10)); + + if (_writeDelay > TimeSpan.Zero) + { + Thread.Sleep(_writeDelay); + } + + lock (_gate) + { + _writes.Add(Encoding.UTF8.GetString(buffer, offset, count)); + } + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + _released.Set(); + } + + base.Dispose(disposing); + } + } + + /// Transport over a . + private sealed class GatedWriteTransport : IStreamTransport + { + private readonly GatedWriteStream _stream; + private bool _isConnected; + private bool _disposed; + + public GatedWriteTransport(TimeSpan? writeDelay = null) => _stream = new GatedWriteStream(writeDelay); + + public IReadOnlyList Writes => _stream.Writes; + + public Action? OnWriteStarted + { + get => _stream.OnWriteStarted; + set => _stream.OnWriteStarted = value; + } + + public bool WaitForWriteToStart(TimeSpan timeout) => _stream.WaitForWriteToStart(timeout); + + public void ReleaseWrites() => _stream.ReleaseWrites(); + + public Stream Stream => _disposed + ? throw new ObjectDisposedException(nameof(GatedWriteTransport)) + : _stream; + + public bool IsConnected => _isConnected && !_disposed; + + public string ConnectionInfo => _isConnected ? "Gated: Connected" : "Gated: Disconnected"; + + public event EventHandler? StatusChanged; + + public Task ConnectAsync() => ConnectAsync(null); + + public Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + if (_disposed) throw new ObjectDisposedException(nameof(GatedWriteTransport)); + _isConnected = true; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(true, ConnectionInfo)); + return Task.CompletedTask; + } + + public Task DisconnectAsync() + { + _stream.ReleaseWrites(); + _isConnected = false; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(false, ConnectionInfo)); + return Task.CompletedTask; + } + + public void Connect() => ConnectAsync().Wait(); + + public void Disconnect() => DisconnectAsync().Wait(); + + public void Dispose() + { + if (_disposed) return; + _stream.ReleaseWrites(); + _isConnected = false; + _disposed = true; + } + } + /// /// Transport whose writes block until released, so the producer's queue stays non-empty and a /// text exchange is guaranteed to still be draining it when a cancellation lands. diff --git a/src/Daqifi.Core/Communication/Producers/IMessageProducer.cs b/src/Daqifi.Core/Communication/Producers/IMessageProducer.cs index 9ee743ed..5435ff17 100644 --- a/src/Daqifi.Core/Communication/Producers/IMessageProducer.cs +++ b/src/Daqifi.Core/Communication/Producers/IMessageProducer.cs @@ -52,6 +52,27 @@ public interface IMessageProducer : IDisposable /// int QueuedMessageCount { get; } + /// + /// Gets a value indicating whether nothing is queued and no write is part-way through + /// reaching the stream. + /// + /// + /// + /// is not a substitute for this. A message is taken off the + /// queue before it is written, so the count reads zero while the producer thread is + /// still inside a blocking write. Anything that needs "the wire is quiet before I take the + /// stream" — the device's text exchange, which swaps the stream's reader — has to ask this + /// instead, or it can swap while a command is still going out and collect that command's reply + /// as part of its own (issue #342). + /// + /// + /// The default implementation is the weaker queue-only answer, so existing implementations keep + /// compiling and behave exactly as they did. Implementations that write on a background thread + /// should override it — does. + /// + /// + bool IsIdle => QueuedMessageCount == 0; + /// /// Gets a value indicating whether the producer is currently running. /// diff --git a/src/Daqifi.Core/Communication/Producers/MessageProducer.cs b/src/Daqifi.Core/Communication/Producers/MessageProducer.cs index c49cd273..d6308d4a 100644 --- a/src/Daqifi.Core/Communication/Producers/MessageProducer.cs +++ b/src/Daqifi.Core/Communication/Producers/MessageProducer.cs @@ -19,6 +19,18 @@ public class MessageProducer : IMessageProducer private readonly ConcurrentQueue> _messageQueue; private readonly ManualResetEventSlim _messageAvailable = new(false); private volatile bool _isRunning; + + /// + /// True while the background thread is part-way through draining a batch, including the time + /// it spends inside the blocking stream write. + /// + /// + /// Claimed before the dequeue that starts a batch, not after: a message leaves the queue + /// before it is written, so setting this afterwards would leave an instant where the queue is + /// empty and nothing yet reports a write in progress — exactly the gap + /// exists to close. + /// + private volatile bool _draining; private bool _disposed; private Thread? _producerThread; @@ -52,6 +64,9 @@ public MessageProducer(Stream stream, ILogger>? logger = null /// public int QueuedMessageCount => _messageQueue.Count; + /// + public bool IsIdle => !_draining && _messageQueue.IsEmpty; + /// /// Gets a value indicating whether the producer is currently running. /// @@ -173,50 +188,62 @@ private void ProcessMessages() _messageAvailable.Wait(100); _messageAvailable.Reset(); - // Process all available messages - while (_messageQueue.TryDequeue(out var message)) + // Claimed around the whole batch rather than around each write, so there is + // never an instant where a message has left the queue but no write is yet + // reported in progress. Conservative in the other direction (it stays set + // between messages of a batch), which is the safe way to be wrong. + _draining = true; + try { - try + // Process all available messages + while (_messageQueue.TryDequeue(out var message)) { - WriteMessageToStream(message); - - // A successful write clears any run of failures the transport has - // accumulated: the link is demonstrably alive. - _healthSink?.ReportIoSuccess(); - } - catch (Exception ex) - { - // Surface the failure but keep draining the queue so a single - // bad write doesn't stall the remaining messages. Tell the transport - // too — it is the only component that can decide a run of failures - // means the device is gone rather than glitching. - // - // A write TIMEOUT is deliberately excluded: it means the device is not - // draining its receive buffer right now (busy, or flow-controlled), not - // that the link is gone. Treating it as evidence of a disconnect could - // tear down a healthy connection to a momentarily busy device — the - // same reason the reader loop treats a read timeout as benign. - var isTimeout = ex is TimeoutException; - if (!isTimeout) + try { - _healthSink?.ReportIoFault(ex); + WriteMessageToStream(message); + + // A successful write clears any run of failures the transport has + // accumulated: the link is demonstrably alive. + _healthSink?.ReportIoSuccess(); } + catch (Exception ex) + { + // Surface the failure but keep draining the queue so a single + // bad write doesn't stall the remaining messages. Tell the transport + // too — it is the only component that can decide a run of failures + // means the device is gone rather than glitching. + // + // A write TIMEOUT is deliberately excluded: it means the device is not + // draining its receive buffer right now (busy, or flow-controlled), not + // that the link is gone. Treating it as evidence of a disconnect could + // tear down a healthy connection to a momentarily busy device — the + // same reason the reader loop treats a read timeout as benign. + var isTimeout = ex is TimeoutException; + if (!isTimeout) + { + _healthSink?.ReportIoFault(ex); + } - // A timeout gets its own greppable message so "the write never - // happened because the device isn't draining" (busy/flow-controlled) - // can be told apart from any other write failure in the logs. - var logMessage = isTimeout - ? "Timed out writing message to the stream; continuing with remaining queued messages." - : "Failed to write message to the stream; continuing with remaining queued messages."; - SafeLog(() => _logger.LogWarning(ex, logMessage)); + // A timeout gets its own greppable message so "the write never + // happened because the device isn't draining" (busy/flow-controlled) + // can be told apart from any other write failure in the logs. + var logMessage = isTimeout + ? "Timed out writing message to the stream; continuing with remaining queued messages." + : "Failed to write message to the stream; continuing with remaining queued messages."; + SafeLog(() => _logger.LogWarning(ex, logMessage)); - // The only signal a caller gets that this specific message was not - // delivered (issue #408). A throwing subscriber must not take down - // the background loop, so this goes through the same SafeLog guard - // used for the logger above. - SafeLog(() => SendFailed?.Invoke(this, new MessageSendFailedEventArgs(message, ex))); + // The only signal a caller gets that this specific message was not + // delivered (issue #408). A throwing subscriber must not take down + // the background loop, so this goes through the same SafeLog guard + // used for the logger above. + SafeLog(() => SendFailed?.Invoke(this, new MessageSendFailedEventArgs(message, ex))); + } } } + finally + { + _draining = false; + } } catch (Exception ex) { diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 48d542e8..182cd85e 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -1077,13 +1077,34 @@ private void MarkOperationInFlight() } /// - /// Stops deferring and sends everything parked while the operation ran, in order. + /// How many times will replay a fresh batch before it + /// stops racing the senders and finishes holding . + /// + private const int MaxFlushRounds = 8; + + /// + /// Sends everything parked while the operation ran, in order, and only then stops deferring. /// /// + /// /// Called before the semaphore is released, so the parked messages are queued ahead of - /// whatever the next operation does. Clearing the flag and taking the list happen together - /// under , which is what guarantees no message is parked into a - /// list that has already been drained. + /// whatever the next operation does. + /// + /// + /// Deferral stays on for the whole replay. Clearing the flag first and replaying + /// afterwards would leave a window where another thread sees "nothing in flight", sends + /// straight to the producer, and overtakes messages parked before it — losing exactly the + /// ordering this mechanism exists to keep. + /// + /// + /// The replay itself runs outside the gate, which is what keeps + /// non-blocking — the whole reason deferral was chosen over making it wait. The cost is that + /// a send can be parked while a replay is running, so this drains in rounds until a round + /// finds the list empty and can stop deferring in the same breath it observes that. A sender + /// fast enough to refill the list every round would loop forever, so after + /// the last round finishes under the gate: senders block for + /// one final replay instead of this spinning for as long as they keep sending. + /// /// /// A parked send that fails is logged and dropped rather than thrown: the caller was told /// the message was accepted before this point, and has never @@ -1093,19 +1114,43 @@ private void MarkOperationInFlight() /// private void FlushDeferredSends() { - List? parked; - lock (_deferralGate) + for (var round = 0; round < MaxFlushRounds; round++) { - _operationInFlight = false; - parked = _deferredSends; - _deferredSends = null; + List? parked; + lock (_deferralGate) + { + parked = _deferredSends; + _deferredSends = null; + + if (parked == null) + { + // Nothing arrived while the previous round was replaying. Deferral stops + // here, atomically with that observation, so no message can be parked into + // a list nobody will drain. + _operationInFlight = false; + return; + } + } + + ReplayDeferredSends(parked); } - if (parked == null) + lock (_deferralGate) { - return; + _operationInFlight = false; + + var remaining = _deferredSends; + _deferredSends = null; + if (remaining != null) + { + ReplayDeferredSends(remaining); + } } + } + /// Sends one batch of parked messages, in order, never throwing. + private void ReplayDeferredSends(List parked) + { foreach (var send in parked) { try @@ -1167,18 +1212,29 @@ private bool TryDeferSend(IOutboundMessage message) /// Short on purpose: it is only covering messages queued microseconds before the exchange /// opened, and the exchange has its own stale-line boundary as a backstop. /// - private static readonly TimeSpan OutboundDrainWait = TimeSpan.FromMilliseconds(250); + internal virtual TimeSpan OutboundDrainWait => TimeSpan.FromMilliseconds(250); /// /// Waits, briefly, for messages queued before this exchange to reach the wire. /// /// + /// /// New sends from other threads are already parked by the time this runs, but anything - /// queued just before the exchange opened is still in the producer's queue. Written after - /// the consumer swap, its reply would land in this exchange's lines instead of the protobuf - /// consumer's — a reply matched to the wrong request. Letting the queue empty first puts - /// those replies back where they belong. Bounded, because a device that is not draining its - /// receive buffer must not stall the exchange: the stale-line boundary still covers it. + /// queued just before the exchange opened is still on its way out. Written after the + /// consumer swap, its reply would land in this exchange's lines instead of the protobuf + /// consumer's — a reply matched to the wrong request. Letting the outbound side go quiet + /// first puts those replies back where they belong. + /// + /// + /// Waits on , never on + /// QueuedMessageCount == 0: the count drops as soon as a message is dequeued, which + /// is before it is written, so a count of zero can mean "still writing". That is the + /// very case this barrier exists to catch. + /// + /// + /// Bounded, because a device that is not draining its receive buffer must not stall the + /// exchange: the stale-line boundary still covers what slips through. + /// /// private async Task DrainOutboundQueueAsync(CancellationToken cancellationToken) { @@ -1189,7 +1245,7 @@ private async Task DrainOutboundQueueAsync(CancellationToken cancellationToken) } var deadline = DateTime.UtcNow + OutboundDrainWait; - while (producer.QueuedMessageCount > 0 && DateTime.UtcNow < deadline) + while (!producer.IsIdle && DateTime.UtcNow < deadline) { await Task.Delay(10, cancellationToken).ConfigureAwait(false); } From 53036b68a1ca0ac8142ca7e694e42ae25415734a Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Sat, 1 Aug 2026 13:39:15 -0600 Subject: [PATCH 3/5] fix(device): drain the deferred backlog without ever holding the gate across I/O (Qodo round 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 1's bound on the flush was paid for in the wrong currency. Capping the rounds and finishing the last one under _deferralGate meant a concurrent Send() could block on replay I/O for as long as that took — measured at 462ms in the new test — which is exactly the guarantee deferral was chosen over waiting to provide. The backlog is now a queue that is drained one message at a time: the message is dequeued under the gate, replayed outside it, always, with no final under-gate stretch. What keeps ordering is no longer the flag alone but the backlog itself — it stays non-null (possibly empty) for the whole drain, and a send arriving mid-drain sees it and parks behind it instead of overtaking. It is nulled only in the same locked moment the drain observes it empty. Termination is still bounded, but by handing off rather than by blocking: after 64 messages the rest goes to a background empty exclusive operation, which is itself a flush. Going through the operation lock is the point — a bare background replay could write into a text exchange that started meanwhile. The backlog stays non-null across the handoff, so ordering survives it. Teardown now discards parked sends: they were addressed to a session that no longer exists, and a backlog outliving its drainer would leave the next session deferring into it. Also fixes a test that discarded the bool from its phase-boundary wait, so the competing sender could race a replay that never started and the ordering assertion would pass vacuously. The result is asserted, and the wait is shorter than the join that follows so the failure names itself. Co-Authored-By: Claude Opus 5 --- ...DaqifiDeviceOperationSerializationTests.cs | 108 +++++++++++- src/Daqifi.Core/Device/DaqifiDevice.cs | 161 +++++++++++++----- 2 files changed, 224 insertions(+), 45 deletions(-) diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs index b7b6aba4..0f027fea 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs @@ -22,6 +22,13 @@ public class DaqifiDeviceOperationSerializationTests { private static readonly TimeSpan DeadlockBudget = TimeSpan.FromSeconds(15); + /// + /// How long a helper thread waits for the phase of the run it is meant to race. Deliberately + /// shorter than the joins that follow, so "the phase never happened" fails as itself instead of + /// as a join timeout. + /// + private static readonly TimeSpan PhaseBoundaryWait = TimeSpan.FromSeconds(5); + // ── Mutual exclusion ──────────────────────────────────────────────────────────────────── [Fact] @@ -485,9 +492,21 @@ public async Task Send_ArrivingDuringTheFlush_DoesNotOvertakeAlreadyDeferredMess using var replayStarted = new ManualResetEventSlim(false); transport.OnWriteStarted = () => replayStarted.Set(); + // The wait's result is captured and asserted, not discarded. If the replay never starts, + // the competitor must not send at all — otherwise "B came last" would be satisfied by a + // race that never happened, and a real regression would show up as a flake instead of a + // failure. + var replayObserved = false; var competitor = new Thread(() => { - replayStarted.Wait(DeadlockBudget); + // Shorter than the Join below, so a phase boundary that never fires surfaces as the + // specific "replay never started" assertion rather than an opaque join timeout. + replayObserved = replayStarted.Wait(PhaseBoundaryWait); + if (!replayObserved) + { + return; + } + device.Send(new TaggedBinaryMessage("B")); }) { @@ -499,7 +518,12 @@ public async Task Send_ArrivingDuringTheFlush_DoesNotOvertakeAlreadyDeferredMess release.SetResult(); await operation.WaitAsync(DeadlockBudget); - Assert.True(competitor.Join(TimeSpan.FromSeconds(10))); + Assert.True(competitor.Join(TimeSpan.FromSeconds(10)), "The competing sender never finished."); + + // Join established the happens-before, so this read is safe. + Assert.True( + replayObserved, + "The replay never started, so the competitor never raced it and the ordering assertion below would be vacuous."); await WaitForWriteAsync(transport, "B"); @@ -509,6 +533,86 @@ public async Task Send_ArrivingDuringTheFlush_DoesNotOvertakeAlreadyDeferredMess device.Disconnect(); } + [Fact] + public async Task Send_DoesNotBlockWhileALargeBacklogIsBeingFlushed() + { + // The flush is bounded so one operation cannot be kept from returning by a fast sender. + // That bound must NOT be paid for by finishing the last stretch under the deferral gate: + // Send() takes that same gate, so it would then block on whatever blocking I/O the replay + // is doing — losing the non-blocking guarantee that is the whole reason deferral was + // chosen over making Send() wait. + // + // The backlog here is deliberately larger than the per-flush bound, so the probe lands in + // the stretch that runs after the bound is hit. + using var transport = new GatedWriteTransport(writeDelay: TimeSpan.FromMilliseconds(10)); + using var device = new DaqifiDevice("Backlog Device", transport); + device.Connect(); + transport.ReleaseWrites(); + + const int backlog = 120; + const int probeAfterWrites = 80; // comfortably past the 64-message per-flush bound + + var entered = new TaskCompletionSource(); + var release = new TaskCompletionSource(); + + var operation = Task.Run(() => device.RunExclusiveAsync(async _ => + { + entered.SetResult(); + await release.Task; + })); + + await entered.Task.WaitAsync(DeadlockBudget); + + await Task.Run(() => + { + for (var i = 0; i < backlog; i++) + { + device.Send(new TaggedBinaryMessage($"m{i}")); + } + }); + + using var deepIntoFlush = new ManualResetEventSlim(false); + var writes = 0; + transport.OnWriteStarted = () => + { + if (Interlocked.Increment(ref writes) >= probeAfterWrites) + { + deepIntoFlush.Set(); + } + }; + + var probeElapsed = TimeSpan.MaxValue; + var probeRan = false; + var probe = new Thread(() => + { + probeRan = deepIntoFlush.Wait(PhaseBoundaryWait); + if (!probeRan) + { + return; + } + + var sw = Stopwatch.StartNew(); + device.Send(new TaggedBinaryMessage("probe")); + sw.Stop(); + probeElapsed = sw.Elapsed; + }) + { + IsBackground = true, + }; + probe.Start(); + + release.SetResult(); + await operation.WaitAsync(DeadlockBudget); + Assert.True(probe.Join(TimeSpan.FromSeconds(30)), "The probing sender never finished."); + + Assert.True(probeRan, "The flush never reached the probe point, so nothing was measured."); + Assert.True( + probeElapsed < TimeSpan.FromMilliseconds(150), + $"Send() blocked for {probeElapsed.TotalMilliseconds:0}ms during the flush; it must never wait on replay I/O."); + + device.Disconnect(); + } + private static bool ConsumerIsRunning(DaqifiDevice device) { var consumer = typeof(DaqifiDevice) diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 182cd85e..8c9c4e9a 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -942,10 +942,16 @@ private void ReleaseLifecycleLock() private bool _operationInFlight; /// - /// Sends parked by because another flow held the operation lock. - /// Flushed, in order, by that flow on its way out. + /// The backlog: sends parked by because another flow held the + /// operation lock. Replayed, in order, by that flow on its way out. /// - private List? _deferredSends; + /// + /// Non-null means "a backlog exists", and that is itself a reason to keep deferring — a + /// message sent straight out while this is draining would overtake messages parked before + /// it. It therefore stays non-null (and may be empty) for the whole drain, and is nulled + /// only in the same locked moment the drain observes it empty. + /// + private Queue? _deferredSends; /// /// Serializes writes on the producer-less path, where writes to the @@ -1077,10 +1083,16 @@ private void MarkOperationInFlight() } /// - /// How many times will replay a fresh batch before it - /// stops racing the senders and finishes holding . + /// How many parked messages one operation will replay on its way out before handing the + /// rest to a background flush. /// - private const int MaxFlushRounds = 8; + /// + /// A bound is needed because the operation holding the device is the one doing the + /// replaying, and senders can refill the backlog while it works — without a bound, a fast + /// enough sender would keep that operation from ever returning, with the operation lock + /// held and everything else queued behind it. + /// + private const int MaxDeferredSendsPerFlush = 64; /// /// Sends everything parked while the operation ran, in order, and only then stops deferring. @@ -1097,13 +1109,20 @@ private void MarkOperationInFlight() /// ordering this mechanism exists to keep. /// /// - /// The replay itself runs outside the gate, which is what keeps - /// non-blocking — the whole reason deferral was chosen over making it wait. The cost is that - /// a send can be parked while a replay is running, so this drains in rounds until a round - /// finds the list empty and can stop deferring in the same breath it observes that. A sender - /// fast enough to refill the list every round would loop forever, so after - /// the last round finishes under the gate: senders block for - /// one final replay instead of this spinning for as long as they keep sending. + /// Every replay runs outside — always, with no exception + /// for a final round. A replayed send can be a blocking stream write, and + /// takes that same gate, so replaying under it would make + /// Send() block on I/O: the exact property deferral was chosen over waiting to + /// preserve. Messages are therefore taken one at a time, and the backlog stays non-null + /// while the drain runs, which is what keeps a concurrent send parking behind it instead of + /// overtaking it. + /// + /// + /// Bounded by so a fast sender cannot keep this + /// operation from returning. Past that the rest is handed to a background flush, which takes + /// the operation lock exactly as any operation does — so it can never replay into somebody + /// else's exchange — and drains the same way. The backlog stays non-null across the handoff, + /// so ordering survives it. /// /// /// A parked send that fails is logged and dropped rather than thrown: the caller was told @@ -1114,56 +1133,96 @@ private void MarkOperationInFlight() /// private void FlushDeferredSends() { - for (var round = 0; round < MaxFlushRounds; round++) + for (var sent = 0; sent < MaxDeferredSendsPerFlush; sent++) { - List? parked; + Action next; lock (_deferralGate) { - parked = _deferredSends; - _deferredSends = null; - - if (parked == null) + if (_deferredSends == null || _deferredSends.Count == 0) { - // Nothing arrived while the previous round was replaying. Deferral stops - // here, atomically with that observation, so no message can be parked into - // a list nobody will drain. + // Drained. Deferral stops here, in the same locked moment that emptiness is + // observed, so no message can be parked into a backlog nobody will drain. + _deferredSends = null; _operationInFlight = false; return; } + + next = _deferredSends.Dequeue(); } - ReplayDeferredSends(parked); + // Outside the gate on purpose: this can block on a stream write, and Send() takes + // the same gate. The backlog is still non-null, so a send arriving now parks behind + // what is being replayed rather than overtaking it. + ReplayDeferredSend(next); } - lock (_deferralGate) - { - _operationInFlight = false; - - var remaining = _deferredSends; - _deferredSends = null; - if (remaining != null) - { - ReplayDeferredSends(remaining); - } - } + HandOffRemainingDrain(); } - /// Sends one batch of parked messages, in order, never throwing. - private void ReplayDeferredSends(List parked) + /// + /// Hands an unfinished backlog to a background flush so the current operation can return. + /// + /// + /// An empty exclusive operation is a flush: it takes the operation lock, and its own + /// exit path drains the backlog exactly as this one did. Going through the lock is the point + /// — a bare background replay could write into a text exchange that started in the meantime. + /// is deliberately left set and the backlog left non-null, + /// so sends keep deferring across the handoff and ordering holds; whichever operation next + /// reaches its exit path (this background one, or a real one that got the lock first) clears + /// them. + /// + private void HandOffRemainingDrain() { - foreach (var send in parked) + _ = Task.Run(async () => { try { - send(); + await RunExclusiveAsync(_ => Task.CompletedTask).ConfigureAwait(false); } catch (Exception ex) { + // The only way in is a disposed device, where nothing else can be running and + // these sends can never reach anything. Drop the backlog rather than leave + // Send() deferring into one nobody will drain. + lock (_deferralGate) + { + _deferredSends = null; + _operationInFlight = false; + } + SafeLog(() => _logger.LogWarning( ex, - "A message deferred while an exclusive operation was running could not be " - + "sent afterwards; it was dropped.")); + "Messages deferred while an exclusive operation was running were dropped: " + + "the device went away before they could be sent.")); } + }); + } + + /// Sends one parked message, never throwing. + private void ReplayDeferredSend(Action send) + { + try + { + send(); + } + catch (Exception ex) + { + SafeLog(() => _logger.LogWarning( + ex, + "A message deferred while an exclusive operation was running could not be " + + "sent afterwards; it was dropped.")); + } + } + + /// + /// Drops any parked sends. Called when the session they belonged to is torn down: those + /// commands were addressed to a connection that no longer exists. + /// + private void DiscardDeferredSends() + { + lock (_deferralGate) + { + _deferredSends = null; } } @@ -1180,11 +1239,22 @@ private void ReleaseOperationLock() } /// - /// Parks a send if another flow currently owns the device, and reports whether it did. + /// Parks a send if it must not go out yet, and reports whether it did. /// /// + /// + /// Two reasons to park. An operation owns the device, so writing now would land inside its + /// exchange; or a backlog is still being replayed, so writing now would overtake messages + /// parked before this one. + /// + /// /// The flow that owns the lock is never deferred — those are the operation's own commands, /// and parking them would leave the operation waiting for itself. + /// + /// + /// Only ever appends: this holds the gate for a queue insert and nothing else, which is what + /// lets promise it will not block. + /// /// private bool TryDeferSend(IOutboundMessage message) { @@ -1195,12 +1265,12 @@ private bool TryDeferSend(IOutboundMessage message) lock (_deferralGate) { - if (!_operationInFlight) + if (!_operationInFlight && _deferredSends == null) { return false; } - (_deferredSends ??= new List()).Add(() => SendNow(message)); + (_deferredSends ??= new Queue()).Enqueue(() => SendNow(message)); return true; } } @@ -1826,6 +1896,11 @@ private void FinishDisconnect(bool lockAcquired, ConnectionStatus finalStatus) State = DeviceState.Disconnected; _isInitialized = false; _isDisconnecting = false; + + // Parked commands belonged to the session that just ended; the producer they were + // headed for has been torn down. Dropping them here also stops a backlog outliving its + // drainer, which would otherwise leave the next session deferring into it. + DiscardDeferredSends(); if (lockAcquired) { try From 9b088892f7c37c1df4546e5b24221c2cd453a8c9 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Sat, 1 Aug 2026 13:51:01 -0600 Subject: [PATCH 4/5] fix(device): reset both halves of the deferral state on teardown (Qodo round 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Last round's teardown change dropped the parked backlog but left the "an operation owns the device" flag set. That is the same hazard by a different route, and a worse one: with the flag stale, every Send() on the next session parks into a fresh backlog that nothing will drain, and because Send() reports success the moment it parks, the caller is told the command is on its way while it silently goes nowhere. Teardown is exactly where it gets stranded. The operation whose exit path would normally clear the flag is the one that failed to finish inside the bounded wait — so the disconnect proceeds without the lock and the flag outlives the session. That is the case Disconnect exists to handle, and it lands on the reconnect path from #379. Both halves now reset together under one lock. Swept the rest of the teardown path. MarkDisconnectedWithoutTeardown — the route taken when the LIFECYCLE lock is abandoned — is deliberately left alone: it tears nothing down, so the session is still owned by the stuck holder, and clearing deferral there would let sends bypass an operation that is genuinely still running on a live session. It reports Disconnected, so Send() throws rather than parking, and the stuck holder's own teardown resets the state when it unwinds. Test drives the abandoned-lock path specifically; a clean disconnect resets the flag via the operation's own exit path and would pass either way. Without the fix it reproduces the silent loss exactly: "'DIO:PORt:STATe' never reached the wire. Writes: ". Co-Authored-By: Claude Opus 5 --- ...DaqifiDeviceOperationSerializationTests.cs | 51 +++++++++++++++++++ src/Daqifi.Core/Device/DaqifiDevice.cs | 36 ++++++++++--- 2 files changed, 80 insertions(+), 7 deletions(-) diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs index 0f027fea..3296c63e 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs @@ -399,6 +399,57 @@ private static int InboundSubscriberCount(DaqifiDevice device) return handler?.GetInvocationList().Length ?? 0; } + // ── Qodo round 3: teardown must reset deferral state ──────────────────────────────────── + + [Fact] + public async Task Send_AfterATeardownThatCouldNotTakeTheLock_IsStillDelivered() + { + // The silent-loss case. Teardown is bounded: when an in-flight operation does not finish + // inside the wait, the disconnect proceeds anyway — and the operation that would normally + // clear the deferral flag on its way out is precisely the one that did not finish. If the + // flag survives the teardown, every Send() on the NEXT session parks into a backlog with + // no drainer, and Send() reports success while the command goes nowhere. + // + // Driven through the abandoned-lock path on purpose. A clean disconnect resets the flag via + // the operation's own exit path and would pass either way. + using var transport = new RecordingTransport(); + using var device = new DaqifiDevice("Stranded Device", transport); + device.Connect(); + + var entered = new TaskCompletionSource(); + var release = new TaskCompletionSource(); + + var wedged = Task.Run(() => device.RunExclusiveAsync(async _ => + { + entered.SetResult(); + await release.Task; + })); + + await entered.Task.WaitAsync(DeadlockBudget); + + // A cancelled token makes the teardown give up on the operation lock immediately — the + // same exit the bounded wait takes when it times out, without waiting out the budget. + using var giveUpImmediately = new CancellationTokenSource(); + await giveUpImmediately.CancelAsync(); + await device.DisconnectAsync(giveUpImmediately.Token).WaitAsync(DeadlockBudget); + + Assert.False(device.IsConnected); + Assert.False(wedged.IsCompleted, "The operation was supposed to still be in flight."); + + // New session. The previous one's deferral state must not follow it here. + device.Connect(); + Assert.True(device.IsConnected); + + await Task.Run(() => device.Send(ScpiMessageProducer.SetDioPortState(4, 1))); + + await WaitForWriteAsync(transport, "DIO:PORt:STATe"); + + release.SetResult(); + await wedged.WaitAsync(DeadlockBudget); + + device.Disconnect(); + } + // ── Qodo round 1, finding 1: the drain must cover in-flight writes ────────────────────── [Fact] diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 8c9c4e9a..419b6ee9 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -1215,14 +1215,36 @@ private void ReplayDeferredSend(Action send) } /// - /// Drops any parked sends. Called when the session they belonged to is torn down: those - /// commands were addressed to a connection that no longer exists. + /// Returns deferral to its resting state: nothing parked, nothing deferring. /// - private void DiscardDeferredSends() + /// + /// + /// Called when a session is torn down. The parked commands were addressed to a connection + /// that no longer exists, and — the part that matters — the "an operation owns the device" + /// flag has to go with them. + /// + /// + /// Both, or neither. Dropping the backlog while leaving the flag set is the same hazard + /// wearing a different hat: the next session would park every into a + /// fresh backlog with nothing left to drain it, and because Send() reports success + /// the moment it parks, the caller would be told the command was on its way while it + /// silently went nowhere. Teardown is exactly where that gets stranded, because the + /// operation whose completion would normally clear the flag is the one that failed to + /// finish inside the bounded wait. + /// + /// + /// Safe even when a wedged operation is still holding the operation lock. That operation + /// still owns the semaphore, so no new operation can begin before it releases, and its own + /// exit path runs before that release — it cannot clear the flag out from under a later + /// operation. + /// + /// + private void ResetDeferralState() { lock (_deferralGate) { _deferredSends = null; + _operationInFlight = false; } } @@ -1897,10 +1919,10 @@ private void FinishDisconnect(bool lockAcquired, ConnectionStatus finalStatus) _isInitialized = false; _isDisconnecting = false; - // Parked commands belonged to the session that just ended; the producer they were - // headed for has been torn down. Dropping them here also stops a backlog outliving its - // drainer, which would otherwise leave the next session deferring into it. - DiscardDeferredSends(); + // Deferral goes back to its resting state with the session. Both halves — the parked + // commands and the "an operation owns the device" flag — or the next session defers + // into a backlog nobody will drain and loses messages silently. + ResetDeferralState(); if (lockAcquired) { try From 8882518efe90f9748bb6c990d1c4de404953188c Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Sat, 1 Aug 2026 19:30:10 -0600 Subject: [PATCH 5/5] fix(device): scope operation ownership to the session that granted it (Qodo round 4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The hazard is real: a flow that acquired the operation lock and then had the transport torn down and replaced underneath it kept bypassing deferral, writing straight into the new session while an unrelated operation owned it. But the cause was not the unconditional reset, and the proposed remedy — gating that reset on teardown having acquired the lock — makes things strictly worse. TryDeferSend asks about ownership BEFORE it looks at _operationInFlight, so the stale flow's bypass never consulted that flag at all. Leaving the flag set does not quiet the stale flow; it only silences every other flow, whose sends then park in a backlog with no drainer. That is round 3's silent loss plus the stale writer, and the round-3 test reproduces it verbatim against the proposed change. What was actually wrong was the ownership predicate. Holding the semaphore and owning the current session are different questions, and teardown is where they diverge. Ownership is now stamped with a session generation that every teardown retires: - Re-entrancy keeps asking the session-blind question. A flow that holds the semaphore must never be told to wait for it — ExecuteTextCommandAsync waits without a timeout, so a wrong answer there is a hang, not a degradation. - Send()'s bypass asks the session-scoped one. A flow from a retired session is not the owner of the current one and queues behind its rules. This also settles the round-3 asymmetry, which was right in outcome and wrong in reasoning. The principle is not "is an operation running" but "did the session end". FinishDisconnect: it ended, so the deferral state describes something that no longer exists — reset unconditionally. MarkDisconnectedWithoutTeardown: nothing was torn down, the stuck holder still owns a live transport, so the state is still live — leave it. Co-Authored-By: Claude Opus 5 --- ...DaqifiDeviceOperationSerializationTests.cs | 69 +++++++++++++ src/Daqifi.Core/Device/DaqifiDevice.cs | 96 +++++++++++++++---- 2 files changed, 144 insertions(+), 21 deletions(-) diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs index 3296c63e..694c4a73 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs @@ -450,6 +450,75 @@ public async Task Send_AfterATeardownThatCouldNotTakeTheLock_IsStillDelivered() device.Disconnect(); } + [Fact] + public async Task Send_FromAFlowThatOutlivedItsSession_NoLongerBypassesDeferral() + { + // Ownership of the device is a property of a SESSION. A flow that acquired the operation + // lock, then had the transport torn down and replaced underneath it, is not the owner of + // the new session — and must not keep skipping deferral as though it were. + // + // The vehicle is the documented fan-out hazard: work started from inside an exclusive + // block inherits that block's execution context, and with it the block's ownership. Here + // that inherited ownership is deliberately made to outlive a teardown, which is precisely + // the stale-owner shape. It must not let the leaked sender write straight through while a + // completely unrelated operation owns the reconnected device. + using var transport = new RecordingTransport(); + using var device = new DaqifiDevice("Outlived Device", transport); + device.Connect(); + + using var sendNow = new ManualResetEventSlim(false); + using var sendDone = new ManualResetEventSlim(false); + Task? leaked = null; + + await device.RunExclusiveAsync(_ => + { + // Inherits this block's context — and therefore its ownership. + leaked = Task.Run(() => + { + sendNow.Wait(DeadlockBudget); + device.Send(ScpiMessageProducer.SetDioPortState(5, 1)); + sendDone.Set(); + }); + return Task.CompletedTask; + }).WaitAsync(DeadlockBudget); + + // Tear the session down and bring a new one up. The leaked sender is now a flow from a + // session that no longer exists. + await device.DisconnectAsync().WaitAsync(DeadlockBudget); + device.Connect(); + Assert.True(device.IsConnected); + + // A genuine operation now owns the reconnected device. + var entered = new TaskCompletionSource(); + var release = new TaskCompletionSource(); + var owner = Task.Run(() => device.RunExclusiveAsync(async _ => + { + entered.SetResult(); + await release.Task; + })); + + await entered.Task.WaitAsync(DeadlockBudget); + + sendNow.Set(); + Assert.True(sendDone.Wait(DeadlockBudget), "The leaked sender never ran."); + + // Still owned by someone else, so the stale flow's message must be parked, not written. + await Task.Delay(250); + Assert.DoesNotContain(transport.Writes, w => w.Contains("DIO:PORt:STATe", StringComparison.Ordinal)); + + release.SetResult(); + await owner.WaitAsync(DeadlockBudget); + if (leaked != null) + { + await leaked.WaitAsync(DeadlockBudget); + } + + // ...and delivered once that operation finishes. + await WaitForWriteAsync(transport, "DIO:PORt:STATe"); + + device.Disconnect(); + } + // ── Qodo round 1, finding 1: the drain must cover in-flight writes ────────────────────── [Fact] diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 419b6ee9..7dedd105 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -437,7 +437,7 @@ protected void WithChannelsLock(Action action) // because the text exchange is still its busiest user. // // SemaphoreSlim chosen over Lock because the holders are async; counter is (1, 1) for - // mutual exclusion. Not reentrant, so re-entry is tracked by _ownsOperationLock below. + // mutual exclusion. Not reentrant, so re-entry is tracked by _operationLockGeneration below. private readonly SemaphoreSlim _textExchangeLock = new(1, 1); // Async-context flag that tracks whether the current logical flow @@ -449,7 +449,7 @@ protected void WithChannelsLock(Action action) // Environment.CurrentManagedThreadId capture wouldn't work — the // value seen before await may not match the value seen after. // - // Distinct from _ownsOperationLock: this one says "a consumer swap is in progress on this + // Distinct from _operationLockGeneration: this one says "a consumer swap is in progress on this // flow" (nesting is a bug), that one says "this flow holds the lock" (nesting is fine). private readonly AsyncLocal _isInsideTextExchange = new(); @@ -918,15 +918,53 @@ private void ReleaseLifecycleLock() #region Operation serialization (issue #342) /// - /// True while the current logical flow owns _textExchangeLock. + /// The session generation in which the current logical flow acquired + /// _textExchangeLock, or 0 when it does not hold it. /// /// - /// Set by and by the text exchange, and read by - /// everything that would otherwise wait on a lock it already holds. - /// rather than a thread id so it survives an await resuming on another thread — the - /// same technique _isInsideLifecycleOperation and _isInsideTextExchange use. + /// Set by and by the text exchange. + /// rather than a thread id so it survives an await + /// resuming on another thread — the same technique _isInsideLifecycleOperation and + /// _isInsideTextExchange use. + /// + /// A generation rather than a bool because holding the lock and owning the current + /// session are different questions, and teardown separates them. See + /// and . + /// + /// + private readonly AsyncLocal _operationLockGeneration = new(); + + /// + /// Bumped by every teardown, retiring the ownership of any flow that took the lock in an + /// earlier session. Starts at 1 so that 0 always means "does not hold the lock". + /// + private int _operationGeneration = 1; + + /// + /// True when this flow acquired the operation lock — in any session. + /// + /// + /// This is the re-entrancy question, and it must stay session-blind. A flow that holds the + /// semaphore must never be told to wait for it, whatever has happened to the session in the + /// meantime: ExecuteTextCommandAsync waits on it without a timeout, so answering + /// "no" to a flow that really does hold it is not a degraded result, it is a hang. /// - private readonly AsyncLocal _ownsOperationLock = new(); + private bool HoldsOperationLock => _operationLockGeneration.Value != 0; + + /// + /// True when this flow acquired the operation lock in the session that is still current. + /// + /// + /// This is the authority question, and it is the one asks before + /// skipping deferral. Exclusivity is a property of a session: once the transport has been + /// torn down and a new one opened, a flow still running from the old session is not the + /// owner of the new one and its sends must queue behind that session's rules like anybody + /// else's. Without this, a flow that outlived its teardown would keep bypassing deferral + /// into a session it has no claim on. + /// + private bool OwnsCurrentSession => + _operationLockGeneration.Value != 0 + && _operationLockGeneration.Value == Volatile.Read(ref _operationGeneration); /// /// Guards and as one unit. @@ -1027,7 +1065,7 @@ public async Task RunExclusiveAsync( // Already ours: run nested, exactly as a reentrant monitor would. This is what lets an // exclusive block call GetSdCardFilesAsync — which opens a text exchange on this same // lock — instead of deadlocking against a non-reentrant semaphore. - if (_ownsOperationLock.Value) + if (HoldsOperationLock) { return await operation(cancellationToken).ConfigureAwait(false); } @@ -1037,7 +1075,7 @@ public async Task RunExclusiveAsync( // Set HERE rather than inside the helper above: an async method's AsyncLocal writes do // not flow back to its caller, only forward to its callees. Assigning it in this frame // is what makes the body — and everything the body awaits — see the ownership. - _ownsOperationLock.Value = true; + _operationLockGeneration.Value = Volatile.Read(ref _operationGeneration); MarkOperationInFlight(); try @@ -1046,7 +1084,7 @@ public async Task RunExclusiveAsync( } finally { - _ownsOperationLock.Value = false; + _operationLockGeneration.Value = 0; FlushDeferredSends(); ReleaseOperationLock(); } @@ -1233,14 +1271,30 @@ private void ReplayDeferredSend(Action send) /// finish inside the bounded wait. /// /// - /// Safe even when a wedged operation is still holding the operation lock. That operation - /// still owns the semaphore, so no new operation can begin before it releases, and its own - /// exit path runs before that release — it cannot clear the flag out from under a later - /// operation. + /// The generation bump is what makes the reset safe when a wedged operation is still + /// holding the lock. Clearing the flag alone would leave that flow bypassing deferral — + /// asks before it looks + /// at the flag — so it would keep sending into the next session as though it still owned + /// it. Retiring the generation ends that claim: the flow keeps the semaphore (only it can + /// release it, and its exit path still must) but stops counting as the owner of a session + /// that has been replaced. + /// + /// + /// Note what is deliberately not done here: the reset is unconditional, and in + /// particular is not gated on teardown having acquired the lock. Gating it would strand + /// exactly when the lock could not be taken — the case a + /// bounded teardown exists for — leaving the next session deferring into a backlog with no + /// drainer. And it would not achieve isolation either, because the stale flow's bypass does + /// not consult that flag at all. What the flag governs is every other flow; gating it + /// would silence them and leave the stale one talking. /// /// private void ResetDeferralState() { + // Retire the outgoing session's ownership before reopening the gate, so no flow can be + // both "not deferring" and "not the owner" at the same instant. + Interlocked.Increment(ref _operationGeneration); + lock (_deferralGate) { _deferredSends = null; @@ -1280,7 +1334,7 @@ private void ReleaseOperationLock() /// private bool TryDeferSend(IOutboundMessage message) { - if (_ownsOperationLock.Value) + if (OwnsCurrentSession) { return false; } @@ -1818,7 +1872,7 @@ private bool AcquireTextExchangeLockForTeardown() // budget on a lock we are holding ourselves and then tear down anyway; run nested // instead, and leave the release to the owner. Reported as "not acquired" precisely so // FinishDisconnect does not release a lock this teardown never took. - if (_ownsOperationLock.Value) + if (HoldsOperationLock) { return false; } @@ -1844,7 +1898,7 @@ private async Task AcquireTextExchangeLockForTeardownAsync(CancellationTok { // See AcquireTextExchangeLockForTeardown: re-entry from a flow that already owns the // lock runs nested rather than waiting on itself. - if (_ownsOperationLock.Value) + if (HoldsOperationLock) { return false; } @@ -2271,7 +2325,7 @@ private async Task> ExecuteTextCommandCoreAsync( // The exchange runs under the device's operation lock. A flow that already owns it — // one inside RunExclusiveAsync, typically — runs nested rather than waiting on a // semaphore it is itself holding, and leaves the release to the owner. - var ownsLock = !_ownsOperationLock.Value; + var ownsLock = !HoldsOperationLock; if (ownsLock) { try @@ -2293,7 +2347,7 @@ private async Task> ExecuteTextCommandCoreAsync( // Assigned in this frame, not in a helper: an async method's AsyncLocal writes flow // forward to its callees but never back to its caller. - _ownsOperationLock.Value = true; + _operationLockGeneration.Value = Volatile.Read(ref _operationGeneration); MarkOperationInFlight(); } @@ -2573,7 +2627,7 @@ private async Task> ExecuteTextCommandCoreAsync( // but proceeds anyway if that acquisition times out). if (ownsLock) { - _ownsOperationLock.Value = false; + _operationLockGeneration.Value = 0; FlushDeferredSends(); ReleaseOperationLock(); }