Skip to content

Commit eda8d64

Browse files
thomhurstthomhurst
committed
refactor: deduplicate lifecycle, fault handling and validation
- Extract shared internal ProcessorLifecycle owning start/ cancel/dispose. The void and result bases cannot share an ancestor (differently typed completion sources), and the previous mirrored ~90 lines of concurrency-sensitive code had already drifted once (one base registered CancelAll, the other Dispose). Each base now supplies only its two typed fan-out loops. - Collapse the StartProcessing extension trampoline into internal instance methods on the bases; call sites keep identical syntax. - Collapse the four identical TaskWrapper catch blocks into two TrySetFromFault overloads so the fault-classification policy exists once per completion-source type. - Reuse GetEnumerableTasks() for OverallTask/Results instead of restating the projection. - Remove ValidateCount/ValidateBatchSize/ValidateParallelism/ ValidateTimeSpan forwarders that only renamed ThrowIfNegative/ThrowIfNegativeOrZero; call the primitives directly. CallerArgumentExpression keeps messages identical. - Build the wrapper/completion-source arrays the same way in all four abstract processor constructors.
1 parent 838180a commit eda8d64

20 files changed

Lines changed: 234 additions & 343 deletions

‎EnumerableAsyncProcessor/Extensions/AsyncProcessorExtensions.cs‎

Lines changed: 0 additions & 20 deletions
This file was deleted.

‎EnumerableAsyncProcessor/ProcessorLifecycle.cs‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
1+
using EnumerableAsyncProcessor.Validation;
2+
3+
namespace EnumerableAsyncProcessor;
4+
5+
/// <summary>
6+
/// Owns the start/cancel/dispose lifecycle shared by the void and result processor base classes,
7+
/// which cannot share a common ancestor because they fan out to differently typed completion sources.
8+
/// </summary>
9+
internal sealed class ProcessorLifecycle
10+
{
11+
private static readonly TimeSpan DisposalTimeout = TimeSpan.FromSeconds(30);
12+
13+
private readonly CancellationTokenSource _cancellationTokenSource;
14+
private readonly Action _trySetCanceledAll;
15+
private readonly Action<Exception> _trySetExceptionAll;
16+
private CancellationTokenRegistration _cancellationTokenRegistration;
17+
private Task? _processTask;
18+
private volatile bool _disposed;
19+
private readonly object _disposeLock = new();
20+
21+
public CancellationToken Token { get; }
22+
23+
public ProcessorLifecycle(CancellationTokenSource cancellationTokenSource, Action trySetCanceledAll, Action<Exception> trySetExceptionAll)
24+
{
25+
ValidationHelper.ValidateCancellationTokenSource(cancellationTokenSource);
26+
27+
_cancellationTokenSource = cancellationTokenSource;
28+
_trySetCanceledAll = trySetCanceledAll;
29+
_trySetExceptionAll = trySetExceptionAll;
30+
Token = cancellationTokenSource.Token;
31+
}
32+
33+
// Cancellation is registered here rather than at construction so that a token cancelled
34+
// while the processor is still being constructed can never fire on a partially built instance.
35+
public void Start(Func<Task> process)
36+
{
37+
_cancellationTokenRegistration = Token.Register(CancelAll);
38+
_processTask = RunProcess(process);
39+
}
40+
41+
private async Task RunProcess(Func<Task> process)
42+
{
43+
try
44+
{
45+
await process().ConfigureAwait(false);
46+
}
47+
catch (OperationCanceledException)
48+
{
49+
_trySetCanceledAll();
50+
}
51+
catch (Exception exception)
52+
{
53+
// A failure outside the per-item wrappers (e.g. cancellation plumbing) would otherwise
54+
// leave awaiters of the per-item tasks hanging forever.
55+
_trySetExceptionAll(exception);
56+
}
57+
}
58+
59+
public void CancelAll()
60+
{
61+
if (_disposed)
62+
return;
63+
64+
CancelAllCore();
65+
}
66+
67+
private void CancelAllCore()
68+
{
69+
if (!_cancellationTokenSource.IsCancellationRequested)
70+
{
71+
_cancellationTokenSource.Cancel();
72+
}
73+
74+
_trySetCanceledAll();
75+
}
76+
77+
public async ValueTask DisposeAsync(Func<ValueTask> disposeAsyncCore)
78+
{
79+
lock (_disposeLock)
80+
{
81+
if (_disposed)
82+
return;
83+
_disposed = true;
84+
}
85+
86+
// Allow the owning processor to dispose its resources first
87+
await disposeAsyncCore().ConfigureAwait(false);
88+
89+
CancelAllCore();
90+
_cancellationTokenRegistration.Dispose();
91+
92+
// Give in-flight tasks a bounded window to observe cancellation and finish
93+
if (_processTask is { IsCompleted: false })
94+
{
95+
try
96+
{
97+
using var timeoutCts = new CancellationTokenSource(DisposalTimeout);
98+
await _processTask.WaitAsync(timeoutCts.Token).ConfigureAwait(false);
99+
}
100+
catch (OperationCanceledException)
101+
{
102+
// Timed out waiting for in-flight tasks - continue with disposal
103+
}
104+
}
105+
106+
_cancellationTokenSource.Dispose();
107+
}
108+
109+
public void Dispose()
110+
{
111+
lock (_disposeLock)
112+
{
113+
if (_disposed)
114+
return;
115+
_disposed = true;
116+
}
117+
118+
// Cancel and release without blocking; in-flight tasks complete against
119+
// already-cancelled completion sources, which is a no-op.
120+
CancelAllCore();
121+
_cancellationTokenRegistration.Dispose();
122+
_cancellationTokenSource.Dispose();
123+
}
124+
}

