From f8138bfe8b657b62ad59ee0c24624921fdf02cc9 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Fri, 31 Jul 2026 13:26:15 -0600 Subject: [PATCH 1/3] fix(device): add an opt-in connect that leaves a running stream alone A DAQiFi unit has one global acquisition, and InitializeAsync unconditionally sends SYSTem:StopStreamData (plus POWer:STATe, STReam:FORmat, and the USB STReam:INTerface routing step). A second session connecting to a device that is already streaming therefore ends the first session's acquisition, silently. Adds DeviceConnectionOptions.PreserveActiveStream (and the matching DaqifiDevice property, plus a DeviceConnectionOptions.Observing preset) which skips exactly the initialization commands that write global stream state. The session still sends SYSTem:ECHO -1 and the read-only identity/capability queries, so it reaches Ready with populated channels. Default behavior is unchanged. Also documents the hazard in the README and DEVICE_INTERFACES, and points multi-session consumers at DaqifiDeviceRegistry. Closes #385 --- README.md | 7 ++ docs/DEVICE_INTERFACES.md | 67 ++++++++++++++++- .../Device/DaqifiDeviceFactoryTests.cs | 23 ++++++ .../Device/DaqifiDeviceInitializeTests.cs | 63 ++++++++++++++++ src/Daqifi.Core/Device/DaqifiDevice.cs | 71 +++++++++++++++++++ src/Daqifi.Core/Device/DaqifiDeviceFactory.cs | 9 ++- .../Device/DaqifiStreamingDevice.cs | 13 ++++ .../Device/DeviceConnectionOptions.cs | 37 ++++++++++ 8 files changed, 288 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 52bcff27..7629e260 100644 --- a/README.md +++ b/README.md @@ -135,6 +135,13 @@ var options = new DeviceConnectionOptions using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760, options); ``` +> **Connecting takes control of the device.** A DAQiFi unit has a single global acquisition, and the +> default connect sequence stops it — so connecting to a device another session is already streaming +> silently ends that session's data. Use `DeviceConnectionOptions.Observing` for a secondary session +> that only needs to look, and `DaqifiDeviceRegistry` to avoid opening the same unit twice in one +> process. See +> [Connecting stops any stream already running](docs/DEVICE_INTERFACES.md#connecting-stops-any-stream-already-running). + ### Device discovery ```csharp diff --git a/docs/DEVICE_INTERFACES.md b/docs/DEVICE_INTERFACES.md index 4c234f61..319a30eb 100644 --- a/docs/DEVICE_INTERFACES.md +++ b/docs/DEVICE_INTERFACES.md @@ -112,7 +112,8 @@ var options = new DeviceConnectionOptions MaxAttempts = 3, ConnectionTimeout = TimeSpan.FromSeconds(10) }, - InitializeDevice = true // Run init sequence after connect + InitializeDevice = true, // Run init sequence after connect + PreserveActiveStream = false // Take control of the device (see the warning below) }; ``` @@ -120,6 +121,13 @@ Pre-configured presets: - `DeviceConnectionOptions.Default` - Standard settings - `DeviceConnectionOptions.Fast` - Quick connection, fewer retries - `DeviceConnectionOptions.Resilient` - More retries, longer timeouts +- `DeviceConnectionOptions.Observing` - Connect without disturbing a stream already running + (see [Connecting stops any stream already running](#connecting-stops-any-stream-already-running)) + +> **Connecting stops whatever the device was streaming.** With the default options, initialization +> takes control of the device. If another session — another app, another process, another machine — +> was already streaming from that unit, its data stops, silently. +> See [Connecting stops any stream already running](#connecting-stops-any-stream-already-running). ## Usage Examples @@ -326,6 +334,59 @@ blocks other threads. Two concurrent `ConnectAsync` calls for the *same* physica open a connection — the loser is detected after connecting and disposed — so serialize your own calls if a single connect attempt matters. +### Connecting stops any stream already running + +A DAQiFi device has **one** acquisition: one ADC, one sample rate, one destination interface. There +is no per-connection stream. So the connect-time initialization sequence, which stops streaming, +sets the power state, fixes the stream format, and (over USB) routes the stream to this connection, +acts on the device as a whole — not on your connection. + +That is the right default when your session owns the device: it also clears a stream orphaned by a +previously crashed session, so a stale acquisition never leaks into a fresh one. But when a +**second** session connects to a device that is already streaming, the same sequence ends the first +session's acquisition. Neither side is told. The first app's data simply stops. + +Realistic ways to hit this, all silent to the victim: + +- A desktop app on USB while a script or second app talks to the same unit over WiFi +- Two instances of the same application +- A second viewer attaching to a unit a logger is already streaming + +**If you only need to look, connect as an observer.** `PreserveActiveStream` skips every +initialization command that writes global stream state: + +```csharp +// A secondary session that must not disturb whoever is already streaming. +var device = await DaqifiDeviceFactory.ConnectTcpAsync( + ip, DaqifiDeviceFactory.DefaultTcpDataPort, DeviceConnectionOptions.Observing); + +// Connecting manually? Set it before InitializeAsync. +device.PreserveActiveStream = true; +await device.InitializeAsync(); +``` + +| Initialization step | Default | `PreserveActiveStream` | +|---|---|---| +| `SYSTem:ECHO -1` | sent | sent — text-mode only, no stream state | +| `SYSTem:StopStreamData` | sent | **skipped** — this is what kills the other session | +| `SYSTem:POWer:STATe 1` | sent | **skipped** | +| `SYSTem:STReam:FORmat 0` | sent | **skipped** | +| `SYSTem:STReam:INTerface` (USB routing) | sent | **skipped** — would steal the stream | +| `SYSTem:SYSInfoPB?` + capability query | sent | sent — read-only | + +The observing session is fully usable for status, metadata, and channel inspection, and reaches +`DeviceState.Ready` exactly as a normal connection does. What it is **not** is configured to stream: +the format and destination interface are left as the other session set them, and frames keep going +wherever they were already going. A session that later wants to stream itself has to take control, +which necessarily stops the other one — reconnect with the default options for that. + +**Limits.** This is a courtesy, not arbitration. It stops *this* library from clobbering a stream; +it cannot stop anything else from doing so, and the firmware does not currently reject or announce +a second controlling session. Within one process, prefer +[`DaqifiDeviceRegistry`](#managing-multiple-devices-daqifideviceregistry) — it refuses to open the +same physical unit twice at all, so the conflict never arises. Across processes there is no +protection beyond both sides opting in to `PreserveActiveStream`. + ### Manual Device Connection (Advanced) For cases where you need more control over the connection process: @@ -341,6 +402,10 @@ await transport.ConnectAsync(new ConnectionRetryOptions { MaxAttempts = 3 }); // Create device with transport using var device = new DaqifiDevice("My Device", transport); device.Connect(); + +// InitializeAsync takes control of the device and stops any stream it was already running. +// Set device.PreserveActiveStream = true first if another session may be streaming — see +// "Connecting stops any stream already running" above. await device.InitializeAsync(); // Now ready to send commands diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceFactoryTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceFactoryTests.cs index 1fc296ba..15f5e757 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceFactoryTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceFactoryTests.cs @@ -20,6 +20,9 @@ public void DeviceConnectionOptions_DefaultValues_AreCorrect() Assert.Equal("DAQiFi Device", options.DeviceName); Assert.Null(options.ConnectionRetry); Assert.True(options.InitializeDevice); + // Connecting takes control of the device (and stops any running stream) unless + // explicitly opted out of — the historical behavior (#385). + Assert.False(options.PreserveActiveStream); } [Fact] @@ -32,6 +35,26 @@ public void DeviceConnectionOptions_Default_ReturnsDefaultOptions() Assert.Equal("DAQiFi Device", options.DeviceName); Assert.Null(options.ConnectionRetry); Assert.True(options.InitializeDevice); + Assert.False(options.PreserveActiveStream); + } + + [Fact] + public void DeviceConnectionOptions_Observing_PreservesActiveStream() + { + // Act + var options = DeviceConnectionOptions.Observing; + + // Assert — the opt-in preset still initializes, it just does so non-disruptively + Assert.True(options.PreserveActiveStream); + Assert.True(options.InitializeDevice); + } + + [Fact] + public void DeviceConnectionOptions_FastAndResilient_DoNotPreserveActiveStream() + { + // Assert — the existing presets keep the historical take-control behavior + Assert.False(DeviceConnectionOptions.Fast.PreserveActiveStream); + Assert.False(DeviceConnectionOptions.Resilient.PreserveActiveStream); } [Fact] diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs index 01192596..4968317c 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs @@ -32,6 +32,69 @@ public async Task InitializeAsync_SendsAllConfigCommands() Assert.Contains(sentData, d => d.Contains("SYSTem:POWer:STATe 1")); Assert.Contains(sentData, d => d.Contains("SYSTem:STReam:FORmat 0")); Assert.Contains(sentData, d => d.Contains("SYSTem:SYSInfoPB?")); + + // The stream-disruptive commands are the default because this session is assumed to + // own the device; the opt-out below must not change that (#385). + Assert.False(device.PreserveActiveStream); + } + + [Fact] + public async Task InitializeAsync_WithPreserveActiveStream_OmitsStreamDisruptiveCommands() + { + // Arrange — a secondary session attaching to a device another session may already be + // streaming from. + var device = new TestableDaqifiDevice("TestDevice") { PreserveActiveStream = true }; + device.Connect(); + + // Act + await device.InitializeAsync(); + + // Assert — nothing that halts, powers, or reconfigures the device's single global + // stream is sent. + var sentData = device.DirectSentMessages.Select(m => m.Data).ToList(); + + Assert.DoesNotContain(sentData, d => d.Contains("SYSTem:StopStreamData")); + Assert.DoesNotContain(sentData, d => d.Contains("SYSTem:POWer:STATe")); + Assert.DoesNotContain(sentData, d => d.Contains("SYSTem:STReam:FORmat")); + + // ...but the session is still fully set up: echo off so its own replies parse, and the + // identity query that populates channels. + Assert.Contains(sentData, d => d.Contains("SYSTem:ECHO -1")); + Assert.Contains(sentData, d => d.Contains("SYSTem:SYSInfoPB?")); + } + + [Fact] + public async Task InitializeAsync_WithPreserveActiveStream_StillProducesReadyPopulatedDevice() + { + // Arrange + var device = new TestableDaqifiDevice("TestDevice") { PreserveActiveStream = true }; + device.Connect(); + + // Act + await device.InitializeAsync(); + + // Assert — skipping the disruptive commands must not cost the caller a usable session + Assert.Equal(DeviceState.Ready, device.State); + Assert.Equal(4, device.Channels.Count); // 2 analog + 2 digital + } + + [Fact] + public async Task InitializeAsync_WithPreserveActiveStream_StreamingUsb_DoesNotRouteStreamToUsb() + { + // Arrange — over USB the streaming device normally claims the stream with + // SYSTem:STReam:INTerface 0, which would take data away from a session already + // receiving it over WiFi. + var device = new TestableStreamingDevice("TestDevice") { PreserveActiveStream = true }; + device.Connect(); + + // Act + await device.InitializeAsync(); + + // Assert + Assert.DoesNotContain(device.SentData, d => d.Contains("SYSTem:STReam:INTerface")); + Assert.Equal(0, device.UsbStepAttemptCount); + Assert.Equal(DeviceState.Ready, device.State); + Assert.Equal(4, device.Channels.Count); } [Fact] diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 3e28d755..149cb15e 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -312,6 +312,52 @@ protected void WithChannelsLock(Action action) /// public DeviceState State { get; private set; } = DeviceState.Disconnected; + /// + /// When true, omits the initialization commands that + /// would halt or re-route a stream the device is already running, so connecting does not + /// disturb a session started elsewhere. Default is false — the historical behavior, + /// where connecting takes control of the device. + /// + /// + /// + /// Streaming is a single global device state: one acquisition at one rate, delivered to one + /// interface. The default initialization sequence stops that stream, sets the device's power + /// state, fixes the stream format, and (over USB) routes the stream to this connection. That + /// is correct when this session owns the device — it also clears a stream orphaned by a + /// previously crashed session — but a second session running it silently ends the + /// first session's acquisition, with no error surfaced to either side. + /// + /// + /// With this set, initialization sends only SYSTem:ECHO -1 (so replies to this + /// connection can be parsed) followed by the read-only identity and capability queries. + /// Skipped are SYSTem:StopStreamData, SYSTem:POWer:STATe 1, + /// SYSTem:STReam:FORmat 0, and the USB SYSTem:STReam:INTerface routing step. + /// + /// + /// The resulting session is fully usable for status, metadata, channel inspection, and any + /// command the caller chooses to send. It is not configured to stream: the device's + /// stream format and destination interface are left exactly as the other session left them, + /// and stream frames continue to go wherever they were already going. A session that later + /// wants to stream itself must take control of the device, which necessarily stops whatever + /// the other session was doing. + /// + /// + /// Read once, when runs; changing it afterwards has no effect. + /// This guards only against this library's own connect sequence — it is not device-side + /// arbitration, and two processes can still fight over one unit. Within a single process, + /// prefer , which refuses to open the same physical unit + /// twice. + /// + /// + public bool PreserveActiveStream { get; set; } + + /// + /// The value captured for the initialization currently in + /// progress. Derived classes read this from to decide + /// whether their own initialization may write global stream state. + /// + protected bool InitializationPreservesActiveStream { get; private set; } + private ConnectionStatus _status; /// @@ -1465,6 +1511,13 @@ public void Dispose() /// 4. Set protobuf message format /// 5. Query device info and block until the device reports its channel configuration /// + /// Steps 2-4 write global device state. Streaming is a single global state on a DAQiFi + /// device, so step 2 stops a stream any session started, not just this one: + /// connecting to a device another session is already streaming silently ends that session's + /// acquisition. That is the right default when this session owns the device (it also clears a + /// stream orphaned by a crashed session). Set before + /// calling this to skip steps 2-4 and connect as a non-disruptive observer instead. + /// /// Rather than returning after a fixed delay, the method awaits the first /// event so callers receive a fully populated device. /// Serial/CDC devices can take noticeably longer than the previous fixed wait to send @@ -1525,6 +1578,11 @@ public virtual async Task InitializeAsync( _messageConsumer.MessageReceived += OnInboundMessageReceived; } + // Snapshot once so every retry attempt — and the derived-class hook further down — + // sees the same decision even if a caller mutates the property mid-initialization. + var preserveActiveStream = PreserveActiveStream; + InitializationPreservesActiveStream = preserveActiveStream; + // Send the text-mode SCPI setup commands via ExecuteTextCommandAsync so that // any -200 execution error response is captured rather than silently discarded // by the protobuf consumer. The protobuf consumer is stopped for the duration @@ -1546,7 +1604,20 @@ public virtual async Task InitializeAsync( initLines = await ExecuteTextCommandAsync(() => { + // Echo is a per-device text-mode setting, not stream state: this session + // needs it off to parse its own replies, and the value is the same one any + // other Core session already set. Safe to send either way. Send(ScpiMessageProducer.DisableDeviceEcho); + + // Everything below writes global stream state. A secondary "observe" + // session must not touch it — StopStreamData ends another session's + // acquisition outright (#385), and the power-state and stream-format + // commands reconfigure the same single acquisition it is running. + if (preserveActiveStream) + { + return; + } + Thread.Sleep(100); Send(ScpiMessageProducer.StopStreaming); diff --git a/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs b/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs index 743588a5..9b88b163 100644 --- a/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs +++ b/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs @@ -384,6 +384,7 @@ private static async Task ConnectWiFiDeviceAsync( ConnectionRetry = effectiveOptions.ConnectionRetry, InitializeDevice = effectiveOptions.InitializeDevice, ChannelPopulationTimeout = effectiveOptions.ChannelPopulationTimeout, + PreserveActiveStream = effectiveOptions.PreserveActiveStream, Logger = effectiveOptions.Logger }; @@ -427,6 +428,7 @@ private static async Task ConnectSerialDeviceAsync( ConnectionRetry = effectiveOptions.ConnectionRetry, InitializeDevice = effectiveOptions.InitializeDevice, ChannelPopulationTimeout = effectiveOptions.ChannelPopulationTimeout, + PreserveActiveStream = effectiveOptions.PreserveActiveStream, Logger = effectiveOptions.Logger }; @@ -470,7 +472,12 @@ private static async Task ConnectWithTransportAsync( // Step 2: Create the device with the transport // Note: Once created, the device owns the transport and will dispose it - device = new DaqifiStreamingDevice(options.DeviceName, transport, options.Logger); + device = new DaqifiStreamingDevice(options.DeviceName, transport, options.Logger) + { + // Read by InitializeAsync below; set before Connect so it is never observed + // half-applied. + PreserveActiveStream = options.PreserveActiveStream + }; // Step 3: Connect the device (starts message producers/consumers) device.Connect(); diff --git a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs index d2fcaa7a..105522d4 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -280,6 +280,11 @@ private void InitializeStreamingDevice() /// This runs inside the base exception handling /// (before the device is marked initialized/ready), so a cancellation or SCPI error here /// leaves the device in a consistent state and re-initializable, rather than falsely Ready. + /// + /// The routing command is global device state: it takes the stream away from whatever + /// interface it was going to, so a second session running it steals another session's data + /// (#385). It is therefore skipped entirely when + /// is set. /// /// A cancellation token to observe while initializing. /// A task representing the asynchronous initialization operation. @@ -296,6 +301,14 @@ protected override async Task OnDeviceInitializingAsync(CancellationToken cancel return; } + // An observe-only session must not re-route the device's single global stream: doing so + // would take the data away from the session that is already receiving it (#385). The + // interface is left exactly as the owning session configured it. + if (InitializationPreservesActiveStream) + { + return; + } + // Direct streaming to the USB interface. Uses ExecuteTextCommandAsync so the // command is sent in text mode (protobuf consumer temporarily stopped) and any // SCPI error response is captured rather than garbling the protobuf stream. diff --git a/src/Daqifi.Core/Device/DeviceConnectionOptions.cs b/src/Daqifi.Core/Device/DeviceConnectionOptions.cs index 3e7fb131..c908e58a 100644 --- a/src/Daqifi.Core/Device/DeviceConnectionOptions.cs +++ b/src/Daqifi.Core/Device/DeviceConnectionOptions.cs @@ -37,6 +37,34 @@ public class DeviceConnectionOptions /// public TimeSpan ChannelPopulationTimeout { get; set; } = TimeSpan.FromSeconds(8); + /// + /// Gets or sets a value indicating whether initialization must leave a stream that is already + /// running on the device untouched. Default is false, which preserves the historical + /// behavior: connecting takes control of the device and stops whatever it was streaming. + /// + /// + /// + /// Streaming is a single global device state — one acquisition, one destination interface — so + /// the default initialization sequence deliberately clears it, which is right for the usual + /// single-session case (it also clears a stream orphaned by a crashed session). When a second + /// session connects to a device another session is already streaming, that same sequence + /// silently ends the first session's acquisition. + /// + /// + /// Set this to true for a secondary "observe" connection. Initialization then omits every + /// command that halts or re-routes the device's stream, and only queries the device for its + /// identity and channel configuration. See + /// for exactly which commands are skipped and + /// what the resulting session can and cannot do. + /// + /// + /// This only protects against this library's connect sequence. Two processes can still + /// fight over one device; within a process, route connections through + /// so the same physical unit is not opened twice at all. + /// + /// + public bool PreserveActiveStream { get; set; } + /// /// Optional logger the constructed device routes its diagnostics through (bad calibration/ /// resolution warnings, SCPI text-exchange timing). When null, the device uses a no-op logger. @@ -65,4 +93,13 @@ public class DeviceConnectionOptions { ConnectionRetry = ConnectionRetryOptions.Resilient }; + + /// + /// Creates a configuration for a secondary session that must not disturb a stream another + /// session may already be running on the device. Sets . + /// + public static DeviceConnectionOptions Observing => new() + { + PreserveActiveStream = true + }; } From 04c25147b6478bf6db022b419f897f74d25862ad Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Fri, 31 Jul 2026 13:45:48 -0600 Subject: [PATCH 2/3] fix(device): pass the observing decision to the init hook instead of storing it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review caught a real race. The PreserveActiveStream decision was held in an instance field between InitializeAsync capturing it and the derived hook reading it, so a second InitializeAsync starting on the same device while the first was still in flight overwrote it. The first initialization's USB routing step then acted on the second one's decision — an observing session could route the stream to itself, which is exactly the stream-stealing this change exists to prevent. Removes the field and passes the decision to OnDeviceInitializingAsync as a parameter, so it lives on the stack and no concurrent operation can reach it. Signature change to the protected hook; DaqifiStreamingDevice is the only overrider. Adds a regression test that overlaps two initializations on one device and asserts each hook receives its own decision (fails as {false, false} against the previous implementation), plus two tests pinning that the decision is read once at the start and not re-read later. No change to the SCPI sent on the wire for either path. --- .../Device/DaqifiDeviceInitializeTests.cs | 162 ++++++++++++++++++ src/Daqifi.Core/Device/DaqifiDevice.cs | 29 ++-- .../Device/DaqifiStreamingDevice.cs | 14 +- 3 files changed, 189 insertions(+), 16 deletions(-) diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs index 4968317c..33529f86 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs @@ -97,6 +97,48 @@ public async Task InitializeAsync_WithPreserveActiveStream_StreamingUsb_DoesNotR Assert.Equal(4, device.Channels.Count); } + [Fact] + public async Task InitializeAsync_WhenPreserveActiveStreamChangesMidInitialization_KeepsTheDecisionItStartedWith() + { + // Arrange — an observing initialization whose flag is flipped to false after it has + // begun (a concurrent initialization on the same instance, or a caller mutating the + // property). The decision belongs to the operation, so the USB routing step must still + // be skipped: honoring the late change would steal a stream this session promised not + // to touch. + var device = new TestableStreamingDevice("TestDevice") { PreserveActiveStream = true }; + device.MutateDuringInitialization = () => device.PreserveActiveStream = false; + device.Connect(); + + // Act + await device.InitializeAsync(); + + // Assert + Assert.False(device.PreserveActiveStream); // the mutation really did land + Assert.DoesNotContain(device.SentData, d => d.Contains("SYSTem:STReam:INTerface")); + Assert.Equal(0, device.UsbStepAttemptCount); + Assert.Equal(DeviceState.Ready, device.State); + } + + [Fact] + public async Task InitializeAsync_WhenPreserveActiveStreamIsSetMidInitialization_StillTakesControl() + { + // Arrange — the mirror case: a normal take-control initialization must not be silently + // downgraded to observing by a flag set after it started, which would leave the stream + // routed somewhere this session cannot read. + var device = new TestableStreamingDevice("TestDevice"); + device.MutateDuringInitialization = () => device.PreserveActiveStream = true; + device.Connect(); + + // Act + await device.InitializeAsync(); + + // Assert + Assert.True(device.PreserveActiveStream); // the mutation really did land + Assert.Contains(device.SentData, d => d.Contains("SYSTem:STReam:INTerface 0")); + Assert.Contains(device.SentData, d => d.Contains("SYSTem:StopStreamData")); + Assert.Equal(DeviceState.Ready, device.State); + } + [Fact] public async Task InitializeAsync_SendsGetDeviceInfo() { @@ -530,6 +572,111 @@ private enum UsbStepBehavior Cancel } + [Fact] + public async Task InitializeAsync_WhenTwoInitializationsOverlapOnOneDevice_EachKeepsItsOwnDecision() + { + // Arrange — the race reported against the first cut of this change: the observing + // decision was held in an instance field, so a second InitializeAsync starting while + // the first was still in flight overwrote it, and the first initialization's USB + // routing step then acted on the second one's decision. Reproduced by holding an + // observing initialization inside its first SCPI exchange — past the point the + // decision is made, before the routing step — while a take-control initialization + // starts on the same instance. + var device = new OverlappingInitDevice("TestDevice"); + device.Connect(); + + device.PreserveActiveStream = true; + var observing = device.InitializeAsync(); + + // Wait until the observing call is parked inside its first exchange. + await device.FirstExchangeEntered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + device.PreserveActiveStream = false; + var takingControl = device.InitializeAsync(); + await takingControl.WaitAsync(TimeSpan.FromSeconds(5)); + + // Act — let the observing initialization finish, now that the flag has moved under it. + device.ReleaseFirstExchange(); + await observing.WaitAsync(TimeSpan.FromSeconds(5)); + + // Assert — one initialization decided to observe and one to take control, and each + // reached its own hook with its own decision. A shared field yields {false, false}. + Assert.Equal( + new[] { false, true }, + device.HookDecisions.OrderBy(flag => flag).ToArray()); + } + + /// + /// A testable USB streaming device that records the decision each initialization passes to + /// OnDeviceInitializingAsync, and can park its first text exchange so a second + /// initialization can be started while the first is still in flight. + /// + private class OverlappingInitDevice : DaqifiStreamingDevice + { + private readonly TaskCompletionSource _release = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _exchangeCount; + + /// Completes once an initialization has entered its first text exchange. + public TaskCompletionSource FirstExchangeEntered { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + /// The decision each initialization handed to the derived hook. + public System.Collections.Concurrent.ConcurrentBag HookDecisions { get; } = new(); + + public override bool IsUsbConnection => true; + + public OverlappingInitDevice(string name) : base(name, (IPAddress?)null) { } + + public void ReleaseFirstExchange() => _release.TrySetResult(); + + public override void Send(IOutboundMessage message) + { + if (message is IOutboundMessage stringMessage && + stringMessage.Data.Contains("SYSInfoPB")) + { + PopulateChannelsFromStatus(new DaqifiOutMessage + { + AnalogInPortNum = 2, + DigitalPortNum = 2 + }); + } + } + + protected override Task OnDeviceInitializingAsync( + bool preserveActiveStream, + CancellationToken cancellationToken) + { + HookDecisions.Add(preserveActiveStream); + return Task.CompletedTask; + } + + protected override async Task> ExecuteTextCommandAsync( + Action setupAction, + int responseTimeoutMs = 1000, + int completionTimeoutMs = 250, + CancellationToken cancellationToken = default, + Func? prepareAsync = null) + { + if (prepareAsync != null) + { + await prepareAsync(cancellationToken).ConfigureAwait(false); + } + + setupAction(); + + // Park only the very first exchange — the one belonging to the initialization the + // test starts first. + if (Interlocked.Increment(ref _exchangeCount) == 1) + { + FirstExchangeEntered.TrySetResult(); + await _release.Task.ConfigureAwait(false); + } + + return Array.Empty(); + } + } + /// /// A testable DaqifiStreamingDevice (always USB) whose base init populates channels on /// GetDeviceInfo and whose USB stream-interface step can be made to succeed, return a SCPI @@ -547,6 +694,15 @@ private class TestableStreamingDevice : DaqifiStreamingDevice /// public int UsbStepAttemptCount { get; private set; } + /// + /// Invoked once, from inside the first text exchange of initialization — i.e. after + /// InitializeAsync has captured its PreserveActiveStream decision but before the + /// derived USB step runs. Lets a test mutate device state mid-initialization. + /// + public Action? MutateDuringInitialization { get; set; } + + private bool _mutationApplied; + public override bool IsUsbConnection => true; public TestableStreamingDevice(string name, UsbStepBehavior usbStepBehavior = UsbStepBehavior.Succeed) @@ -585,6 +741,12 @@ protected override async Task> ExecuteTextCommandAsync( await prepareAsync(cancellationToken).ConfigureAwait(false); } + if (!_mutationApplied && MutateDuringInitialization != null) + { + _mutationApplied = true; + MutateDuringInitialization(); + } + var before = _sent.Count; setupAction(); var sentThisCall = _sent.Skip(before).ToList(); diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 149cb15e..c4c8f9da 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -351,13 +351,6 @@ protected void WithChannelsLock(Action action) /// public bool PreserveActiveStream { get; set; } - /// - /// The value captured for the initialization currently in - /// progress. Derived classes read this from to decide - /// whether their own initialization may write global stream state. - /// - protected bool InitializationPreservesActiveStream { get; private set; } - private ConnectionStatus _status; /// @@ -1578,10 +1571,13 @@ public virtual async Task InitializeAsync( _messageConsumer.MessageReceived += OnInboundMessageReceived; } - // Snapshot once so every retry attempt — and the derived-class hook further down — - // sees the same decision even if a caller mutates the property mid-initialization. + // Snapshot into a local, once. Every retry attempt and the derived-class hook + // further down are then handed the same decision explicitly, so it cannot be + // changed out from under an initialization already in flight — neither by a caller + // mutating the property nor by a second concurrent InitializeAsync on this same + // instance. The decision belongs to this operation, so it lives on the stack rather + // than in a field. var preserveActiveStream = PreserveActiveStream; - InitializationPreservesActiveStream = preserveActiveStream; // Send the text-mode SCPI setup commands via ExecuteTextCommandAsync so that // any -200 execution error response is captured rather than silently discarded @@ -1674,7 +1670,7 @@ await WaitForChannelsPopulatedAsync( // this try/catch so a failure there leaves the device in a consistent terminal state // rather than a falsely-ready device. _isInitialized is only set after it succeeds, // so a failed init can be safely retried. - await OnDeviceInitializingAsync(cancellationToken).ConfigureAwait(false); + await OnDeviceInitializingAsync(preserveActiveStream, cancellationToken).ConfigureAwait(false); _isInitialized = true; State = DeviceState.Ready; @@ -1705,9 +1701,18 @@ await WaitForChannelsPopulatedAsync( /// state and other faults set — rather than a falsely-ready /// device, and the failed initialization can be retried. The base implementation does nothing. /// + /// + /// The decision for this initialization, passed + /// explicitly rather than read from the device so it cannot change while initialization is in + /// flight. When true, an override must not send any command that writes global stream + /// state — stopping, reconfiguring, or re-routing the stream would disturb a session that is + /// already using the device. + /// /// A cancellation token to observe. /// A task representing the asynchronous operation. - protected virtual Task OnDeviceInitializingAsync(CancellationToken cancellationToken) => Task.CompletedTask; + protected virtual Task OnDeviceInitializingAsync( + bool preserveActiveStream, + CancellationToken cancellationToken) => Task.CompletedTask; /// /// Runs the capability-document read during initialization, absorbing any failure. diff --git a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs index 105522d4..26e80e73 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -283,9 +283,13 @@ private void InitializeStreamingDevice() /// /// The routing command is global device state: it takes the stream away from whatever /// interface it was going to, so a second session running it steals another session's data - /// (#385). It is therefore skipped entirely when - /// is set. + /// (#385). It is therefore skipped entirely when + /// is set. /// + /// + /// When true, this initialization must leave a stream another session is already + /// running untouched, so the routing command is not sent at all. + /// /// A cancellation token to observe while initializing. /// A task representing the asynchronous initialization operation. /// @@ -294,7 +298,9 @@ private void InitializeStreamingDevice() /// command because it still has the interface set from a prior WiFi-streaming session, /// within the tight response window right after connect. /// - protected override async Task OnDeviceInitializingAsync(CancellationToken cancellationToken) + protected override async Task OnDeviceInitializingAsync( + bool preserveActiveStream, + CancellationToken cancellationToken) { if (!IsUsbConnection) { @@ -304,7 +310,7 @@ protected override async Task OnDeviceInitializingAsync(CancellationToken cancel // An observe-only session must not re-route the device's single global stream: doing so // would take the data away from the session that is already receiving it (#385). The // interface is left exactly as the owning session configured it. - if (InitializationPreservesActiveStream) + if (preserveActiveStream) { return; } From 1049355aaad67fafca0faaaa80db707a6885a1d9 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Fri, 31 Jul 2026 14:03:44 -0600 Subject: [PATCH 3/3] fix(device): never report Ready for a cancelled initialization Review caught that an observing initialization could swallow a late cancellation. After channels populate, nothing on that path is guaranteed to observe the token: the capability read returns early on firmware that does not advertise the document, channel population can short-circuit when the status arrives synchronously, and the observing hook returns immediately because it has no work to do. A caller that cancelled in that window got a successfully completed task and a device reporting Ready. Enforces the invariant in InitializeAsync, immediately before the transition to Ready, rather than patching the one early return that was flagged. The same hole exists on the pre-existing non-USB early return and in any third-party override with no awaitable work; guarding the transition covers all of them. Cancellation here follows the path already documented for this method: state reverts to Connected, _isInitialized stays false, and the initialization can be retried. Adds a regression test that cancels at exactly that seam, covering both the observing and take-control paths. It fails without the guard with "No exception was thrown". No change to the SCPI sent on the wire. --- .../Device/DaqifiDeviceInitializeTests.cs | 75 +++++++++++++++++++ src/Daqifi.Core/Device/DaqifiDevice.cs | 9 +++ .../Device/DaqifiStreamingDevice.cs | 4 + 3 files changed, 88 insertions(+) diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs index 33529f86..7549e20d 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs @@ -572,6 +572,81 @@ private enum UsbStepBehavior Cancel } + [Theory] + [InlineData(true)] // observing: the hook returns immediately, no awaitable work + [InlineData(false)] // take-control: the hook does send SCPI + public async Task InitializeAsync_WhenCancelledAfterChannelsPopulate_DoesNotReportReady(bool preserveActiveStream) + { + // Arrange — cancellation lands at the one seam nothing was guaranteed to observe: + // after channels populate, during the capability read. Firmware that does not + // advertise the capability document returns from that read without touching the + // token, and on the observing path the derived hook then returns immediately too, so + // a device whose caller had already cancelled still reached Ready. + using var cts = new CancellationTokenSource(); + var device = new CancelDuringCapabilityReadDevice("TestDevice", cts) + { + PreserveActiveStream = preserveActiveStream + }; + device.Connect(); + + // Act & Assert — a cancelled initialization reports cancellation, not success. + await Assert.ThrowsAsync( + () => device.InitializeAsync(TimeSpan.FromSeconds(5), cts.Token)); + + Assert.NotEqual(DeviceState.Ready, device.State); + Assert.Equal(DeviceState.Connected, device.State); + } + + /// + /// A testable USB streaming device that cancels the supplied token from inside the + /// capability read — i.e. after channels have populated and before the derived + /// initialization hook — and returns without observing the token itself, exactly as the + /// real read does on firmware that does not advertise a capability document. + /// + private class CancelDuringCapabilityReadDevice : DaqifiStreamingDevice + { + private readonly CancellationTokenSource _cts; + + public override bool IsUsbConnection => true; + + public CancelDuringCapabilityReadDevice(string name, CancellationTokenSource cts) + : base(name, (IPAddress?)null) + { + _cts = cts; + } + + public override void Send(IOutboundMessage message) + { + if (message is IOutboundMessage stringMessage && + stringMessage.Data.Contains("SYSInfoPB")) + { + PopulateChannelsFromStatus(new DaqifiOutMessage + { + AnalogInPortNum = 2, + DigitalPortNum = 2 + }); + } + } + + public override Task ReadCapabilityDocumentAsync( + CancellationToken cancellationToken = default) + { + _cts.Cancel(); + return Task.FromResult(null); + } + + protected override Task> ExecuteTextCommandAsync( + Action setupAction, + int responseTimeoutMs = 1000, + int completionTimeoutMs = 250, + CancellationToken cancellationToken = default, + Func? prepareAsync = null) + { + setupAction(); + return Task.FromResult>(Array.Empty()); + } + } + [Fact] public async Task InitializeAsync_WhenTwoInitializationsOverlapOnOneDevice_EachKeepsItsOwnDecision() { diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index c4c8f9da..6847265e 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -1672,6 +1672,15 @@ await WaitForChannelsPopulatedAsync( // so a failed init can be safely retried. await OnDeviceInitializingAsync(preserveActiveStream, cancellationToken).ConfigureAwait(false); + // A cancelled initialization must never report Ready. Nothing above is guaranteed + // to observe the token: the capability read returns early on firmware that does not + // advertise the document, channel population can short-circuit when the status + // arrives synchronously, and a derived hook may legitimately have no awaitable work + // (the observing path and the non-USB path both return immediately). So the + // invariant is enforced here, at the one transition that matters, rather than relying + // on every path and every override to check for itself. + cancellationToken.ThrowIfCancellationRequested(); + _isInitialized = true; State = DeviceState.Ready; } diff --git a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs index 26e80e73..48643941 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -310,6 +310,10 @@ protected override async Task OnDeviceInitializingAsync( // An observe-only session must not re-route the device's single global stream: doing so // would take the data away from the session that is already receiving it (#385). The // interface is left exactly as the owning session configured it. + // + // Returning without observing the token is deliberate: there is no work to abandon, and + // InitializeAsync re-checks cancellation before it marks the device Ready, so a token + // cancelled during this hook is still honored. if (preserveActiveStream) { return;