Skip to content

Commit 88d2391

Browse files
thomhurstthomhurst
committed
refactor: Remove isIOBound logic and all Channel usage
- Removed all isIOBound parameters and logic from ParallelExtensions - Removed System.Threading.Channels package reference - Refactored AsyncEnumerable processors to use semaphores instead of Channels - Removed ChannelItemTaskWrapper structs from TaskWrapper.cs - Simplified parallel processing to always use Task.Run for consistency This simplifies the codebase by removing the I/O vs CPU-bound distinction and eliminates the dependency on System.Threading.Channels, using standard semaphore-based concurrency control instead.
1 parent f59a247 commit 88d2391

14 files changed

Lines changed: 94 additions & 415 deletions

‎EnumerableAsyncProcessor/EnumerableAsyncProcessor.csproj‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,11 +23,6 @@
2323

2424
<ItemGroup Condition="'$(TargetFrameworkIdentifier)'!='.NETCoreApp'">
2525
<PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="6.0.0" />
26-
<PackageReference Include="System.Threading.Channels" Version="6.0.0" />
27-
</ItemGroup>
28-
29-
<ItemGroup Condition="'$(TargetFrameworkIdentifier)'=='.NETCoreApp' AND '$(TargetFramework)'=='net6.0'">
30-
<PackageReference Include="System.Threading.Channels" Version="6.0.0" />
3126
</ItemGroup>
3227

3328
<ItemGroup>

‎EnumerableAsyncProcessor/Extensions/ParallelExtensions.cs‎

Lines changed: 14 additions & 178 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,4 @@
1-
#if NET6_0_OR_GREATER
2-
using System.Threading.Channels;
3-
#endif
4-
5-
namespace EnumerableAsyncProcessor.Extensions;
1+
namespace EnumerableAsyncProcessor.Extensions;
62