‎EnumerableAsyncProcessor/RunnableProcessors/Abstract/AbstractAsyncProcessor.cs‎

Lines changed: 5 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -12,19 +12,13 @@ public abstract class AbstractAsyncProcessor : AbstractAsyncProcessorBase
1212

1313
protected AbstractAsyncProcessor(int count, Func<Task> taskSelector, CancellationTokenSource cancellationTokenSource) : base(cancellationTokenSource)
1414
{
15-
ValidationHelper.ValidateCount(count);
15+
ValidationHelper.ThrowIfNegative(count);
1616
ValidationHelper.ThrowIfNull(taskSelector);
1717

18-
var taskWrappers = new ActionTaskWrapper[count];
19-
_taskCompletionSources = new TaskCompletionSource[count];
18+
TaskWrappers = Enumerable.Range(0, count)
19+
.Select(_ => new ActionTaskWrapper(taskSelector, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)))
20+
.ToArray();
2021

21-
for (var i = 0; i < count; i++)
22-
{
23-
var taskCompletionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
24-
_taskCompletionSources[i] = taskCompletionSource;
25-
taskWrappers[i] = new ActionTaskWrapper(taskSelector, taskCompletionSource);
26-
}
27-
28-
TaskWrappers = taskWrappers;
22+
_taskCompletionSources = TaskWrappers.Select(x => x.TaskCompletionSource).ToArray();
2923
}
3024
}

‎EnumerableAsyncProcessor/RunnableProcessors/Abstract/AbstractAsyncProcessorBase.cs‎

Lines changed: 17 additions & 92 deletions
Original file line numberDiff line numberDiff line change
@@ -1,65 +1,30 @@
11
using System.Runtime.CompilerServices;
22
using EnumerableAsyncProcessor.Interfaces;
3-
using EnumerableAsyncProcessor.Validation;
43

54
namespace EnumerableAsyncProcessor.RunnableProcessors.Abstract;
65

