Skip to content

Commit 6d672d2

Browse files
thomhurstthomhurst
authored
Merge pull request #307 from thomhurst/fix/deadlock-issues
fix: Fix deadlock issues in async processors
2 parents 4291d89 + 5d1547b commit 6d672d2

5 files changed

Lines changed: 112 additions & 27 deletions

File tree

‎.claude/settings.local.json‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,16 @@
1212
"Bash(timeout 30 dotnet run:*)",
1313
"Bash(.EnumerableAsyncProcessor.Example.exe)",
1414
"Bash(EnumerableAsyncProcessor.Example.exe)",
15-
"Bash(del \"EnumerableAsyncProcessor.UnitTests\\ValidationTests.cs\")"
15+
"Bash(del \"EnumerableAsyncProcessor.UnitTests\\ValidationTests.cs\")",
16+
"Bash(mkdir:*)",
17+
"Bash(git add:*)",
18+
"Bash(dotnet build)",
19+
"Bash(dotnet build:*)",
20+
"Bash(dotnet test)",
21+
"Bash(dotnet test:*)",
22+
"Bash(rg:*)",
23+
"Bash(find:*)",
24+
"Bash(timeout 30 dotnet test --no-build --verbosity minimal)"
1625
],
1726
"deny": []
1827
}

‎EnumerableAsyncProcessor/RunnableProcessors/Abstract/AbstractAsyncProcessorBase.cs‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -148,12 +148,16 @@ protected virtual ValueTask DisposeAsyncCore()
148148

149149
public void Dispose()
150150
{
151-
// Use async disposal with ConfigureAwait(false) to avoid deadlocks
152-
// and add a timeout to prevent indefinite blocking
151+
// Use Task.Run to avoid deadlocks by running async disposal on thread pool
152+
// Add timeout to prevent indefinite blocking
153153
try
154154
{
155-
var disposeTask = DisposeAsync().ConfigureAwait(false);
156-
disposeTask.GetAwaiter().GetResult();
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+
}
157161
}
158162
catch
159163
{

‎EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,10 +37,17 @@ public async Task ExecuteAsync()
3737
{
3838
await channel.Writer.WriteAsync(item, cancellationToken).ConfigureAwait(false);
3939
}
40+
channel.Writer.Complete();
4041
}
41-
finally
42+
catch (OperationCanceledException)
4243
{
43-
channel.Writer.Complete();
44+
channel.Writer.TryComplete();
45+
throw;
46+
}
47+
catch (Exception ex)
48+
{
49+
channel.Writer.TryComplete(ex);
50+
throw;
4451
}
4552
}, cancellationToken);
4653

‎EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableChannelBasedProcessor.cs‎

Lines changed: 77 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -62,9 +62,32 @@ private async IAsyncEnumerable<TOutput> ExecuteWithoutOrderPreservationAsync(Can
6262
// Complete output when all processing is done
6363
var completionTask = Task.Run(async () =>
6464
{
65-
await producerTask.ConfigureAwait(false);
66-
await Task.WhenAll(consumerTasks).ConfigureAwait(false);
67-
outputChannel.Writer.Complete();
65+
Exception? exception = null;
66+
try
67+
{
68+
await producerTask.ConfigureAwait(false);
69+
await Task.WhenAll(consumerTasks).ConfigureAwait(false);
70+
}
71+
catch (Exception ex)
72+
{
73+
exception = ex;
74+
}
75+
finally
76+
{
77+
if (exception != null)
78+
{
79+
outputChannel.Writer.TryComplete(exception);
80+
}
81+
else
82+
{
83+
outputChannel.Writer.TryComplete();
84+
}
85+
}
86+
87+
if (exception != null)
88+
{
89+
throw exception;
90+
}
6891
}, cancellationToken);
6992

7093
// Yield results as they complete
@@ -138,7 +161,36 @@ private async IAsyncEnumerable<TOutput> ExecuteWithOrderPreservationAsync(Cancel
138161
await Task.Delay(10, cancellationToken).ConfigureAwait(false);
139162
}
140163
}
141-
outputChannel.Writer.Complete();
164+
}, cancellationToken);
165+
166+
// Ensure channel completion when all tasks finish
167+
var channelCompletionTask = Task.Run(async () =>
168+
{
169+
Exception? exception = null;
170+
try
171+
{
172+
await yieldingTask.ConfigureAwait(false);
173+
}
174+
catch (Exception ex)
175+
{
176+
exception = ex;
177+
}
178+
finally
179+
{
180+
if (exception != null)
181+
{
182+
outputChannel.Writer.TryComplete(exception);
183+
}
184+
else
185+
{
186+
outputChannel.Writer.TryComplete();
187+
}
188+
}
189+
190+
if (exception != null)
191+
{
192+
throw exception;
193+
}
142194
}, cancellationToken);
143195

144196
// Yield from output channel
@@ -160,21 +212,30 @@ private async Task ProcessOrderedItemAsync(
160212
SemaphoreSlim semaphore,
161213
CancellationToken cancellationToken)
162214
{
215+
var tcs = new TaskCompletionSource<TOutput>();
216+
217+
await orderingLock.WaitAsync(cancellationToken).ConfigureAwait(false);
218+
try
219+
{
220+
orderingDictionary[index] = tcs;
221+
}
222+
finally
223+
{
224+
orderingLock.Release();
225+
}
226+
163227
try
164228
{
165229
var result = await _taskSelector(item).ConfigureAwait(false);
166-
167-
await orderingLock.WaitAsync(cancellationToken).ConfigureAwait(false);
168-
try
169-
{
170-
var tcs = new TaskCompletionSource<TOutput>();
171-
tcs.SetResult(result);
172-
orderingDictionary[index] = tcs;
173-
}
174-
finally
175-
{
176-
orderingLock.Release();
177-
}
230+
tcs.TrySetResult(result);
231+
}
232+
catch (OperationCanceledException)
233+
{
234+
tcs.TrySetCanceled();
235+
}
236+
catch (Exception ex)
237+
{
238+
tcs.TrySetException(ex);
178239
}
179240
finally
180241
{

‎EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/Abstract/ResultAbstractAsyncProcessorBase.cs‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -155,12 +155,16 @@ protected virtual ValueTask DisposeAsyncCore()
155155

156156
public void Dispose()
157157
{
158-
// Use async disposal with ConfigureAwait(false) to avoid deadlocks
159-
// and add a timeout to prevent indefinite blocking
158+
// Use Task.Run to avoid deadlocks by running async disposal on thread pool
159+
// Add timeout to prevent indefinite blocking
160160
try
161161
{
162-
var disposeTask = DisposeAsync().ConfigureAwait(false);
163-
disposeTask.GetAwaiter().GetResult();
162+
var disposeTask = Task.Run(async () => await DisposeAsync().ConfigureAwait(false));
163+
if (!disposeTask.Wait(TimeSpan.FromSeconds(30)))
164+
{
165+
// Log warning if disposal times out, but don't throw
166+
// as per IDisposable pattern
167+
}
164168
}
165169
catch
166170
{

0 commit comments

Comments
 (0)