Skip to content

feat: Optimize parallel processing with optional thread pool scheduling #316

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 1 commit into from
Aug 11, 2025
Merged

feat: Optimize parallel processing with optional thread pool scheduling #316

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
62 changes: 0 additions & 62 deletions EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,68 +195,6 @@ public async Task SelectAsync_WithEmptyAsyncEnumerable_ReturnsEmptyResults()
await Assert.That(results).IsEmpty();
}

[Test]
public async Task ForEachAsync_ProcessInParallelUnbounded_ProcessesAllItems()
{
var processedItems = new List<int>();
var maxConcurrency = 0;
var currentConcurrency = 0;
var asyncEnumerable = GenerateAsyncEnumerable(50);

await asyncEnumerable
.ForEachAsync(async item =>
{
var current = Interlocked.Increment(ref currentConcurrency);
lock (processedItems)
{
if (current > maxConcurrency)
maxConcurrency = current;
}

await Task.Delay(10);
lock (processedItems)
{
processedItems.Add(item);
}
Interlocked.Decrement(ref currentConcurrency);
})
.ProcessInParallelUnbounded()
.ExecuteAsync();

await Assert.That(processedItems.Count).IsEqualTo(50);
await Assert.That(processedItems.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 50));
// Should have high concurrency since it's unbounded (at least 2)
await Assert.That(maxConcurrency).IsGreaterThan(2);
}

[Test]
public async Task SelectAsync_ProcessInParallelUnbounded_ReturnsAllResults()
{
var asyncEnumerable = GenerateAsyncEnumerable(30);
var maxConcurrency = 0;
var currentConcurrency = 0;

var results = await asyncEnumerable
.SelectAsync(async item =>
{
var current = Interlocked.Increment(ref currentConcurrency);
if (current > maxConcurrency)
maxConcurrency = current;

await Task.Delay(10);
Interlocked.Decrement(ref currentConcurrency);
return item * 3;
})
.ProcessInParallelUnbounded()
.ExecuteAsync()
.ToListAsync();

await Assert.That(results.Count).IsEqualTo(30);
await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 30).Select(x => x * 3));
// Should have high concurrency since it's unbounded (at least 2)
await Assert.That(maxConcurrency).IsGreaterThan(2);
}

[Test]
public async Task ForEachAsync_WithException_PropagatesException()
{
Expand Down
2 changes: 1 addition & 1 deletion EnumerableAsyncProcessor.UnitTests/ParallelPerformanceTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ public async Task MeasureParallelProcessingTime_200ItemsWithDirectThreadSleep()
Thread.Sleep(sleepMilliseconds);
return Task.CompletedTask;
})
.ProcessInParallel();
.ProcessInParallel(scheduleOnThreadPool: true);

await processor;

