Skip to content

fix: Maximize parallel throughput #312

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
thomhurst merged 2 commits into from
Aug 10, 2025
Merged

fix: Maximize parallel throughput #312

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f40ea5e
fix: Maximize parallel throughput by removing Task.Yield bottleneck
thomhurst Aug 10, 2025
2f4caa3
fix: Resolve unobserved task exceptions in AsyncEnumerable processors
thomhurst Aug 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions ...ableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ internal AsyncEnumerableParallelProcessor(
public async Task ExecuteAsync()
{
var cancellationToken = _cancellationTokenSource.Token;
var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
using var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
var tasks = new List<Task>();

try
Expand All @@ -39,8 +39,7 @@ public async Task ExecuteAsync()
{
try
{
// Yield to ensure we don't block the thread if _taskSelector is synchronous
await Task.Yield();
// Removed Task.Yield - parallelism is now handled at the processor level
await _taskSelector(capturedItem).ConfigureAwait(false);
}
finally
Expand All @@ -51,12 +50,15 @@ public async Task ExecuteAsync()

tasks.Add(task);
}

await Task.WhenAll(tasks).ConfigureAwait(false);
}
finally
{
semaphore.Dispose();
// Always wait for all tasks to complete before the using block disposes the semaphore
// This ensures the semaphore is not disposed while tasks are still running
if (tasks.Count > 0)
{
await Task.WhenAll(tasks).ConfigureAwait(false);
}
}
}
}
Expand Down
46 changes: 26 additions & 20 deletions ...ableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ internal ResultAsyncEnumerableParallelProcessor(
public async IAsyncEnumerable<TOutput> ExecuteAsync()
{
var cancellationToken = _cancellationTokenSource.Token;
var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
using var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
var tasks = new List<Task<TOutput>>();

try
Expand All @@ -36,7 +36,18 @@ public async IAsyncEnumerable<TOutput> ExecuteAsync()
await semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);

var capturedItem = item;
var task = ProcessItemAsync(capturedItem, semaphore, cancellationToken);
// Use Task.Run to ensure parallelism and prevent blocking
var task = Task.Run(async () =>
{
try
{
return await _taskSelector(capturedItem).ConfigureAwait(false);
}
finally
{
semaphore.Release();
}
}, cancellationToken);
tasks.Add(task);

// Yield completed results
Expand All @@ -56,24 +67,19 @@ public async IAsyncEnumerable<TOutput> ExecuteAsync()
}
finally
{
semaphore.Dispose();
}
}

private async Task<TOutput> ProcessItemAsync(
TInput item,
SemaphoreSlim semaphore,
CancellationToken cancellationToken)
{
try
{
// Yield to ensure we don't block the thread if _taskSelector is synchronous
await Task.Yield();
return await _taskSelector(item).ConfigureAwait(false);
}
finally
{
semaphore.Release();
// Ensure all tasks complete before the using block disposes the semaphore
// This handles cancellation or exception scenarios
if (tasks.Count > 0)
{
try
{
await Task.WhenAll(tasks).ConfigureAwait(false);
}
catch
{
// Ignore exceptions here as they've already been handled
}
}
}
}
}
Expand Down
22 changes: 10 additions & 12 deletions EnumerableAsyncProcessor/RunnableProcessors/ParallelAsyncProcessor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,33 +16,31 @@ internal override async Task Process()
// If no concurrency limit, process all tasks in parallel
if (_maxConcurrency == null)
{
await Task.WhenAll(TaskWrappers.Select(taskWrapper => taskWrapper.Process(CancellationToken))).ConfigureAwait(false);
// Use Task.Run to ensure all tasks start immediately on thread pool threads
// This prevents synchronous code in user delegates from blocking other tasks
await Task.WhenAll(TaskWrappers.Select(taskWrapper =>
Task.Run(() => taskWrapper.Process(CancellationToken), CancellationToken)
)).ConfigureAwait(false);
return;
}

