Skip to content

Commit a303676

Browse files
thomhurstthomhurst
committed
fix: correctness issues in processor core
- Materialize task wrappers once at construction. Input enumerables were lazily re-enumerated up to 5 times (validation Any(), Process, WhenAll, CancelAll, dispose), breaking one-shot sources and re-running side effects. Removes the ConcurrentDictionary identity workaround and the eager .Any() probe of lazy enumerables. - Fix disposal never cancelling: _disposed was set before CancelAll, whose guard then made it a no-op, so dispose waited up to 30s for tasks to finish naturally. - Result processors registered sync-over-async Dispose as the cancellation callback, blocking the cancelling thread up to 30s. Both bases now register CancelAll, and only at Start() so a token cancelled during construction cannot fire on a partially constructed instance (NRE). - Route Process() faults into unfinished completion sources instead of discarding them; awaiters previously hung forever if item enumeration or cancellation plumbing threw. - Create TaskCompletionSources with RunContinuationsAsynchronously to stop user continuations running inline on the completing thread. - Preserve all task exceptions via InnerExceptions instead of GetBaseException, which dropped all but the first fault. - Sync Dispose no longer blocks via Task.Run().Wait(30s); it cancels and releases immediately. - Consistent cancellation token flow in rate-limited and timed processors (was a mix of None and missing tokens).
1 parent 445d406 commit a303676

18 files changed

Lines changed: 319 additions & 445 deletions

‎EnumerableAsyncProcessor/Extensions/AsyncProcessorExtensions.cs‎

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

@@ -8,13 +8,13 @@ internal static class AsyncProcessorExtensions
88
{
99
internal static IAsyncProcessor StartProcessing(this AbstractAsyncProcessorBase processor)
1010
{
11-
_ = processor.Process();
11+
processor.Start();
1212
return processor;
1313
}
14-
14+
1515
internal static IAsyncProcessor<T1> StartProcessing<T1>(this ResultAbstractAsyncProcessorBase<T1> processor)
1616
{
17-
_ = processor.Process();
17+
processor.Start();
1818
return processor;
1919
}
20-
}
20+
}

‎EnumerableAsyncProcessor/RunnableProcessors/Abstract/AbstractAsyncProcessor.cs‎

Lines changed: 11 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -1,41 +1,30 @@
1-
using System.Collections.Concurrent;
2-
using System.Diagnostics.CodeAnalysis;
31
using EnumerableAsyncProcessor.Validation;
42

53
namespace EnumerableAsyncProcessor.RunnableProcessors.Abstract;
64

75
public abstract class AbstractAsyncProcessor : AbstractAsyncProcessorBase
86
{
9-
private readonly ConcurrentDictionary<int, TaskCompletionSource> _taskCompletionSources = [];
7+
protected readonly IReadOnlyList<ActionTaskWrapper> TaskWrappers;
108

11-
protected readonly IEnumerable<ActionTaskWrapper> TaskWrappers;
9+
private readonly TaskCompletionSource[] _taskCompletionSources;
1210

13-
[field: AllowNull, MaybeNull]
14-
protected override IEnumerable<TaskCompletionSource> EnumerableTaskCompletionSources
15-
=> field ??= TaskWrappers.Select(x => x.TaskCompletionSource);
11+
protected override IReadOnlyList<TaskCompletionSource> EnumerableTaskCompletionSources => _taskCompletionSources;
1612

17-
1813
protected AbstractAsyncProcessor(int count, Func<Task> taskSelector, CancellationTokenSource cancellationTokenSource) : base(cancellationTokenSource)
1914
{
2015
ValidationHelper.ValidateCount(count);
2116
ValidationHelper.ThrowIfNull(taskSelector);
2217

23-
// Provide optimization for empty collections
24-
if (count == 0)
25-
{
26-
TaskWrappers = [];
27-
return;
28-
}
18+
var taskWrappers = new ActionTaskWrapper[count];
19+
_taskCompletionSources = new TaskCompletionSource[count];
2920

30-
// Provide performance warning for very large collections
31-
var warning = ValidationHelper.GetPerformanceWarning(count);
32-
if (warning != null)
21+
for (var i = 0; i < count; i++)
3322
{
34-
// In a real application, you might want to log this warning
35-
// For now, we'll just store it as a comment that could be used by logging
36-
_ = warning;
23+
var taskCompletionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
24+
_taskCompletionSources[i] = taskCompletionSource;
25+
taskWrappers[i] = new ActionTaskWrapper(taskSelector, taskCompletionSource);
3726
}
3827

39-
TaskWrappers = Enumerable.Range(0, count).Select(index => new ActionTaskWrapper(taskSelector, _taskCompletionSources.GetOrAdd(index, new TaskCompletionSource())));
28+
TaskWrappers = taskWrappers;
4029
}
41-
}
30+
}

‎EnumerableAsyncProcessor/RunnableProcessors/Abstract/AbstractAsyncProcessorBase.cs‎