Expand Down
34 changes: 22 additions & 12 deletions EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,17 @@ public IAsyncProcessor ProcessInParallel(int levelOfParallelism, TimeSpan timeSp
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel()
{
return ProcessInParallel(null);
return ProcessInParallel(null, false);
}

/// <summary>
/// Process tasks in parallel without concurrency limits.
/// </summary>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel(bool scheduleOnThreadPool)
{
return ProcessInParallel(null, scheduleOnThreadPool);
}

/// <summary>
Expand All @@ -48,23 +58,23 @@ public IAsyncProcessor ProcessInParallel()
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel(int? maxConcurrency)
{
return new ParallelAsyncProcessor(_count, _taskSelector, _cancellationTokenSource, maxConcurrency).StartProcessing();
return ProcessInParallel(maxConcurrency, false);
}

public IAsyncProcessor ProcessOneAtATime()
/// <summary>
/// Process tasks in parallel with specified concurrency limit.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
{
return new OneAtATimeAsyncProcessor(_count, _taskSelector, _cancellationTokenSource).StartProcessing();
return new ParallelAsyncProcessor(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
}

/// <summary>
/// Process ALL tasks in parallel without any concurrency limits.
/// WARNING: Use with caution - can overwhelm system resources with large task counts.
/// Ideal for scenarios requiring maximum parallelism like running thousands of unit tests.
/// </summary>
/// <returns>An async processor with unbounded parallelism.</returns>
public IAsyncProcessor ProcessInParallelUnbounded()
public IAsyncProcessor ProcessOneAtATime()
{
return new UnboundedParallelAsyncProcessor(_count, _taskSelector, _cancellationTokenSource).StartProcessing();
return new OneAtATimeAsyncProcessor(_count, _taskSelector, _cancellationTokenSource).StartProcessing();
}

}
34 changes: 22 additions & 12 deletions EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder_1.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,17 @@ public IAsyncProcessor<TOutput> ProcessInParallel(int levelOfParallelism, TimeSp
/// <returns>An async processor configured for parallel execution that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallel()
{
return ProcessInParallel(null);
return ProcessInParallel(null, false);
}

/// <summary>
/// Process tasks in parallel without concurrency limits and return results.
/// </summary>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
/// <returns>An async processor configured for parallel execution that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallel(bool scheduleOnThreadPool)
{
return ProcessInParallel(null, scheduleOnThreadPool);
}

/// <summary>
Expand All @@ -48,23 +58,23 @@ public IAsyncProcessor<TOutput> ProcessInParallel()
/// <returns>An async processor configured for parallel execution that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency)
{
return new ResultParallelAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource, maxConcurrency).StartProcessing();
return ProcessInParallel(maxConcurrency, false);
}

public IAsyncProcessor<TOutput> ProcessOneAtATime()
/// <summary>
/// Process tasks in parallel with specified concurrency limit and return results.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
/// <returns>An async processor configured for parallel execution that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
{
return new ResultOneAtATimeAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource).StartProcessing();
return new ResultParallelAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
}

/// <summary>
/// Process ALL tasks in parallel without any concurrency limits and return results.
/// WARNING: Use with caution - can overwhelm system resources with large task counts.
/// Ideal for scenarios requiring maximum parallelism like running thousands of unit tests.
/// </summary>
/// <returns>An async processor with unbounded parallelism that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallelUnbounded()
public IAsyncProcessor<TOutput> ProcessOneAtATime()
{
return new ResultUnboundedParallelAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource).StartProcessing();
return new ResultOneAtATimeAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource).StartProcessing();
}

}
10 changes: 0 additions & 10 deletions EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,15 +47,5 @@ public IAsyncEnumerableProcessor ProcessOneAtATime()
_items, _taskSelector, _cancellationTokenSource);
}

/// <summary>
/// Process ALL items in parallel without any concurrency limits.
/// WARNING: Use with caution - can overwhelm system resources with large async enumerables.
/// </summary>
public IAsyncEnumerableProcessor ProcessInParallelUnbounded()
{
return new AsyncEnumerableUnboundedParallelProcessor<TInput>(
_items, _taskSelector, _cancellationTokenSource);
}

}
#endif
10 changes: 0 additions & 10 deletions EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,15 +47,5 @@ public IAsyncEnumerableProcessor<TOutput> ProcessOneAtATime()
_items, _taskSelector, _cancellationTokenSource);
}

