diff --git a/src/RedShirt.Example.JobWorker.Core/Services/ExecutionState/JobLoaderStateService.cs b/src/RedShirt.Example.JobWorker.Core/Services/ExecutionState/JobLoaderStateService.cs index 2573bff..77d68e1 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/ExecutionState/JobLoaderStateService.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/ExecutionState/JobLoaderStateService.cs @@ -5,7 +5,12 @@ namespace RedShirt.Example.JobWorker.Core.Services.ExecutionState; /// internal interface IJobLoaderStateReaderService { - bool HasLoaderStarted(); + /// + /// Thread-safe addition of callback actions invoked once the loader has both started and stopped. + /// If the loader is already finished, the callback is invoked immediately. + /// + /// + void AddOnFinishCallback(Action callback); bool IsLoaderFinished(); } @@ -24,8 +29,7 @@ internal interface IJobLoaderStateService : IJobLoaderStateReaderService internal sealed class JobLoaderStateService : IJobLoaderStateService { /// - /// Multithreading protection. - /// Feels a little silly for a service that only sets booleans to true, but it makes automated audits happy. + /// Multithreading protection for start/stop flags and finish callbacks. /// private readonly Lock _lock = new(); @@ -33,35 +37,89 @@ internal sealed class JobLoaderStateService : IJobLoaderStateService private bool _isStarted; + private Action? _onFinishCallbacks; + + private bool IsFinishedUnsafe() + { + return _isStarted && _isFinished; + } + + /// + /// Detaches finish callbacks if the loader is finished. + /// Must be safe to call with or without the caller already holding + /// ( is reentrant). + /// + private Action? TakeCallbacksIfFinished() + { + lock (_lock) + { + if (!IsFinishedUnsafe()) + { + return null; + } + + return Interlocked.Exchange(ref _onFinishCallbacks, null); + } + } + + private static void InvokeCallbacks(Action? callbacks) + { + if (callbacks is null) + { + return; + } + + foreach (var invocation in callbacks.GetInvocationList()) + { + ((Action) invocation)(); + } + } + public void ReportLoaderStart() { + Action? callbacks; lock (_lock) { _isStarted = true; + callbacks = TakeCallbacksIfFinished(); } + + InvokeCallbacks(callbacks); } public void ReportLoaderStop() { + Action? callbacks; lock (_lock) { _isFinished = true; + callbacks = TakeCallbacksIfFinished(); } + + InvokeCallbacks(callbacks); } - public bool HasLoaderStarted() + public bool IsLoaderFinished() { lock (_lock) { - return _isStarted; + return IsFinishedUnsafe(); } } - public bool IsLoaderFinished() + public void AddOnFinishCallback(Action callback) { + ArgumentNullException.ThrowIfNull(callback); + lock (_lock) { - return _isStarted && _isFinished; + if (!IsFinishedUnsafe()) + { + _onFinishCallbacks += callback; + return; + } } + + callback(); } } \ No newline at end of file diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index c520865..14606ea 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -15,7 +15,7 @@ namespace RedShirt.Example.JobWorker.Core.Services.Jobs; /// If you choose to use one polling mode when applying this template by pruning the other one, then you may want to /// prune in this method as well. /// -internal interface IJobRepository +internal interface IJobRepository : IDisposable { Task> GetAllIdempotencyBlockedJobsAsync(CancellationToken cancellationToken = default); Task> GetAllInFlightJobsAsync(CancellationToken cancellationToken = default); @@ -61,33 +61,50 @@ Task LoadAsync(IReadOnlyList intakeItems, Task WaitForJobDemandAsync(TimeSpan waitDuration, CancellationToken cancellationToken = default); } -internal sealed class JobRepository( - IExecutionEndArbiter executionEndArbiter, - IJobLoaderStateReaderService jobLoaderStateService, - ISourceMessageSorter sorter, - IOptions options) - : IJobRepository +internal sealed class JobRepository : IJobRepository { private readonly Lock _callbackLock = new(); + + private readonly CancellationTokenSource _cancellationTokenSource = new(); + + private readonly IExecutionEndArbiter _executionEndArbiter; + private readonly Lock _generalGate = new(); private readonly SemaphoreSlim _inactiveJobsListSemaphore = new(1, 1); + private readonly IJobLoaderStateReaderService _jobLoaderStateService; /// /// Indicates that jobs are available to be pulled by JobExecutor instances via GetNextJobAsync. /// private readonly AsyncManualResetEvent _jobsAvailableEvent = new(); + /// + /// Guards Set/Reset of together with enqueue onto + /// and . + /// Lock order: then this gate. + /// + private readonly Lock _jobsAvailableGate = new(); + /// /// Signalled when the repository has no watched jobs OR the repository was unable to produce an inactive job for a /// worker request. /// private readonly AsyncManualResetEvent _jobsDemandEvent = new(); + private readonly IOptions _options; + /// /// Signalled when the repository has no watched jobs. /// Starts signalled because the repository begins empty. /// private readonly AsyncManualResetEvent _repositoryEmptyEvent = new(true); + /// + /// Guards Set/Reset of . + /// + private readonly Lock _repositoryEmptyGate = new(); + + private readonly ISourceMessageSorter _sorter; + private readonly Lock _tallyLock = new(); /// @@ -98,6 +115,8 @@ internal sealed class JobRepository( private readonly SemaphoreSlim _watchedJobsListSemaphore = new(1, 1); + private bool _disposed; + private Action? _idempotencyBlockedJobsCallbacks; private int _idempotencyBlockedTally; @@ -114,9 +133,71 @@ internal sealed class JobRepository( private int _inactiveJobsTally; + /// + /// Notes if is set. + /// Created because of optimization paranoia to avoid unnecessary event sets/resets to + /// . + /// Use should be gated behind . + /// + private bool _jobsAvailableEventIsSet; + + /// + /// Notes if is set. + /// Created because of optimization paranoia to avoid unnecessary event sets/resets to + /// . + /// Use should be gated behind . + /// + private bool _repositoryEmptyEventIsSet = true; + private Action? _watchedJobsCallbacks; private int _watchedJobsTally; + private void OnExecutionEndArbiterStop(Exception? exception) + { + ConsiderInterruptingEventWaits(); + } + + /// + /// Check to see if we should cancel the local CancellationTokenSource to interrupt method invocations that are waiting + /// on an event. + /// + private void ConsiderInterruptingEventWaits() + { + lock (_generalGate) + { + if (_disposed) + { + return; + } + + if (!HaveReasonToExpectFutureJobs()) + { + _cancellationTokenSource.Cancel(); + } + } + } + + private void Dispose(bool disposing) + { + if (!disposing) + { + return; + } + + lock (_generalGate) + { + if (_disposed) + { + return; + } + + _disposed = true; + } + + _cancellationTokenSource.Cancel(); + _cancellationTokenSource.Dispose(); + } + private void NotifyInactiveCountUpdate(int count) { Action? callbacks; @@ -150,11 +231,80 @@ private void NotifyWatchedJobsUpdate(int count) callbacks?.Invoke(count); } - private async Task TryGetUnblockedJobAsync(CancellationToken cancellationToken) + /// + /// Align with whether inactive jobs or shortlisted jobs exist. + /// Assumed to run while holding . + /// + private void SyncRepositoryEmptyEvent() + { + lock (_repositoryEmptyGate) + { + bool isEmptyCondition; + lock (_tallyLock) + { + isEmptyCondition = _watchedJobsTally == 0; + } + + if (isEmptyCondition) + { + // Job list is empty + + // ReSharper disable once InvertIf + if (!_repositoryEmptyEventIsSet) + { + _repositoryEmptyEvent.Set(); + _repositoryEmptyEventIsSet = true; + } + } + else if (_repositoryEmptyEventIsSet) + { + _repositoryEmptyEvent.Reset(); + _repositoryEmptyEventIsSet = false; + } + } + } + + /// + /// Align with whether inactive jobs or shortlisted jobs exist. + /// Should be run after state has been updated and a data store has been updated. + /// + private void SyncJobsAvailableEvent() + { + lock (_jobsAvailableGate) + { + bool isEmptyCondition; + lock (_tallyLock) + { + isEmptyCondition = _inactiveJobsTally == 0; + } + + isEmptyCondition &= _unblockedJobsQueue.IsEmpty; + + if (isEmptyCondition) + { + // Job list is empty + + // ReSharper disable once InvertIf + if (_jobsAvailableEventIsSet) + { + _jobsAvailableEvent.Reset(); + _jobsAvailableEventIsSet = false; + } + } + else if (!_jobsAvailableEventIsSet) + { + _jobsAvailableEvent.Set(); + _jobsAvailableEventIsSet = true; + } + } + } + + private TryGetJobResponse TryGetUnblockedJobAsync() { IJobRepositoryEntry? result; var iterated = false; + // ReSharper disable once InconsistentlySynchronizedField while (_unblockedJobsQueue.TryDequeue(out result)) { iterated = true; @@ -167,23 +317,12 @@ private async Task TryGetUnblockedJobAsync(CancellationToken } } + // Need to set state before thinking about syncing events + result?.State = JobState.Active; + if (iterated) { - // Check to see if we emptied the queue, but only if we actually dequeued something - // Assume that the event is up to date and doesn't need a redundant reset. - await _inactiveJobsListSemaphore.WaitAsync(cancellationToken); - try - { - if (_inactiveJobsList.Count == 0 && _unblockedJobsQueue.IsEmpty) - { - // Jobs are no longer available - _jobsAvailableEvent.Reset(); - } - } - finally - { - _inactiveJobsListSemaphore.Release(); - } + SyncJobsAvailableEvent(); } if (result is null @@ -215,12 +354,6 @@ private async Task TryGetInactiveJobAsync(CancellationToken c if (result is not null) { _inactiveJobsList.RemoveAt(0); - - if (_inactiveJobsList.Count == 0 && _unblockedJobsQueue.IsEmpty) - { - // Jobs are no longer available - _jobsAvailableEvent.Reset(); - } } } finally @@ -228,6 +361,14 @@ private async Task TryGetInactiveJobAsync(CancellationToken c _inactiveJobsListSemaphore.Release(); } + // Need to set state before thinking about syncing events + result?.State = JobState.Active; + + if (result is not null) + { + SyncJobsAvailableEvent(); + } + return new TryGetJobResponse { Success = result is not null, @@ -248,11 +389,14 @@ private void OnEntryStateUpdateUnblocked(IJobRepositoryEntry job, JobState? oldS return; } - // Identified as a newly-unblocked job. - // Shortlist the job for re-execution in memory. - _unblockedJobsQueue.Enqueue(job); - // Tell any active invocations of GetNextJobAsync that there is something available. - _jobsAvailableEvent.Set(); + lock (_jobsAvailableGate) + { + // Identified as a newly-unblocked job. + // Shortlist the job for re-execution in memory. + _unblockedJobsQueue.Enqueue(job); + // Tell any active invocations of GetNextJobAsync that there is something available. + _jobsAvailableEvent.Set(); + } } /// @@ -334,20 +478,119 @@ private void OnEntryStateUpdateTallies(IJobRepositoryEntry job, JobState? oldSta } } + var consideringToConsiderCancellingWaitEvents = false; // Yes, this variable name is very silly if (updatedInactive) { NotifyInactiveCountUpdate(localTallyInactive); + consideringToConsiderCancellingWaitEvents |= localTallyInactive == 0; } if (updatedIdempotencyBlocked) { NotifyIdempotencyBlockedCountUpdate(localTallyIdempotencyBlocked); + consideringToConsiderCancellingWaitEvents |= localTallyIdempotencyBlocked == 0; } if (updatedWatched) { NotifyWatchedJobsUpdate(localTallyWatched); } + + if (consideringToConsiderCancellingWaitEvents) + { + ConsiderInterruptingEventWaits(); + } + + /* + * Note: Although tallies are updated here, SyncJobsAvailableEvent should not be invoked here. + * SyncJobsAvailableEvent reads off these tallies that suggest a state, but in practice the events are used + * for more concrete realities and tallies are set before these realities are implemented (read: before the + * inactive jobs list or unblocked jobs queue is updated). Therefore, SyncJobsAvailableEvent should only be + * invoked when these sources of truth have been updated. + */ + } + + private bool HaveReasonToExpectFutureJobs() + { + if ( + // If execution is still running, then we have every reason to believe that there will be more incoming jobs. + // Note: Using the raw IExecutionEndArbiter because we want to avoid a circular dependency. + _executionEndArbiter.ShouldKeepRunning() + // If the job loader is not yet finished, then there may be more incoming jobs. + || !_jobLoaderStateService.IsLoaderFinished()) + { + return true; + } + + lock (_tallyLock) + { + // Confirm whether there are any inactive jobs, or jobs that may become inactive again. + return _inactiveJobsTally > 0 + || _idempotencyBlockedTally > 0; + } + } + + private async Task DoOperationWithLinkedToken(Func operation, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + + /* + * Construct a linked CTS to tie it to _cancellationTokenSource. + * + * Note: During development, I experimented with a GetLinkedToken method, + * but that ended up being invalid. The reason it was invalid is that + * disposing a linked source unregisters it from _cancellationTokenSource. + * The CancellationToken that was being returned was no longer hooked to that source, + * making the end result just the baseline cancellation token with extra steps. + */ + + CancellationTokenSource? linkedCts = null; + try + { + CancellationToken linkedToken; + lock (_generalGate) + { + if (_disposed) + { + throw new OperationCanceledException(); + } + + linkedCts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, _cancellationTokenSource.Token); + linkedToken = linkedCts.Token; + } + + await operation(linkedToken); + } + finally + { + linkedCts?.Dispose(); + } + } + + private Task DoWaitForAvailableJobsAsync(CancellationToken cancellationToken) + { + return _jobsAvailableEvent.WaitAsync(cancellationToken); + } + + private Task DoWaitForEmptyRepositoryAsync(CancellationToken cancellationToken) + { + return _repositoryEmptyEvent.WaitAsync(cancellationToken); + } + + public JobRepository(IExecutionEndArbiter executionEndArbiter, + IJobLoaderStateReaderService jobLoaderStateReaderService, + ISourceMessageSorter sourceMessageSorter, + IOptions options) + { + _executionEndArbiter = executionEndArbiter; + _jobLoaderStateService = jobLoaderStateReaderService; + _sorter = sourceMessageSorter; + _options = options; + + executionEndArbiter.AddOnStopCallback(OnExecutionEndArbiterStop); + jobLoaderStateReaderService.AddOnFinishCallback(ConsiderInterruptingEventWaits); } internal List WatchedJobs { get; } = []; @@ -355,7 +598,6 @@ private void OnEntryStateUpdateTallies(IJobRepositoryEntry job, JobState? oldSta public async Task> GetAllInFlightJobsAsync(CancellationToken cancellationToken = default) { await _watchedJobsListSemaphore.WaitAsync(cancellationToken); - try { var items = WatchedJobs @@ -412,18 +654,18 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo do { // Try shortlist of unblocked jobs - if (await TryGetUnblockedJobAsync(cancellationToken) is {Success: true} unblockedAttemptResult) + if (TryGetUnblockedJobAsync() is {Success: true, Result: { } formerlyUnblockedJobResult}) { - result = unblockedAttemptResult.Result!; + result = formerlyUnblockedJobResult; // Continue out of loop iteration to abort via do-while condition continue; } - if (await TryGetInactiveJobAsync(cancellationToken) is {Success: true} inactiveAttemptResult) + if (await TryGetInactiveJobAsync(cancellationToken) is + {Success: true, Result: { } formerlyInactiveJobResult}) { - result = inactiveAttemptResult.Result!; - + result = formerlyInactiveJobResult; // Continue out of loop iteration to abort via do-while condition continue; } @@ -431,14 +673,7 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo // If execution has reached here, then there are currently no available jobs to be handed out. // Is it because we've been asked to stop running? - if ( - // Note: Using the raw IExecutionEndArbiter because we want to avoid a circular dependency - !executionEndArbiter.ShouldKeepRunning() - // Confirm that the job loader thread has finished and will not be loading any more jobs - && jobLoaderStateService.IsLoaderFinished() - // Confirm that there are no more jobs in the background. - // This was already implied by the overall method structure, but now that the loader is finished we want to guarantee it - && await GetInactiveJobCountAsync(cancellationToken) == 0) + if (!HaveReasonToExpectFutureJobs()) { // It IS because we've been asked to stop running! // We have also confirmed that the job loader is fully finished, and no more jobs are incoming @@ -446,17 +681,21 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo } // Note that there's a demand. - // Only the JobLoader should care about this via the IJobRepository.WaitForJobDemandAsync method + // Only the loader mode should care about this via the IJobRepository.WaitForJobDemandAsync method _jobsDemandEvent.Set(); - // Wait for jobs to arrive - // The milliseconds timeout is necessary due to timing problems that came up during unit testing - // I can't say that I'm thrilled with it, though... - await _jobsAvailableEvent.WaitAsync(TimeSpan.FromMilliseconds(250), cancellationToken); + try + { + await DoOperationWithLinkedToken(DoWaitForAvailableJobsAsync, cancellationToken); + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + // Exception from a cancelled internal CTS suggests shutdown + // Manually break from loop so that invoking executors can abort + break; + } } while (result is null); - result.State = JobState.Active; - return result; } @@ -474,7 +713,6 @@ public async Task LoadAsync(IReadOnlyList intakeItems, try { await _watchedJobsListSemaphore.WaitAsync(cancellationToken); - try { // ReSharper disable once ForeachCanBePartlyConvertedToQueryUsingAnotherGetEnumerator @@ -495,8 +733,6 @@ public async Task LoadAsync(IReadOnlyList intakeItems, WatchedJobs.Add(job); _jobsDemandEvent.Reset(); - // Once jobs are added, then the repository is either no longer empty or continues to not be empty. - _repositoryEmptyEvent.Reset(); } } finally @@ -511,15 +747,15 @@ public async Task LoadAsync(IReadOnlyList intakeItems, * * Needs to be compatible with Batch mode, at least for the time being. * * We're assuming that we're not working with enormous datasets for our backlog size. */ - _inactiveJobsList = sorter.GetSortedListOfJobs(_inactiveJobsList); + _inactiveJobsList = _sorter.GetSortedListOfJobs(_inactiveJobsList); + SyncJobsAvailableEvent(); + SyncRepositoryEmptyEvent(); } finally { _inactiveJobsListSemaphore.Release(); } - _jobsAvailableEvent.Set(); - NotifyWatchedJobsUpdate(await GetWatchedJobsCountAsync(cancellationToken)); } @@ -530,32 +766,27 @@ public async Task RemoveJobAsync(IJobRepositoryEntry job, CancellationToken canc await _inactiveJobsListSemaphore.WaitAsync(cancellationToken); try { - if (_inactiveJobsList.Remove(job) - && _inactiveJobsList.Count == 0 - && _unblockedJobsQueue.IsEmpty) - { - // Jobs are no longer available - _jobsAvailableEvent.Reset(); - } + _inactiveJobsList.Remove(job); + SyncJobsAvailableEvent(); } finally { _inactiveJobsListSemaphore.Release(); } + var watchedIsNowEmpty = false; await _watchedJobsListSemaphore.WaitAsync(cancellationToken); try { - WatchedJobs.Remove(job); - - if (WatchedJobs.Count == 0) + if (WatchedJobs.Remove(job) && WatchedJobs.Count == 0) { + // Just reached zero. // Avoid possible race condition in JobLoader // If there's nothing to grab, then there must be an executor about to demand something. _jobsDemandEvent.Set(); // No watched jobs is the very definition of an empty repository - _repositoryEmptyEvent.Set(); + watchedIsNowEmpty = true; } } finally @@ -564,6 +795,10 @@ public async Task RemoveJobAsync(IJobRepositoryEntry job, CancellationToken canc } job.Dispose(); + if (watchedIsNowEmpty) + { + SyncRepositoryEmptyEvent(); + } } public void SubscribeToInactiveCountUpdate(Action callback) @@ -620,28 +855,39 @@ public void SubscribeToWatchedJobsUpdate(Action callback) } } - public Task WaitForJobDemandAsync(TimeSpan waitDuration, CancellationToken cancellationToken = default) + public async Task WaitForJobDemandAsync(TimeSpan waitDuration, CancellationToken cancellationToken = default) { - return _jobsDemandEvent.WaitAsync(waitDuration, cancellationToken); + var result = false; + + try + { + await DoOperationWithLinkedToken( + async linkedToken => { result = await _jobsDemandEvent.WaitAsync(waitDuration, linkedToken); }, + cancellationToken); + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + // Pass, use default false + } + + return result; } public async Task WaitForEmptyRepositoryAsync(CancellationToken cancellationToken = default) { - int count; - do + try { - // Short timeout mirrors GetNextJobAsync: avoids missing a Set/Reset edge under concurrency - await _repositoryEmptyEvent.WaitAsync(TimeSpan.FromMilliseconds(250), cancellationToken); - lock (_tallyLock) - { - count = _watchedJobsTally; - } - } while (count > 0); + await DoOperationWithLinkedToken(DoWaitForEmptyRepositoryAsync, cancellationToken); + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + // Pass + } } public int GetBacklogMaxCount() { - return options.Value.EffectiveBacklogSize; + return _options.Value.EffectiveBacklogSize; } public async Task GetInactiveJobCountAsync(CancellationToken cancellationToken = default) @@ -660,6 +906,13 @@ public async Task GetInactiveJobCountAsync(CancellationToken cancellationTo } } + public void Dispose() + { + Dispose(true); + // ReSharper disable once GCSuppressFinalizeForTypeWithoutDestructor + GC.SuppressFinalize(this); + } + private sealed class TryGetJobResponse { public required bool Success { get; init; } diff --git a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/ExecutionState/JobLoaderStateServiceTests.cs b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/ExecutionState/JobLoaderStateServiceTests.cs index 5c696e7..132a59b 100644 --- a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/ExecutionState/JobLoaderStateServiceTests.cs +++ b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/ExecutionState/JobLoaderStateServiceTests.cs @@ -5,64 +5,154 @@ namespace RedShirt.Example.JobWorker.Core.UnitTests.Tests.Services.ExecutionStat // Generated by Cursor public class JobLoaderStateServiceTests { + [Fact] + public void AddOnFinishCallback_AfterStartAndStop_InvokesImmediately() + { + var service = new JobLoaderStateService(); + var invocations = 0; + + service.ReportLoaderStart(); + service.ReportLoaderStop(); + service.AddOnFinishCallback(() => invocations++); + + Assert.Equal(1, invocations); + } + + [Fact] + public void AddOnFinishCallback_AfterStartOnly_DoesNotInvokeUntilStop() + { + var service = new JobLoaderStateService(); + var invocations = 0; + + service.ReportLoaderStart(); + service.AddOnFinishCallback(() => invocations++); + + Assert.Equal(0, invocations); + + service.ReportLoaderStop(); + + Assert.Equal(1, invocations); + } + + [Fact] + public void AddOnFinishCallback_AfterStopOnly_DoesNotInvokeUntilStart() + { + var service = new JobLoaderStateService(); + var invocations = 0; + + service.ReportLoaderStop(); + service.AddOnFinishCallback(() => invocations++); + + Assert.Equal(0, invocations); + + service.ReportLoaderStart(); + + Assert.Equal(1, invocations); + } + /// - /// Concurrent start/stop/read calls should not throw and should eventually report finished. + /// Subscribers racing with start/stop must each run exactly once (immediate or deferred). /// [Fact(Timeout = 2000)] - public async Task TestConcurrentAccess_EventuallyFinished() + public async Task AddOnFinishCallback_ConcurrentSubscribe_EachRunsOnce() { var service = new JobLoaderStateService(); + var invocations = 0; var cancellationToken = TestContext.Current.CancellationToken; + const int subscriberCount = 64; - var startTask = Task.Run(() => + var subscribeTask = Task.Run(() => { - for (var i = 0; i < 100; i++) - { - service.ReportLoaderStart(); - _ = service.IsLoaderFinished(); - } + Parallel.For(0, subscriberCount, new ParallelOptions {CancellationToken = cancellationToken}, + _ => service.AddOnFinishCallback(() => Interlocked.Increment(ref invocations))); }, cancellationToken); - var stopTask = Task.Run(() => + var reportTask = Task.Run(() => { - for (var i = 0; i < 100; i++) - { - service.ReportLoaderStop(); - _ = service.IsLoaderFinished(); - } + service.ReportLoaderStart(); + service.ReportLoaderStop(); }, cancellationToken); - await Task.WhenAll(startTask, stopTask); + await Task.WhenAll(subscribeTask, reportTask); + Assert.Equal(subscriberCount, invocations); Assert.True(service.IsLoaderFinished()); } [Fact] - public void TestHasLoaderStarted_AfterStart_True() + public void AddOnFinishCallback_InvokesRegisteredCallbacksOnce() { var service = new JobLoaderStateService(); + var invocations = 0; + + service.AddOnFinishCallback(() => invocations++); + service.AddOnFinishCallback(() => invocations++); service.ReportLoaderStart(); + service.ReportLoaderStop(); + service.ReportLoaderStop(); - Assert.True(service.HasLoaderStarted()); + Assert.Equal(2, invocations); } - [Fact] - public void TestHasLoaderStarted_AfterStopOnly_False() + [Fact(Timeout = 2000)] + public async Task AddOnFinishCallback_IsThreadSafeAndRunsOnce() { var service = new JobLoaderStateService(); + var invocations = 0; + var cancellationToken = TestContext.Current.CancellationToken; - service.ReportLoaderStop(); + service.AddOnFinishCallback(() => Interlocked.Increment(ref invocations)); + + await Task.WhenAll( + Enumerable.Range(0, 32).Select(_ => Task.Run(() => + { + service.ReportLoaderStart(); + service.ReportLoaderStop(); + }, cancellationToken))); - Assert.False(service.HasLoaderStarted()); + Assert.Equal(1, invocations); + Assert.True(service.IsLoaderFinished()); } [Fact] - public void TestHasLoaderStarted_InitialState_False() + public void AddOnFinishCallback_WhenNull_ThrowsArgumentNullException() { var service = new JobLoaderStateService(); - Assert.False(service.HasLoaderStarted()); + Assert.Throws(() => service.AddOnFinishCallback(null!)); + } + + /// + /// Concurrent start/stop/read calls should not throw and should eventually report finished. + /// + [Fact(Timeout = 2000)] + public async Task TestConcurrentAccess_EventuallyFinished() + { + var service = new JobLoaderStateService(); + var cancellationToken = TestContext.Current.CancellationToken; + + var startTask = Task.Run(() => + { + for (var i = 0; i < 100; i++) + { + service.ReportLoaderStart(); + _ = service.IsLoaderFinished(); + } + }, cancellationToken); + + var stopTask = Task.Run(() => + { + for (var i = 0; i < 100; i++) + { + service.ReportLoaderStop(); + _ = service.IsLoaderFinished(); + } + }, cancellationToken); + + await Task.WhenAll(startTask, stopTask); + + Assert.True(service.IsLoaderFinished()); } /// diff --git a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs index e9098ca..084d963 100644 --- a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs +++ b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs @@ -7,11 +7,28 @@ using RedShirt.Example.JobWorker.Core.Services.SourceMessages; using RedShirt.Example.JobWorker.Core.Utility; using System.Diagnostics; +using Range = Moq.Range; namespace RedShirt.Example.JobWorker.Core.UnitTests.Tests.Services.Jobs; public class JobRepositoryTests { + private static void SetupConstructionCallbacks( + Mock executionEndArbiter, + Mock jobLoaderStateService) + { + executionEndArbiter.Setup(a => a.AddOnStopCallback(It.IsAny>())); + jobLoaderStateService.Setup(s => s.AddOnFinishCallback(It.IsAny())); + } + + private static void VerifyConstructionCallbacks( + Mock executionEndArbiter, + Mock jobLoaderStateService) + { + executionEndArbiter.Verify(a => a.AddOnStopCallback(It.IsAny>()), Times.Once); + jobLoaderStateService.Verify(s => s.AddOnFinishCallback(It.IsAny()), Times.Once); + } + private static JobRepository CreateRepository( Mock? executionEndArbiter = null, Mock? jobLoaderStateService = null, @@ -19,17 +36,24 @@ private static JobRepository CreateRepository( int backlogSize = 10) { executionEndArbiter ??= new Mock(MockBehavior.Strict); + executionEndArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); jobLoaderStateService ??= new Mock(MockBehavior.Strict); sorter ??= new Mock(); sorter .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); - return new JobRepository( + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); + + var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(new JobRepository.ConfigurationModel {BacklogSize = backlogSize})); + + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); + + return jobRepository; } private static Mock CreateJobModel(string messageId) @@ -39,6 +63,26 @@ private static Mock CreateJobModel(string messageId) return jobModel; } + [Fact] + public void Constructor_RegistersStopAndFinishCallbacks() + { + var executionEndArbiter = new Mock(MockBehavior.Strict); + var jobLoaderStateService = new Mock(MockBehavior.Strict); + var sorter = new Mock(MockBehavior.Strict); + + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); + + _ = new JobRepository( + executionEndArbiter.Object, + jobLoaderStateService.Object, + sorter.Object, + Options.Create(new JobRepository.ConfigurationModel {BacklogSize = 0})); + + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); + executionEndArbiter.VerifyNoOtherCalls(); + jobLoaderStateService.VerifyNoOtherCalls(); + } + [Fact(Timeout = 2000)] public async Task GetNextJobAsync_WhenOnlyUnblockedJobIsDisposed_FallsBackToInactiveQueue() { @@ -130,6 +174,7 @@ public async Task LoadAsync_WhenResponseHasNoItems_DoesNotTouchWatchedJobs() var jobLoaderStateService = new Mock(MockBehavior.Strict); var sorter = new Mock(MockBehavior.Strict); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, @@ -138,6 +183,7 @@ public async Task LoadAsync_WhenResponseHasNoItems_DoesNotTouchWatchedJobs() { BacklogSize = 0 })); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); await jobRepository.LoadAsync([], TestContext.Current.CancellationToken); @@ -213,11 +259,13 @@ public async Task TestGetAllIdempotencyBlockedJobsAsync() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var expectedBlockedJobs = new List>(); @@ -287,11 +335,13 @@ public async Task TestGetAllInFlightJobsAsync() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var expectedJobs = new List>(); @@ -366,11 +416,13 @@ public void TestGetBacklogMaxCount(int backlogSize, int expectedEffectiveBatchSi .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); Assert.Equal(expectedEffectiveBatchSize, jobRepository.GetBacklogMaxCount()); } @@ -392,10 +444,12 @@ public async Task TestGetCountsAsync() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); Mock job; @@ -451,11 +505,13 @@ public async Task TestGetNextJobAsync_Null() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); Assert.Null(await jobRepository.GetNextJobAsync(TestContext.Current.CancellationToken)); @@ -482,11 +538,13 @@ public async Task TestLoadJobs(int responseSize) .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -550,11 +608,13 @@ public async Task TestLoadJobsAndWaitForJob(int responseSize) .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -623,11 +683,13 @@ public async Task TestLoadJobsAndWaitForJob_RequeuedBacklog() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -720,11 +782,13 @@ public async Task TestLoadJobsAndWaitForJob_RequeuedBacklogThenEmpty() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -849,11 +913,13 @@ public async Task TestLoadJobsAndWaitForJob_ThenRemove() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -935,7 +1001,8 @@ public async Task TestLoadJobsAndWaitForJob_UntilEmpty(int responseSize) var jobLoaderStateService = new Mock(MockBehavior.Strict); jobLoaderStateService .Setup(s => s.IsLoaderFinished()) - .Returns(true); + // ReSharper disable once AccessToModifiedClosure + .Returns(() => readyToEnd); var options = new JobRepository.ConfigurationModel { @@ -947,11 +1014,13 @@ public async Task TestLoadJobsAndWaitForJob_UntilEmpty(int responseSize) .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -1007,9 +1076,27 @@ public async Task TestLoadJobsAndWaitForJob_UntilEmpty(int responseSize) DateTime.UtcNow + TimeSpan.FromMilliseconds(250)); } - await Task.Delay(TimeSpan.FromMilliseconds(250), TestContext.Current.CancellationToken); + await Task.Delay(TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken); + // Confirm that the task hasn't yet finished after that delay + Assert.False(retrievedJobsTask.IsCompleted); + // Mark things as ready to finish readyToEnd = true; + var jobLoaderStateServiceCallbackInvocation = Assert.Single(jobLoaderStateService.Invocations, + i => i.Method.Name == nameof(jobLoaderStateService.Object.AddOnFinishCallback)); + // At this point I'm just having fun with the ridiculous variable names in this test method. + var jobLoaderStateServiceCallbackInvocationArgumentAsCallback = + jobLoaderStateServiceCallbackInvocation.Arguments[0] as Action; + Assert.NotNull(jobLoaderStateServiceCallbackInvocationArgumentAsCallback); + jobLoaderStateServiceCallbackInvocationArgumentAsCallback(); + + /* + * Now that we've cancelled the mock-application, the task should be finishable through the callbacks. + */ + await Task.Delay(TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken); var retrievedJobs = await retrievedJobsTask; + + /* Verify Results */ + Assert.NotNull(retrievedJobs); Assert.NotEmpty(retrievedJobs); @@ -1025,7 +1112,10 @@ public async Task TestLoadJobsAndWaitForJob_UntilEmpty(int responseSize) Assert.Same(expectedEnvelope.RawJobModel, currentJob.RawJobModel); } - jobLoaderStateService.Verify(s => s.IsLoaderFinished(), Times.Once); + // Verify that state service was called a reasonable number of times (1-4). + // It's going to be at least once, during some debugging with some extra statements it was 4. + // At this point in the test, we're happy as long as it wasn't invoked a million times + jobLoaderStateService.Verify(s => s.IsLoaderFinished(), Times.Between(1, 4, Range.Inclusive)); } /// @@ -1056,11 +1146,13 @@ public async Task TestLoadJobsAndWaitForJob_VerifySetToActive(int responseSize) .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -1123,11 +1215,13 @@ public async Task TestRemoveJobsAsync() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var job = new Mock(); job.SetupProperty(j => j.State, JobState.Active); @@ -1161,11 +1255,13 @@ public async Task TestRemoveJobsAsyncB() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var job = new Mock(); job.SetupProperty(j => j.State, JobState.Active); @@ -1209,11 +1305,13 @@ public async Task TestWaitForDemand_False() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); // Start waiting for there to be a job demand var stopwatch = Stopwatch.StartNew(); @@ -1247,11 +1345,13 @@ public async Task TestWaitForDemand_True() var sorter = new Mock(MockBehavior.Strict); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); // Start waiting for there to be a job demand var demandTask = Task.Run( @@ -1280,6 +1380,7 @@ public async Task UnblockingJob_ShortlistsJobAheadOfInactiveQueue() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, @@ -1288,6 +1389,7 @@ public async Task UnblockingJob_ShortlistsJobAheadOfInactiveQueue() { BacklogSize = 0 })); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var unblockedModel = new Mock(MockBehavior.Strict); unblockedModel.Setup(m => m.MessageId).Returns("unblocked"); @@ -1325,6 +1427,7 @@ await jobRepository.LoadAsync( var nextJob = await jobRepository.GetNextJobAsync(TestContext.Current.CancellationToken); + Assert.NotNull(nextJob); Assert.Same(unblockedEntry, nextJob); Assert.Equal(JobState.Active, nextJob.State); } @@ -1333,6 +1436,7 @@ await jobRepository.LoadAsync( public async Task WaitForEmptyRepositoryAsync_CompletesWhenLastJobRemoved() { var executionEndArbiter = new Mock(MockBehavior.Strict); + executionEndArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); var jobLoaderStateService = new Mock(MockBehavior.Strict); var options = new JobRepository.ConfigurationModel { @@ -1343,11 +1447,13 @@ public async Task WaitForEmptyRepositoryAsync_CompletesWhenLastJobRemoved() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobModel1 = new Mock(MockBehavior.Strict); jobModel1.Setup(m => m.MessageId).Returns(Guid.NewGuid().ToString()); @@ -1412,11 +1518,13 @@ public async Task WaitForEmptyRepositoryAsync_HonorsCancellation() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobModel = new Mock(MockBehavior.Strict); jobModel.Setup(m => m.MessageId).Returns(Guid.NewGuid().ToString()); @@ -1458,11 +1566,13 @@ public async Task WaitForEmptyRepositoryAsync_WhenAlreadyEmpty_ReturnsImmediatel .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var stopwatch = Stopwatch.StartNew(); await jobRepository.WaitForEmptyRepositoryAsync(TestContext.Current.CancellationToken);