Lines changed: 78 additions & 84 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,3 @@
1-
using System.Diagnostics.CodeAnalysis;
21
using System.Runtime.CompilerServices;
32
using EnumerableAsyncProcessor.Interfaces;
43
using EnumerableAsyncProcessor.Validation;
@@ -7,35 +6,65 @@ namespace EnumerableAsyncProcessor.RunnableProcessors.Abstract;
76

87
public abstract class AbstractAsyncProcessorBase : IAsyncProcessor, IAsyncDisposable, IDisposable
98
{
10-
protected abstract IEnumerable<TaskCompletionSource> EnumerableTaskCompletionSources { get; }
9+
private static readonly TimeSpan DisposalTimeout = TimeSpan.FromSeconds(30);
10+
11+
protected abstract IReadOnlyList<TaskCompletionSource> EnumerableTaskCompletionSources { get; }
1112
protected readonly CancellationToken CancellationToken;
1213

13-
[field: MaybeNull, AllowNull]
14-
private IEnumerable<Task> EnumerableTasks => field ??= EnumerableTaskCompletionSources.Select(x => x.Task);
15-
1614
private readonly CancellationTokenSource _cancellationTokenSource;
15+
private CancellationTokenRegistration _cancellationTokenRegistration;
16+
private Task? _overallTask;
17+
private Task? _processTask;
1718
private volatile bool _disposed;
1819
private readonly object _disposeLock = new();
1920

20-
[field: AllowNull, MaybeNull]
21-
private Task OverallTask => field ??= Task.WhenAll(EnumerableTasks);
22-
21+
private Task OverallTask => _overallTask ??= Task.WhenAll(EnumerableTaskCompletionSources.Select(x => x.Task));
22+
2323
protected AbstractAsyncProcessorBase(CancellationTokenSource cancellationTokenSource)
2424
{
2525
ValidationHelper.ValidateCancellationTokenSource(cancellationTokenSource);
26-
27-
CancellationToken = cancellationTokenSource.Token;
28-
CancellationToken.Register(CancelAll);
29-
CancellationToken.ThrowIfCancellationRequested();
30-
26+
3127
_cancellationTokenSource = cancellationTokenSource;
28+
CancellationToken = cancellationTokenSource.Token;
3229
}
3330

3431
internal abstract Task Process();
35-
32+
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()
36+
{
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+
}
63+
}
64+
3665
public IEnumerable<Task> GetEnumerableTasks()
3766
{
38-
return EnumerableTasks;
67+
return EnumerableTaskCompletionSources.Select(x => x.Task);
3968
}
4069

4170
public TaskAwaiter GetAwaiter()
@@ -47,28 +76,30 @@ public Task WaitAsync()
4776
{
4877
return OverallTask;
4978
}
50-
79+
5180
public void CancelAll()
5281
{
5382
if (_disposed)
5483
return;
55-
84+
85+
CancelAllCore();
86+
}
87+
88+
private void CancelAllCore()
89+
{
5690
if (!_cancellationTokenSource.IsCancellationRequested)
5791
{
5892
_cancellationTokenSource.Cancel();
5993
}
6094

61-
foreach (var tcs in EnumerableTaskCompletionSources)
95+
foreach (var taskCompletionSource in EnumerableTaskCompletionSources)
6296
{
63-
tcs.TrySetCanceled(CancellationToken);
97+
taskCompletionSource.TrySetCanceled(CancellationToken);
6498
}
6599
}
66100

67101
public async ValueTask DisposeAsync()
68102
{
69-
if (_disposed)
70-
return;
71-
72103
lock (_disposeLock)
73104
{
74105
if (_disposed)
@@ -79,60 +110,24 @@ public async ValueTask DisposeAsync()
79110
// Allow derived classes to dispose their resources first
80111
await DisposeAsyncCore().ConfigureAwait(false);
81112

82-
// Cancel all operations
83-
CancelAll();
113+
CancelAllCore();
114+
_cancellationTokenRegistration.Dispose();
84115

85-
// Wait for all running tasks to complete with timeout
86-
try
116+
// Give in-flight tasks a bounded window to observe cancellation and finish
117+
if (_processTask is { IsCompleted: false })
87118
{
88-
using var timeoutCts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
89-
var allTasks = EnumerableTasks.ToList();
90-
91-
if (allTasks.Count > 0)
119+
try
92120
{
93-
var completionTasks = allTasks.Select(async task =>
94-
{
95-
try
96-
{
97-
await task.WaitAsync(timeoutCts.Token).ConfigureAwait(false);
98-
}
99-
catch (OperationCanceledException)
100-
{
101-
// Expected when cancelled - ignore
102-
}
103-
catch (Exception)
104-
{
105-
// Task exceptions are expected - ignore during disposal
106-
}
107-
}).ToList();
108-
109-
if (completionTasks.Count > 0)
110-
{
111-
try
112-
{
113-
await Task.WhenAll(completionTasks).ConfigureAwait(false);
114-
}
115-
catch (OperationCanceledException)
116-
{
117-
// Timeout occurred - continue with disposal
118-
}
119-
}
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
120127
}
121-
}
122-
catch (Exception)
123-
{
124-
// Swallow exceptions during disposal cleanup
125128
}
126129

127-
// Dispose the cancellation token source
128-
try
129-
{
130-
_cancellationTokenSource.Dispose();
131-
}
132-
catch (Exception)
133-
{
134-
// Swallow disposal exceptions
135-
}
130+
_cancellationTokenSource.Dispose();
136131

137132
GC.SuppressFinalize(this);
138133
}
@@ -148,20 +143,19 @@ protected virtual ValueTask DisposeAsyncCore()
148143

149144
public void Dispose()
150145
{
151-
// Use Task.Run to avoid deadlocks by running async disposal on thread pool
152-
// Add timeout to prevent indefinite blocking
153-
try
154-
{
155-
var disposeTask = Task.Run(async () => await DisposeAsync().ConfigureAwait(false));
156-
if (!disposeTask.Wait(TimeSpan.FromSeconds(30)))
157-
{
158-
// Log warning if disposal times out, but don't throw
159-
// as per IDisposable pattern
160-
}
161-
}
162-
catch
146+
lock (_disposeLock)
163147
{
164-
// Suppress exceptions during disposal as per IDisposable pattern
148+
if (_disposed)
149+
return;
150+
_disposed = true;
165151
}
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+
159+
GC.SuppressFinalize(this);
166160
}
167-
}
161+
}