// Use semaphore for concurrency throttling
using var semaphore = new SemaphoreSlim(_maxConcurrency.Value, _maxConcurrency.Value);

var tasks = TaskWrappers.Select(async taskWrapper =>
// Materialize tasks immediately to ensure they all start in parallel (up to concurrency limit)
// Use Task.Run to prevent synchronous code from blocking thread pool threads
var tasks = TaskWrappers.Select(taskWrapper => Task.Run(async () =>
{
await semaphore.WaitAsync(CancellationToken).ConfigureAwait(false);
try
{
var task = taskWrapper.Process(CancellationToken);

// Fast-path for already completed tasks
if (task.IsCompleted)
{
return;
}

await task.ConfigureAwait(false);
await taskWrapper.Process(CancellationToken).ConfigureAwait(false);
}
finally
{
semaphore.Release();
}
});
}, CancellationToken)).ToList(); // Force immediate task creation

await Task.WhenAll(tasks).ConfigureAwait(false);
}
Expand Down
26 changes: 9 additions & 17 deletions EnumerableAsyncProcessor/RunnableProcessors/ParallelAsyncProcessor_1.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,39 +16,31 @@ internal override async Task Process()
// If no concurrency limit, process all tasks in parallel
if (_maxConcurrency == null)
{
// Use Task.Run to ensure all tasks start immediately on thread pool threads
// This prevents synchronous code in user delegates from blocking other tasks
await Task.WhenAll(TaskWrappers.Select(taskWrapper =>
{
var task = taskWrapper.Process(CancellationToken);
// Fast-path for already completed tasks
if (task.IsCompleted)
{
}
return task;
})).ConfigureAwait(false);
Task.Run(() => taskWrapper.Process(CancellationToken), CancellationToken)
)).ConfigureAwait(false);
return;
}

// Use semaphore for concurrency throttling
using var semaphore = new SemaphoreSlim(_maxConcurrency.Value, _maxConcurrency.Value);

var tasks = TaskWrappers.Select(async taskWrapper =>
// Materialize tasks immediately to ensure they all start in parallel (up to concurrency limit)
// Use Task.Run to prevent synchronous code from blocking thread pool threads
var tasks = TaskWrappers.Select(taskWrapper => Task.Run(async () =>
{
await semaphore.WaitAsync(CancellationToken).ConfigureAwait(false);
try
{
var task = taskWrapper.Process(CancellationToken);
// Fast-path for already completed tasks
if (task.IsCompleted)
{
return;
}
await task.ConfigureAwait(false);
await taskWrapper.Process(CancellationToken).ConfigureAwait(false);
}
finally
{
semaphore.Release();
}
});
}, CancellationToken)).ToList(); // Force immediate task creation

await Task.WhenAll(tasks).ConfigureAwait(false);
}
Expand Down
26 changes: 9 additions & 17 deletions ...rableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultParallelAsyncProcessor_1.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,39 +16,31 @@ internal override async Task Process()
// If no concurrency limit, process all tasks in parallel
if (_maxConcurrency == null)
{
// Use Task.Run to ensure all tasks start immediately on thread pool threads
// This prevents synchronous code in user delegates from blocking other tasks
await Task.WhenAll(TaskWrappers.Select(taskWrapper =>
{
var task = taskWrapper.Process(CancellationToken);
// Fast-path for already completed tasks
if (task.IsCompleted)
{
}
return task;
})).ConfigureAwait(false);
Task.Run(() => taskWrapper.Process(CancellationToken), CancellationToken)
)).ConfigureAwait(false);
return;
}

// Use semaphore for concurrency throttling
using var semaphore = new SemaphoreSlim(_maxConcurrency.Value, _maxConcurrency.Value);

