diff --git a/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceLiveStreamTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceLiveStreamTests.cs index 4ada0c8..970941f 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceLiveStreamTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceLiveStreamTests.cs @@ -84,6 +84,31 @@ public async Task StreamSamplesAsync_ConsumerFallsBehind_DropsOldest_AndCountsDr Assert.True(device.DroppedLiveSampleCount > 0, "drop-oldest should have dropped and counted overflow samples"); } + [Fact] + public async Task StreamSamplesAsync_WithCancellation_EndsEnumeration_ButNotDeviceStream() + { + var device = CreateStreaming(analogCount: 1); + AnalogChannel(device, 0).IsEnabled = true; + device.StartStreaming(); + + using var cts = new CancellationTokenSource(); + + // The token supplied by WithCancellation rather than by the argument. This device method + // hands back LiveSampleStream's async iterator as-is, so the token has to reach that + // iterator's [EnumeratorCancellation] parameter; re-wrapping the forward in another + // iterator without the attribute would drop it silently and hang here instead. + var enumeration = Task.Run(async () => + { + await foreach (var _ in device.StreamSamplesAsync().WithCancellation(cts.Token)) { } + }); + + cts.Cancel(); + + await Assert.ThrowsAnyAsync( + () => enumeration.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(device.IsStreaming); + } + [Fact] public async Task StreamSamplesAsync_InvalidBufferCapacity_Throws() { diff --git a/src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs b/src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs new file mode 100644 index 0000000..e5303ea --- /dev/null +++ b/src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs @@ -0,0 +1,312 @@ +using Daqifi.Core.Channel; +using Daqifi.Core.Communication.Messages; +using Daqifi.Core.Device; +using Daqifi.Core.Device.Internal; +using Daqifi.Core.Device.SdCard; +using System; +using System.Collections.Generic; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using Xunit; + +namespace Daqifi.Core.Tests.Device.Internal; + +/// +/// Unit tests for , the pull-based live-sample view extracted from +/// (#344). +/// +/// +/// +/// What a consumer observes end to end — samples yielded in order, cancellation ending enumeration +/// without stopping the device, drop-oldest under backpressure, the deferred capacity throw — is +/// already pinned through the device by DaqifiStreamingDeviceLiveStreamTests. Those are +/// deliberately untouched: they are the evidence that the extraction changed nothing, so they are +/// not repeated here. +/// +/// +/// These add what only a direct test can see: the subscription bookkeeping (every channel +/// subscribed once at start and unsubscribed on every exit path, including cancellation), that the +/// channel set is snapshotted exactly once per enumeration, and that the drop counter is +/// device-wide rather than per-enumeration. The fake host below throws on every member outside +/// this block's remit, so a future change that sends a command, takes the channels lock, or +/// touches streaming state fails loudly rather than passing quietly. +/// +/// +public class LiveSampleStreamTests +{ + /// + /// Upper bound on every await in this file. The collaborator's reads are unbounded by design — + /// a live stream waits for the next sample — so a regression in cancellation, in delivery, or + /// in argument validation would otherwise park a test forever and stall the whole run. Every + /// wait here is bounded so such a regression surfaces as a fast + /// instead. Generous enough not to flake on a loaded CI agent. + /// + private static readonly TimeSpan MoveNextTimeout = TimeSpan.FromSeconds(5); + + [Fact] + public void Constructor_NullHost_Throws() + { + Assert.Throws(() => new LiveSampleStream(null!)); + } + + [Fact] + public async Task Enumeration_SubscribesToEveryChannelInTheSnapshot() + { + var host = new FakeHost(new FakeChannel(0), new FakeChannel(1)); + var stream = new LiveSampleStream(host); + + await using var e = stream.StreamSamplesAsync(CancellationToken.None).GetAsyncEnumerator(); + var moveNext = e.MoveNextAsync(); // runs the body: subscribes synchronously, then awaits + + Assert.All(host.Channels, c => Assert.Equal(1, c.SubscriberCount)); + + // Every subscribed channel must reach the same buffer, not just the first. + host.Channels[1].RaiseSample(2.5); + Assert.True(await moveNext.AsTask().WaitAsync(MoveNextTimeout)); + Assert.Same(host.Channels[1], e.Current.Channel); + Assert.Equal(2.5, e.Current.Sample.Value); + } + + [Fact] + public async Task Enumeration_Disposed_UnsubscribesFromEveryChannel() + { + var host = new FakeHost(new FakeChannel(0), new FakeChannel(1)); + var stream = new LiveSampleStream(host); + + var e = stream.StreamSamplesAsync(CancellationToken.None).GetAsyncEnumerator(); + var moveNext = e.MoveNextAsync(); + host.Channels[0].RaiseSample(1.0); + Assert.True(await moveNext.AsTask().WaitAsync(MoveNextTimeout)); + + await e.DisposeAsync(); + + // A leaked handler would keep the decode path writing into a dead buffer for the rest of + // the device's life — one per enumeration a consumer ever started. + Assert.All(host.Channels, c => Assert.Equal(0, c.SubscriberCount)); + } + + [Fact] + public async Task Enumeration_EndedByCancellation_StillUnsubscribes() + { + var host = new FakeHost(new FakeChannel(0)); + var stream = new LiveSampleStream(host); + + using var cts = new CancellationTokenSource(); + var e = stream.StreamSamplesAsync(cts.Token).GetAsyncEnumerator(); + var moveNext = e.MoveNextAsync(); + cts.Cancel(); + + // Bounded on purpose: if cancellation ever stops ending the read, an unbounded await here + // would park forever and hang the whole run. The timeout turns that into a fast, named + // failure (TimeoutException instead of the expected OperationCanceledException). + await Assert.ThrowsAnyAsync( + () => moveNext.AsTask().WaitAsync(MoveNextTimeout)); + await e.DisposeAsync(); + + // The unsubscribe lives in a finally, so the throwing exit path has to clean up too. + Assert.Equal(0, host.Channels[0].SubscriberCount); + } + + [Fact] + public async Task Enumeration_SnapshotsTheChannelsOnce_AndIgnoresLaterArrivals() + { + var host = new FakeHost(new FakeChannel(0)); + var stream = new LiveSampleStream(host); + + await using var e = stream.StreamSamplesAsync(CancellationToken.None).GetAsyncEnumerator(); + var moveNext = e.MoveNextAsync(); + + Assert.Equal(1, host.SnapshotCalls); + + // A channel that appears after the enumeration started is not observed by it — that is the + // documented "observes the channels present when it starts" contract, and it is also what + // makes the unsubscribe list above exactly right. + var late = new FakeChannel(1); + host.Add(late); + late.RaiseSample(9.0); + Assert.Equal(0, late.SubscriberCount); + + host.Channels[0].RaiseSample(1.0); + Assert.True(await moveNext.AsTask().WaitAsync(MoveNextTimeout)); + Assert.Equal(1.0, e.Current.Sample.Value); + Assert.Equal(1, host.SnapshotCalls); + } + + [Fact] + public async Task ConcurrentEnumerations_EachGetTheirOwnBuffer() + { + var host = new FakeHost(new FakeChannel(0)); + var stream = new LiveSampleStream(host); + + await using var first = stream.StreamSamplesAsync(CancellationToken.None).GetAsyncEnumerator(); + await using var second = stream.StreamSamplesAsync(CancellationToken.None).GetAsyncEnumerator(); + var firstMove = first.MoveNextAsync(); + var secondMove = second.MoveNextAsync(); + + Assert.Equal(2, host.Channels[0].SubscriberCount); + + host.Channels[0].RaiseSample(4.0); + + // One sample, delivered to both consumers — neither steals it from the other. + Assert.True(await firstMove.AsTask().WaitAsync(MoveNextTimeout)); + Assert.True(await secondMove.AsTask().WaitAsync(MoveNextTimeout)); + Assert.Equal(4.0, first.Current.Sample.Value); + Assert.Equal(4.0, second.Current.Sample.Value); + } + + [Fact] + public async Task DroppedSampleCount_AccumulatesAcrossEnumerations() + { + var host = new FakeHost(new FakeChannel(0)); + var stream = new LiveSampleStream(host); + + Assert.Equal(0, stream.DroppedSampleCount); + + var afterFirst = await OverflowOnce(stream, host); + Assert.True(afterFirst > 0, "drop-oldest should have dropped and counted overflow samples"); + + var afterSecond = await OverflowOnce(stream, host); + + // The counter is a device-wide health signal, so a second enumeration adds to it rather + // than starting over. + Assert.True(afterSecond > afterFirst, $"expected the count to keep growing, got {afterFirst} then {afterSecond}"); + } + + [Fact] + public async Task InvalidBufferCapacity_ThrowsOnFirstMoveNext_NotAtTheCall() + { + var host = new FakeHost(new FakeChannel(0)); + var stream = new LiveSampleStream(host); + + // An async iterator defers its body, so nothing runs — and nothing is subscribed — until + // the first MoveNextAsync. The device forwards this iterator as-is to keep that timing. + var enumerable = stream.StreamSamplesAsync(CancellationToken.None, bufferCapacity: 0); + Assert.Equal(0, host.SnapshotCalls); + + // Bounded on purpose: were the validation to stop throwing, this enumeration would block + // on an empty buffer that nothing ever writes to, so an unbounded await would hang the run + // rather than fail it. + var ex = await Assert.ThrowsAsync( + () => ConsumeAsync().WaitAsync(MoveNextTimeout)); + Assert.Equal("bufferCapacity", ex.ParamName); + + async Task ConsumeAsync() + { + await foreach (var _ in enumerable) { } + } + } + + #region Helpers + + /// + /// Runs one enumeration whose reader is parked, floods it past its two-slot buffer, then + /// returns the collaborator's drop count after that enumeration has been disposed. + /// + private static async Task OverflowOnce(LiveSampleStream stream, FakeHost host) + { + var e = stream.StreamSamplesAsync(CancellationToken.None, bufferCapacity: 2).GetAsyncEnumerator(); + var moveNext = e.MoveNextAsync(); // subscribes; reader is awaiting, not consuming synchronously + for (var i = 0; i < 20; i++) + { + host.Channels[0].RaiseSample(i); + } + + Assert.True(await moveNext.AsTask().WaitAsync(MoveNextTimeout)); + await e.DisposeAsync(); + return stream.DroppedSampleCount; + } + + private sealed class FakeChannel : IChannel + { + private EventHandler? _sampleReceived; + + public FakeChannel(int channelNumber) + { + ChannelNumber = channelNumber; + Name = "ch" + channelNumber; + } + + public int SubscriberCount { get; private set; } + + public event EventHandler? SampleReceived + { + add { _sampleReceived += value; SubscriberCount++; } + remove { _sampleReceived -= value; SubscriberCount--; } + } + + public void RaiseSample(double value) + { + var sample = new DataSample(DateTime.UtcNow, value); + ActiveSample = sample; + _sampleReceived?.Invoke(this, new SampleReceivedEventArgs(this, sample)); + } + + public int ChannelNumber { get; } + public string Name { get; set; } + public bool IsEnabled { get; set; } = true; + public ChannelType Type => ChannelType.Analog; + public ChannelDirection Direction { get; set; } = ChannelDirection.Input; + public IDataSample? ActiveSample { get; private set; } + + public void SetActiveSample(double value, DateTime timestamp) => throw new NotSupportedException(); + public void SetActiveSample(IDataSample sample) => throw new NotSupportedException(); + } + + private sealed class FakeHost : IDeviceOperationHost + { + private readonly List _channels; + + public FakeHost(params FakeChannel[] channels) + { + _channels = new List(channels); + } + + public IReadOnlyList Channels => _channels; + + public int SnapshotCalls { get; private set; } + + public void Add(FakeChannel channel) => _channels.Add(channel); + + public IReadOnlyList SnapshotChannels() + { + SnapshotCalls++; + return _channels.ToArray(); + } + + // Outside this block's remit — reaching for any of these is a regression, not a refinement. + // In particular the live path must never send a command or take the channels lock: it is an + // adapter over events the decoder already raises, and it runs on the decode thread. + public bool IsConnected => throw new NotSupportedException(); + public bool IsUsbConnection => throw new NotSupportedException(); + public bool IsStreaming { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + public int StreamingFrequency => throw new NotSupportedException(); + public DeviceMetadata Metadata => throw new NotSupportedException(); + public TimeSpan SdCardDownloadTimeout => throw new NotSupportedException(); + public TimeSpan SdCardTransferIdleTimeout => throw new NotSupportedException(); + public void StopStreaming() => throw new NotSupportedException(); + public void Send(IOutboundMessage message) => throw new NotSupportedException(); + public void Disconnect() => throw new NotSupportedException(); + public void WithChannelsLock(Action action) => throw new NotSupportedException(); + public Task> ExecuteTextCommandAsync( + Action setupAction, + int responseTimeoutMs = 1000, + int completionTimeoutMs = 250, + CancellationToken cancellationToken = default, + Func? prepareAsync = null, + Func? finalizeAsync = null) => throw new NotSupportedException(); + public Task ExecuteRawCaptureAsync( + Func rawAction, + CancellationToken cancellationToken = default) => throw new NotSupportedException(); + public void EnsureSupported(DeviceFeature feature) => throw new NotSupportedException(); + public FeatureNotSupportedException CreateFeatureNotSupportedException(DeviceFeature feature) + => throw new NotSupportedException(); + public void RaiseLowSdSpaceWarning(LowSdSpaceWarningEventArgs e) => throw new NotSupportedException(); + public void RaiseStreamFrameDiscarded(StreamFrameDiscardedEventArgs e) => throw new NotSupportedException(); + public void RaiseGapDetected(TimestampGapEventArgs e) => throw new NotSupportedException(); + public void RaiseRawStreamFrame(DaqifiOutMessage message) => throw new NotSupportedException(); + public void RaiseStreamDecodeFailure(Exception error) => throw new NotSupportedException(); + } + + #endregion +} diff --git a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs index cc73557..2796188 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -15,9 +15,7 @@ using System.IO; using System.Linq; using System.Net; -using System.Runtime.CompilerServices; using System.Threading; -using System.Threading.Channels; using System.Threading.Tasks; #nullable enable @@ -198,6 +196,7 @@ private void InitializeStreamingDevice() // which a field initializer cannot reference. Every constructor routes through this // method, so they are always in place before the device is handed to a caller. _frameDecoder = new StreamFrameDecoder(this); + _liveSampleStream = new LiveSampleStream(this); _channelControl = new ChannelControlOperations(this); _administration = new DeviceAdministrationOperations(this); _networkOperations = new NetworkConfigurationOperations(this); @@ -633,14 +632,12 @@ protected override Task RestoreSessionSnapshotAsync( /// public const int DefaultLiveSampleBufferCapacity = 4096; - private long _droppedLiveSampleCount; - /// /// Gets the cumulative number of live samples dropped across all /// enumerations because a consumer could not keep up with the incoming rate (drop-oldest policy). /// A non-zero and growing value means a live consumer is too slow for the current stream rate. /// - public long DroppedLiveSampleCount => Interlocked.Read(ref _droppedLiveSampleCount); + public long DroppedLiveSampleCount => _liveSampleStream.DroppedSampleCount; /// /// Exposes decoded live samples as an for pull-based @@ -649,6 +646,7 @@ protected override Task RestoreSessionSnapshotAsync( /// per-channel and raw-frame events are unaffected. /// /// + /// /// Samples are buffered in a bounded channel with a drop-oldest overflow policy: if the /// consumer falls behind, the oldest buffered samples are discarded (memory never grows /// unbounded) and is incremented — the decode thread that @@ -656,6 +654,14 @@ protected override Task RestoreSessionSnapshotAsync( /// cancelling ends it promptly (surfaced as /// ) and unsubscribes, but does not stop the /// device's stream — call for that. + /// + /// + /// This returns 's async iterator directly rather than wrapping + /// it in one of its own, which is what keeps the two deferred behaviors a caller can observe + /// exactly as they were: WithCancellation still reaches the iterator's own + /// [EnumeratorCancellation] parameter, and an invalid + /// still throws on the first MoveNextAsync rather than at the call. + /// /// /// Ends enumeration when cancelled. /// @@ -663,51 +669,10 @@ protected override Task RestoreSessionSnapshotAsync( /// /// An async stream of (channel + decoded sample). /// is less than 1. - public async IAsyncEnumerable StreamSamplesAsync( - [EnumeratorCancellation] CancellationToken cancellationToken = default, + public IAsyncEnumerable StreamSamplesAsync( + CancellationToken cancellationToken = default, int? bufferCapacity = null) - { - var capacity = bufferCapacity ?? DefaultLiveSampleBufferCapacity; - if (capacity < 1) - { - throw new ArgumentOutOfRangeException( - nameof(bufferCapacity), capacity, "Buffer capacity must be at least 1."); - } - - var buffer = System.Threading.Channels.Channel.CreateBounded( - new BoundedChannelOptions(capacity) - { - FullMode = BoundedChannelFullMode.DropOldest, - SingleReader = true, - SingleWriter = false, - }, - _ => Interlocked.Increment(ref _droppedLiveSampleCount)); - - void OnSample(object? sender, SampleReceivedEventArgs e) => - buffer.Writer.TryWrite(new LiveSample(e.Channel, e.Sample)); - - var channels = SnapshotChannels(); - foreach (var channel in channels) - { - channel.SampleReceived += OnSample; - } - - try - { - await foreach (var sample in buffer.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) - { - yield return sample; - } - } - finally - { - foreach (var channel in channels) - { - channel.SampleReceived -= OnSample; - } - buffer.Writer.TryComplete(); - } - } + => _liveSampleStream.StreamSamplesAsync(cancellationToken, bufferCapacity); /// /// Handles a streaming data frame by handing it to , which @@ -942,6 +907,9 @@ public void SetAdcCalibrationOffset(int channelNumber, double calB) /// The streaming hot path: frame screening, timestamps, gaps, per-channel decode. private StreamFrameDecoder _frameDecoder = null!; + /// The pull-based live-sample view: bounded buffer, drop-oldest, drop counter. + private LiveSampleStream _liveSampleStream = null!; + /// Channel enable/disable, DIO, PWM and analog output (). private ChannelControlOperations _channelControl = null!; diff --git a/src/Daqifi.Core/Device/Internal/LiveSampleStream.cs b/src/Daqifi.Core/Device/Internal/LiveSampleStream.cs new file mode 100644 index 0000000..43e3908 --- /dev/null +++ b/src/Daqifi.Core/Device/Internal/LiveSampleStream.cs @@ -0,0 +1,97 @@ +using Daqifi.Core.Channel; +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Channels; +using System.Threading.Tasks; + +#nullable enable + +namespace Daqifi.Core.Device.Internal +{ + /// + /// The pull-based live-sample view of a streaming device — the bounded buffer behind + /// and the drop counter that goes with + /// it — extracted from (#344) so the device delegates + /// rather than hosts it. + /// + /// + /// + /// This is an adapter over the push-based events, not a + /// second decode path: the decoder still raises those events exactly as before, and each + /// enumeration simply subscribes to them for its lifetime. Nothing here runs on the decode + /// thread beyond a non-blocking . + /// + /// + /// The channels to subscribe to are read through , so the + /// enumeration observes the same channel collection — under the same lock — that the rest of + /// the device does. + /// + /// + internal sealed class LiveSampleStream + { + private readonly IDeviceOperationHost _host; + + /// + /// Cumulative drop count across every enumeration this collaborator has served. Lives here + /// rather than per-enumeration because the device exposes it as a device-wide health signal. + /// + private long _droppedSampleCount; + + internal LiveSampleStream(IDeviceOperationHost host) + { + _host = host ?? throw new ArgumentNullException(nameof(host)); + } + + /// + internal long DroppedSampleCount => Interlocked.Read(ref _droppedSampleCount); + + /// + internal async IAsyncEnumerable StreamSamplesAsync( + [EnumeratorCancellation] CancellationToken cancellationToken = default, + int? bufferCapacity = null) + { + var capacity = bufferCapacity ?? DaqifiStreamingDevice.DefaultLiveSampleBufferCapacity; + if (capacity < 1) + { + throw new ArgumentOutOfRangeException( + nameof(bufferCapacity), capacity, "Buffer capacity must be at least 1."); + } + + var buffer = System.Threading.Channels.Channel.CreateBounded( + new BoundedChannelOptions(capacity) + { + FullMode = BoundedChannelFullMode.DropOldest, + SingleReader = true, + SingleWriter = false, + }, + _ => Interlocked.Increment(ref _droppedSampleCount)); + + void OnSample(object? sender, SampleReceivedEventArgs e) => + buffer.Writer.TryWrite(new LiveSample(e.Channel, e.Sample)); + + var channels = _host.SnapshotChannels(); + foreach (var channel in channels) + { + channel.SampleReceived += OnSample; + } + + try + { + await foreach (var sample in buffer.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return sample; + } + } + finally + { + foreach (var channel in channels) + { + channel.SampleReceived -= OnSample; + } + buffer.Writer.TryComplete(); + } + } + } +}