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..694c4a73 --- /dev/null +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceOperationSerializationTests.cs @@ -0,0 +1,1179 @@ +using Daqifi.Core.Communication.Consumers; +using Daqifi.Core.Communication.Messages; +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); + + /// + /// 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] + 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; + } + + // ── 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(); + } + + [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] + 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(); + + // 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(() => + { + // 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")); + }) + { + IsBackground = true, + }; + competitor.Start(); + + transport.ReleaseWrites(); + release.SetResult(); + + await operation.WaitAsync(DeadlockBudget); + 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"); + + var order = transport.Writes.Where(w => w.Length <= 2).ToList(); + Assert.Equal(new[] { "A1", "A2", "A3", "B" }, order); + + 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) + .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] + 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 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 (writes().Any(w => w.Contains(fragment, StringComparison.Ordinal))) + { + return; + } + + await Task.Delay(20); + } + + Assert.Fail($"'{fragment}' never reached the wire. Writes: {string.Join(" | ", 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); + } + + /// + /// 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. + /// + 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/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 00f25f6e..7dedd105 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 _operationLockGeneration 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 _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(); /// @@ -904,6 +915,490 @@ private void ReleaseLifecycleLock() #endregion + #region Operation serialization (issue #342) + + /// + /// 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. + /// 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 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. + /// + /// + /// 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; + + /// + /// The backlog: sends parked by because another flow held the + /// operation lock. Replayed, in order, by that flow on its way out. + /// + /// + /// 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 + /// 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 (HoldsOperationLock) + { + 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. + _operationLockGeneration.Value = Volatile.Read(ref _operationGeneration); + MarkOperationInFlight(); + + try + { + return await operation(cancellationToken).ConfigureAwait(false); + } + finally + { + _operationLockGeneration.Value = 0; + 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; + } + } + + /// + /// How many parked messages one operation will replay on its way out before handing the + /// rest to a background flush. + /// + /// + /// 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. + /// + /// + /// + /// Called before the semaphore is released, so the parked messages are queued ahead of + /// 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. + /// + /// + /// 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 + /// 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() + { + for (var sent = 0; sent < MaxDeferredSendsPerFlush; sent++) + { + Action next; + lock (_deferralGate) + { + if (_deferredSends == null || _deferredSends.Count == 0) + { + // 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(); + } + + // 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); + } + + HandOffRemainingDrain(); + } + + /// + /// 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() + { + _ = Task.Run(async () => + { + try + { + 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, + "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.")); + } + } + + /// + /// Returns deferral to its resting state: nothing parked, nothing deferring. + /// + /// + /// + /// 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. + /// + /// + /// 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; + _operationInFlight = false; + } + } + + private void ReleaseOperationLock() + { + try + { + _textExchangeLock.Release(); + } + catch (ObjectDisposedException) + { + // Raced a Dispose that already tore the semaphore down. + } + } + + /// + /// 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) + { + if (OwnsCurrentSession) + { + return false; + } + + lock (_deferralGate) + { + if (!_operationInFlight && _deferredSends == null) + { + return false; + } + + (_deferredSends ??= new Queue()).Enqueue(() => 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. + /// + 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 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) + { + var producer = _messageProducer; + if (producer == null) + { + return; + } + + var deadline = DateTime.UtcNow + OutboundDrainWait; + while (!producer.IsIdle && DateTime.UtcNow < deadline) + { + await Task.Delay(10, cancellationToken).ConfigureAwait(false); + } + } + + #endregion + /// /// Connects to the device. /// @@ -1372,6 +1867,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 (HoldsOperationLock) + { + return false; + } + try { return _textExchangeLock.Wait(TextExchangeTeardownWait); @@ -1391,6 +1896,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 (HoldsOperationLock) + { + return false; + } + try { return await _textExchangeLock.WaitAsync(TextExchangeTeardownWait, cancellationToken) @@ -1460,6 +1972,11 @@ private void FinishDisconnect(bool lockAcquired, ConnectionStatus finalStatus) State = DeviceState.Disconnected; _isInitialized = false; _isDisconnecting = false; + + // 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 @@ -1475,6 +1992,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 +2021,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 +2060,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 +2322,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 = !HoldsOperationLock; + 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. + _operationLockGeneration.Value = Volatile.Read(ref _operationGeneration); + MarkOperationInFlight(); } _isInsideTextExchange.Value = true; @@ -1842,6 +2414,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 +2619,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) { + _operationLockGeneration.Value = 0; + 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