From e39e64dd159e6e6addda6305693bfa0bfefcf711 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 11 Jul 2026 17:55:53 +0000 Subject: [PATCH 1/3] Fix TryUnsubscribe and TryRemoveJob ownership validation TryUnsubscribe now returns false when a JobInfo is not registered in the session job list before treating it as a user job. TryRemoveJob returns false when the job is not found instead of always reporting success. Adds regression tests for foreign and already-removed jobs. Stabilizes TryUnsubscribe_UserJob_CancelsJob by keeping the session in Running phase. Co-authored-by: DevAM --- .../SessionApiTests.cs | 41 ++++++++++++++++++ .../UnsubscribeTests.cs | 42 ++++++++++++++++++- .../src/Session/Session.cs | 31 ++++++++++++-- 3 files changed, 109 insertions(+), 5 deletions(-) diff --git a/NetworkInspector.Sessions.Tests/SessionApiTests.cs b/NetworkInspector.Sessions.Tests/SessionApiTests.cs index cefcd0e..1f4dbe1 100644 --- a/NetworkInspector.Sessions.Tests/SessionApiTests.cs +++ b/NetworkInspector.Sessions.Tests/SessionApiTests.cs @@ -84,6 +84,47 @@ public async Task TryRemoveJob_TerminalJob_Succeeds() session.Shutdown(); } + [Test] + public async Task TryRemoveJob_AlreadyRemoved_ReturnsFalse() + { + using Stack stack = TestHarness.CreateStack(); + using TestFrameSource source = TestFrameSource.WithUdpFrames(3); + + using Session session = new(stack); + session.TryAddFrameSource(source, out _); + session.TryStart(); + session.WaitForCompletion(); + + JobInfo sourceJob = session.GetJobs().First(j => j.UiName == source.UiName); + _WaitForCondition(() => sourceJob.Status is JobStatus.Completed or JobStatus.Cancelled); + + await Assert.That(session.TryRemoveJob(sourceJob)).IsTrue(); + await Assert.That(session.TryRemoveJob(sourceJob)).IsFalse(); + + session.Shutdown(); + } + + [Test] + public async Task TryRemoveJob_ForeignJob_ReturnsFalse() + { + using Stack stack = TestHarness.CreateStack(); + using Session session = new(stack); + + using Job foreignJob = new( + new JobId(999), + "Foreign", + "Not in session", + _ => { }, + static (_, _) => { }); + foreignJob.Start(); + foreignJob.Join(); + JobInfo foreignInfo = new(foreignJob); + + bool removed = session.TryRemoveJob(foreignInfo); + + await Assert.That(removed).IsFalse(); + } + [Test] public async Task TryRemoveJob_RunningJob_ThrowsSessionException() { diff --git a/NetworkInspector.Sessions.Tests/UnsubscribeTests.cs b/NetworkInspector.Sessions.Tests/UnsubscribeTests.cs index f4d56d1..2ccd50e 100644 --- a/NetworkInspector.Sessions.Tests/UnsubscribeTests.cs +++ b/NetworkInspector.Sessions.Tests/UnsubscribeTests.cs @@ -213,15 +213,17 @@ public async Task Shutdown_ListenerStatus_IsSessionEnded() public async Task TryUnsubscribe_UserJob_CancelsJob() { using Stack stack = TestHarness.CreateStack(); - using TestFrameSource source = TestFrameSource.WithUdpFrames(5); + using BlockingTestFrameSource source = new(10); using Session session = new(stack); session.TryAddFrameSource(source, out _); session.TryStart(); + _WaitForCondition(() => session.Phase == SessionPhase.Running); + // Add a long-running user job. using ManualResetEventSlim gate = new(false); - session.TryAddJob("TestJob", "Long running job", ct => + bool added = session.TryAddJob("TestJob", "Long running job", ct => { try { @@ -230,6 +232,7 @@ public async Task TryUnsubscribe_UserJob_CancelsJob() catch (OperationCanceledException) { /* expected */ } }, out JobInfo? jobInfo); + await Assert.That(added).IsTrue(); await Assert.That(jobInfo).IsNotNull(); // Wait for the job to start. @@ -243,6 +246,7 @@ public async Task TryUnsubscribe_UserJob_CancelsJob() _WaitForCondition( () => jobInfo!.Status is JobStatus.Cancelled or JobStatus.Completed); + source.Release(); session.Shutdown(); } @@ -272,6 +276,40 @@ public async Task TryUnsubscribe_TerminalJob_ReturnsFalse() session.Shutdown(); } + [Test] + public async Task TryUnsubscribe_ForeignJob_ReturnsFalse() + { + using Stack stack = TestHarness.CreateStack(); + using BlockingTestFrameSource source = new(10); + + using Session session = new(stack); + session.TryAddFrameSource(source, out _); + session.TryStart(); + + _WaitForCondition(() => session.Phase == SessionPhase.Running); + + using ManualResetEventSlim gate = new(false); + using Job foreignJob = new( + new JobId(999), + "Foreign", + "Not in session", + ct => gate.Wait(ct), + static (_, _) => { }); + foreignJob.Start(); + JobInfo foreignInfo = new(foreignJob); + + bool result = session.TryUnsubscribe(foreignInfo); + + await Assert.That(result).IsFalse(); + await Assert.That(foreignJob.Status).IsEqualTo(JobStatus.Running); + + gate.Set(); + foreignJob.Join(); + + source.Release(); + session.Shutdown(); + } + [Test] public async Task TryUnsubscribe_IdlePhase_ReturnsFalse() { diff --git a/NetworkInspector.Sessions/src/Session/Session.cs b/NetworkInspector.Sessions/src/Session/Session.cs index d89a55b..f1ff907 100644 --- a/NetworkInspector.Sessions/src/Session/Session.cs +++ b/NetworkInspector.Sessions/src/Session/Session.cs @@ -407,8 +407,11 @@ public bool TryRemoveJob(JobInfo job) } // Remove from the unified list and notify listeners. - // If the job is not in the list (e.g. already removed), this is a no-op. - _AllJobs.Remove(job); + if (!_AllJobs.Remove(job)) + { + return false; + } + _NotifyAllListeners(NotifyFlags.JobRemoved); return true; } @@ -448,12 +451,34 @@ public bool TryUnsubscribe(JobInfo job) return _TryUnsubscribeListener(slot, info); } - // Must be a user job — cancel it. + // Must be a user job — cancel only if owned by this session. + if (!_ContainsJob(job)) + { + return false; + } + return _TryUnsubscribeUserJob(job); } // ── Unsubscribe helpers ────────────────────────────────────────────────── + /// + /// Returns when is registered in + /// (reference identity). + /// + private bool _ContainsJob(JobInfo job) + { + foreach (JobInfo entry in _AllJobs.Current) + { + if (ReferenceEquals(entry, job)) + { + return true; + } + } + + return false; + } + /// /// Finds the whose /// matches the given . Returns if not found. From ffd74c937ddadd1ee528ee085a72c06583de872c Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 11 Jul 2026 18:43:35 +0000 Subject: [PATCH 2/3] Session review remediation: listener wait doc, SourceInfos snapshot, parse helpers - Document event-gated ManualResetEventSlim listener wait loop in class XML - Initialize SpinLock in constructor; remove CA2213 pragma - Add _SourceInfos SnapshotList; GetFrameSources returns CurrentSnapshot without copy - Extract _RegisterListenerSlot for listener registration without CA2000 pragma - Add _TryParseFrameUnderLock for non-throwing reparse in TryGetPacket - Record PacketToFrameMap before PacketStore in source loop - Preserve ownership validation (_ContainsJob, TryRemoveJob, TryUnsubscribe) Co-authored-by: DevAM --- .../src/Session/Session.cs | 114 ++++++++++++------ 1 file changed, 76 insertions(+), 38 deletions(-) diff --git a/NetworkInspector.Sessions/src/Session/Session.cs b/NetworkInspector.Sessions/src/Session/Session.cs index f1ff907..8e69d51 100644 --- a/NetworkInspector.Sessions/src/Session/Session.cs +++ b/NetworkInspector.Sessions/src/Session/Session.cs @@ -18,11 +18,12 @@ namespace NetworkInspector.Sessions; /// /// Listener-thread model (pull-based): /// Each subscription receives a dedicated -/// with a SpinWait poll loop. Producers set atomic flags; -/// the listener thread reads and clears flags via Interlocked.Exchange, -/// then pulls data from the shared . No queues, no -/// signal objects, no batch copies. Natural coalescing: multiple events between -/// two poll cycles merge into a single flag read. +/// with an event-gated wait loop backed by +/// . Producers set atomic flags and signal the +/// wake event; the listener thread reads and clears flags via Interlocked.Exchange, +/// then pulls data from the shared . No queues, no batch +/// copies. Natural coalescing: multiple events between two wake cycles merge into a +/// single flag read. /// /// /// @@ -77,9 +78,7 @@ public sealed class Session : ISession, ISessionReader // SpinLock protects Stack parsing which mutates protocol-instance state. // enableThreadOwnerTracking=false removes re-entrancy overhead on hot path. -#pragma warning disable CA2213 // SpinLock is a mutable value type, not a disposable object - private SpinLock _ParseLock = new(enableThreadOwnerTracking: false); -#pragma warning restore CA2213 + private SpinLock _ParseLock; // Globally unique PacketId counter (shared across all source threads). // Allocated INSIDE the SpinLock to guarantee correctness after a reparse @@ -113,6 +112,9 @@ public sealed class Session : ISession, ISessionReader // Sources registered before Start(). Re-used on Restart(). private readonly SnapshotList _SourceEntries = new(); + // Public read-only views of frame sources for GetFrameSources(). + private readonly SnapshotList _SourceInfos = new(); + // Running source jobs (populated at Start() time, replaced on Restart()). // Non-readonly: reassigned by _StartInternal() on each start/restart cycle. private Job[] _SourceJobs = []; @@ -164,6 +166,7 @@ public Session(Stack stack) { _Stack = stack ?? throw new ArgumentNullException(nameof(stack)); _FrameInterfaceRegistry = stack.FrameInterfaceRegistry; + _ParseLock = new(enableThreadOwnerTracking: false); } // -- ISessionReader: Status -- @@ -201,16 +204,8 @@ public bool TryAddFrameSource(IFrameSource source, [NotNullWhen(true)] out Frame // -- ISessionReader: Source info -- /// - public IReadOnlyList GetFrameSources() - { - ReadOnlySpan entries = _SourceEntries.Current; - FrameSourceInfo[] snapshot = new FrameSourceInfo[entries.Length]; - for (int i = 0; i < entries.Length; i++) - { - snapshot[i] = entries[i].Info; - } - return snapshot; - } + /// Returns the current immutable snapshot array; no per-call allocation copy. + public IReadOnlyList GetFrameSources() => _SourceInfos.CurrentSnapshot; // -- ISession: Listener management -- @@ -237,10 +232,7 @@ public bool TryAddListener(ISessionListener listener, [NotNullWhen(true)] out Li ListenerId listenerId = _State.AllocateListenerId(); JobId jobId = _State.AllocateJobId(); - // CA2000: ListenerSlot is stored in _ListenerSlots and disposed during Shutdown(). -#pragma warning disable CA2000 - ListenerSlot slot = new(jobId, listener, this, _OnJobStatusChanged); -#pragma warning restore CA2000 + ListenerSlot slot = _RegisterListenerSlot(jobId, listener); info = new ListenerInfo() { @@ -256,13 +248,8 @@ public bool TryAddListener(ISessionListener listener, [NotNullWhen(true)] out Li JobInfo slotJobInfo = slot.Info; info.UnsubscribeCallback = () => TryUnsubscribe(slotJobInfo); - // Add to snapshot list so source threads can set flags on it. - _ListenerSlots.Add(slot); // Track the public view for GetListeners(). _ListenerInfos.Add(info); - // Register the listener's job in the unified job list so callers - // can observe listener job status via GetJobs(). - _AllJobs.Add(slot.Info); // Start the slot in any non-Idle phase. During Idle, _StartInternal() // will start all pending slots. In all active phases (Running, @@ -321,15 +308,14 @@ public bool TryGetPacket(PacketId id, [NotNullWhen(true)] out Packet? packet) Frame? raFrame = raSource.FrameById(frameId); if (raFrame is not null) { - // _ParseFrameUnderLock uses ParseFrameIndexed when PacketIndex is active. - packet = _ParseFrameUnderLock(raFrame.Value, id); - if (packet is not null) + if (!_TryParseFrameUnderLock(raFrame.Value, id, out packet)) { - // Cache the re-parsed packet so subsequent lookups avoid re-parsing. - _PacketStore.Store(id, packet); - return true; + return false; } - return false; + + // Cache the re-parsed packet so subsequent lookups avoid re-parsing. + _PacketStore.Store(id, packet); + return true; } } @@ -1099,6 +1085,7 @@ private FrameSourceInfo _AddFrameSourceInternal(IFrameSource source) // unified job list without creating a second JobInfo wrapper. FrameSourceEntry entry = new(info, source, job); _SourceEntries.Add(entry); + _SourceInfos.Add(info); _AllJobs.Add(entry.JobInfo); // Wire the convenience API: FrameSourceInfo.Stop() → TryUnsubscribe(job). @@ -1220,8 +1207,8 @@ private static NetworkInspector.Core.Index.PacketIndex _CreatePacketIndex(Stack /// Start the source and register its capture interface(s). /// Pull frames one by one via . /// Parse each frame under the shared . - /// Store the packet in the . /// Record the PacketId -> FrameId mapping. + /// Store the packet in the . /// Increment global counters and set . /// On source exhaustion: set SourceCompleted and (if last) AllSourcesCompleted flags. /// @@ -1258,9 +1245,6 @@ private void _RunSourceLoop( Packet packet = _ParseFrameUnderLock(capturedFrame, packetId: null); - // Store packet in the shared PacketStore — all listeners read from here. - _PacketStore.Store(packet.Id, packet); - // Record PacketId -> FrameId mapping for random access re-parse. // Failure means the PacketId is invalid or the map capacity is exceeded. if (!_Mapping.Record(packet.Id, capturedFrame.Id, sourceInfo.Id)) @@ -1270,6 +1254,9 @@ private void _RunSourceLoop( $"The packet map capacity ({PacketToFrameMap.MaxEntries}) may have been exceeded."); } + // Store packet in the shared PacketStore — all listeners read from here. + _PacketStore.Store(packet.Id, packet); + // Update global atomic counters. Interlocked.Increment(ref _PacketCount); Interlocked.Increment(ref _FrameCount); @@ -1305,8 +1292,59 @@ private void _RunSourceLoop( } } + /// + /// Creates a and registers it in session registries. + /// Ownership transfers to before return so disposal is session-managed. + /// + private ListenerSlot _RegisterListenerSlot(JobId jobId, ISessionListener listener) + { + ListenerSlot slot = new(jobId, listener, this, _OnJobStatusChanged); + _ListenerSlots.Add(slot); + _AllJobs.Add(slot.Info); + return slot; + } + // -- Parse helper -- + /// + /// Attempts to parse a frame under with a fixed . + /// Returns without throwing when parsing fails (stale frame, protocol error). + /// + private bool _TryParseFrameUnderLock(Frame frame, PacketId packetId, [NotNullWhen(true)] out Packet? packet) + { + bool lockTaken = false; + try + { + _ParseLock.Enter(ref lockTaken); + PacketIndex? index = _PacketIndex; + try + { + if (index is not null) + { + packet = Packet.ParseFrameIndexed(packetId, _Stack, frame, index); + } + else + { + packet = Packet.ParseFrame(packetId, _Stack, frame); + } + + return true; + } + catch + { + packet = null; + return false; + } + } + finally + { + if (lockTaken) + { + _ParseLock.Exit(useMemoryBarrier: false); + } + } + } + /// /// Parses a frame under . /// When is , allocates the next sequential id. From 8ccdec806d9930f51ca044c421b9dc56dd4a39b3 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 11 Jul 2026 18:45:56 +0000 Subject: [PATCH 3/3] Address Sessions review findings across API, docs, and coverage - Validate TryUnsubscribe and TryRemoveJob ownership (prior commit) - Add Try-parse reparse path and record mapping before store - Remove analyzer suppressions; initialize SpinLock in constructor - Guard job and listener ID generators against overflow - Use SnapshotList for GetFrameSources without per-call allocation - Update ISession and Session XML documentation - Add #region structure to Job and ListenerSlot - Consolidate test wait helpers; add exit-point coverage tests - Document listener notification cost and ThreadWaitHelper usage Co-authored-by: DevAM --- .../ExitPointCoverageTests.cs | 133 ++++++++++++++++++ .../GlobalUsings.cs | 1 + .../Helpers/WaitHelper.cs | 27 ++++ .../NotifyFlagsTests.cs | 2 + .../SessionApiTests.cs | 48 ++++--- .../SessionIntegrationTests.cs | 37 ++--- .../UnsubscribeTests.cs | 50 ++----- NetworkInspector.Sessions/src/Jobs/Job.cs | 37 ++++- .../src/Listeners/ListenerSlot.cs | 31 +++- .../src/Session/ISession.cs | 9 +- .../src/Session/Session.cs | 4 +- .../src/Session/SessionState.cs | 20 ++- .../src/ThreadWaitHelper.cs | 2 + 13 files changed, 304 insertions(+), 97 deletions(-) create mode 100644 NetworkInspector.Sessions.Tests/ExitPointCoverageTests.cs create mode 100644 NetworkInspector.Sessions.Tests/Helpers/WaitHelper.cs diff --git a/NetworkInspector.Sessions.Tests/ExitPointCoverageTests.cs b/NetworkInspector.Sessions.Tests/ExitPointCoverageTests.cs new file mode 100644 index 0000000..b55c6fd --- /dev/null +++ b/NetworkInspector.Sessions.Tests/ExitPointCoverageTests.cs @@ -0,0 +1,133 @@ +// Copyright © 2026 DevAM. All rights reserved. Licensed under MIT license. See license in the repository root for license information. + +using FieldInfo = System.Reflection.FieldInfo; + +namespace NetworkInspector.Sessions.Tests; + +/// Exit-point coverage for session error paths. +[NotInParallel(nameof(ExitPointCoverageTests))] +internal sealed class ExitPointCoverageTests +{ + [Test] + public async Task TryGetPacket_ReparseWithMismatchedStack_ReturnsFalse() + { + using Stack stack = TestHarness.CreateStack(); + using TestFrameSource source = TestFrameSource.WithUdpFrames(3); + + using Session session = new(stack); + session.TryAddFrameSource(source, out _); + session.TryStart(); + session.WaitForCompletion(); + + _GetPacketStore(session).Clear(); + + Stack wrongStack = TestHarness.CreateStack(); + FieldInfo stackField = typeof(Session).GetField( + "_Stack", + BindingFlags.Instance | BindingFlags.NonPublic)!; + Stack originalStack = (Stack)stackField.GetValue(session)!; + try + { + stackField.SetValue(session, wrongStack); + + bool found = session.TryGetPacket(new PacketId(0), out Packet? packet); + + await Assert.That(found).IsFalse(); + await Assert.That(packet).IsNull(); + } + finally + { + stackField.SetValue(session, originalStack); + wrongStack.Dispose(); + } + } + + [Test] + public async Task RunSourceLoop_MappingCapacityExceeded_FailsSourceJob() + { + using Stack stack = TestHarness.CreateStack(); + using TestFrameSource source = TestFrameSource.WithUdpFrames(1); + + using Session session = new(stack); + session.TryAddFrameSource(source, out _); + + FieldInfo nextPacketIdField = typeof(Session).GetField( + "_NextPacketId", + BindingFlags.Instance | BindingFlags.NonPublic)!; + nextPacketIdField.SetValue(session, PacketToFrameMap.MaxEntries); + + session.TryStart(); + + JobInfo sourceJob = session.GetJobs().First(j => j.UiName == source.UiName); + WaitHelper.WaitUntil(() => sourceJob.Status == JobStatus.Failed); + + await Assert.That(sourceJob.FailureException).IsNotNull(); + await Assert.That(sourceJob.FailureException!.Message) + .Contains(PacketToFrameMap.MaxEntries.ToString(CultureInfo.InvariantCulture)); + } + + [Test] + public async Task RunSourceLoop_WithoutPacketIndex_UsesNonIndexedParse() + { + using Stack stack = TestHarness.CreateStack(); + using BlockingTestFrameSource source = new(3); + + using Session session = new(stack); + session.TryAddFrameSource(source, out _); + session.TryStart(); + + WaitHelper.WaitUntil(() => session.Phase == SessionPhase.Running); + + _SetPacketIndex(session, null); + source.Release(); + + WaitHelper.WaitUntil(() => session.PacketCount >= 3); + + await Assert.That(session.PacketIndex).IsNull(); + await Assert.That(session.PacketCount).IsGreaterThanOrEqualTo(3); + } + + [Test] + public async Task AllocateListenerId_AtCapacity_ThrowsInvalidOperationException() + { + using Stack stack = TestHarness.CreateStack(); + using Session session = new(stack); + + FieldInfo stateField = typeof(Session).GetField( + "_State", + BindingFlags.Instance | BindingFlags.NonPublic)!; + SessionState state = (SessionState)stateField.GetValue(session)!; + FieldInfo nextListenerIdField = typeof(SessionState).GetField( + "_NextListenerId", + BindingFlags.Instance | BindingFlags.NonPublic)!; + nextListenerIdField.SetValue(state, (long)int.MaxValue); + + TestSessionListener listener = new(); + + try + { + session.TryAddListener(listener, out _); + throw new InvalidOperationException("Expected InvalidOperationException was not thrown."); + } + catch (InvalidOperationException ex) + { + await Assert.That(ex.Message).Contains("listener ID"); + } + } + + private static void _SetPacketIndex(Session session, PacketIndex? index) + { + FieldInfo field = typeof(Session).GetField( + "_PacketIndex", + BindingFlags.Instance | BindingFlags.NonPublic)!; + field.SetValue(session, index); + } + + private static PacketStore _GetPacketStore(Session session) + { + FieldInfo field = typeof(Session).GetField( + "_PacketStore", + BindingFlags.Instance | BindingFlags.NonPublic)!; + return (PacketStore)field.GetValue(session)!; + } +} diff --git a/NetworkInspector.Sessions.Tests/GlobalUsings.cs b/NetworkInspector.Sessions.Tests/GlobalUsings.cs index 9946041..559aecd 100644 --- a/NetworkInspector.Sessions.Tests/GlobalUsings.cs +++ b/NetworkInspector.Sessions.Tests/GlobalUsings.cs @@ -37,6 +37,7 @@ #endregion #region Test Framework +global using NetworkInspector.Sessions.Tests.Helpers; global using TUnit.Assertions; global using TUnit.Assertions.Extensions; global using TUnit.Core; diff --git a/NetworkInspector.Sessions.Tests/Helpers/WaitHelper.cs b/NetworkInspector.Sessions.Tests/Helpers/WaitHelper.cs new file mode 100644 index 0000000..dc412a7 --- /dev/null +++ b/NetworkInspector.Sessions.Tests/Helpers/WaitHelper.cs @@ -0,0 +1,27 @@ +// Copyright © 2026 DevAM. All rights reserved. Licensed under MIT license. See license in the repository root for license information. + +namespace NetworkInspector.Sessions.Tests.Helpers; + +/// Shared spin-wait helpers for session integration tests. +internal static class WaitHelper +{ + /// + /// Spins for up to until + /// returns . + /// + internal static void WaitUntil(Func condition, int timeoutMs = 5000) + { + Stopwatch sw = Stopwatch.StartNew(); + SpinWait wait = new(); + while (!condition()) + { + if (sw.ElapsedMilliseconds > timeoutMs) + { + throw new TimeoutException( + $"Condition was not met within {timeoutMs.ToString(CultureInfo.InvariantCulture)} ms."); + } + + wait.SpinOnce(); + } + } +} diff --git a/NetworkInspector.Sessions.Tests/NotifyFlagsTests.cs b/NetworkInspector.Sessions.Tests/NotifyFlagsTests.cs index f1ef2bf..d9d7ef2 100644 --- a/NetworkInspector.Sessions.Tests/NotifyFlagsTests.cs +++ b/NetworkInspector.Sessions.Tests/NotifyFlagsTests.cs @@ -20,8 +20,10 @@ public async Task Flags_IndividualBitsAreDistinct() NotifyFlags.AllSourcesCompleted, NotifyFlags.JobAdded, NotifyFlags.JobStatusChanged, + NotifyFlags.JobRemoved, NotifyFlags.PhaseChanged, NotifyFlags.ShuttingDown, + NotifyFlags.StackChanged, ]; for (int i = 0; i < allFlags.Length; i++) diff --git a/NetworkInspector.Sessions.Tests/SessionApiTests.cs b/NetworkInspector.Sessions.Tests/SessionApiTests.cs index 1f4dbe1..6c3718a 100644 --- a/NetworkInspector.Sessions.Tests/SessionApiTests.cs +++ b/NetworkInspector.Sessions.Tests/SessionApiTests.cs @@ -74,7 +74,7 @@ public async Task TryRemoveJob_TerminalJob_Succeeds() session.WaitForCompletion(); JobInfo sourceJob = session.GetJobs().First(j => j.UiName == source.UiName); - _WaitForCondition(() => sourceJob.Status is JobStatus.Completed or JobStatus.Cancelled); + WaitHelper.WaitUntil(() => sourceJob.Status is JobStatus.Completed or JobStatus.Cancelled); bool removed = session.TryRemoveJob(sourceJob); @@ -96,7 +96,7 @@ public async Task TryRemoveJob_AlreadyRemoved_ReturnsFalse() session.WaitForCompletion(); JobInfo sourceJob = session.GetJobs().First(j => j.UiName == source.UiName); - _WaitForCondition(() => sourceJob.Status is JobStatus.Completed or JobStatus.Cancelled); + WaitHelper.WaitUntil(() => sourceJob.Status is JobStatus.Completed or JobStatus.Cancelled); await Assert.That(session.TryRemoveJob(sourceJob)).IsTrue(); await Assert.That(session.TryRemoveJob(sourceJob)).IsFalse(); @@ -136,7 +136,7 @@ public async Task TryRemoveJob_RunningJob_ThrowsSessionException() session.TryStart(); JobInfo sourceJob = session.GetJobs().First(j => j.UiName == source.UiName); - _WaitForCondition(() => sourceJob.Status == JobStatus.Running); + WaitHelper.WaitUntil(() => sourceJob.Status == JobStatus.Running); try { @@ -363,6 +363,34 @@ public async Task TryGetPacket_AfterStoreClear_ReparsesWithIndex() session.Shutdown(); } + [Test] + public async Task AllocateJobId_AtCapacity_ThrowsInvalidOperationException() + { + using Stack stack = TestHarness.CreateStack(); + using Session session = new(stack); + + FieldInfo stateField = typeof(Session).GetField( + "_State", + BindingFlags.Instance | BindingFlags.NonPublic)!; + SessionState state = (SessionState)stateField.GetValue(session)!; + FieldInfo nextJobIdField = typeof(SessionState).GetField( + "_NextJobId", + BindingFlags.Instance | BindingFlags.NonPublic)!; + nextJobIdField.SetValue(state, (long)int.MaxValue); + + using TestFrameSource source = TestFrameSource.WithUdpFrames(1); + + try + { + session.TryAddFrameSource(source, out _); + throw new InvalidOperationException("Expected InvalidOperationException was not thrown."); + } + catch (InvalidOperationException ex) + { + await Assert.That(ex.Message).Contains("job ID"); + } + } + [Test] public async Task UseAfterDispose_ThrowsSessionException() { @@ -389,20 +417,6 @@ private static PacketStore _GetPacketStore(Session session) return (PacketStore)field.GetValue(session)!; } - private static void _WaitForCondition(Func condition, int timeoutMs = 5000) - { - Stopwatch sw = Stopwatch.StartNew(); - SpinWait wait = new(); - while (!condition()) - { - if (sw.ElapsedMilliseconds > timeoutMs) - { - throw new TimeoutException($"Condition was not met within {timeoutMs} ms."); - } - wait.SpinOnce(); - } - } - private sealed class EmptyNameListener : ISessionListener { public string UiName => " "; diff --git a/NetworkInspector.Sessions.Tests/SessionIntegrationTests.cs b/NetworkInspector.Sessions.Tests/SessionIntegrationTests.cs index f35d8ff..c429468 100644 --- a/NetworkInspector.Sessions.Tests/SessionIntegrationTests.cs +++ b/NetworkInspector.Sessions.Tests/SessionIntegrationTests.cs @@ -22,7 +22,7 @@ public async Task StartAndRun_ListenerSeesAllPackets() session.WaitForCompletion(); // Give the listener slot time to process remaining flags. - _WaitForCondition(() => listener.TotalPacketsSeen >= frameCount); + WaitHelper.WaitUntil(() => listener.TotalPacketsSeen >= frameCount); session.Shutdown(); @@ -46,7 +46,7 @@ public async Task Listener_ReceivesAllSourcesCompleted() session.WaitForCompletion(); // Wait for the AllSourcesCompleted flag to propagate. - _WaitForCondition(() => listener.AllSourcesCompletedCount > 0); + WaitHelper.WaitUntil(() => listener.AllSourcesCompletedCount > 0); session.Shutdown(); @@ -68,7 +68,7 @@ public async Task Listener_ReceivesPhaseChanged() session.WaitForCompletion(); // Wait for the listener to process all flags. - _WaitForCondition(() => listener.AllSourcesCompletedCount > 0); + WaitHelper.WaitUntil(() => listener.AllSourcesCompletedCount > 0); session.Shutdown(); @@ -91,7 +91,7 @@ public async Task Shutdown_ListenerReceivesShuttingDownAndUnsubscribed() session.Shutdown(); // Wait for listener thread to drain. - _WaitForCondition(() => listener.UnsubscribedCount > 0); + WaitHelper.WaitUntil(() => listener.UnsubscribedCount > 0); await Assert.That(listener.ShuttingDownCount).IsGreaterThanOrEqualTo(1); await Assert.That(listener.UnsubscribedCount).IsEqualTo(1); @@ -150,7 +150,7 @@ public async Task Restart_ClearsCountersAndReparsesFromSources() session.WaitForCompletion(); // Wait for listener to catch up. - _WaitForCondition(() => listener.TotalPacketsSeen >= frameCount); + WaitHelper.WaitUntil(() => listener.TotalPacketsSeen >= frameCount); await Assert.That(session.PacketCount).IsEqualTo(frameCount); @@ -167,11 +167,11 @@ public async Task Restart_ClearsCountersAndReparsesFromSources() await Assert.That(session.PacketCount).IsEqualTo(frameCount); // Listener receives re-parsed packets via StackChanged + NewPackets notification. - _WaitForCondition(() => listener.TotalPacketsSeen >= frameCount * 2); + WaitHelper.WaitUntil(() => listener.TotalPacketsSeen >= frameCount * 2); await Assert.That(listener.TotalPacketsSeen).IsGreaterThanOrEqualTo(frameCount * 2); // Listener received exactly one OnStackChanged callback. - _WaitForCondition(() => listener.StackChangedCount >= 1); + WaitHelper.WaitUntil(() => listener.StackChangedCount >= 1); await Assert.That(listener.StackChangedCount).IsEqualTo(1); session.Shutdown(); @@ -217,7 +217,7 @@ public async Task MultipleListeners_AllReceivePackets() session.WaitForCompletion(); // Both listeners should see all packets. - _WaitForCondition(() => + WaitHelper.WaitUntil(() => listener1.TotalPacketsSeen >= frameCount && listener2.TotalPacketsSeen >= frameCount); @@ -245,7 +245,7 @@ public async Task Dispose_ImplicitShutdown() session.Dispose(); // Listener should have been notified and unsubscribed. - _WaitForCondition(() => listener.UnsubscribedCount > 0); + WaitHelper.WaitUntil(() => listener.UnsubscribedCount > 0); await Assert.That(listener.UnsubscribedCount).IsEqualTo(1); } @@ -295,23 +295,4 @@ public async Task GetFrameSources_ReturnsRegisteredSources() await Assert.That(sources.Count).IsEqualTo(1); await Assert.That(sources[0].UiName).IsEqualTo("TestSource"); } - - /// - /// Spins for up to ~5 seconds until returns true. - /// Throws if the condition is not met. - /// - private static void _WaitForCondition(Func condition, int timeoutMs = 5000) - { - Stopwatch sw = Stopwatch.StartNew(); - SpinWait wait = new(); - while (!condition()) - { - if (sw.ElapsedMilliseconds > timeoutMs) - { - throw new TimeoutException( - $"Condition was not met within {timeoutMs} ms."); - } - wait.SpinOnce(); - } - } } diff --git a/NetworkInspector.Sessions.Tests/UnsubscribeTests.cs b/NetworkInspector.Sessions.Tests/UnsubscribeTests.cs index 2ccd50e..7ccb20d 100644 --- a/NetworkInspector.Sessions.Tests/UnsubscribeTests.cs +++ b/NetworkInspector.Sessions.Tests/UnsubscribeTests.cs @@ -24,7 +24,7 @@ public async Task TryUnsubscribe_Source_StopsFrameReading() session.TryStart(); // Wait for initial frames to be consumed. - _WaitForCondition(() => session.PacketCount >= initialFrames); + WaitHelper.WaitUntil(() => session.PacketCount >= initialFrames); // Unsubscribe the source — should cancel its job. JobInfo sourceJob = session.GetJobs().First( @@ -34,7 +34,7 @@ public async Task TryUnsubscribe_Source_StopsFrameReading() await Assert.That(result).IsTrue(); // Wait for the source job to reach a terminal state. - _WaitForCondition( + WaitHelper.WaitUntil( () => sourceJob.Status is JobStatus.Cancelled or JobStatus.Completed); // The source should have stopped producing frames shortly after unsubscribe. @@ -81,7 +81,7 @@ public async Task TryUnsubscribe_LastSource_TransitionsToStopped() session.TryAddListener(listener, out _); session.TryStart(); - _WaitForCondition(() => session.PacketCount >= initialFrames); + WaitHelper.WaitUntil(() => session.PacketCount >= initialFrames); // Unsubscribe the only source. JobInfo sourceJob = session.GetJobs().First( @@ -91,7 +91,7 @@ public async Task TryUnsubscribe_LastSource_TransitionsToStopped() await Assert.That(result).IsTrue(); // Session should transition to Stopped since all sources are done. - _WaitForCondition( + WaitHelper.WaitUntil( () => session.Phase == SessionPhase.Stopped); await Assert.That(session.Phase).IsEqualTo(SessionPhase.Stopped); @@ -111,14 +111,14 @@ public async Task FrameSourceInfo_Stop_ConvenienceApi() session.TryAddFrameSource(source, out FrameSourceInfo? sourceInfo); session.TryStart(); - _WaitForCondition(() => session.PacketCount >= initialFrames); + WaitHelper.WaitUntil(() => session.PacketCount >= initialFrames); // Use convenience API. await Assert.That(sourceInfo!.IsStoppable).IsTrue(); sourceInfo.Stop(); // Wait for the source to stop. - _WaitForCondition( + WaitHelper.WaitUntil( () => session.Phase == SessionPhase.Stopped); await Assert.That(session.PacketCount).IsGreaterThanOrEqualTo(initialFrames); @@ -141,7 +141,7 @@ public async Task TryUnsubscribe_Listener_CallsOnUnsubscribed() session.TryAddListener(listener, out ListenerInfo? listenerInfo); session.TryStart(); - _WaitForCondition(() => session.PacketCount >= frameCount); + WaitHelper.WaitUntil(() => session.PacketCount >= frameCount); // Find the listener's job. JobInfo listenerJob = session.GetJobs().First( @@ -151,7 +151,7 @@ public async Task TryUnsubscribe_Listener_CallsOnUnsubscribed() await Assert.That(result).IsTrue(); // OnUnsubscribed should have been called. - _WaitForCondition(() => listener.UnsubscribedCount > 0); + WaitHelper.WaitUntil(() => listener.UnsubscribedCount > 0); await Assert.That(listener.UnsubscribedCount).IsEqualTo(1); // Status should be Unsubscribed. @@ -176,13 +176,13 @@ public async Task ListenerInfo_Unsubscribe_ConvenienceApi() session.TryAddListener(listener, out ListenerInfo? listenerInfo); session.TryStart(); - _WaitForCondition(() => session.PacketCount >= frameCount); + WaitHelper.WaitUntil(() => session.PacketCount >= frameCount); // Use convenience API. listenerInfo!.Unsubscribe(); // OnUnsubscribed should have been called. - _WaitForCondition(() => listener.UnsubscribedCount > 0); + WaitHelper.WaitUntil(() => listener.UnsubscribedCount > 0); await Assert.That(listener.UnsubscribedCount).IsEqualTo(1); await Assert.That(listenerInfo.Status).IsEqualTo(SubscriptionStatus.Unsubscribed); @@ -219,7 +219,7 @@ public async Task TryUnsubscribe_UserJob_CancelsJob() session.TryAddFrameSource(source, out _); session.TryStart(); - _WaitForCondition(() => session.Phase == SessionPhase.Running); + WaitHelper.WaitUntil(() => session.Phase == SessionPhase.Running); // Add a long-running user job. using ManualResetEventSlim gate = new(false); @@ -236,14 +236,14 @@ public async Task TryUnsubscribe_UserJob_CancelsJob() await Assert.That(jobInfo).IsNotNull(); // Wait for the job to start. - _WaitForCondition(() => jobInfo!.Status == JobStatus.Running); + WaitHelper.WaitUntil(() => jobInfo!.Status == JobStatus.Running); // Unsubscribe the user job. bool result = session.TryUnsubscribe(jobInfo!); await Assert.That(result).IsTrue(); - _WaitForCondition( + WaitHelper.WaitUntil( () => jobInfo!.Status is JobStatus.Cancelled or JobStatus.Completed); source.Release(); @@ -266,7 +266,7 @@ public async Task TryUnsubscribe_TerminalJob_ReturnsFalse() // Source job is already completed. JobInfo sourceJob = session.GetJobs().First( j => j.UiName == source.UiName); - _WaitForCondition( + WaitHelper.WaitUntil( () => sourceJob.Status is JobStatus.Completed or JobStatus.Cancelled); bool result = session.TryUnsubscribe(sourceJob); @@ -286,7 +286,7 @@ public async Task TryUnsubscribe_ForeignJob_ReturnsFalse() session.TryAddFrameSource(source, out _); session.TryStart(); - _WaitForCondition(() => session.Phase == SessionPhase.Running); + WaitHelper.WaitUntil(() => session.Phase == SessionPhase.Running); using ManualResetEventSlim gate = new(false); using Job foreignJob = new( @@ -327,24 +327,4 @@ public async Task TryUnsubscribe_IdlePhase_ReturnsFalse() await Assert.That(result).IsFalse(); } - - // ── Helpers ─────────────────────────────────────────────────────────────── - - /// - /// Spins for up to ~5 seconds until returns true. - /// - private static void _WaitForCondition(Func condition, int timeoutMs = 5000) - { - Stopwatch sw = Stopwatch.StartNew(); - SpinWait wait = new(); - while (!condition()) - { - if (sw.ElapsedMilliseconds > timeoutMs) - { - throw new TimeoutException( - $"Condition was not met within {timeoutMs} ms."); - } - wait.SpinOnce(); - } - } } diff --git a/NetworkInspector.Sessions/src/Jobs/Job.cs b/NetworkInspector.Sessions/src/Jobs/Job.cs index 9d08296..2c2a7f6 100644 --- a/NetworkInspector.Sessions/src/Jobs/Job.cs +++ b/NetworkInspector.Sessions/src/Jobs/Job.cs @@ -1,6 +1,7 @@ // Copyright © 2026 DevAM. All rights reserved. Licensed under MIT license. See license in the repository root for license information. namespace NetworkInspector.Sessions.Jobs; + /// /// Represents a single unit of work in the session. /// Each job runs on its own dedicated background thread. @@ -14,6 +15,8 @@ namespace NetworkInspector.Sessions.Jobs; /// internal sealed class Job : IDisposable { + #region Fields + private readonly Action _Work; private readonly CancellationTokenSource _Cts = new(); private readonly Action _OnStatusChanged; @@ -34,6 +37,11 @@ internal sealed class Job : IDisposable private DateTimeOffset _EndTimeValue; private int _EndTimeSet; // 0 = not set, 1 = set private Exception? _FailureException; + + #endregion + + #region Lifecycle + /// Creates a job with the given identity and work delegate. internal Job( JobId id, @@ -48,7 +56,11 @@ internal Job( _Work = work; _OnStatusChanged = onStatusChanged; } - // ── Identity ───────────────────────────────────────────────────────────── + + #endregion + + #region Identity + /// Unique job identifier within the session. internal JobId Id { @@ -64,7 +76,11 @@ internal string Description { get; } - // ── Observable state (thread-safe reads) ───────────────────────────────── + + #endregion + + #region Observable state + /// Current execution status. Volatile read — always current. internal JobStatus Status => (JobStatus)Volatile.Read(ref _Status); /// When the job thread started. Null if not yet started. @@ -95,7 +111,11 @@ internal DateTimeOffset? EndTime } /// Exception that caused job failure, if any. internal Exception? FailureException => Volatile.Read(ref _FailureException!); - // ── Lifecycle ───────────────────────────────────────────────────────────── + + #endregion + + #region Public API + /// /// Starts the job on a new background thread. /// Transitions status from to . @@ -181,9 +201,14 @@ public void Dispose() _Completed.Dispose(); _Cts.Dispose(); } - // ── Private implementation ──────────────────────────────────────────────── - private static bool _IsTerminal(JobStatus status) => + + #endregion + + #region Private helpers + + private static bool _IsTerminal(JobStatus status) => status is JobStatus.Completed or JobStatus.Cancelled or JobStatus.Failed; + /// /// Entry point for the job thread. Executes the work delegate and updates /// status regardless of outcome. @@ -229,5 +254,7 @@ private void _SetStatus(JobStatus status) } _OnStatusChanged(this, status); } + + #endregion } diff --git a/NetworkInspector.Sessions/src/Listeners/ListenerSlot.cs b/NetworkInspector.Sessions/src/Listeners/ListenerSlot.cs index fc2c92f..75d6429 100644 --- a/NetworkInspector.Sessions/src/Listeners/ListenerSlot.cs +++ b/NetworkInspector.Sessions/src/Listeners/ListenerSlot.cs @@ -1,6 +1,7 @@ // Copyright © 2026 DevAM. All rights reserved. Licensed under MIT license. See license in the repository root for license information. namespace NetworkInspector.Sessions.Listeners; + /// /// Session-internal bridge: holds the notification flags and pull cursor /// for a single . @@ -25,9 +26,8 @@ namespace NetworkInspector.Sessions.Listeners; /// internal sealed class ListenerSlot : IDisposable { - // ── Atomic flag field ──────────────────────────────────────────────────── - // Written by producers via Interlocked.Or, read+cleared by consumer via Interlocked.Exchange. - // Public (internal) for direct OR access from source threads without method-call overhead. + #region Fields + internal int Flags; private readonly ISessionListener _Listener; private readonly ISessionReader _SessionReader; @@ -37,6 +37,11 @@ internal sealed class ListenerSlot : IDisposable private long _PacketCursor; // 0 = OnUnsubscribed not yet invoked; 1 = invoked (RunLoop finally or coordinator fallback). private int _OnUnsubscribedInvoked; + + #endregion + + #region Lifecycle + /// /// Creates a listener slot that delivers notifications to /// and pulls data from . @@ -60,7 +65,11 @@ internal ListenerSlot( // in the unified Session job list. Info = new JobInfo(_Job); } - // ── Identity ───────────────────────────────────────────────────────────── + + #endregion + + #region Identity + /// The underlying job's identifier. internal JobId Id => _Job.Id; /// @@ -84,7 +93,11 @@ internal ListenerInfo? ListenerInfo } /// User-visible listener name. internal string UiName => _Listener.UiName; - // ── Lifecycle ──────────────────────────────────────────────────────────── + + #endregion + + #region Public API + /// Starts the listener slot's background thread. internal void Start() { @@ -144,7 +157,11 @@ internal void Notify(NotifyFlags flags) [MethodImpl(MethodImplOptions.AggressiveInlining)] internal void ResetPacketCursor() => Volatile.Write(ref _PacketCursor, 0); - // ── Event-driven dispatch loop ────────────────────────────────────────── + + #endregion + + #region Private helpers + /// /// The work delegate executed by the underlying thread. /// Waits on when no flags are pending, then dispatches to @@ -260,5 +277,7 @@ private void _DispatchFlags(NotifyFlags notify) _Listener.OnShuttingDown(); } } + + #endregion } diff --git a/NetworkInspector.Sessions/src/Session/ISession.cs b/NetworkInspector.Sessions/src/Session/ISession.cs index 5ccf8bf..40cb134 100644 --- a/NetworkInspector.Sessions/src/Session/ISession.cs +++ b/NetworkInspector.Sessions/src/Session/ISession.cs @@ -28,7 +28,8 @@ public interface ISession : ISessionReader, IDisposable /// /// Attempts to register a session listener. May be called while the session is - /// or . + /// , , or + /// . /// Returns if the session is shutting down or stopped. /// /// @@ -178,8 +179,10 @@ bool TryAddJob(string uiName, string description, Action work /// /// If is , waits indefinitely for /// graceful completion. If a is provided, source jobs are - /// given that long to finish before they are force-cancelled. - /// Shutdown(TimeSpan.Zero) is equivalent to an immediate forced shutdown. + /// given that long to finish before shutdown teardown continues; jobs that are still + /// running remain cancelled but may not have exited yet — inspect job status via + /// . Shutdown(TimeSpan.Zero) skips waiting + /// for source completion. /// /// /// Idempotent — safe to call multiple times. diff --git a/NetworkInspector.Sessions/src/Session/Session.cs b/NetworkInspector.Sessions/src/Session/Session.cs index 8e69d51..f6dcd58 100644 --- a/NetworkInspector.Sessions/src/Session/Session.cs +++ b/NetworkInspector.Sessions/src/Session/Session.cs @@ -1381,7 +1381,9 @@ private Packet _ParseFrameUnderLock(Frame frame, PacketId? packetId) /// Non-blocking, lock-free. Safe to call from any thread. /// /// - /// is set once per parsed frame (O(frames × listeners) atomic ORs). + /// is set once per parsed frame + /// (O(frames × listeners) atomic ORs). Listener slots coalesce duplicate + /// flags between wake cycles; each frame still issues one OR per listener. /// [MethodImpl(MethodImplOptions.AggressiveInlining)] private void _NotifyAllListeners(NotifyFlags flags) diff --git a/NetworkInspector.Sessions/src/Session/SessionState.cs b/NetworkInspector.Sessions/src/Session/SessionState.cs index 77509d9..81a112c 100644 --- a/NetworkInspector.Sessions/src/Session/SessionState.cs +++ b/NetworkInspector.Sessions/src/Session/SessionState.cs @@ -27,9 +27,25 @@ internal void SetPhase(SessionPhase phase) /// Allocates the next unique job ID. Thread-safe. internal JobId AllocateJobId() - => new((int)Interlocked.Increment(ref _NextJobId) - 1); + { + long next = Interlocked.Increment(ref _NextJobId); + if (next > int.MaxValue) + { + throw new InvalidOperationException("Maximum job ID count exceeded."); + } + + return new((int)next - 1); + } /// Allocates the next unique listener ID. Thread-safe. internal ListenerId AllocateListenerId() - => new((int)Interlocked.Increment(ref _NextListenerId) - 1); + { + long next = Interlocked.Increment(ref _NextListenerId); + if (next > int.MaxValue) + { + throw new InvalidOperationException("Maximum listener ID count exceeded."); + } + + return new((int)next - 1); + } } diff --git a/NetworkInspector.Sessions/src/ThreadWaitHelper.cs b/NetworkInspector.Sessions/src/ThreadWaitHelper.cs index a185e75..6e8b769 100644 --- a/NetworkInspector.Sessions/src/ThreadWaitHelper.cs +++ b/NetworkInspector.Sessions/src/ThreadWaitHelper.cs @@ -4,6 +4,8 @@ namespace NetworkInspector.Sessions; /// /// Spin-then-sleep wait helpers for non-hot-path blocking (job join, shutdown drain). +/// uses a tight loop for concurrent +/// callers where minimal latency is preferred. /// Caps busy-spin before yielding to avoid burning a core when work is slow or stuck. /// internal static class ThreadWaitHelper