Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
11 changes: 11 additions & 0 deletions EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder.cs
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,17 @@ public IAsyncProcessor ProcessOneAtATime()
{
return new OneAtATimeAsyncProcessor(_count, _taskSelector, _cancellationTokenSource).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()
{
return new UnboundedParallelAsyncProcessor(_count, _taskSelector, _cancellationTokenSource).StartProcessing();
}

#if NET6_0_OR_GREATER
/// <summary>
Expand Down
11 changes: 11 additions & 0 deletions EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder_1.cs
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,17 @@ public IAsyncProcessor<TOutput> ProcessOneAtATime()
{
return new ResultOneAtATimeAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource).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()
{
return new ResultUnboundedParallelAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource).StartProcessing();
}

#if NET6_0_OR_GREATER
/// <summary>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,18 @@ public IAsyncProcessor ProcessOneAtATime()
return new OneAtATimeAsyncProcessor<TInput>(_items, _taskSelector, _cancellationTokenSource)
.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()
{
return new UnboundedParallelAsyncProcessor<TInput>(_items, _taskSelector, _cancellationTokenSource)
.StartProcessing();
}

#if NET6_0_OR_GREATER
/// <summary>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,17 @@ public IAsyncProcessor<TOutput> ProcessOneAtATime()
{
return new ResultOneAtATimeAsyncProcessor<TInput, TOutput>(_items, _taskSelector, _cancellationTokenSource).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()
{
return new ResultUnboundedParallelAsyncProcessor<TInput, TOutput>(_items, _taskSelector, _cancellationTokenSource).StartProcessing();
}

#if NET6_0_OR_GREATER
/// <summary>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,9 @@ public class IOBoundParallelAsyncProcessor : AbstractAsyncProcessor

internal IOBoundParallelAsyncProcessor(int count, Func<Task> taskSelector, CancellationTokenSource cancellationTokenSource, int? maxConcurrency = null) : base(count, taskSelector, cancellationTokenSource)
{
// For I/O operations, allow much higher concurrency - default to 10x processor count or minimum 100
_maxConcurrency = maxConcurrency ?? Math.Max(100, Environment.ProcessorCount * 10);
// For I/O operations, allow much higher concurrency - default to 64x processor count or minimum 1000
// This enables scenarios like running 10,000 unit tests in parallel
_maxConcurrency = maxConcurrency ?? Math.Max(1000, Environment.ProcessorCount * 64);
}

internal override async Task Process()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,9 @@ public class IOBoundParallelAsyncProcessor<TInput> : AbstractAsyncProcessor<TInp

internal IOBoundParallelAsyncProcessor(IEnumerable<TInput> items, Func<TInput, Task> taskSelector, CancellationTokenSource cancellationTokenSource, int? maxConcurrency = null) : base(items, taskSelector, cancellationTokenSource)
{
// For I/O operations, allow much higher concurrency - default to 10x processor count or minimum 100
_maxConcurrency = maxConcurrency ?? Math.Max(100, Environment.ProcessorCount * 10);
// For I/O operations, allow much higher concurrency - default to 64x processor count or minimum 1000
// This enables scenarios like running 10,000 unit tests in parallel
_maxConcurrency = maxConcurrency ?? Math.Max(1000, Environment.ProcessorCount * 64);
}

internal override async Task Process()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,9 @@ public class ResultIOBoundParallelAsyncProcessor<TOutput> : ResultAbstractAsyncP

internal ResultIOBoundParallelAsyncProcessor(int count, Func<Task<TOutput>> taskSelector, CancellationTokenSource cancellationTokenSource, int? maxConcurrency = null) : base(count, taskSelector, cancellationTokenSource)
{
// For I/O operations, allow much higher concurrency - default to 10x processor count or minimum 100
_maxConcurrency = maxConcurrency ?? Math.Max(100, Environment.ProcessorCount * 10);
// For I/O operations, allow much higher concurrency - default to 64x processor count or minimum 1000
// This enables scenarios like running 10,000 unit tests in parallel
_maxConcurrency = maxConcurrency ?? Math.Max(1000, Environment.ProcessorCount * 64);
}

internal override async Task Process()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,9 @@ public class ResultIOBoundParallelAsyncProcessor<TInput, TOutput> : ResultAbstra

internal ResultIOBoundParallelAsyncProcessor(IEnumerable<TInput> items, Func<TInput, Task<TOutput>> taskSelector, CancellationTokenSource cancellationTokenSource, int? maxConcurrency = null) : base(items, taskSelector, cancellationTokenSource)
{
// For I/O operations, allow much higher concurrency - default to 10x processor count or minimum 100
_maxConcurrency = maxConcurrency ?? Math.Max(100, Environment.ProcessorCount * 10);
// For I/O operations, allow much higher concurrency - default to 64x processor count or minimum 1000
// This enables scenarios like running 10,000 unit tests in parallel
_maxConcurrency = maxConcurrency ?? Math.Max(1000, Environment.ProcessorCount * 64);
}

internal override async Task Process()
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
using EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors.Abstract;

namespace EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors;

/// <summary>
/// A specialized parallel processor that starts ALL tasks immediately without any concurrency limits and returns results.
/// WARNING: Use with caution - this can overwhelm system resources with large task counts.
/// Ideal for scenarios where you need maximum parallelism and have sufficient resources.
/// </summary>
public class ResultUnboundedParallelAsyncProcessor<TOutput> : ResultAbstractAsyncProcessor<TOutput>
{
internal ResultUnboundedParallelAsyncProcessor(int count, Func<Task<TOutput>> taskSelector, CancellationTokenSource cancellationTokenSource) : base(count, taskSelector, cancellationTokenSource)
{
}

internal override Task Process()
{
// Start ALL tasks immediately without any throttling
// This provides true unbounded parallelism
var tasks = TaskWrappers.Select(taskWrapper =>
{
var task = taskWrapper.Process(CancellationToken);
// Fast-path for already completed tasks
if (task.IsCompleted)
{
return task;
}
return task;
});

return Task.WhenAll(tasks);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
using EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors.Abstract;

namespace EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors;

/// <summary>
/// A specialized parallel processor that starts ALL tasks immediately without any concurrency limits and returns results.
/// WARNING: Use with caution - this can overwhelm system resources with large task counts.
/// Ideal for scenarios where you need maximum parallelism and have sufficient resources.
/// </summary>
public class ResultUnboundedParallelAsyncProcessor<TInput, TOutput> : ResultAbstractAsyncProcessor<TInput, TOutput>
{
internal ResultUnboundedParallelAsyncProcessor(IEnumerable<TInput> items, Func<TInput, Task<TOutput>> taskSelector, CancellationTokenSource cancellationTokenSource) : base(items, taskSelector, cancellationTokenSource)
{
}

internal override Task Process()
{
// Start ALL tasks immediately without any throttling
// This provides true unbounded parallelism
var tasks = TaskWrappers.Select(taskWrapper =>
{
var task = taskWrapper.Process(CancellationToken);
// Fast-path for already completed tasks
if (task.IsCompleted)
{
return task;
}
return task;
});

return Task.WhenAll(tasks);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
using EnumerableAsyncProcessor.RunnableProcessors.Abstract;

namespace EnumerableAsyncProcessor.RunnableProcessors;

/// <summary>
/// A specialized parallel processor that starts ALL tasks immediately without any concurrency limits.
/// WARNING: Use with caution - this can overwhelm system resources with large task counts.
/// Ideal for scenarios where you need maximum parallelism and have sufficient resources.
/// </summary>
public class UnboundedParallelAsyncProcessor : AbstractAsyncProcessor
{
internal UnboundedParallelAsyncProcessor(int count, Func<Task> taskSelector, CancellationTokenSource cancellationTokenSource) : base(count, taskSelector, cancellationTokenSource)
{
}

internal override Task Process()
{
// Start ALL tasks immediately without any throttling
// This provides true unbounded parallelism
var tasks = TaskWrappers.Select(taskWrapper =>
{
var task = taskWrapper.Process(CancellationToken);
// Fast-path for already completed tasks
if (task.IsCompleted)
{
return task;
}
return task;
});

return Task.WhenAll(tasks);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
using EnumerableAsyncProcessor.RunnableProcessors.Abstract;

namespace EnumerableAsyncProcessor.RunnableProcessors;

/// <summary>
/// A specialized parallel processor that starts ALL tasks immediately without any concurrency limits.
/// WARNING: Use with caution - this can overwhelm system resources with large task counts.
/// Ideal for scenarios where you need maximum parallelism and have sufficient resources.
/// </summary>
public class UnboundedParallelAsyncProcessor<TInput> : AbstractAsyncProcessor<TInput>
{
internal UnboundedParallelAsyncProcessor(IEnumerable<TInput> items, Func<TInput, Task> taskSelector, CancellationTokenSource cancellationTokenSource) : base(items, taskSelector, cancellationTokenSource)
{
}

internal override Task Process()
{
// Start ALL tasks immediately without any throttling
// This provides true unbounded parallelism
var tasks = TaskWrappers.Select(taskWrapper =>
{
var task = taskWrapper.Process(CancellationToken);
// Fast-path for already completed tasks
if (task.IsCompleted)
{
return task;
}
return task;
});

return Task.WhenAll(tasks);
}
}