diff --git a/README.md b/README.md index 7629e260..dd37b058 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,7 @@ using Daqifi.Core.Channel; // Connect — transport and device initialization handled for you. The factory returns the base // DaqifiDevice type, but the constructed instance is always a DaqifiStreamingDevice. -using var device = (DaqifiStreamingDevice)await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = (DaqifiStreamingDevice)await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); // Subscribe to decoded, per-channel samples var ai0 = device.GetChannelsSnapshot().First(c => c.Type == ChannelType.Analog && c.ChannelNumber == 0); @@ -97,7 +97,7 @@ Pick whichever transport fits your setup — each snippet is a standalone, copy- **TCP with a resilient retry preset** (5 retries, longer timeouts): ```csharp -using var device = await DaqifiDeviceFactory.ConnectTcpAsync( +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync( "192.168.1.100", 9760, DeviceConnectionOptions.Resilient); ``` @@ -106,7 +106,7 @@ using var device = await DaqifiDeviceFactory.ConnectTcpAsync( ```csharp // Replace with your OS-specific port: // Windows: "COM3" • macOS: "/dev/cu.usbmodem1" • Linux: "/dev/ttyACM0" -using var device = await DaqifiDeviceFactory.ConnectSerialAsync("COM3"); +await using var device = await DaqifiDeviceFactory.ConnectSerialAsync("COM3"); ``` **From a discovered device:** @@ -114,7 +114,7 @@ using var device = await DaqifiDeviceFactory.ConnectSerialAsync("COM3"); ```csharp using var finder = new WiFiDeviceFinder(); var devices = await finder.DiscoverAsync(TimeSpan.FromSeconds(5)); -using var device = await DaqifiDeviceFactory.ConnectFromDeviceInfoAsync(devices.First()); +await using var device = await DaqifiDeviceFactory.ConnectFromDeviceInfoAsync(devices.First()); ``` ### Custom retry options @@ -132,7 +132,7 @@ var options = new DeviceConnectionOptions }, InitializeDevice = true }; -using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760, options); +await 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 diff --git a/docs/DEVICE_INTERFACES.md b/docs/DEVICE_INTERFACES.md index 5bc4266b..bb719fb2 100644 --- a/docs/DEVICE_INTERFACES.md +++ b/docs/DEVICE_INTERFACES.md @@ -15,7 +15,7 @@ using Daqifi.Core.Device; using Daqifi.Core.Communication.Producers; // Connect to a device (handles transport, connection, and initialization) -using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); // Subscribe to incoming data device.MessageReceived += (sender, e) => @@ -150,7 +150,7 @@ foreach (var deviceInfo in devices) // Connect to the first discovered device if (devices.Any()) { - using var device = await DaqifiDeviceFactory.ConnectFromDeviceInfoAsync(devices.First()); + await using var device = await DaqifiDeviceFactory.ConnectFromDeviceInfoAsync(devices.First()); // Device is ready to use } ``` @@ -396,29 +396,51 @@ For cases where you need more control over the connection process: using Daqifi.Core.Device; using Daqifi.Core.Communication.Transport; -// Create and connect transport manually +// Create and connect transport manually. Every step takes a CancellationToken, so a user who +// cancels stops the attempt where it stands instead of waiting out the retries. var transport = new TcpStreamTransport("192.168.1.100", 9760); -await transport.ConnectAsync(new ConnectionRetryOptions { MaxAttempts = 3 }); +await transport.ConnectAsync(new ConnectionRetryOptions { MaxAttempts = 3 }, cancellationToken); // Create device with transport -using var device = new DaqifiDevice("My Device", transport); -device.Connect(); +await using var device = new DaqifiDevice("My Device", transport); +await device.ConnectAsync(cancellationToken); // 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(); +await device.InitializeAsync(cancellationToken: cancellationToken); // Now ready to send commands device.Send(ScpiMessageProducer.GetDeviceInfo); ``` +### Connecting and disconnecting without blocking + +`ConnectAsync`, `DisconnectAsync` and `DisposeAsync` are the non-blocking forms of `Connect`, +`Disconnect` and `Dispose`. Prefer them on a UI thread: the synchronous disconnect waits (up to ten +seconds) for any command exchange still in flight, which on a UI thread is a visible freeze. + +```csharp +// Disposal never blocks the caller — this is the recommended pattern. +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +``` + +A cancelled `ConnectAsync` throws `OperationCanceledException` and leaves nothing half-open — if the +transport had already come up when the cancel landed, it is closed again. + +`DisconnectAsync` treats its token differently, and deliberately: cancelling it skips the wait for an +in-flight command exchange and proceeds straight to teardown. It never aborts the disconnect and +never throws `OperationCanceledException`, because a teardown abandoned part-way would leave the +device in an indeterminate state. On return the device is always disconnected. + +The synchronous `Connect`, `Disconnect` and `Dispose` remain, unchanged, for existing callers. + ### Error Handling ```csharp try { - using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); + await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); device.Send(ScpiMessageProducer.StartStreaming(100)); } catch (ArgumentOutOfRangeException ex) when (ex.ParamName == "port") @@ -485,7 +507,7 @@ Notes: ### Connection Status Monitoring ```csharp -using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); device.StatusChanged += (sender, args) => { @@ -565,7 +587,7 @@ The failures that feed this escalation are also reported individually, as they h After initialization, device metadata is populated: ```csharp -using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); // Access device information Console.WriteLine($"Part Number: {device.Metadata.PartNumber}"); @@ -600,7 +622,7 @@ using Daqifi.Core.Device; // DaqifiDeviceFactory methods return the base DaqifiDevice type, but the constructed instance is // always a DaqifiStreamingDevice — cast (or pattern-match with `is`) to reach its streaming API. -using var device = (DaqifiStreamingDevice)await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = (DaqifiStreamingDevice)await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); var ai0 = device.GetChannelsSnapshot().First(c => c.Type == ChannelType.Analog && c.ChannelNumber == 0); ai0.SampleReceived += (sender, e) => @@ -632,7 +654,7 @@ cancellation and backpressure instead of hand-building an event/queue bridge. Ea the decoded `IDataSample` with the `IChannel` that produced it. ```csharp -using var device = (DaqifiStreamingDevice)await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = (DaqifiStreamingDevice)await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); device.EnableChannels(device.GetChannelsSnapshot().Where(c => c.Type == ChannelType.Analog)); device.StreamingFrequency = 100; // Hz @@ -655,7 +677,7 @@ call `StopStreaming()` for that. This is additive: `SampleReceived` and `Message #### Raw protobuf frames ```csharp -using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); var sampleCount = 0; device.MessageReceived += (sender, e) => @@ -767,7 +789,7 @@ to reach these members — the same way as `INetworkConfigurable` below. ```csharp using Daqifi.Core.Device.Diagnostics; -using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); if (device is IDeviceDiagnostics diagnostics) { diff --git a/src/Daqifi.Core.Tests/Communication/Transport/ConnectRetryExecutorTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/ConnectRetryExecutorTests.cs index 624f1349..6120b0b2 100644 --- a/src/Daqifi.Core.Tests/Communication/Transport/ConnectRetryExecutorTests.cs +++ b/src/Daqifi.Core.Tests/Communication/Transport/ConnectRetryExecutorTests.cs @@ -30,7 +30,7 @@ public async Task ExecuteAsync_SucceedsFirstAttempt_ReportsConnectedOnce() await ConnectRetryExecutor.ExecuteAsync( FastRetry(3), - connectAttempt: _ => { attempts++; return Task.CompletedTask; }, + connectAttempt: (_, _) => { attempts++; return Task.CompletedTask; }, onAttemptFailed: () => failedCleanups++, onStatusChanged: (c, e) => statuses.Add((c, e))); @@ -49,7 +49,7 @@ public async Task ExecuteAsync_ReceivesResolvedOptions_InConnectAttempt() await ConnectRetryExecutor.ExecuteAsync( options, - connectAttempt: o => { seen = o; return Task.CompletedTask; }, + connectAttempt: (o, _) => { seen = o; return Task.CompletedTask; }, onAttemptFailed: () => { }, onStatusChanged: (_, _) => { }); @@ -66,7 +66,7 @@ public async Task ExecuteAsync_NullOptions_UsesSingleNoRetryAttempt() var thrown = await Assert.ThrowsAsync(() => ConnectRetryExecutor.ExecuteAsync( retryOptions: null, - connectAttempt: _ => { attempts++; throw boom; }, + connectAttempt: (_, _) => { attempts++; throw boom; }, onAttemptFailed: () => failedCleanups++, onStatusChanged: (_, _) => { })); @@ -84,7 +84,7 @@ public async Task ExecuteAsync_RetriesUntilSuccess_CleansUpEachFailedAttempt() await ConnectRetryExecutor.ExecuteAsync( FastRetry(3), - connectAttempt: _ => + connectAttempt: (_, _) => { attempts++; if (attempts < 3) @@ -120,7 +120,7 @@ public async Task ExecuteAsync_AllAttemptsFail_ThrowsLastExceptionAndReportsItUn var thrown = await Assert.ThrowsAsync(() => ConnectRetryExecutor.ExecuteAsync( FastRetry(3), - connectAttempt: _ => + connectAttempt: (_, _) => { attempts++; throw attempts < 3 ? new InvalidOperationException($"fail {attempts}") : lastBoom; @@ -155,7 +155,7 @@ public async Task ExecuteAsync_RetryDisabledWithMultipleMaxAttempts_TriesOnce() await Assert.ThrowsAsync(() => ConnectRetryExecutor.ExecuteAsync( options, - connectAttempt: _ => { attempts++; throw new InvalidOperationException("boom"); }, + connectAttempt: (_, _) => { attempts++; throw new InvalidOperationException("boom"); }, onAttemptFailed: () => { }, onStatusChanged: (_, _) => { })); @@ -179,7 +179,7 @@ public async Task ExecuteAsync_AppliesBackoffDelayBetweenAttempts() await ConnectRetryExecutor.ExecuteAsync( options, - connectAttempt: _ => + connectAttempt: (_, _) => { attempts++; if (attempts == 1) @@ -198,4 +198,115 @@ await ConnectRetryExecutor.ExecuteAsync( Assert.True(sw.ElapsedMilliseconds >= 80, $"Expected a backoff delay before retry, but only {sw.ElapsedMilliseconds}ms elapsed."); } + + [Fact] + public async Task ExecuteAsync_TokenAlreadyCanceled_NeverAttempts() + { + var attempts = 0; + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => + ConnectRetryExecutor.ExecuteAsync( + FastRetry(3), + connectAttempt: (_, _) => { attempts++; return Task.CompletedTask; }, + onAttemptFailed: () => { }, + onStatusChanged: (_, _) => { }, + cancellationToken: cts.Token)); + + Assert.Equal(0, attempts); + } + + [Fact] + public async Task ExecuteAsync_CanceledAttempt_IsNotRetried() + { + // A cancelled attempt is the caller walking away, not a transient failure — the loop must + // stop rather than burn through the remaining attempts. This is what lets an auto-reconnect + // loop (issue #379) be torn down promptly. + var attempts = 0; + var failedCleanups = 0; + var statuses = new List<(bool Connected, Exception? Error)>(); + using var cts = new CancellationTokenSource(); + + await Assert.ThrowsAnyAsync(() => + ConnectRetryExecutor.ExecuteAsync( + FastRetry(5), + connectAttempt: (_, token) => + { + attempts++; + cts.Cancel(); + token.ThrowIfCancellationRequested(); + return Task.CompletedTask; + }, + onAttemptFailed: () => failedCleanups++, + onStatusChanged: (c, e) => statuses.Add((c, e)), + cancellationToken: cts.Token)); + + Assert.Equal(1, attempts); + // The half-built handle is still cleaned up, and the transport is reported disconnected. + Assert.Equal(1, failedCleanups); + var status = Assert.Single(statuses); + Assert.False(status.Connected); + Assert.IsAssignableFrom(status.Error); + } + + [Fact] + public async Task ExecuteAsync_CanceledWhileAnAttemptFailedNormally_DoesNotDialAgain() + { + // Zero backoff means there is no Task.Delay to notice the cancellation, so the per-iteration + // check is the only thing standing between a cancelled caller and another connection attempt. + var attempts = 0; + using var cts = new CancellationTokenSource(); + + await Assert.ThrowsAnyAsync(() => + ConnectRetryExecutor.ExecuteAsync( + FastRetry(5), + connectAttempt: (_, _) => + { + attempts++; + cts.Cancel(); + throw new InvalidOperationException("boom"); + }, + onAttemptFailed: () => { }, + onStatusChanged: (_, _) => { }, + cancellationToken: cts.Token)); + + Assert.Equal(1, attempts); + } + + [Fact] + public async Task ExecuteAsync_CanceledDuringBackoffDelay_StopsWaiting() + { + var options = new ConnectionRetryOptions + { + Enabled = true, + MaxAttempts = 3, + InitialDelay = TimeSpan.FromSeconds(30), + MaxDelay = TimeSpan.FromSeconds(30), + BackoffMultiplier = 1.0 + }; + + var attempts = 0; + using var cts = new CancellationTokenSource(); + var sw = Stopwatch.StartNew(); + + var run = ConnectRetryExecutor.ExecuteAsync( + options, + connectAttempt: (_, _) => + { + attempts++; + throw new InvalidOperationException("boom"); + }, + onAttemptFailed: () => { }, + onStatusChanged: (_, _) => { }, + cancellationToken: cts.Token); + + cts.CancelAfter(TimeSpan.FromMilliseconds(100)); + await Assert.ThrowsAnyAsync(() => run); + sw.Stop(); + + Assert.Equal(1, attempts); + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(10), + $"The 30s backoff delay was not cancellable ({sw.ElapsedMilliseconds}ms elapsed)."); + } } diff --git a/src/Daqifi.Core.Tests/Communication/Transport/StreamTransportCancellationTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/StreamTransportCancellationTests.cs new file mode 100644 index 00000000..0fb8a02d --- /dev/null +++ b/src/Daqifi.Core.Tests/Communication/Transport/StreamTransportCancellationTests.cs @@ -0,0 +1,175 @@ +using Daqifi.Core.Communication.Transport; +using System.Diagnostics; +using System.Net; + +namespace Daqifi.Core.Tests.Communication.Transport; + +/// +/// Pins the cancellation contract added to +/// in issue #341: an in-flight dial can be abandoned, a caller-driven cancel is never disguised as +/// a connect timeout, and a transport written against the pre-#341 interface still works. +/// +public class StreamTransportCancellationTests +{ + [Fact] + public async Task TcpConnectAsync_CanceledMidAttempt_ThrowsOperationCanceledPromptly() + { + // A never-completing connect task plus a 30s timeout: without the token being honored this + // would sit here for the full half-minute. + using var transport = new TcpStreamTransport(IPAddress.Parse("192.0.2.1"), 9760); + transport.ConnectTaskFactory = _ => Task.Delay(Timeout.Infinite); + var options = new ConnectionRetryOptions + { + Enabled = false, + MaxAttempts = 1, + ConnectionTimeout = TimeSpan.FromSeconds(30) + }; + + using var cts = new CancellationTokenSource(); + var connect = transport.ConnectAsync(options, cts.Token); + + var sw = Stopwatch.StartNew(); + cts.CancelAfter(TimeSpan.FromMilliseconds(100)); + + var thrown = await Assert.ThrowsAnyAsync(() => connect); + sw.Stop(); + + Assert.IsNotType(thrown); + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(10), + $"Cancellation took {sw.ElapsedMilliseconds}ms; the connect timeout was not short-circuited."); + Assert.False(transport.IsConnected); + } + + [Fact] + public async Task TcpConnectAsync_TokenAlreadyCanceled_NeverStartsAnAttempt() + { + using var transport = new TcpStreamTransport(IPAddress.Parse("192.0.2.1"), 9760); + var attempts = 0; + transport.ConnectTaskFactory = _ => + { + attempts++; + return Task.Delay(Timeout.Infinite); + }; + + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => transport.ConnectAsync(cts.Token)); + Assert.Equal(0, attempts); + } + + [Fact] + public async Task TcpConnectAsync_TimeoutWithALiveToken_StillSurfacesAsTimeoutException() + { + // Regression guard for the linked cancellation source: now that the caller's token shares a + // source with the timeout, the two must still be told apart. A timeout is a device problem + // the caller must see as such (daqifi-desktop#517), not a TaskCanceledException. + using var transport = new TcpStreamTransport(IPAddress.Parse("192.0.2.1"), 9760); + transport.ConnectTaskFactory = _ => Task.Delay(Timeout.Infinite); + var options = new ConnectionRetryOptions + { + Enabled = false, + MaxAttempts = 1, + ConnectionTimeout = TimeSpan.FromMilliseconds(250) + }; + + using var cts = new CancellationTokenSource(); + + var ex = await Assert.ThrowsAsync(() => transport.ConnectAsync(options, cts.Token)); + Assert.Contains("192.0.2.1:9760", ex.Message); + Assert.IsAssignableFrom(ex.InnerException); + } + + [Fact] + public async Task SerialConnectAsync_TokenAlreadyCanceled_ThrowsWithoutOpeningThePort() + { + // A port name nothing can open: if the token were ignored the attempt would fail with an + // IO/argument exception from Open() instead of the cancellation the caller asked for. + using var transport = new SerialStreamTransport("/dev/null-daqifi-does-not-exist"); + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => transport.ConnectAsync(cts.Token)); + Assert.False(transport.IsConnected); + } + + [Fact] + public async Task LegacyTransport_WrittenBeforeTheTokenExisted_StillConnectsThroughTheNewOverload() + { + // Source-compatibility guard: this transport implements only the pre-#341 members. It must + // keep compiling, and the cancellable overload must fall back to the uncancellable one. + IStreamTransport transport = new LegacyStreamTransport(); + + await transport.ConnectAsync(CancellationToken.None); + + Assert.True(transport.IsConnected); + transport.Dispose(); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task LegacyTransport_WithAnAlreadyCanceledToken_RefusesToOpenAnything(bool withRetryOptions) + { + // A transport that cannot abandon an in-flight dial can still decline to start one, and it + // must: opening a connection the caller has already given up on has real side effects — a + // serial open pulses DTR and resets the MCU. Both cancellable overloads share the default + // implementation, so both are checked. + using var transport = new LegacyStreamTransport(); + IStreamTransport asInterface = transport; + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => withRetryOptions + ? asInterface.ConnectAsync(null, cts.Token) + : asInterface.ConnectAsync(cts.Token)); + + Assert.Equal(0, transport.ConnectCalls); + Assert.False(transport.IsConnected); + } + + /// + /// An implementation frozen at the pre-#341 shape — it does not + /// override either cancellable overload, so it exercises their default implementations. + /// + private sealed class LegacyStreamTransport : IStreamTransport + { + private readonly MemoryStream _stream = new(); + private bool _isConnected; + + /// How many times the transport actually tried to open something. + public int ConnectCalls { get; private set; } + + public Stream Stream => _stream; + public bool IsConnected => _isConnected; + public string ConnectionInfo => "Legacy"; + + public event EventHandler? StatusChanged; + + public Task ConnectAsync() => ConnectAsync(null); + + public Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + ConnectCalls++; + _isConnected = true; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(true, ConnectionInfo)); + return Task.CompletedTask; + } + + public Task DisconnectAsync() + { + _isConnected = false; + return Task.CompletedTask; + } + + public void Connect() => ConnectAsync().GetAwaiter().GetResult(); + + public void Disconnect() => DisconnectAsync().GetAwaiter().GetResult(); + + public void Dispose() + { + _isConnected = false; + _stream.Dispose(); + } + } +} diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs new file mode 100644 index 00000000..10fca68e --- /dev/null +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs @@ -0,0 +1,753 @@ +using Daqifi.Core.Communication.Producers; +using Daqifi.Core.Communication.Transport; +using Daqifi.Core.Device; +using System.Diagnostics; + +namespace Daqifi.Core.Tests.Device; + +/// +/// Covers the cancellable async connect/disconnect surface and +/// added in issue #341, together with the guarantee that the pre-existing synchronous entry +/// points still behave exactly as they did. +/// +public class DaqifiDeviceAsyncLifecycleTests +{ + [Fact] + public async Task ConnectAsync_Succeeds_ReportsConnectedAndStartsSending() + { + using var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + await using var device = new DaqifiDevice("Mock Device", transport); + + var statusChanges = new List(); + device.StatusChanged += (_, args) => statusChanges.Add(args.Status); + + await device.ConnectAsync(); + + Assert.True(device.IsConnected); + Assert.Equal(ConnectionStatus.Connected, device.Status); + Assert.Equal(DeviceState.Connected, device.State); + Assert.Equal(new[] { ConnectionStatus.Connecting, ConnectionStatus.Connected }, statusChanges); + + // The producer really is running over the transport's stream, not just flagged as such. + device.Send(ScpiMessageProducer.GetDeviceInfo); + Assert.True(await transport.WaitForWrittenTextAsync("SYSTem:SYSInfoPB?")); + } + + [Fact] + public async Task ConnectAsync_TokenAlreadyCanceled_NeverTouchesTheTransport() + { + using var transport = new GatedMockTransport(); + using var device = new DaqifiDevice("Mock Device", transport); + using var cts = new CancellationTokenSource(); + cts.Cancel(); + + await Assert.ThrowsAnyAsync(() => device.ConnectAsync(cts.Token)); + + Assert.Equal(0, transport.ConnectAttempts); + Assert.False(device.IsConnected); + // The device never even claimed to be connecting, so no consumer sees a spurious transition. + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + } + + [Fact] + public async Task ConnectAsync_CanceledMidAttempt_AbandonsItAndReportsDisconnected() + { + // The transport blocks inside ConnectAsync until its gate opens — the stand-in for a real + // dial that has not answered yet. Cancelling must break that wait rather than wait it out. + using var transport = new GatedMockTransport(); + using var device = new DaqifiDevice("Mock Device", transport); + using var cts = new CancellationTokenSource(); + + var statusChanges = new List(); + device.StatusChanged += (_, args) => statusChanges.Add(args.Status); + + var connect = device.ConnectAsync(cts.Token); + Assert.True(await transport.WaitForConnectEnteredAsync()); + Assert.False(connect.IsCompleted); + + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => connect); + Assert.False(device.IsConnected); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.Equal(DeviceState.Disconnected, device.State); + Assert.Equal(new[] { ConnectionStatus.Connecting, ConnectionStatus.Disconnected }, statusChanges); + } + + [Fact] + public async Task ConnectAsync_CanceledAfterTheTransportOpened_ClosesItAgain() + { + // A cancel that lands in the window between "transport up" and "pumps started" must not + // leak a live connection owned by a device that reports itself disconnected. + using var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + using var cts = new CancellationTokenSource(); + transport.OnConnected = () => cts.Cancel(); + + using var device = new DaqifiDevice("Mock Device", transport); + + await Assert.ThrowsAnyAsync(() => device.ConnectAsync(cts.Token)); + + Assert.Equal(1, transport.ConnectAttempts); + Assert.Equal(1, transport.DisconnectCount); + Assert.False(transport.IsConnected); + Assert.False(device.IsConnected); + } + + [Fact] + public async Task ConnectAsync_TransportThrows_ReportsDisconnectedAndRethrows() + { + using var transport = new GatedMockTransport { ConnectFailure = new InvalidOperationException("no route") }; + using var device = new DaqifiDevice("Mock Device", transport); + + var thrown = await Assert.ThrowsAsync(() => device.ConnectAsync()); + + Assert.Equal("no route", thrown.Message); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.Equal(DeviceState.Disconnected, device.State); + } + + [Fact] + public async Task DisconnectAsync_TearsDownTheTransportAndAllowsReconnect() + { + using var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + using var device = new DaqifiDevice("Mock Device", transport); + + await device.ConnectAsync(); + await device.DisconnectAsync(); + + Assert.Equal(1, transport.DisconnectCount); + Assert.False(transport.IsConnected); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.Equal(DeviceState.Disconnected, device.State); + + // Still reusable afterwards — teardown must not have poisoned anything. + await device.ConnectAsync(); + Assert.True(device.IsConnected); + } + + [Fact] + public async Task DisconnectAsync_NeverReportsLost() + { + using var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + using var device = new DaqifiDevice("Mock Device", transport); + await device.ConnectAsync(); + + var statusChanges = new List(); + device.StatusChanged += (_, args) => statusChanges.Add(args.Status); + + await device.DisconnectAsync(); + + Assert.DoesNotContain(ConnectionStatus.Lost, statusChanges); + Assert.Contains(ConnectionStatus.Disconnected, statusChanges); + } + + [Fact] + public async Task DisconnectAsync_WhileATextExchangeHoldsTheLock_CancelingSkipsTheWait() + { + // The acceptance criterion behind #341: the sync Disconnect() can sit on the text-exchange + // lock for its full budget, which on a UI thread is a multi-second freeze. Cancelling the + // async one gives up that courtesy wait immediately and tears down anyway. + using var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + using var device = new TextExchangeProbeDevice("Mock Device", transport); + await device.ConnectAsync(); + + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var heldExchange = device.HoldTextExchangeAsync(entered, responseTimeoutMs: 3000); + Assert.Same(entered.Task, await Task.WhenAny(entered.Task, Task.Delay(TimeSpan.FromSeconds(10)))); + + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + var sw = Stopwatch.StartNew(); + var disconnect = device.DisconnectAsync(cts.Token); + var finishedFirst = await Task.WhenAny(disconnect, heldExchange); + sw.Stop(); + + Assert.Same(disconnect, finishedFirst); + await disconnect; + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(2), + $"DisconnectAsync waited {sw.ElapsedMilliseconds}ms despite a cancelled token."); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + + // Let the abandoned exchange unwind; it is expected to fail now that the device is gone. + try + { + await heldExchange; + } + catch (Exception) + { + // Losing the race with teardown is the documented outcome, not a test failure. + } + } + + [Fact] + public async Task AwaitUsing_DisconnectsAndDisposesTheTransport() + { + var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + + await using (var device = new DaqifiDevice("Mock Device", transport)) + { + await device.ConnectAsync(); + Assert.True(device.IsConnected); + } + + Assert.True(transport.IsDisposed); + Assert.Equal(1, transport.DisconnectCount); + } + + [Fact] + public async Task DisposeAsync_IsIdempotentAndInterchangeableWithDispose() + { + var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + var device = new DaqifiDevice("Mock Device", transport); + await device.ConnectAsync(); + + await device.DisposeAsync(); + await device.DisposeAsync(); + device.Dispose(); + + // Exactly one teardown, no matter how many times (or which way) it is disposed. + Assert.Equal(1, transport.DisconnectCount); + Assert.True(transport.IsDisposed); + } + + [Fact] + public async Task DisposeAsync_OnANeverConnectedDevice_StillDisposesTheTransport() + { + var transport = new GatedMockTransport(); + var device = new DaqifiDevice("Mock Device", transport); + + await device.DisposeAsync(); + + Assert.True(transport.IsDisposed); + Assert.Equal(0, transport.DisconnectCount); + } + + [Fact] + public async Task Dispose_CalledWhileDisposeAsyncIsStillTearingDown_DoesNotStartASecondTeardown() + { + // Regression for the disposal overlap race. DisposeAsync spends real awaited time inside + // DisconnectAsync, and a flag published only at the end of teardown leaves that whole + // window open — a concurrent Dispose() would sail through and dispose the transport and the + // text-exchange semaphore a second time. + var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + var device = new DaqifiDevice("Mock Device", transport); + await device.ConnectAsync(); + + transport.HoldFirstDisconnect(); + var disposeAsync = device.DisposeAsync(); + Assert.True(await transport.WaitForDisconnectEnteredAsync(), + "DisposeAsync never reached the transport disconnect."); + + // The async teardown is parked mid-flight. A Dispose() arriving now must bounce off the + // gate immediately rather than run its own teardown. + var sw = Stopwatch.StartNew(); + device.Dispose(); + sw.Stop(); + + transport.ReleaseDisconnect(); + await disposeAsync; + + Assert.Equal(1, transport.DisconnectCount); + Assert.Equal(1, transport.DisposeCount); + Assert.Equal(0, transport.SyncDisconnectCalls); + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(1), + $"Dispose() spent {sw.ElapsedMilliseconds}ms; it should have returned at once as the loser."); + } + + [Fact] + public async Task Dispose_WhenTheDisconnectThrows_StillReleasesTheResources() + { + // Claiming the gate up front means there is no second chance at disposal, so teardown has + // to release the handles even when the disconnect fails on the way out. + var transport = new ThrowOnDisconnectTransport(); + var device = new DaqifiDevice("Mock Device", transport); + device.Connect(); + + Assert.Throws(() => device.Dispose()); + + Assert.True(transport.IsDisposed); + await Task.CompletedTask; + } + + [Fact] + public async Task DisposeAsync_AfterSyncDispose_IsANoOp() + { + var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + var device = new DaqifiDevice("Mock Device", transport); + await device.ConnectAsync(); + + device.Dispose(); + await device.DisposeAsync(); + + Assert.Equal(1, transport.DisconnectCount); + } + + // ---- The async connect path must carry everything the sync one does ---- + // + // #415 added the background-error surface and its per-session throttle reset to Connect(). + // Every test it shipped drives Connect(), because ConnectAsync() did not exist yet — so + // nothing else in the suite would notice if the async path lost either one when the two + // were folded into a shared step set. These two tests are that guard. + + [Fact] + public async Task ConnectAsync_WiresUpTheBackgroundErrorSurface() + { + using var transport = new ScriptedErrorTransport(); + using var device = new DaqifiDevice("Erroring Device", transport); + + var raised = new ManualResetEventSlim(false); + DeviceErrorEventArgs? captured = null; + device.ErrorOccurred += (_, e) => + { + captured = e; + raised.Set(); + }; + + await device.ConnectAsync(); + transport.ScriptedStream.FailReads = true; + + Assert.True(raised.Wait(TimeSpan.FromSeconds(10)), + "a read failure after ConnectAsync never reached an ErrorOccurred subscriber."); + Assert.Equal(DeviceErrorSource.MessageConsumer, captured!.Source); + } + + [Fact] + public async Task ConnectAsync_OpensAFreshErrorThrottleSessionOnReconnect() + { + // The throttle collapses repeats of the same (source, exception type), so a reconnect must + // clear it or the new session's first failure is swallowed. + // + // Nothing here depends on how fast the machine is. Against the default five-second window + // it would: the test has to observe two errors inside one window, so a slow run makes the + // second error due anyway (passing while proving nothing), and a tight bound fails a + // correct implementation on a loaded runner. Widening the window to ten minutes takes the + // clock out of it — a fresh session raises immediately, and a surviving bucket stays shut + // for far longer than any run of this test. + using var transport = new ScriptedErrorTransport(); + using var device = new DaqifiDevice("Erroring Device", transport); + + var throttle = new DeviceErrorThrottle(TimeSpan.FromMinutes(10)); + device.SetErrorThrottleForTesting(throttle); + + var gate = new object(); + var firstSessionRaised = new ManualResetEventSlim(false); + var secondSessionRaised = new ManualResetEventSlim(false); + var inSecondSession = false; + long firstRaisedAt = 0; + long secondRaisedAt = 0; + DeviceErrorEventArgs? secondSessionFirstError = null; + + device.ErrorOccurred += (_, e) => + { + lock (gate) + { + if (inSecondSession) + { + // Only the session's *first* raise carries the "nothing suppressed behind me" + // claim; later ones legitimately report collapsed occurrences. + if (secondSessionFirstError != null) + { + return; + } + + secondSessionFirstError = e; + secondRaisedAt = Stopwatch.GetTimestamp(); + secondSessionRaised.Set(); + return; + } + + if (firstRaisedAt == 0) + { + firstRaisedAt = Stopwatch.GetTimestamp(); + firstSessionRaised.Set(); + } + } + }; + + // Session one: fail reads until the throttle has opened a window and collapsed at least one + // repeat behind it, so a surviving bucket would have a non-zero count to carry forward. + await device.ConnectAsync(); + transport.ScriptedStream.FailReads = true; + Assert.True(firstSessionRaised.Wait(TimeSpan.FromSeconds(10)), + "the first session never reported an error."); + + // Quiesce and tear down. Disconnect stops the consumer, so no session-one raise can still + // be in flight when the flag flips below. + transport.ScriptedStream.FailReads = false; + await device.DisconnectAsync(); + + // Session two: same failure, same bucket key. + await device.ConnectAsync(); + lock (gate) + { + inSecondSession = true; + } + + transport.ScriptedStream.FailReads = true; + + // Primary guard. Without the reset the bucket is still shut — for another ten minutes — so + // this failure never arrives and the wait is what trips. + Assert.True(secondSessionRaised.Wait(TimeSpan.FromSeconds(10)), + "the reconnected session never reported an error at all: its first failure was " + + "collapsed into the throttle window the previous session opened, so ConnectAsync " + + "did not reset the error throttle."); + + // Corroborating guard, independent of the clock in a different way: a reset clears the + // bucket, so a fresh session's first raise has nothing collapsed behind it. A bucket that + // survived would report the previous session's count here whenever it eventually fired. + Assert.True( + secondSessionFirstError!.SuppressedCount == 0, + $"The reconnected session's first error reported {secondSessionFirstError.SuppressedCount} " + + "suppressed occurrence(s), so the throttle bucket survived the reconnect."); + + // Backstop for the test's own premise rather than for the code: if the two errors somehow + // landed a whole throttle window apart, the second raise was due regardless and the run + // proved nothing. Ten minutes makes that unreachable in practice — a run that slow has + // failed on the waits above long before — but assert it rather than assume it, so the test + // can never report a pass it did not earn. + var gap = Stopwatch.GetElapsedTime(firstRaisedAt, secondRaisedAt); + Assert.True( + gap < throttle.Interval, + $"Inconclusive run: {gap.TotalSeconds:0.##}s separated the two errors, which is beyond " + + $"the {throttle.Interval.TotalMinutes:0.##}-minute throttle window this test installs. " + + "The second raise would have been due even without the reset, so this run cannot " + + "distinguish the two cases. Failing rather than reporting a pass that guards nothing."); + } + + /// + /// Wraps #415's in a transport, so the + /// async connect path can be pointed at a stream whose reads fail on demand. + /// + private sealed class ScriptedErrorTransport : IStreamTransport + { + private bool _isConnected; + private bool _disposed; + + public DeviceErrorSurfaceTests.ScriptedStream ScriptedStream { get; } = new(); + + public Stream Stream => _disposed + ? throw new ObjectDisposedException(nameof(ScriptedErrorTransport)) + : ScriptedStream; + + public bool IsConnected => _isConnected && !_disposed; + + public string ConnectionInfo => _isConnected ? "Scripted: Connected" : "Scripted: Disconnected"; + + public event EventHandler? StatusChanged; + + public Task ConnectAsync() => ConnectAsync(null); + + public Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + _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().GetAwaiter().GetResult(); + + public void Disconnect() => DisconnectAsync().GetAwaiter().GetResult(); + + public void Dispose() + { + if (_disposed) + { + return; + } + + _isConnected = false; + _disposed = true; + ScriptedStream.Dispose(); + } + } + + // ---- Synchronous parity: the pre-#341 entry points must behave identically ---- + + [Fact] + public void Connect_Disconnect_StillDriveTheTransportSynchronously() + { + using var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + using var device = new DaqifiDevice("Mock Device", transport); + + var statusChanges = new List(); + device.StatusChanged += (_, args) => statusChanges.Add(args.Status); + + device.Connect(); + + Assert.True(device.IsConnected); + Assert.Equal(DeviceState.Connected, device.State); + Assert.Equal(1, transport.SyncConnectCalls); + + device.Disconnect(); + + Assert.False(device.IsConnected); + Assert.Equal(DeviceState.Disconnected, device.State); + Assert.Equal(1, transport.SyncDisconnectCalls); + Assert.Equal( + new[] { ConnectionStatus.Connecting, ConnectionStatus.Connected, ConnectionStatus.Disconnected }, + statusChanges); + } + + [Fact] + public void Connect_WhenTheTransportThrows_PropagatesTheOriginalExceptionUnwrapped() + { + using var transport = new GatedMockTransport { ConnectFailure = new InvalidOperationException("no route") }; + using var device = new DaqifiDevice("Mock Device", transport); + + var thrown = Assert.Throws(() => device.Connect()); + + Assert.Equal("no route", thrown.Message); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.Equal(DeviceState.Disconnected, device.State); + } + + [Fact] + public void Dispose_StillDisconnectsAndDisposesTheTransport() + { + var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + var device = new DaqifiDevice("Mock Device", transport); + device.Connect(); + + device.Dispose(); + + Assert.Equal(1, transport.SyncDisconnectCalls); + Assert.True(transport.IsDisposed); + } + + /// + /// Transport whose close fails, standing in for a serial port that throws on the way out. + /// + private sealed class ThrowOnDisconnectTransport : IStreamTransport + { + private readonly MemoryStream _stream = new(); + private bool _isConnected; + + public bool IsDisposed { get; private set; } + + public Stream Stream => _stream; + public bool IsConnected => _isConnected; + public string ConnectionInfo => "Throwing"; + + public event EventHandler? StatusChanged; + + public Task ConnectAsync() => ConnectAsync(null); + + public Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + _isConnected = true; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(true, ConnectionInfo)); + return Task.CompletedTask; + } + + public Task DisconnectAsync() => throw new IOException("the port went away"); + + public void Connect() => ConnectAsync().GetAwaiter().GetResult(); + + public void Disconnect() => DisconnectAsync().GetAwaiter().GetResult(); + + public void Dispose() + { + IsDisposed = true; + _isConnected = false; + _stream.Dispose(); + } + } + + /// + /// Exposes the protected text-exchange seam so a test can hold the device-wide text exchange + /// lock while a disconnect is attempted. + /// + private sealed class TextExchangeProbeDevice(string name, IStreamTransport transport) + : DaqifiDevice(name, transport) + { + /// + /// Runs a text exchange that holds the lock for roughly + /// (nothing ever replies on the mock stream), signalling from + /// inside the lock. + /// + public Task HoldTextExchangeAsync(TaskCompletionSource entered, int responseTimeoutMs) => + ExecuteTextCommandAsync( + () => entered.TrySetResult(), + responseTimeoutMs: responseTimeoutMs, + completionTimeoutMs: 50); + } + + /// + /// Mock transport whose connect can be held open, made to fail, or observed — everything the + /// cancellation paths need without a real socket or serial port. + /// + private sealed class GatedMockTransport : IStreamTransport + { + private readonly MemoryStream _stream = new(); + private readonly TaskCompletionSource _connectGate = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _connectEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _disconnectGate = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _disconnectEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); + private volatile bool _isConnected; + private volatile bool _disposed; + private int _connectAttempts; + private int _disconnectCount; + private int _syncConnectCalls; + private int _syncDisconnectCalls; + private int _disposeCount; + private int _disconnectGateArmed; + + /// When set, every connect attempt throws this instead of succeeding. + public Exception? ConnectFailure { get; init; } + + /// Invoked after the transport reports itself connected, before ConnectAsync returns. + public Action? OnConnected { get; set; } + + public int ConnectAttempts => Volatile.Read(ref _connectAttempts); + public int DisconnectCount => Volatile.Read(ref _disconnectCount); + public int SyncConnectCalls => Volatile.Read(ref _syncConnectCalls); + public int SyncDisconnectCalls => Volatile.Read(ref _syncDisconnectCalls); + public int DisposeCount => Volatile.Read(ref _disposeCount); + public bool IsDisposed => _disposed; + + /// + /// Parks the FIRST disconnect until is called. Only the + /// first, so a regression that starts a second teardown fails an assertion instead of + /// deadlocking the test run. + /// + public void HoldFirstDisconnect() => Interlocked.Exchange(ref _disconnectGateArmed, 1); + + public void ReleaseDisconnect() => _disconnectGate.TrySetResult(); + + public Task WaitForDisconnectEnteredAsync() => WaitAsync(_disconnectEntered.Task); + + public Stream Stream => _disposed + ? throw new ObjectDisposedException(nameof(GatedMockTransport)) + : _stream; + + public bool IsConnected => _isConnected && !_disposed; + + public string ConnectionInfo => _isConnected ? "Gated: Connected" : "Gated: Disconnected"; + + public event EventHandler? StatusChanged; + + public void OpenConnectGate() => _connectGate.TrySetResult(); + + public Task WaitForConnectEnteredAsync() => WaitAsync(_connectEntered.Task); + + public async Task WaitForWrittenTextAsync(string expected) + { + var deadline = DateTime.UtcNow.AddSeconds(5); + while (DateTime.UtcNow < deadline) + { + try + { + if (System.Text.Encoding.UTF8.GetString(_stream.ToArray()).Contains(expected)) + { + return true; + } + } + catch (Exception) + { + // Snapshotting a MemoryStream the producer thread is writing to can tear; + // just look again on the next poll. + } + + await Task.Delay(25); + } + + return false; + } + + public Task ConnectAsync() => ConnectAsync(null, CancellationToken.None); + + public Task ConnectAsync(ConnectionRetryOptions? retryOptions) => + ConnectAsync(retryOptions, CancellationToken.None); + + public Task ConnectAsync(CancellationToken cancellationToken) => + ConnectAsync(null, cancellationToken); + + public async Task ConnectAsync(ConnectionRetryOptions? retryOptions, CancellationToken cancellationToken) + { + ObjectDisposedException.ThrowIf(_disposed, this); + + Interlocked.Increment(ref _connectAttempts); + _connectEntered.TrySetResult(); + + if (ConnectFailure != null) + { + throw ConnectFailure; + } + + await _connectGate.Task.WaitAsync(cancellationToken).ConfigureAwait(false); + + _isConnected = true; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(true, ConnectionInfo)); + OnConnected?.Invoke(); + } + + public async Task DisconnectAsync() + { + if (!_isConnected) + { + return; + } + + _disconnectEntered.TrySetResult(); + + if (Interlocked.Exchange(ref _disconnectGateArmed, 0) == 1) + { + await _disconnectGate.Task.ConfigureAwait(false); + } + + Interlocked.Increment(ref _disconnectCount); + _isConnected = false; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(false, ConnectionInfo)); + } + + public void Connect() + { + Interlocked.Increment(ref _syncConnectCalls); + ConnectAsync().GetAwaiter().GetResult(); + } + + public void Disconnect() + { + Interlocked.Increment(ref _syncDisconnectCalls); + DisconnectAsync().GetAwaiter().GetResult(); + } + + public void Dispose() + { + Interlocked.Increment(ref _disposeCount); + + if (_disposed) + { + return; + } + + _isConnected = false; + _disposed = true; + _stream.Dispose(); + } + + private static async Task WaitAsync(Task task) + { + return ReferenceEquals(task, await Task.WhenAny(task, Task.Delay(TimeSpan.FromSeconds(5)))); + } + } +} diff --git a/src/Daqifi.Core/Communication/Transport/ConnectRetryExecutor.cs b/src/Daqifi.Core/Communication/Transport/ConnectRetryExecutor.cs index 85ad0121..d3cd39b7 100644 --- a/src/Daqifi.Core/Communication/Transport/ConnectRetryExecutor.cs +++ b/src/Daqifi.Core/Communication/Transport/ConnectRetryExecutor.cs @@ -18,7 +18,8 @@ internal static class ConnectRetryExecutor /// Retry configuration, or null for a single no-retry attempt. /// /// Opens the transport handle. Receives the resolved options so it can honor the - /// connection timeout. Throwing signals a failed attempt. + /// connection timeout, and the caller's cancellation token so it can abandon an + /// attempt already in flight. Throwing signals a failed attempt. /// /// /// Disposes/nulls the transport handle after a failed attempt, before the next @@ -29,34 +30,59 @@ internal static class ConnectRetryExecutor /// (false, error) on each failure (a retry-in-progress exception between /// attempts, the real exception on the terminal failure). /// + /// + /// Observed between attempts, while waiting out a backoff delay, and by + /// itself. A cancellation is never treated as a retryable + /// attempt failure: the loop stops immediately and the + /// is surfaced to the caller, which is what lets an + /// auto-reconnect loop be torn down promptly rather than after the remaining attempts. + /// + /// Thrown when the operation is canceled. public static async Task ExecuteAsync( ConnectionRetryOptions? retryOptions, - Func connectAttempt, + Func connectAttempt, Action onAttemptFailed, - Action onStatusChanged) + Action onStatusChanged, + CancellationToken cancellationToken = default) { var options = retryOptions ?? ConnectionRetryOptions.NoRetry; var maxAttempts = options.Enabled ? options.MaxAttempts : 1; Exception? lastException = null; + cancellationToken.ThrowIfCancellationRequested(); + for (var attempt = 1; attempt <= maxAttempts; attempt++) { try { + // Checked every iteration, not just before the delay: a retry policy configured + // with no backoff skips the delay entirely, and without this a cancellation that + // arrived during the previous attempt would be answered with another dial. + cancellationToken.ThrowIfCancellationRequested(); + // Calculate delay for this attempt if (attempt > 1) { var delay = options.CalculateDelay(attempt); if (delay > TimeSpan.Zero) { - await Task.Delay(delay); + await Task.Delay(delay, cancellationToken); } } - await connectAttempt(options); + await connectAttempt(options, cancellationToken); onStatusChanged(true, null); return; // Success! } + catch (OperationCanceledException ex) when (cancellationToken.IsCancellationRequested) + { + // The caller gave up on this connect. Clean up the half-built handle exactly as a + // failed attempt would, report the transport as disconnected, and stop — retrying + // after a cancellation would keep the device dialling long after the caller left. + onAttemptFailed(); + onStatusChanged(false, ex); + throw; + } catch (Exception ex) { lastException = ex; diff --git a/src/Daqifi.Core/Communication/Transport/IStreamTransport.cs b/src/Daqifi.Core/Communication/Transport/IStreamTransport.cs index ebeae07c..192e5dd0 100644 --- a/src/Daqifi.Core/Communication/Transport/IStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/IStreamTransport.cs @@ -44,6 +44,46 @@ public interface IStreamTransport : IDisposable /// A task representing the asynchronous connect operation. Task ConnectAsync(ConnectionRetryOptions? retryOptions); + /// + /// Establishes the transport connection, abandoning the attempt if + /// is signalled. + /// + /// A cancellation token to observe while connecting. + /// A task representing the asynchronous connect operation. + /// Thrown when the attempt is canceled. + Task ConnectAsync(CancellationToken cancellationToken) => ConnectAsync(null, cancellationToken); + + /// + /// Establishes the transport connection with retry support, abandoning the attempt if + /// is signalled — including while waiting out the + /// backoff delay between retries. + /// + /// + /// + /// This is the cancellable form of . It has a + /// default implementation, so an existing implementation keeps + /// compiling and working unchanged. That default honors the token only before the attempt + /// starts and then forwards to the uncancellable overload, which cannot be interrupted once it + /// is running. + /// + /// + /// The pre-check is not a formality: opening a connection the caller has already given up on has + /// real side effects — a serial open pulses DTR and resets the MCU — so refusing to start is the + /// one part of this contract every implementation can keep. Implementations that can also + /// abandon an attempt already in flight should override this member; the transports shipped in + /// daqifi-core do. + /// + /// + /// Configuration for retry behavior. If null, uses default single attempt. + /// A cancellation token to observe while connecting. + /// A task representing the asynchronous connect operation. + /// Thrown when the attempt is canceled. + Task ConnectAsync(ConnectionRetryOptions? retryOptions, CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + return ConnectAsync(retryOptions); + } + /// /// Closes the transport connection. /// diff --git a/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs b/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs index 789b39dd..507104d9 100644 --- a/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs @@ -237,6 +237,25 @@ public async Task ConnectAsync() /// Configuration for retry behavior. If null, uses default single attempt. /// A task representing the asynchronous connect operation. public async Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + await ConnectAsync(retryOptions, CancellationToken.None); + } + + /// + public async Task ConnectAsync(CancellationToken cancellationToken) + { + await ConnectAsync(null, cancellationToken); + } + + /// + /// + /// Cancellation is observed between attempts and while waiting out a backoff delay, and is + /// checked immediately before each . It cannot interrupt the + /// Open call itself — the framework offers no cancellable form of it — so on a port that + /// hangs open, cancellation takes effect when that call returns. In practice opening a serial + /// port either succeeds or fails quickly; the long wait worth cancelling is the retry loop. + /// + public async Task ConnectAsync(ConnectionRetryOptions? retryOptions, CancellationToken cancellationToken) { ThrowIfDisposed(); @@ -245,7 +264,7 @@ public async Task ConnectAsync(ConnectionRetryOptions? retryOptions) await ConnectRetryExecutor.ExecuteAsync( retryOptions, - connectAttempt: options => + connectAttempt: (options, attemptToken) => { var timeout = (int)options.ConnectionTimeout.TotalMilliseconds; _serialPort = new SerialPort(_portName, _baudRate, _parity, _dataBits, _stopBits) @@ -256,6 +275,11 @@ await ConnectRetryExecutor.ExecuteAsync( RtsEnable = _enableRts }; + // Last chance to bail before an uninterruptible open — on a serial port a + // DTR-triggered MCU reset also fires here, so not opening at all is the only way to + // honor a cancel that arrived while the previous attempt was backing off. + attemptToken.ThrowIfCancellationRequested(); + _serialPort.Open(); // After a successful open, swap the connect timeouts for the (shorter) @@ -269,7 +293,8 @@ await ConnectRetryExecutor.ExecuteAsync( _serialPort?.Dispose(); _serialPort = null; }, - onStatusChanged: OnStatusChanged); + onStatusChanged: OnStatusChanged, + cancellationToken: cancellationToken); StartDropDetection(); } diff --git a/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs b/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs index 8a5be806..03ce761f 100644 --- a/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs @@ -212,6 +212,22 @@ public async Task ConnectAsync() /// . /// public async Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + await ConnectAsync(retryOptions, CancellationToken.None); + } + + /// + public async Task ConnectAsync(CancellationToken cancellationToken) + { + await ConnectAsync(null, cancellationToken); + } + + /// + /// + /// Thrown when the final connection attempt does not complete within + /// . + /// + public async Task ConnectAsync(ConnectionRetryOptions? retryOptions, CancellationToken cancellationToken) { ThrowIfDisposed(); @@ -220,7 +236,7 @@ public async Task ConnectAsync(ConnectionRetryOptions? retryOptions) await ConnectRetryExecutor.ExecuteAsync( retryOptions, - connectAttempt: async options => + connectAttempt: async (options, attemptToken) => { _tcpClient = _localInterface != null ? new TcpClient(new IPEndPoint(_localInterface, 0)) @@ -231,8 +247,13 @@ await ConnectRetryExecutor.ExecuteAsync( _tcpClient.ReceiveTimeout = timeout; _tcpClient.SendTimeout = timeout; - // Add connection timeout to prevent long waits - using var cts = new CancellationTokenSource(options.ConnectionTimeout); + // Two independent reasons to stop waiting — the connect timeout and the caller's + // token — are linked into one source so a caller can abandon a dial that would + // otherwise sit here for the full timeout. Which one fired is disambiguated below, + // because the two mean very different things to the caller. + using var cts = CancellationTokenSource.CreateLinkedTokenSource(attemptToken); + cts.CancelAfter(options.ConnectionTimeout); + var connectTask = ConnectTaskFactory != null ? ConnectTaskFactory(_tcpClient) : Hostname != null @@ -243,11 +264,14 @@ await ConnectRetryExecutor.ExecuteAsync( { await connectTask.WaitAsync(cts.Token); } - catch (OperationCanceledException oce) when (cts.IsCancellationRequested) + catch (OperationCanceledException oce) + when (cts.IsCancellationRequested && !attemptToken.IsCancellationRequested) { - // The only cancellation source here is the timeout token, so surface the - // failure as what it actually is — a connect timeout — rather than a - // misleading TaskCanceledException (daqifi-desktop#517). + // The timeout — not the caller — ended the wait, so surface the failure as what + // it actually is rather than a misleading TaskCanceledException + // (daqifi-desktop#517). A caller-driven cancel falls through this filter and + // propagates as OperationCanceledException, which is what callers expect from a + // token they signalled themselves. throw new TimeoutException( $"TCP connect to {Hostname ?? _endPoint.Address.ToString()}:{_endPoint.Port} " + $"timed out after {options.ConnectionTimeout.TotalSeconds:0.###}s.", oce); @@ -270,7 +294,8 @@ await ConnectRetryExecutor.ExecuteAsync( _tcpClient = null; _networkStream = null; }, - onStatusChanged: OnStatusChanged); + onStatusChanged: OnStatusChanged, + cancellationToken: cancellationToken); _watchdog.Arm(); } diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 133f4d8c..a817d504 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -26,7 +26,7 @@ namespace Daqifi.Core.Device /// Represents a DAQiFi device that can be connected to and communicated with. /// This is the base implementation of the IDevice interface. /// - public class DaqifiDevice : IDevice, IDisposable + public class DaqifiDevice : IDevice, IDisposable, IAsyncDisposable { /// /// Gets the name of the device. @@ -375,7 +375,15 @@ protected void WithChannelsLock(Action action) protected IStreamTransport? Transport => _transport; private IProtocolHandler? _protocolHandler; + + // "Teardown has finished" — read by ExecuteTextCommandCoreAsync to reject work on a dead + // device. Distinct from _disposeClaimed below, which marks teardown as *started*. private bool _disposed; + + // Disposal gate, 0 until a caller claims teardown. Interlocked rather than a bool because + // Dispose() and DisposeAsync() are documented as interchangeable and DisposeAsync spends + // real awaited time inside the window a plain flag would leave open. See TryClaimDisposal. + private int _disposeClaimed; private bool _isDisconnecting; private bool _isInitialized; private readonly List _channels = new(); @@ -433,7 +441,15 @@ protected void WithChannelsLock(Action action) // Environment.CurrentManagedThreadId capture wouldn't work — the // value seen before await may not match the value seen after. private readonly AsyncLocal _isInsideTextExchange = new(); - + + /// + /// How long / wait to acquire + /// _textExchangeLock before tearing down anyway. See the remarks on + /// for how the budget is derived. + /// + private static readonly TimeSpan TextExchangeTeardownWait = TimeSpan.FromSeconds(10); + + /// /// Gets the current connection status of the device. /// @@ -540,7 +556,25 @@ private set /// Collapses repeated background failures so a systematic fault stays visible without /// storming . See that event for the documented policy. /// - private readonly DeviceErrorThrottle _errorThrottle = new(); + private DeviceErrorThrottle _errorThrottle = new(); + + /// + /// Test seam: replaces the background-error throttle, so a test can widen the collapsing + /// window far beyond the default instead of racing it. Never called in production. + /// + /// + /// A test that wants to prove a reconnect clears the throttle otherwise has to observe two + /// errors inside the default five-second window — which makes the result depend on how + /// quickly the machine happened to run, in both directions: too slow and the second error + /// is due anyway (the test passes without proving anything), tighten the bound and a loaded + /// CI box fails a correct implementation. Widening the window removes the clock from the + /// question entirely. Call before connecting; the field is read from background threads. + /// + /// The throttle to use. + internal void SetErrorThrottleForTesting(DeviceErrorThrottle throttle) + { + _errorThrottle = throttle ?? throw new ArgumentNullException(nameof(throttle)); + } /// /// Initializes a new instance of the class. @@ -594,65 +628,183 @@ public DaqifiDevice(string name, IStreamTransport transport, ILogger? logger = n /// /// Connects to the device. /// + /// + /// Opens the transport on the calling thread. is the + /// non-blocking, cancellable equivalent and is preferred on a UI thread; this overload is + /// kept for existing callers and behaves exactly as it always has. + /// public void Connect() { - Status = ConnectionStatus.Connecting; - State = DeviceState.Connecting; - - // A reconnect is a new session: its first background failure should be reported - // immediately rather than collapsed into a throttle window the previous session opened. - _errorThrottle.Reset(); + BeginConnect(); try { // Connect transport if available _transport?.Connect(); - // Create message producer and consumer from transport if needed + CompleteConnect(); + } + catch + { + FailConnect(); + throw; + } + } + + /// + /// Connects to the device, abandoning the attempt if + /// is signalled. + /// + /// + /// The asynchronous counterpart to : it never blocks the calling + /// thread on the transport handshake, and the token is threaded all the way down to + /// , + /// so an attempt can be given up mid-flight — including between retries, where the retry + /// loop would otherwise keep dialling. A cancel that lands after the transport has come up + /// closes it again, so a cancelled attempt never leaves a half-open connection behind. + /// + /// A cancellation token to observe while connecting. + /// A task representing the asynchronous connect operation. + /// Thrown when the attempt is canceled. + // Deliberately not virtual, matching Connect(). A virtual async twin of a non-virtual sync + // method is a trap: a subclass would override this one, leave Connect() unintercepted, and + // get different behavior depending on which entry point the caller reached for. + public async Task ConnectAsync(CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + + BeginConnect(); + + var transportConnected = false; + try + { if (_transport != null) { - // The reader/writer loops are the first thing to notice a device that has - // gone away; a transport that can act on that gets told (issue #382). - var healthSink = _transport as ITransportHealthSink; + await _transport.ConnectAsync(cancellationToken).ConfigureAwait(false); + transportConnected = true; + } - if (_messageProducer == null) - { - _messageProducer = new MessageProducer(_transport.Stream, healthSink: healthSink); - _messageProducer.SendFailed += OnMessageSendFailed; - } + cancellationToken.ThrowIfCancellationRequested(); - if (_messageConsumer == null) - { - _messageConsumer = new StreamMessageConsumer( - _transport.Stream, - new ProtobufMessageParser(), - healthSink: healthSink); - } + CompleteConnect(); + } + catch (OperationCanceledException) + { + FailConnect(); - // Read/parse/dispatch failures used to be raised into an event with no - // subscribers (issue #378). Subscribe here rather than alongside - // MessageReceived: that one is attached and detached around every consumer - // swap, and error visibility must not have holes in it. '-=' first keeps a - // reconnect on the same consumer instance from double-subscribing. - _messageConsumer.ErrorOccurred -= OnConsumerErrorOccurred; - _messageConsumer.ErrorOccurred += OnConsumerErrorOccurred; + // The transport opened and then the caller gave up — close it rather than leave a + // live connection owned by a device that reports itself disconnected. Matters most + // for a reconnect loop (issue #379), where the leaked handle would still be holding + // the serial port when the next attempt tries to open it. + if (transportConnected) + { + await SafeDisconnectTransportAsync().ConfigureAwait(false); } - // Start message producer and consumer if available - _messageProducer?.Start(); - _messageConsumer?.Start(); - - Status = ConnectionStatus.Connected; - State = DeviceState.Connected; + throw; } catch { - Status = ConnectionStatus.Disconnected; - State = DeviceState.Disconnected; + FailConnect(); throw; } } + /// + /// Marks the device as connecting and opens a fresh error-reporting session. Shared entry + /// point for and . + /// + private void BeginConnect() + { + Status = ConnectionStatus.Connecting; + State = DeviceState.Connecting; + + // A reconnect is a new session: its first background failure should be reported + // immediately rather than collapsed into a throttle window the previous session opened. + // Lives here, not in Connect(), so the async path resets it too — the factory connects + // through ConnectAsync, so leaving it on the sync path alone would mean the primary + // connect path silently kept the previous session's throttle state (issue #378). + _errorThrottle.Reset(); + } + + /// + /// Builds (when needed) and starts the message pumps over the now-open transport, then + /// marks the device connected. Shared by and + /// so the sync and async paths cannot drift apart. + /// + private void CompleteConnect() + { + // Create message producer and consumer from transport if needed + if (_transport != null) + { + // The reader/writer loops are the first thing to notice a device that has + // gone away; a transport that can act on that gets told (issue #382). + var healthSink = _transport as ITransportHealthSink; + + if (_messageProducer == null) + { + _messageProducer = new MessageProducer(_transport.Stream, healthSink: healthSink); + _messageProducer.SendFailed += OnMessageSendFailed; + } + + if (_messageConsumer == null) + { + _messageConsumer = new StreamMessageConsumer( + _transport.Stream, + new ProtobufMessageParser(), + healthSink: healthSink); + } + + // Read/parse/dispatch failures used to be raised into an event with no + // subscribers (issue #378). Subscribe here rather than alongside + // MessageReceived: that one is attached and detached around every consumer + // swap, and error visibility must not have holes in it. '-=' first keeps a + // reconnect on the same consumer instance from double-subscribing. + _messageConsumer.ErrorOccurred -= OnConsumerErrorOccurred; + _messageConsumer.ErrorOccurred += OnConsumerErrorOccurred; + } + + // Start message producer and consumer if available + _messageProducer?.Start(); + _messageConsumer?.Start(); + + Status = ConnectionStatus.Connected; + State = DeviceState.Connected; + } + + /// + /// Rolls the device's reported state back to disconnected after a failed or abandoned + /// connect attempt. + /// + private void FailConnect() + { + Status = ConnectionStatus.Disconnected; + State = DeviceState.Disconnected; + } + + /// + /// Best-effort transport close used on the cancellation path, where an exception must not + /// replace the the caller is waiting for. + /// + private async Task SafeDisconnectTransportAsync() + { + if (_transport == null) + { + return; + } + + try + { + await _transport.DisconnectAsync().ConfigureAwait(false); + } + catch (Exception ex) + { + SafeLog(() => _logger.LogDebug( + ex, + "Failed to close the transport after a cancelled connect attempt.")); + } + } + /// /// Disconnects from the device. /// @@ -669,77 +821,180 @@ public void Connect() /// of responseTimeoutMs*5 = 5s by default + safety margin) and /// most custom-timeout callers; on timeout the in-flight exchange /// sees _isDisconnecting == true via the post-acquisition - /// validation and bails out cleanly. Callers wanting non-blocking - /// disconnect should drive this off a Task.Run. + /// validation and bails out cleanly. Callers wanting a non-blocking + /// disconnect should use . /// public void Disconnect() { _isDisconnecting = true; - // Best-effort coordination with ExecuteTextCommandAsync — - // acquire the lock so we don't tear the transport out from - // under an in-flight text exchange. The lock IS released in - // the finally below when acquired (so a future Connect() - // followed by ExecuteTextCommandAsync isn't blocked); a - // stuck exchange that holds past the timeout drops to the - // _isDisconnecting validation path inside the exchange. - var lockAcquired = false; + var lockAcquired = AcquireTextExchangeLockForTeardown(); + try { - lockAcquired = _textExchangeLock.Wait(TimeSpan.FromSeconds(10)); + StopMessagePumps(); + + // Disconnect transport if available + _transport?.Disconnect(); } - catch (ObjectDisposedException) + finally { - // Disconnect called after Dispose — nothing to coordinate. + FinishDisconnect(lockAcquired); } + } + + /// + /// Disconnects from the device without blocking the calling thread. + /// + /// + /// + /// The asynchronous counterpart to , and the one to use on a UI + /// thread: the courtesy wait for an in-flight text exchange (up to + /// ) is awaited rather than blocked on, and the + /// remaining teardown — joining the reader/writer threads and closing the transport — is + /// pushed off the caller's thread. + /// + /// + /// What the token does. It shortens the courtesy wait: cancelling stops waiting for + /// the in-flight exchange and proceeds straight to teardown, exactly as the timeout does. + /// It never aborts the disconnect itself, and this method does not throw + /// — a teardown abandoned half-way would leave + /// producers, consumers and the transport in an indeterminate state, which is strictly + /// worse than finishing. On return the device is always disconnected. + /// + /// + /// is therefore raised on a thread pool thread rather than the + /// caller's; marshal to your UI thread in the handler if that matters. + /// + /// + /// A cancellation token to observe while disconnecting. + /// A task representing the asynchronous disconnect operation. + // Not virtual, for the same reason as ConnectAsync. + public async Task DisconnectAsync(CancellationToken cancellationToken = default) + { + _isDisconnecting = true; + var lockAcquired = await AcquireTextExchangeLockForTeardownAsync(cancellationToken) + .ConfigureAwait(false); try { - // Unsubscribe from message consumer/producer events - if (_messageConsumer != null) - { - _messageConsumer.MessageReceived -= OnInboundMessageReceived; - _messageConsumer.ErrorOccurred -= OnConsumerErrorOccurred; - } + // StopSafely joins the reader/writer threads with a bounded timeout, and closing a + // serial port whose device has gone away can stall: neither belongs on a UI thread, + // so the whole teardown runs on the thread pool. The ConfigureAwait(false) below + // keeps the transport close and the finally block there too. + await Task.Run(StopMessagePumps, CancellationToken.None).ConfigureAwait(false); - if (_messageProducer != null) + if (_transport != null) { - _messageProducer.SendFailed -= OnMessageSendFailed; + await _transport.DisconnectAsync().ConfigureAwait(false); } + } + finally + { + FinishDisconnect(lockAcquired); + } + } - // Stop message consumer and producer safely if available - _messageConsumer?.StopSafely(); - _messageProducer?.StopSafely(); - - // Null the producer/consumer so a subsequent Connect() - // rebuilds them against the transport's current Stream. - // SerialStreamTransport.Stream returns _serialPort.BaseStream, - // which is a new instance after Disconnect() → Connect() - // reopens the port; reusing the old producer/consumer would - // leave them bound to the previous (disposed) BaseStream - // and any Send() would silently no-op. Surfaced by PR #200's - // post-reconnect readiness probe (LAN chip-info returning - // null on every attempt because Send went to a dead stream). - _messageConsumer = null; - _messageProducer = null; + /// + /// Best-effort coordination with ExecuteTextCommandAsync before teardown: acquire + /// the lock so the transport is not torn out from under an in-flight text exchange. The + /// lock IS released by when acquired (so a future + /// followed by ExecuteTextCommandAsync isn't blocked); a stuck + /// exchange that holds past the timeout drops to the _isDisconnecting validation + /// path inside the exchange. + /// + /// true when the lock was acquired and must be released after teardown. + private bool AcquireTextExchangeLockForTeardown() + { + try + { + return _textExchangeLock.Wait(TextExchangeTeardownWait); + } + catch (ObjectDisposedException) + { + // Disconnect called after Dispose — nothing to coordinate. + return false; + } + } - // Disconnect transport if available - _transport?.Disconnect(); + /// + /// + /// Shortens the wait. A cancellation is swallowed rather than propagated: teardown must + /// still run, and the in-flight exchange sees _isDisconnecting == true and bails out + /// on its own — the same outcome as letting the wait time out. + /// + private async Task AcquireTextExchangeLockForTeardownAsync(CancellationToken cancellationToken) + { + try + { + return await _textExchangeLock.WaitAsync(TextExchangeTeardownWait, cancellationToken) + .ConfigureAwait(false); } - finally + catch (ObjectDisposedException) + { + // DisconnectAsync called after Dispose — nothing to coordinate. + return false; + } + catch (OperationCanceledException) + { + return false; + } + } + + /// + /// Unsubscribes, stops and drops the message producer/consumer. Shared by + /// and . + /// + private void StopMessagePumps() + { + // Unsubscribe from message consumer/producer events + if (_messageConsumer != null) + { + _messageConsumer.MessageReceived -= OnInboundMessageReceived; + _messageConsumer.ErrorOccurred -= OnConsumerErrorOccurred; + } + + if (_messageProducer != null) + { + _messageProducer.SendFailed -= OnMessageSendFailed; + } + + // Stop message consumer and producer safely if available + _messageConsumer?.StopSafely(); + _messageProducer?.StopSafely(); + + // Null the producer/consumer so a subsequent Connect() + // rebuilds them against the transport's current Stream. + // SerialStreamTransport.Stream returns _serialPort.BaseStream, + // which is a new instance after Disconnect() → Connect() + // reopens the port; reusing the old producer/consumer would + // leave them bound to the previous (disposed) BaseStream + // and any Send() would silently no-op. Surfaced by PR #200's + // post-reconnect readiness probe (LAN chip-info returning + // null on every attempt because Send went to a dead stream). + _messageConsumer = null; + _messageProducer = null; + } + + /// + /// Settles the device's reported state after teardown and releases the text-exchange lock + /// if it was taken. Runs from a finally in both disconnect paths, so the device can + /// never be left reporting Connected because teardown threw. + /// + /// Whether the text-exchange lock was acquired before teardown. + private void FinishDisconnect(bool lockAcquired) + { + Status = ConnectionStatus.Disconnected; + State = DeviceState.Disconnected; + _isInitialized = false; + _isDisconnecting = false; + if (lockAcquired) { - Status = ConnectionStatus.Disconnected; - State = DeviceState.Disconnected; - _isInitialized = false; - _isDisconnecting = false; - if (lockAcquired) + try + { + _textExchangeLock.Release(); + } + catch (ObjectDisposedException) { - try - { - _textExchangeLock.Release(); - } - catch (ObjectDisposedException) - { - } } } } @@ -1722,17 +1977,94 @@ protected void RaiseDeviceError(DeviceErrorSource source, Exception error, byte[ /// /// Disposes the device and releases resources. /// + /// + /// Disconnects first, which blocks the calling thread. is the + /// non-blocking equivalent — prefer await using over using on a UI thread. + /// public void Dispose() { - if (!_disposed) + if (!TryClaimDisposal()) + { + return; + } + + try { Disconnect(); - _messageConsumer?.Dispose(); - _messageProducer?.Dispose(); - _transport?.Dispose(); - _textExchangeLock.Dispose(); - _disposed = true; } + finally + { + ReleaseResources(); + } + } + + /// + /// Disposes the device and releases resources without blocking the calling thread, so + /// await using var device = ... is safe on a UI thread. + /// + /// + /// Equivalent to except that the disconnect it performs first runs + /// through . Safe to call more than once, and safe to mix with + /// — whichever runs first wins and the other becomes a no-op. + /// + /// A task representing the asynchronous dispose operation. + public async ValueTask DisposeAsync() + { + if (!TryClaimDisposal()) + { + return; + } + + try + { + await DisconnectAsync().ConfigureAwait(false); + } + finally + { + ReleaseResources(); + } + } + + /// + /// Claims the right to tear this device down, atomically and exactly once. + /// + /// + /// + /// The gate is taken at the start of disposal rather than published at the end, + /// because spends real time awaiting + /// . A plain "have we finished disposing?" flag leaves that + /// entire window open, so a concurrent — an await using scope + /// unwinding while another shutdown path fires, say — would sail through the check and run + /// a second teardown, disposing the transport and the text-exchange semaphore twice. + /// + /// + /// The loser returns immediately rather than waiting for the winner to finish. That is the + /// normal contract for a redundant call, and the + /// alternative — having block on the in-flight + /// — would reintroduce on the calling thread exactly the stall + /// this class now exists to avoid. + /// + /// + /// true for the first caller only. + private bool TryClaimDisposal() => Interlocked.Exchange(ref _disposeClaimed, 1) == 0; + + /// + /// Releases everything the device owns once it is already disconnected. Shared tail of + /// and . + /// + /// + /// The transport is already closed by the preceding disconnect, so + /// on it does not block here. Runs from a + /// finally so that a disconnect which throws — a serial close can — still releases + /// the handles rather than leaking them on a device that can never be disposed again. + /// + private void ReleaseResources() + { + _messageConsumer?.Dispose(); + _messageProducer?.Dispose(); + _transport?.Dispose(); + _textExchangeLock.Dispose(); + _disposed = true; } /// diff --git a/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs b/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs index 9b88b163..126fa3b3 100644 --- a/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs +++ b/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs @@ -465,8 +465,9 @@ private static async Task ConnectWithTransportAsync( { cancellationToken.ThrowIfCancellationRequested(); - // Step 1: Connect the transport - await transport.ConnectAsync(options.ConnectionRetry).ConfigureAwait(false); + // Step 1: Connect the transport. The token goes all the way down now, so a caller who + // gives up mid-dial stops the attempt instead of waiting out the retry loop (#341). + await transport.ConnectAsync(options.ConnectionRetry, cancellationToken).ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); @@ -480,7 +481,7 @@ private static async Task ConnectWithTransportAsync( }; // Step 3: Connect the device (starts message producers/consumers) - device.Connect(); + await device.ConnectAsync(cancellationToken).ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); @@ -499,7 +500,7 @@ await device.InitializeAsync(options.ChannelPopulationTimeout, cancellationToken // If device was created, it owns the transport and will dispose it if (device != null) { - device.Dispose(); + await device.DisposeAsync().ConfigureAwait(false); } else { diff --git a/src/Daqifi.Core/Device/IDevice.cs b/src/Daqifi.Core/Device/IDevice.cs index 89481fb2..e4f44e4f 100644 --- a/src/Daqifi.Core/Device/IDevice.cs +++ b/src/Daqifi.Core/Device/IDevice.cs @@ -1,6 +1,8 @@ using Daqifi.Core.Communication.Messages; using System; using System.Net; +using System.Threading; +using System.Threading.Tasks; #nullable enable @@ -53,15 +55,67 @@ public interface IDevice event EventHandler ErrorOccurred; /// - /// Connects to the device. + /// Connects to the device, blocking the calling thread until the connection is open. /// + /// + /// Prefer on a UI thread, or whenever the attempt needs to be + /// abandonable. + /// void Connect(); /// - /// Disconnects from the device. + /// Disconnects from the device, blocking the calling thread until teardown completes. /// + /// + /// Prefer on a UI thread — teardown waits for any in-flight + /// command exchange to finish, which can take seconds. + /// void Disconnect(); + /// + /// Connects to the device, abandoning the attempt if + /// is signalled. + /// + /// + /// The default implementation simply calls on the calling thread, so + /// an existing implementation keeps compiling and working unchanged — + /// it just cannot honor the token beyond the check made before the attempt starts. + /// overrides it with a genuinely asynchronous, cancellable + /// implementation. + /// + /// A cancellation token to observe while connecting. + /// A task representing the asynchronous connect operation. + /// Thrown when the attempt is canceled. + Task ConnectAsync(CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + Connect(); + return Task.CompletedTask; + } + + /// + /// Disconnects from the device without blocking the calling thread. + /// + /// + /// The default implementation simply calls on the calling thread, + /// so an existing implementation keeps compiling and working + /// unchanged. overrides it with a genuinely asynchronous + /// implementation; see that override for what the token does — teardown always runs to + /// completion, so cancellation shortens the wait rather than aborting the disconnect. + /// + /// + /// A cancellation token to observe while disconnecting. Ignored by this default + /// implementation: aborting a teardown part-way would leave the device in an + /// indeterminate state, which is worse than finishing it. + /// + /// A task representing the asynchronous disconnect operation. + Task DisconnectAsync(CancellationToken cancellationToken = default) + { + _ = cancellationToken; + Disconnect(); + return Task.CompletedTask; + } + /// /// Sends a message to the device. ///