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..7545ad3 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs @@ -39,48 +39,108 @@ 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 AsyncManualResetEvent _doNotWaitIfSetEvent = new(); private readonly ConcurrentQueue _jobs = new(); + private readonly AsyncManualResetEvent _jobsAreAvailableIfSetEvent = new(); + private readonly Lock _lock = new(); private bool _done; -#pragma warning disable S2325 + /// + /// 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() -#pragma warning restore S2325 { - _done = true; - _doNotWaitIfSetEvent.Set(); + lock (_lock) + { + // Mark as done + _done = true; + _jobsAreAvailableIfSetEvent.Set(); + } + } + + /// + /// 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()); } -#pragma warning disable S2325 + public void Load(IJobSourceResponse jobSourceResponse) -#pragma warning disable S2325 { - _jobs.Enqueue(jobSourceResponse); - _doNotWaitIfSetEvent.Set(); + lock (_lock) + { + _jobs.Enqueue(jobSourceResponse); + UpdateEventUnsafe(); + } } public async Task GetNextAsync(CancellationToken cancellationToken = default) { + IJobSourceResponse? response; while (true) { - await _doNotWaitIfSetEvent.WaitAsync(cancellationToken); - if (_jobs.TryDequeue(out var jobSourceResponse)) - { - return jobSourceResponse; - } + // Wait until there are jobs available. + // ReSharper disable once InconsistentlySynchronizedField + await _jobsAreAvailableIfSetEvent.WaitAsync(cancellationToken); - _doNotWaitIfSetEvent.Reset(); - - if (_done) + lock (_lock) { - return null; + try + { + if (_jobs.TryDequeue(out response)) + { + break; + } + } + finally + { + UpdateEventUnsafe(); + } + + if (_done) + { + // Return null, indicating to consumers that things are done + break; + } } } + + return response; } } \ No newline at end of file