‎EnumerableAsyncProcessor/RunnableProcessors/Abstract/AbstractAsyncProcessor_1.cs‎

Lines changed: 12 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -1,43 +1,25 @@
1-
using System.Collections.Concurrent;
2-
using System.Diagnostics.CodeAnalysis;
31
using EnumerableAsyncProcessor.Validation;
42

53
namespace EnumerableAsyncProcessor.RunnableProcessors.Abstract;
64

75
public abstract class AbstractAsyncProcessor<TInput> : AbstractAsyncProcessorBase
86
{
9-
private readonly ConcurrentDictionary<int, TaskCompletionSource> _taskCompletionSources = [];
10-
11-
protected readonly IEnumerable<ItemTaskWrapper<TInput>> TaskWrappers;
7+
protected readonly IReadOnlyList<ItemTaskWrapper<TInput>> TaskWrappers;
8+
9+
private readonly TaskCompletionSource[] _taskCompletionSources;
10+
11+
protected override IReadOnlyList<TaskCompletionSource> EnumerableTaskCompletionSources => _taskCompletionSources;
1212

13-
[field: AllowNull, MaybeNull]
14-
protected override IEnumerable<TaskCompletionSource> EnumerableTaskCompletionSources
15-
=> field ??= TaskWrappers.Select(x => x.TaskCompletionSource);
16-
1713
protected AbstractAsyncProcessor(IEnumerable<TInput> items, Func<TInput, Task> taskSelector, CancellationTokenSource cancellationTokenSource) : base(cancellationTokenSource)
1814
{
19-
var isEmpty = ValidationHelper.ValidateEnumerable(items);
15+
ValidationHelper.ThrowIfNull(items);
2016
ValidationHelper.ThrowIfNull(taskSelector);
2117

22-
// Provide optimization for empty collections
23-
if (isEmpty)
24-
{
25-
TaskWrappers = [];
26-
return;
27-
}
28-
29-
// Get count for performance warnings if collection implements ICollection
30-
if (items is ICollection<TInput> collection)
31-
{
32-
var warning = ValidationHelper.GetPerformanceWarning(collection.Count);
33-
if (warning != null)
34-
{
35-
// In a real application, you might want to log this warning
36-
// For now, we'll just store it as a comment that could be used by logging
37-
_ = warning;
38-
}
39-
}
18+
// Materialize once so one-shot or side-effecting enumerables are only enumerated a single time
19+
TaskWrappers = items
20+
.Select(item => new ItemTaskWrapper<TInput>(item, taskSelector, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)))
21+
.ToArray();
4022

41-
TaskWrappers = items.Select((item, index) => new ItemTaskWrapper<TInput>(item, taskSelector, _taskCompletionSources.GetOrAdd(index, new TaskCompletionSource())));
23+
_taskCompletionSources = TaskWrappers.Select(x => x.TaskCompletionSource).ToArray();
4224
}
43-
}
25+
}

‎EnumerableAsyncProcessor/RunnableProcessors/RateLimitedParallelAsyncProcessor.cs‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,9 @@ internal RateLimitedParallelAsyncProcessor(int count, Func<Task> taskSelector, i
1717

1818
internal override Task Process()
1919
{
20-
// For rate-limited processing, we want strict parallelism control
21-
return TaskWrappers.InParallelAsync(_levelsOfParallelism,
22-
async taskWrapper =>
23-
{
24-
await Task.Run(() => taskWrapper.Process(CancellationToken), CancellationToken).ConfigureAwait(false);
25-
}, CancellationToken);
20+
// Task.Run guards the shared worker slots against synchronous code in user delegates
21+
return TaskWrappers.InParallelAsync(_levelsOfParallelism,
22+
taskWrapper => Task.Run(() => taskWrapper.Process(CancellationToken), CancellationToken),
23+
CancellationToken);
2624
}
2725
}

0 commit comments

Comments
 (0)