/// <summary>
/// Process ALL items in parallel without any concurrency limits and return results.
/// WARNING: Use with caution - can overwhelm system resources with large async enumerables.
/// </summary>
public IAsyncEnumerableProcessor<TOutput> ProcessInParallelUnbounded()
{
return new ResultAsyncEnumerableUnboundedParallelProcessor<TInput, TOutput>(
_items, _taskSelector, _cancellationTokenSource);
}

}
#endif
35 changes: 22 additions & 13 deletions EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_1.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,17 @@ public IAsyncProcessor ProcessInParallel(int levelOfParallelism, TimeSpan timeSp
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel()
{
return ProcessInParallel(null);
return ProcessInParallel(null, false);
}

/// <summary>
/// Process items in parallel without concurrency limits.
/// </summary>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel(bool scheduleOnThreadPool)
{
return ProcessInParallel(null, scheduleOnThreadPool);
}

/// <summary>
Expand All @@ -51,25 +61,24 @@ public IAsyncProcessor ProcessInParallel()
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel(int? maxConcurrency)
{
return new ParallelAsyncProcessor<TInput>(_items, _taskSelector, _cancellationTokenSource, maxConcurrency)
.StartProcessing();
return ProcessInParallel(maxConcurrency, false);
}

public IAsyncProcessor ProcessOneAtATime()
/// <summary>
/// Process items in parallel with specified concurrency limit.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
{
return new OneAtATimeAsyncProcessor<TInput>(_items, _taskSelector, _cancellationTokenSource)
return new ParallelAsyncProcessor<TInput>(_items, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool)
.StartProcessing();
}

/// <summary>
/// Process ALL items in parallel without any concurrency limits.
/// WARNING: Use with caution - can overwhelm system resources with large item counts.
/// Ideal for scenarios requiring maximum parallelism like running thousands of unit tests.
/// </summary>
/// <returns>An async processor with unbounded parallelism.</returns>
public IAsyncProcessor ProcessInParallelUnbounded()
public IAsyncProcessor ProcessOneAtATime()
{
return new UnboundedParallelAsyncProcessor<TInput>(_items, _taskSelector, _cancellationTokenSource)
return new OneAtATimeAsyncProcessor<TInput>(_items, _taskSelector, _cancellationTokenSource)
.StartProcessing();
}

Expand Down
34 changes: 22 additions & 12 deletions EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_2.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,17 @@ public IAsyncProcessor<TOutput> ProcessInParallel(int levelOfParallelism, TimeSp
/// <returns>An async processor configured for parallel execution that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallel()
{
return ProcessInParallel(null);
return ProcessInParallel(null, false);
}

/// <summary>
/// Process items in parallel without concurrency limits and return results.
/// </summary>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
/// <returns>An async processor configured for parallel execution that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallel(bool scheduleOnThreadPool)
{
return ProcessInParallel(null, scheduleOnThreadPool);
}

/// <summary>
Expand All @@ -48,23 +58,23 @@ public IAsyncProcessor<TOutput> ProcessInParallel()
/// <returns>An async processor configured for parallel execution that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency)
{
return new ResultParallelAsyncProcessor<TInput, TOutput>(_items, _taskSelector, _cancellationTokenSource, maxConcurrency).StartProcessing();
return ProcessInParallel(maxConcurrency, false);
}

public IAsyncProcessor<TOutput> ProcessOneAtATime()
/// <summary>
/// Process items in parallel with specified concurrency limit and return results.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
/// <returns>An async processor configured for parallel execution that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
{
return new ResultOneAtATimeAsyncProcessor<TInput, TOutput>(_items, _taskSelector, _cancellationTokenSource).StartProcessing();
return new ResultParallelAsyncProcessor<TInput, TOutput>(_items, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
}

/// <summary>
/// Process ALL items in parallel without any concurrency limits and return results.
/// WARNING: Use with caution - can overwhelm system resources with large item counts.
/// Ideal for scenarios requiring maximum parallelism like running thousands of unit tests.
/// </summary>
/// <returns>An async processor with unbounded parallelism that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallelUnbounded()
public IAsyncProcessor<TOutput> ProcessOneAtATime()
{
return new ResultUnboundedParallelAsyncProcessor<TInput, TOutput>(_items, _taskSelector, _cancellationTokenSource).StartProcessing();
return new ResultOneAtATimeAsyncProcessor<TInput, TOutput>(_items, _taskSelector, _cancellationTokenSource).StartProcessing();
}

}
54 changes: 0 additions & 54 deletions ...Processor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableUnboundedParallelProcessor.cs
View file
Open in desktop

This file was deleted.

Loading
Loading