73
public static class ParallelExtensions
84
{
@@ -11,87 +7,47 @@ public static async Task InParallelAsync<TSource, TResult>(
117
int levelOfParallelism,
128
Func<TSource, Task<TResult>> taskSelector)
139
{
14-
await InParallelAsync(source, levelOfParallelism, taskSelector, CancellationToken.None, true).ConfigureAwait(false);
10+
await InParallelAsync(source, levelOfParallelism, taskSelector, CancellationToken.None).ConfigureAwait(false);
1511
}
1612

1713
public static async Task InParallelAsync<TSource, TResult>(
1814
this IEnumerable<TSource> source,
1915
int levelOfParallelism,
2016
Func<TSource, Task<TResult>> taskSelector,
2117
CancellationToken cancellationToken)
22-
{
23-
await InParallelAsync(source, levelOfParallelism, taskSelector, cancellationToken, true).ConfigureAwait(false);
24-
}
25-
26-
public static async Task InParallelAsync<TSource, TResult>(
27-
this IEnumerable<TSource> source,
28-
int levelOfParallelism,
29-
Func<TSource, Task<TResult>> taskSelector,
30-
CancellationToken cancellationToken,
31-
bool isIOBound)
3218
{
3319
if (levelOfParallelism <= 0)
3420
{
35-
// For I/O-bound tasks, allow much higher concurrency
36-
levelOfParallelism = isIOBound ? Math.Max(100, Environment.ProcessorCount * 10) : Environment.ProcessorCount;
21+
levelOfParallelism = Environment.ProcessorCount;
3722
}
3823

39-
// For high-concurrency I/O operations, use a channel-based approach to reduce contention
40-
#if NET6_0_OR_GREATER
41-
if (isIOBound && levelOfParallelism > Environment.ProcessorCount * 4)
42-
{
43-
await ProcessWithChannelAsync(source, levelOfParallelism, taskSelector, cancellationToken).ConfigureAwait(false);
44-
return;
45-
}
46-
#endif
47-
4824
using var parallelLock = new SemaphoreSlim(initialCount:levelOfParallelism, maxCount:levelOfParallelism);
4925

50-
await Task.WhenAll(source.Select(item => ProcessAsync(item, taskSelector, parallelLock, cancellationToken, isIOBound))).ConfigureAwait(false);
26+
await Task.WhenAll(source.Select(item => ProcessAsync(item, taskSelector, parallelLock, cancellationToken))).ConfigureAwait(false);
5127
}
5228

5329
public static async Task InParallelAsync<TSource>(
5430
this IEnumerable<TSource> source,
5531
int levelOfParallelism,
5632
Func<TSource, Task> taskSelector)
5733
{
58-
await InParallelAsync(source, levelOfParallelism, taskSelector, CancellationToken.None, true).ConfigureAwait(false);
34+
await InParallelAsync(source, levelOfParallelism, taskSelector, CancellationToken.None).ConfigureAwait(false);
5935
}
6036

6137
public static async Task InParallelAsync<TSource>(
6238
this IEnumerable<TSource> source,
6339
int levelOfParallelism,
6440
Func<TSource, Task> taskSelector,
6541
CancellationToken cancellationToken)
66-
{
67-
await InParallelAsync(source, levelOfParallelism, taskSelector, cancellationToken, true).ConfigureAwait(false);
68-
}
69-
70-
public static async Task InParallelAsync<TSource>(
71-
this IEnumerable<TSource> source,
72-
int levelOfParallelism,
73-
Func<TSource, Task> taskSelector,
74-
CancellationToken cancellationToken,
75-
bool isIOBound)
7642
{
7743
if (levelOfParallelism <= 0)
7844
{
79-
// For I/O-bound tasks, allow much higher concurrency
80-
levelOfParallelism = isIOBound ? Math.Max(100, Environment.ProcessorCount * 10) : Environment.ProcessorCount;
81-
}
82-
83-
// For high-concurrency I/O operations, use a channel-based approach to reduce contention
84-
#if NET6_0_OR_GREATER
85-
if (isIOBound && levelOfParallelism > Environment.ProcessorCount * 4)
86-
{
87-
await ProcessWithChannelAsync(source, levelOfParallelism, taskSelector, cancellationToken).ConfigureAwait(false);
88-
return;
45+
levelOfParallelism = Environment.ProcessorCount;
8946
}
90-
#endif
9147

9248
using var parallelLock = new SemaphoreSlim(initialCount:levelOfParallelism, maxCount:levelOfParallelism);
9349

94-
await Task.WhenAll(source.Select(item => ProcessAsync(item, taskSelector, parallelLock, cancellationToken, isIOBound))).ConfigureAwait(false);
50+
await Task.WhenAll(source.Select(item => ProcessAsync(item, taskSelector, parallelLock, cancellationToken))).ConfigureAwait(false);
9551
}
9652

9753
// Overloads for CPU-bound processing
@@ -101,7 +57,7 @@ public static async Task InParallelAsync<TSource, TResult>(
10157
Func<TSource, TResult> taskSelector,
10258
CancellationToken cancellationToken = default)
10359
{
104-
await InParallelAsync(source, levelOfParallelism, item => Task.FromResult(taskSelector(item)), cancellationToken, false).ConfigureAwait(false);
60+
await InParallelAsync(source, levelOfParallelism, item => Task.FromResult(taskSelector(item)), cancellationToken).ConfigureAwait(false);
10561
}
10662

10763
public static async Task InParallelAsync<TSource>(
@@ -110,119 +66,14 @@ public static async Task InParallelAsync<TSource>(
11066
Action<TSource> taskSelector,
11167
CancellationToken cancellationToken = default)
11268
{
113-
await InParallelAsync(source, levelOfParallelism, item => { taskSelector(item); return Task.CompletedTask; }, cancellationToken, false).ConfigureAwait(false);
114-
}
115-
116-
#if NET6_0_OR_GREATER
117-
private static async Task ProcessWithChannelAsync<TSource, TResult>(
118-
IEnumerable<TSource> source,
119-
int levelOfParallelism,
120-
Func<TSource, Task<TResult>> taskSelector,
121-
CancellationToken cancellationToken)
122-
{
123-
var channel = Channel.CreateUnbounded<TSource>();
124-
var writer = channel.Writer;
125-
var reader = channel.Reader;
126-
127-
// Start producer
128-
var producerTask = Task.Run(async () =>
129-
{
130-
try
131-
{
132-
foreach (var item in source)
133-
{
134-
if (cancellationToken.IsCancellationRequested)
135-
break;
136-
await writer.WriteAsync(item, cancellationToken).ConfigureAwait(false);
137-
}
138-
}
139-
finally
140-
{
141-
writer.Complete();
142-
}
143-
}, cancellationToken);
144-
145-
// Start consumers
146-
var consumerTasks = Enumerable.Range(0, Math.Min(levelOfParallelism, Environment.ProcessorCount))
147-
.Select(_ => Task.Run(async () =>
148-
{
149-
await foreach (var item in reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
150-
{
151-
var task = taskSelector(item);
152-
if (task.IsCompleted)
153-
{
154-
// Fast-path for already completed tasks
155-
var result = await task.ConfigureAwait(false);
156-
}
157-
else
158-
{
159-
await task.ConfigureAwait(false);
160-
}
161-
}
162-
}, cancellationToken))
163-
.ToArray();
164-
165-
await Task.WhenAll(consumerTasks.Concat(new[] { producerTask })).ConfigureAwait(false);
166-
}
167-
168-
private static async Task ProcessWithChannelAsync<TSource>(
169-
IEnumerable<TSource> source,
170-
int levelOfParallelism,
171-
Func<TSource, Task> taskSelector,
172-
CancellationToken cancellationToken)
173-
{
174-
var channel = Channel.CreateUnbounded<TSource>();
175-
var writer = channel.Writer;
176-
var reader = channel.Reader;
177-
178-
// Start producer
179-
var producerTask = Task.Run(async () =>
180-
{
181-
try
182-
{
183-
foreach (var item in source)
184-
{
185-
if (cancellationToken.IsCancellationRequested)
186-
break;
187-
await writer.WriteAsync(item, cancellationToken).ConfigureAwait(false);
188-
}
189-
}
190-
finally
191-
{
192-
writer.Complete();
193-
}
194-
}, cancellationToken);
195-
196-
// Start consumers
197-
var consumerTasks = Enumerable.Range(0, Math.Min(levelOfParallelism, Environment.ProcessorCount))
198-
.Select(_ => Task.Run(async () =>
199-
{
200-
await foreach (var item in reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
201-
{
202-
var task = taskSelector(item);
203-
if (task.IsCompleted)
204-
{
205-
// Fast-path for already completed tasks
206-
await task.ConfigureAwait(false);
207-
}
208-
else
209-
{
210-
await task.ConfigureAwait(false);
211-
}
212-
}
213-
}, cancellationToken))
214-
.ToArray();
215-
216-
await Task.WhenAll(consumerTasks.Concat(new[] { producerTask })).ConfigureAwait(false);
69+
await InParallelAsync(source, levelOfParallelism, item => { taskSelector(item); return Task.CompletedTask; }, cancellationToken).ConfigureAwait(false);
21770
}
218-
#endif
21971

22072
private static async Task<TResult> ProcessAsync<TSource, TResult>(
22173
TSource item,
22274
Func<TSource, Task<TResult>> taskSelector,
22375
SemaphoreSlim parallelLock,
224-
CancellationToken cancellationToken,
225-
bool isIOBound)
76+
CancellationToken cancellationToken)
22677
{
22778
await parallelLock.WaitAsync(cancellationToken).ConfigureAwait(false);
22879

@@ -241,13 +92,7 @@ private static async Task<TResult> ProcessAsync<TSource, TResult>(
24192
return task.Result;
24293
}
24394

244-
// For I/O-bound tasks, don't use Task.Run wrapper
245-
if (isIOBound)
246-
{
247-
return await task.ConfigureAwait(false);
248-
}
249-
250-
// For CPU-bound tasks, use Task.Run to offload to ThreadPool
95+
// Use Task.Run to offload to ThreadPool
25196
return await Task.Run(async () => await task.ConfigureAwait(false), cancellationToken).ConfigureAwait(false);
25297
}
25398
finally
@@ -260,8 +105,7 @@ private static async Task ProcessAsync<TSource>(
260105
TSource item,
261106
Func<TSource, Task> taskSelector,
262107
SemaphoreSlim parallelLock,
263-
CancellationToken cancellationToken,
264-
bool isIOBound)
108+
CancellationToken cancellationToken)
265109
{
266110
await parallelLock.WaitAsync(cancellationToken).ConfigureAwait(false);
267111

@@ -280,16 +124,8 @@ private static async Task ProcessAsync<TSource>(
280124
return;
281125
}
282126

283-
// For I/O-bound tasks, don't use Task.Run wrapper
284-
if (isIOBound)
285-
{
286-
await task.ConfigureAwait(false);
287-
}
288-
else
289-
{
290-
// For CPU-bound tasks, use Task.Run to offload to ThreadPool
291-
await Task.Run(async () => await task.ConfigureAwait(false), cancellationToken).ConfigureAwait(false);
292-
}
127+
// Use Task.Run to offload to ThreadPool
128+
await Task.Run(async () => await task.ConfigureAwait(false), cancellationToken).ConfigureAwait(false);
293129
}
294130
finally
295131
{

‎EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs‎

Lines changed: 27 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,4 @@
11
#if NET6_0_OR_GREATER
2-
using System.Threading.Channels;
32
using EnumerableAsyncProcessor.Extensions;
43

54
namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable;
@@ -25,48 +24,40 @@ internal AsyncEnumerableParallelProcessor(
2524

2625
public async Task ExecuteAsync()
2726
{
28-
var channel = Channel.CreateUnbounded<TInput>();
2927
var cancellationToken = _cancellationTokenSource.Token;
28+
var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
29+
var tasks = new List<Task>();
3030

31-
// Producer task - enumerate the async enumerable and write to channel
32-
var producerTask = Task.Run(async () =>
31+
try
3332
{
34-
try
33+
await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false))
3534
{
36-
await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false))
35+
await semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
36+
37+
var capturedItem = item;
38+
var task = Task.Run(async () =>
3739
{
38-
await channel.Writer.WriteAsync(item, cancellationToken).ConfigureAwait(false);
39-
}
40-
channel.Writer.Complete();
40+
try
41+
{
42+
// Yield to ensure we don't block the thread if _taskSelector is synchronous
43+
await Task.Yield();
44+
await _taskSelector(capturedItem).ConfigureAwait(false);
45+
}
46+
finally
47+
{
48+
semaphore.Release();
49+
}
50+
}, cancellationToken);
51+
52+
tasks.Add(task);
4153
}
42-
catch (OperationCanceledException)
43-
{
44-
channel.Writer.TryComplete();
45-
throw;
46-
}
47-
catch (Exception ex)
48-
{
49-
channel.Writer.TryComplete(ex);
50-
throw;
51-
}
52-
}, cancellationToken);
5354

54-
// Consumer tasks - process items from the channel in parallel
55-
var consumerTasks = Enumerable.Range(0, _maxConcurrency)
56-
.Select(_ => Task.Run(async () =>
57-
{
58-
await foreach (var item in channel.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
59-
{
60-
// Yield to ensure we don't block the thread if _taskSelector is synchronous
61-
await Task.Yield();
62-
await _taskSelector(item).ConfigureAwait(false);
63-
}
64-
}, cancellationToken))
65-
.ToArray();
66-
67-
// Wait for producer and all consumers to complete
68-
await producerTask.ConfigureAwait(false);
69-
await Task.WhenAll(consumerTasks).ConfigureAwait(false);
55+
await Task.WhenAll(tasks).ConfigureAwait(false);
56+
}
57+
finally
58+
{
59+
semaphore.Dispose();
60+
}
7061
}
7162
}
7263
#endif

0 commit comments

Comments
 (0)