76
public abstract class AbstractAsyncProcessorBase : IAsyncProcessor, IAsyncDisposable, IDisposable
87
{
9-
private static readonly TimeSpan DisposalTimeout = TimeSpan.FromSeconds(30);
10-
118
protected abstract IReadOnlyList<TaskCompletionSource> EnumerableTaskCompletionSources { get; }
129
protected readonly CancellationToken CancellationToken;
1310

14-
private readonly CancellationTokenSource _cancellationTokenSource;
15-
private CancellationTokenRegistration _cancellationTokenRegistration;
11+
private readonly ProcessorLifecycle _lifecycle;
1612
private Task? _overallTask;
17-
private Task? _processTask;
18-
private volatile bool _disposed;
19-
private readonly object _disposeLock = new();
2013

21-
private Task OverallTask => _overallTask ??= Task.WhenAll(EnumerableTaskCompletionSources.Select(x => x.Task));
14+
private Task OverallTask => _overallTask ??= Task.WhenAll(GetEnumerableTasks());
2215

2316
protected AbstractAsyncProcessorBase(CancellationTokenSource cancellationTokenSource)
2417
{
25-
ValidationHelper.ValidateCancellationTokenSource(cancellationTokenSource);
26-
27-
_cancellationTokenSource = cancellationTokenSource;
28-
CancellationToken = cancellationTokenSource.Token;
18+
_lifecycle = new ProcessorLifecycle(cancellationTokenSource, TrySetCanceledAll, TrySetExceptionAll);
19+
CancellationToken = _lifecycle.Token;
2920
}
3021

3122
internal abstract Task Process();
3223

33-
// Cancellation is registered here rather than in the constructor so that a token cancelled
34-
// during construction can never invoke CancelAll on a partially constructed instance.
35-
internal void Start()
24+
internal IAsyncProcessor StartProcessing()
3625
{
37-
_cancellationTokenRegistration = CancellationToken.Register(CancelAll);
38-
_processTask = RunProcess();
39-
}
40-
41-
private async Task RunProcess()
42-
{
43-
try
44-
{
45-
await Process().ConfigureAwait(false);
46-
}
47-
catch (OperationCanceledException)
48-
{
49-
foreach (var taskCompletionSource in EnumerableTaskCompletionSources)
50-
{
51-
taskCompletionSource.TrySetCanceled(CancellationToken);
52-
}
53-
}
54-
catch (Exception exception)
55-
{
56-
// A failure outside the per-item wrappers (e.g. cancellation plumbing) would otherwise
57-
// leave awaiters of the per-item tasks hanging forever.
58-
foreach (var taskCompletionSource in EnumerableTaskCompletionSources)
59-
{
60-
taskCompletionSource.TrySetException(exception);
61-
}
62-
}
26+
_lifecycle.Start(Process);
27+
return this;
6328
}
6429

6530
public IEnumerable<Task> GetEnumerableTasks()
@@ -79,56 +44,28 @@ public Task WaitAsync()
7944

8045
public void CancelAll()
8146
{
82-
if (_disposed)
83-
return;
84-
85-
CancelAllCore();
47+
_lifecycle.CancelAll();
8648
}
8749

88-
private void CancelAllCore()
50+
private void TrySetCanceledAll()
8951
{
90-
if (!_cancellationTokenSource.IsCancellationRequested)
91-
{
92-
_cancellationTokenSource.Cancel();
93-
}
94-
9552
foreach (var taskCompletionSource in EnumerableTaskCompletionSources)
9653
{
9754
taskCompletionSource.TrySetCanceled(CancellationToken);
9855
}
9956
}
10057

101-
public async ValueTask DisposeAsync()
58+
private void TrySetExceptionAll(Exception exception)
10259
{
103-
lock (_disposeLock)
104-
{
105-
if (_disposed)
106-
return;
107-
_disposed = true;
108-
}
109-
110-
// Allow derived classes to dispose their resources first
111-
await DisposeAsyncCore().ConfigureAwait(false);
112-
113-
CancelAllCore();
114-
_cancellationTokenRegistration.Dispose();
115-
116-
// Give in-flight tasks a bounded window to observe cancellation and finish
117-
if (_processTask is { IsCompleted: false })
60+
foreach (var taskCompletionSource in EnumerableTaskCompletionSources)
11861
{
119-
try
120-
{
121-
using var timeoutCts = new CancellationTokenSource(DisposalTimeout);
122-
await _processTask.WaitAsync(timeoutCts.Token).ConfigureAwait(false);
123-
}
124-
catch (OperationCanceledException)
125-
{
126-
// Timed out waiting for in-flight tasks - continue with disposal
127-
}
62+
taskCompletionSource.TrySetException(exception);
12863
}
64+
}
12965

130-
_cancellationTokenSource.Dispose();
131-
66+
public async ValueTask DisposeAsync()
67+
{
68+
await _lifecycle.DisposeAsync(DisposeAsyncCore).ConfigureAwait(false);
13269
GC.SuppressFinalize(this);
13370
}
13471

@@ -143,19 +80,7 @@ protected virtual ValueTask DisposeAsyncCore()
14380

14481
public void Dispose()
14582
{
146-
lock (_disposeLock)
147-
{
148-
if (_disposed)
149-
return;
150-
_disposed = true;
151-
}
152-
153-
// Cancel and release without blocking; in-flight tasks complete against
154-
// already-cancelled completion sources, which is a no-op.
155-
CancelAllCore();
156-
_cancellationTokenRegistration.Dispose();
157-
_cancellationTokenSource.Dispose();
158-
83+
_lifecycle.Dispose();
15984
GC.SuppressFinalize(this);
16085
}
16186
}