var tasks = TaskWrappers.Select(async taskWrapper =>
// Materialize tasks immediately to ensure they all start in parallel (up to concurrency limit)
// Use Task.Run to prevent synchronous code from blocking thread pool threads
var tasks = TaskWrappers.Select(taskWrapper => Task.Run(async () =>
{
await semaphore.WaitAsync(CancellationToken).ConfigureAwait(false);
try
{
var task = taskWrapper.Process(CancellationToken);
// Fast-path for already completed tasks
if (task.IsCompleted)
{
return;
}
await task.ConfigureAwait(false);
await taskWrapper.Process(CancellationToken).ConfigureAwait(false);
}
finally
{
semaphore.Release();
}
});
}, CancellationToken)).ToList(); // Force immediate task creation

await Task.WhenAll(tasks).ConfigureAwait(false);
}
Expand Down
26 changes: 9 additions & 17 deletions ...rableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultParallelAsyncProcessor_2.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,39 +16,31 @@ internal override async Task Process()
// If no concurrency limit, process all tasks in parallel
if (_maxConcurrency == null)
{
// Use Task.Run to ensure all tasks start immediately on thread pool threads
// This prevents synchronous code in user delegates from blocking other tasks
await Task.WhenAll(TaskWrappers.Select(taskWrapper =>
{
var task = taskWrapper.Process(CancellationToken);
// Fast-path for already completed tasks
if (task.IsCompleted)
{
}
return task;
})).ConfigureAwait(false);
Task.Run(() => taskWrapper.Process(CancellationToken), CancellationToken)
)).ConfigureAwait(false);
return;
}

// Use semaphore for concurrency throttling
using var semaphore = new SemaphoreSlim(_maxConcurrency.Value, _maxConcurrency.Value);

var tasks = TaskWrappers.Select(async taskWrapper =>
// Materialize tasks immediately to ensure they all start in parallel (up to concurrency limit)
// Use Task.Run to prevent synchronous code from blocking thread pool threads
var tasks = TaskWrappers.Select(taskWrapper => Task.Run(async () =>
{
await semaphore.WaitAsync(CancellationToken).ConfigureAwait(false);
try
{
var task = taskWrapper.Process(CancellationToken);
// Fast-path for already completed tasks
if (task.IsCompleted)
{
return;
}
await task.ConfigureAwait(false);
await taskWrapper.Process(CancellationToken).ConfigureAwait(false);
}
finally
{
semaphore.Release();
}
});
}, CancellationToken)).ToList(); // Force immediate task creation

await Task.WhenAll(tasks).ConfigureAwait(false);
}
Expand Down
12 changes: 4 additions & 8 deletions EnumerableAsyncProcessor/TaskWrapper.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,7 @@ public async Task Process(CancellationToken cancellationToken)

try
{
// Yield to ensure we don't block the calling thread if TaskFactory is synchronous
await Task.Yield();
// Removed Task.Yield - parallelism is now handled at the processor level
var task = TaskFactory.Invoke();

// Fast-path for already completed tasks
Expand Down Expand Up @@ -123,8 +122,7 @@ public async Task Process(CancellationToken cancellationToken)

try
{
// Yield to ensure we don't block the calling thread if TaskFactory is synchronous
await Task.Yield();
// Removed Task.Yield - parallelism is now handled at the processor level
var task = TaskFactory.Invoke(Input);

// Fast-path for already completed tasks
Expand Down Expand Up @@ -222,8 +220,7 @@ public async Task Process(CancellationToken cancellationToken)

try
{
// Yield to ensure we don't block the calling thread if TaskFactory is synchronous
await Task.Yield();
// Removed Task.Yield - parallelism is now handled at the processor level
var task = TaskFactory.Invoke(Input);

// Fast-path for already completed tasks
Expand Down Expand Up @@ -318,8 +315,7 @@ public async Task Process(CancellationToken cancellationToken)

try
{
// Yield to ensure we don't block the calling thread if TaskFactory is synchronous
await Task.Yield();
// Removed Task.Yield - parallelism is now handled at the processor level
var task = TaskFactory.Invoke();

// Fast-path for already completed tasks
Expand Down