From 0edd824b20a3d1d6a42f66206ef6eaa0cdb670a4 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sun, 30 Aug 2026 23:34:44 -0700 Subject: [PATCH 1/6] S2325 --- .../Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs | 4 ---- 1 file changed, 4 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs index 316d0fa..48adc2b 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs @@ -45,9 +45,7 @@ internal class JobSubscriberIntakeQueue : IJobSubscriberIntakeQueue private readonly ConcurrentQueue _jobs = new(); private bool _done; -#pragma warning disable S2325 private void Cancel() -#pragma warning restore S2325 { _done = true; _doNotWaitIfSetEvent.Set(); @@ -57,9 +55,7 @@ public JobSubscriberIntakeQueue(IExecutionEndArbiter executionEndArbiter) { executionEndArbiter.AddOnStopCallback(_ => Cancel()); } -#pragma warning disable S2325 public void Load(IJobSourceResponse jobSourceResponse) -#pragma warning disable S2325 { _jobs.Enqueue(jobSourceResponse); _doNotWaitIfSetEvent.Set(); From 4d2b21bbc64836d753013b1e7d1375232a624104 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sun, 30 Aug 2026 23:36:28 -0700 Subject: [PATCH 2/6] pivot --- .../Jobs/Subscriptions/JobSubscriberIntakeQueue.cs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs index 48adc2b..5d09783 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs @@ -55,6 +55,7 @@ public JobSubscriberIntakeQueue(IExecutionEndArbiter executionEndArbiter) { executionEndArbiter.AddOnStopCallback(_ => Cancel()); } + public void Load(IJobSourceResponse jobSourceResponse) { _jobs.Enqueue(jobSourceResponse); @@ -71,12 +72,14 @@ public void Load(IJobSourceResponse jobSourceResponse) return jobSourceResponse; } - _doNotWaitIfSetEvent.Reset(); - if (_done) { + // Return null, indicating to consumers that things are done return null; } + + // If we are not done, then indicate a demand. + _doNotWaitIfSetEvent.Reset(); } } } \ No newline at end of file From 0296851a29152cc0c2e4307bb5b7c9754ce3cb8b Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sun, 30 Aug 2026 23:37:08 -0700 Subject: [PATCH 3/6] rename --- .../Jobs/Subscriptions/JobSubscriberIntakeQueue.cs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs index 5d09783..3f9ab79 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs @@ -41,14 +41,14 @@ public interface IJobSubscriberIntakeQueue internal class JobSubscriberIntakeQueue : IJobSubscriberIntakeQueue { - private readonly AsyncManualResetEvent _doNotWaitIfSetEvent = new(); + private readonly AsyncManualResetEvent _jobsAreAvailableIfSetEvent = new(); private readonly ConcurrentQueue _jobs = new(); private bool _done; private void Cancel() { _done = true; - _doNotWaitIfSetEvent.Set(); + _jobsAreAvailableIfSetEvent.Set(); } public JobSubscriberIntakeQueue(IExecutionEndArbiter executionEndArbiter) @@ -59,14 +59,14 @@ public JobSubscriberIntakeQueue(IExecutionEndArbiter executionEndArbiter) public void Load(IJobSourceResponse jobSourceResponse) { _jobs.Enqueue(jobSourceResponse); - _doNotWaitIfSetEvent.Set(); + _jobsAreAvailableIfSetEvent.Set(); } public async Task GetNextAsync(CancellationToken cancellationToken = default) { while (true) { - await _doNotWaitIfSetEvent.WaitAsync(cancellationToken); + await _jobsAreAvailableIfSetEvent.WaitAsync(cancellationToken); if (_jobs.TryDequeue(out var jobSourceResponse)) { return jobSourceResponse; @@ -79,7 +79,7 @@ public void Load(IJobSourceResponse jobSourceResponse) } // If we are not done, then indicate a demand. - _doNotWaitIfSetEvent.Reset(); + _jobsAreAvailableIfSetEvent.Reset(); } } } \ No newline at end of file From 5d03171131eec88d6ec8b09a68154ba15bc53416 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sun, 30 Aug 2026 23:49:33 -0700 Subject: [PATCH 4/6] checkpoint --- .../Subscriptions/JobSubscriberIntakeQueue.cs | 73 +++++++++++++++++-- 1 file changed, 66 insertions(+), 7 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs index 3f9ab79..0ce8f6b 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs @@ -41,16 +41,61 @@ public interface IJobSubscriberIntakeQueue internal class JobSubscriberIntakeQueue : IJobSubscriberIntakeQueue { - private readonly AsyncManualResetEvent _jobsAreAvailableIfSetEvent = new(); private readonly ConcurrentQueue _jobs = new(); + private readonly AsyncManualResetEvent _jobsAreAvailableIfSetEvent = new(); + private readonly Lock _lock = new(); private bool _done; + private bool _jobsAreAvailableIfSetEventIsSet; private void Cancel() { _done = true; + // ReSharper disable once InconsistentlySynchronizedField _jobsAreAvailableIfSetEvent.Set(); } + /// + /// Update event state. + /// + private void UpdateEvent() + { + lock (_lock) + { + UpdateEventUnsafe(); + } + } + + /// + /// Update event state. Assumed to be run behind a lock. + /// + private void UpdateEventUnsafe() + { + if (_done) + { + // Already done, keep event in locked-in state. + return; + } + + if (_jobs.IsEmpty) + { + // ReSharper disable once InvertIf + if (_jobsAreAvailableIfSetEventIsSet) + { + _jobsAreAvailableIfSetEvent.Reset(); + _jobsAreAvailableIfSetEventIsSet = false; + } + } + else + { + // ReSharper disable once InvertIf + if (!_jobsAreAvailableIfSetEventIsSet) + { + _jobsAreAvailableIfSetEvent.Set(); + _jobsAreAvailableIfSetEventIsSet = true; + } + } + } + public JobSubscriberIntakeQueue(IExecutionEndArbiter executionEndArbiter) { executionEndArbiter.AddOnStopCallback(_ => Cancel()); @@ -58,28 +103,42 @@ public JobSubscriberIntakeQueue(IExecutionEndArbiter executionEndArbiter) public void Load(IJobSourceResponse jobSourceResponse) { - _jobs.Enqueue(jobSourceResponse); - _jobsAreAvailableIfSetEvent.Set(); + lock (_lock) + { + _jobs.Enqueue(jobSourceResponse); + UpdateEventUnsafe(); + } } public async Task GetNextAsync(CancellationToken cancellationToken = default) { + IJobSourceResponse? response; while (true) { + // ReSharper disable once InconsistentlySynchronizedField await _jobsAreAvailableIfSetEvent.WaitAsync(cancellationToken); - if (_jobs.TryDequeue(out var jobSourceResponse)) + + // ReSharper disable once InconsistentlySynchronizedField + if (_jobs.TryDequeue(out response)) { - return jobSourceResponse; + break; } if (_done) { // Return null, indicating to consumers that things are done - return null; + break; } // If we are not done, then indicate a demand. - _jobsAreAvailableIfSetEvent.Reset(); + // ReSharper disable once InconsistentlySynchronizedField } + + if (!_done) + { + UpdateEvent(); + } + + return response; } } \ No newline at end of file From 55df0deb74970b5a66a75f746640611dd6b198cf Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Mon, 31 Aug 2026 00:16:05 -0700 Subject: [PATCH 5/6] finalize --- .../Subscriptions/JobSubscriberIntakeQueue.cs | 51 ++++++++++++------- 1 file changed, 33 insertions(+), 18 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs index 0ce8f6b..8ed336b 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs @@ -39,19 +39,30 @@ public interface IJobSubscriberIntakeQueue void Load(IJobSourceResponse jobSourceResponse); } +/// +/// Subscriber intake queue implementation. Probably a bit more thread-safe than it really needs to be... +/// internal class JobSubscriberIntakeQueue : IJobSubscriberIntakeQueue { private readonly ConcurrentQueue _jobs = new(); private readonly AsyncManualResetEvent _jobsAreAvailableIfSetEvent = new(); private readonly Lock _lock = new(); private bool _done; + + /// + /// Like in other places in this codebase, using a bool out of complete paranoia of redundant Set/Resets having an + /// impact. + /// private bool _jobsAreAvailableIfSetEventIsSet; private void Cancel() { - _done = true; - // ReSharper disable once InconsistentlySynchronizedField - _jobsAreAvailableIfSetEvent.Set(); + lock (_lock) + { + // Mark as done + _done = true; + _jobsAreAvailableIfSetEvent.Set(); + } } /// @@ -115,27 +126,31 @@ public void Load(IJobSourceResponse jobSourceResponse) IJobSourceResponse? response; while (true) { + // Wait until there are jobs available. // ReSharper disable once InconsistentlySynchronizedField await _jobsAreAvailableIfSetEvent.WaitAsync(cancellationToken); - // ReSharper disable once InconsistentlySynchronizedField - if (_jobs.TryDequeue(out response)) - { - break; - } - - if (_done) + lock (_lock) { - // Return null, indicating to consumers that things are done - break; + try + { + if (_jobs.TryDequeue(out response)) + { + break; + } + } + finally + { + UpdateEvent(); + } + + if (_done) + { + // Return null, indicating to consumers that things are done + break; + } } - // If we are not done, then indicate a demand. - // ReSharper disable once InconsistentlySynchronizedField - } - - if (!_done) - { UpdateEvent(); } From c8739a5036a62395432b76c6f4599ed9214d0675 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Mon, 31 Aug 2026 00:21:23 -0700 Subject: [PATCH 6/6] cleanup of our cleanup --- .../Subscriptions/JobSubscriberIntakeQueue.cs | 15 +-------------- 1 file changed, 1 insertion(+), 14 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs index 8ed336b..7545ad3 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs @@ -65,17 +65,6 @@ private void Cancel() } } - /// - /// Update event state. - /// - private void UpdateEvent() - { - lock (_lock) - { - UpdateEventUnsafe(); - } - } - /// /// Update event state. Assumed to be run behind a lock. /// @@ -141,7 +130,7 @@ public void Load(IJobSourceResponse jobSourceResponse) } finally { - UpdateEvent(); + UpdateEventUnsafe(); } if (_done) @@ -150,8 +139,6 @@ public void Load(IJobSourceResponse jobSourceResponse) break; } } - - UpdateEvent(); } return response;