‎EnumerableAsyncProcessor/RunnableProcessors/BatchAsyncProcessor.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ public class BatchAsyncProcessor : AbstractAsyncProcessor
99

1010
internal BatchAsyncProcessor(int batchSize, int count, Func<Task> taskSelector, CancellationTokenSource cancellationTokenSource) : base(count, taskSelector, cancellationTokenSource)
1111
{
12-
ValidationHelper.ValidateBatchSize(batchSize);
12+
ValidationHelper.ThrowIfNegativeOrZero(batchSize);
1313

1414
_batchSize = batchSize;
1515
}

‎EnumerableAsyncProcessor/RunnableProcessors/BatchAsyncProcessor_1.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ public class BatchAsyncProcessor<TInput> : AbstractAsyncProcessor<TInput>
1010
internal BatchAsyncProcessor(int batchSize, IEnumerable<TInput> items, Func<TInput, Task> taskSelector,
1111
CancellationTokenSource cancellationTokenSource) : base(items, taskSelector, cancellationTokenSource)
1212
{
13-
ValidationHelper.ValidateBatchSize(batchSize);
13+
ValidationHelper.ThrowIfNegativeOrZero(batchSize);
1414

1515
_batchSize = batchSize;
1616
}

‎EnumerableAsyncProcessor/RunnableProcessors/RateLimitedParallelAsyncProcessor.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
using EnumerableAsyncProcessor.Extensions;
1+
using EnumerableAsyncProcessor.Extensions;
22
using EnumerableAsyncProcessor.RunnableProcessors.Abstract;
33
using EnumerableAsyncProcessor.Validation;
44

@@ -10,7 +10,7 @@ public class RateLimitedParallelAsyncProcessor : AbstractAsyncProcessor
1010

1111
internal RateLimitedParallelAsyncProcessor(int count, Func<Task> taskSelector, int levelsOfParallelism, CancellationTokenSource cancellationTokenSource) : base(count, taskSelector, cancellationTokenSource)
1212
{
13-
ValidationHelper.ValidateParallelism(levelsOfParallelism);
13+
ValidationHelper.ThrowIfNegativeOrZero(levelsOfParallelism);
1414

1515
_levelsOfParallelism = levelsOfParallelism;
1616
}

‎EnumerableAsyncProcessor/RunnableProcessors/RateLimitedParallelAsyncProcessor_1.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ public class RateLimitedParallelAsyncProcessor<TInput> : AbstractAsyncProcessor<
1010

1111
internal RateLimitedParallelAsyncProcessor(IEnumerable<TInput> items, Func<TInput, Task> taskSelector, int levelsOfParallelism, CancellationTokenSource cancellationTokenSource) : base(items, taskSelector, cancellationTokenSource)
1212
{
13-
ValidationHelper.ValidateParallelism(levelsOfParallelism);
13+
ValidationHelper.ThrowIfNegativeOrZero(levelsOfParallelism);
1414

1515
_levelsOfParallelism = levelsOfParallelism;
1616
}

0 commit comments

Comments
 (0)