fix: Fix critical deadlock and hanging issues in async processors

## Changes Made:

### 1. Fixed cancellation handling in EnumerableExtensions
- Added proper CancellationToken support to ToIAsyncEnumerable method
- While loop now properly responds to cancellation requests
- Prevents infinite hanging when tasks don't complete

### 2. Fixed semaphore release bug in ParallelExtensions
- Only releases semaphore if it was successfully acquired
- Prevents SemaphoreFullException when cancellation occurs during WaitAsync
- Tracks acquisition state properly in both ProcessAsync overloads

### 3. Removed unnecessary Task.Run wrapping
- Eliminated redundant Task.Run in rate-limited processors
- Removed Task.Run wrapping for already-async operations in ParallelExtensions
- Fixed Task.Run + Task.Yield redundancy (just using Task.Yield where needed)

### 4. Improved unbounded processor implementations
- AsyncEnumerableUnboundedParallelProcessor now truly unbounded (no artificial limits)
- Trusts .NET ThreadPool to manage resources appropriately
- Enhanced documentation warnings about resource risks
- Maintains backward compatibility

### 5. Performance improvements
- Removed wasteful counting/enumeration operations
- Eliminated double context switching from Task.Run + Task.Yield
- Let runtime handle thread pool scheduling optimally

These fixes address the critical issues identified in the deadlock analysis while maintaining the library's performance characteristics and backward compatibility.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: thomhurst

EnumerableAsyncProcessor/Extensions/EnumerableExtensions.cs

@@ -21,21 +21,47 @@ public static ItemActionAsyncProcessorBuilder<T> ForEachAsync<T>(this IEnumerabl
             .ForEachAsync(taskSelector, cancellationToken);
     }
 
-    internal static async IAsyncEnumerable<T> ToIAsyncEnumerable<T>(this IEnumerable<Task<T>> tasks)
+    internal static async IAsyncEnumerable<T> ToIAsyncEnumerable<T>(this IEnumerable<Task<T>> tasks, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
 #if NET9_0_OR_GREATER
-        await foreach (var task in Task.WhenEach(tasks).ConfigureAwait(false))
+        await foreach (var task in Task.WhenEach(tasks).ConfigureAwait(false).WithCancellation(cancellationToken))
         {
             yield return task.Result;
         }
 #else
         var managedTasksList = tasks.ToList();
 
+        // Create a cancellation task that will complete when cancellation is requested
+        using var cancellationTcs = new CancellationTokenSource();
+        var cancellationTask = Task.Delay(Timeout.Infinite, cancellationTcs.Token);
+        
+        // Register callback to trigger the cancellation task
+        using var registration = cancellationToken.Register(() => cancellationTcs.Cancel());
+
         while (managedTasksList.Count != 0)
         {
-            var finishedTask = await Task.WhenAny(managedTasksList).ConfigureAwait(false);
-            managedTasksList.Remove(finishedTask);
-            yield return await finishedTask.ConfigureAwait(false);
+            // Check for cancellation before each iteration
+            cancellationToken.ThrowIfCancellationRequested();
+            
+            // Include the cancellation task in WhenAny
+            var allTasks = new List<Task>(managedTasksList.Count + 1);
+            allTasks.AddRange(managedTasksList);
+            allTasks.Add(cancellationTask);
+            
+            var finishedTask = await Task.WhenAny(allTasks).ConfigureAwait(false);
+            
+            // If the cancellation task completed, throw cancellation
+            if (finishedTask == cancellationTask)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                // This should not happen as cancellation should throw above, but as a safety measure:
+                throw new OperationCanceledException(cancellationToken);
+            }
+            
+            // Remove and yield the completed task
+            var completedTask = (Task<T>)finishedTask;
+            managedTasksList.Remove(completedTask);
+            yield return await completedTask.ConfigureAwait(false);
         }
 #endif
     }

EnumerableAsyncProcessor/Extensions/ParallelExtensions.cs

@@ -75,10 +75,13 @@ private static async Task<TResult> ProcessAsync<TSource, TResult>(
         SemaphoreSlim parallelLock,
         CancellationToken cancellationToken)
     {
-        await parallelLock.WaitAsync(cancellationToken).ConfigureAwait(false);
+        var semaphoreAcquired = false;
 
         try
         {
+            await parallelLock.WaitAsync(cancellationToken).ConfigureAwait(false);
+            semaphoreAcquired = true;
+            
             cancellationToken.ThrowIfCancellationRequested();
             var task = taskSelector(item);
 
@@ -92,12 +95,15 @@ private static async Task<TResult> ProcessAsync<TSource, TResult>(
                 return task.Result;
             }
 
-            // Use Task.Run to offload to ThreadPool
-            return await Task.Run(async () => await task.ConfigureAwait(false), cancellationToken).ConfigureAwait(false);
+            // Await the task directly - it's already async so no need for Task.Run
+            return await task.ConfigureAwait(false);
         }
         finally
         {
-            parallelLock.Release();
+            if (semaphoreAcquired)
+            {
+                parallelLock.Release();
+            }
         }
     }
 
@@ -107,10 +113,13 @@ private static async Task ProcessAsync<TSource>(
         SemaphoreSlim parallelLock,
         CancellationToken cancellationToken)
     {
-        await parallelLock.WaitAsync(cancellationToken).ConfigureAwait(false);
+        var semaphoreAcquired = false;
 
         try
         {
+            await parallelLock.WaitAsync(cancellationToken).ConfigureAwait(false);
+            semaphoreAcquired = true;
+            
             cancellationToken.ThrowIfCancellationRequested();
             var task = taskSelector(item);
 
@@ -124,12 +133,15 @@ private static async Task ProcessAsync<TSource>(
                 return;
             }
 
-            // Use Task.Run to offload to ThreadPool
-            await Task.Run(async () => await task.ConfigureAwait(false), cancellationToken).ConfigureAwait(false);
+            // Await the task directly - it's already async so no need for Task.Run
+            await task.ConfigureAwait(false);
         }
         finally
         {
-            parallelLock.Release();
+            if (semaphoreAcquired)
+            {
+                parallelLock.Release();
+            }
         }
     }
 }

EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableUnboundedParallelProcessor.cs

@@ -31,21 +31,24 @@ public async Task ExecuteAsync()
         var tasks = new List<Task>();
 
         // Start a task for each item immediately as it arrives
-        // No throttling or concurrency control
+        // No throttling or concurrency control - let the ThreadPool manage resources
         await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false))
         {
-            // Start task immediately without waiting
-            // Use Task.Run to ensure we don't block if _taskSelector is synchronous
-            var task = Task.Run(async () =>
-            {
-                await Task.Yield();
-                await _taskSelector(item).ConfigureAwait(false);
-            }, cancellationToken);
+            var capturedItem = item;
+            // Create task that yields first to prevent blocking if _taskSelector is synchronous
+            var task = ProcessItemAsync(capturedItem, cancellationToken);
             tasks.Add(task);
         }
 
         // Wait for all tasks to complete
         await Task.WhenAll(tasks).ConfigureAwait(false);
     }
+    
+    private async Task ProcessItemAsync(TInput item, CancellationToken cancellationToken)
+    {
+        // Yield to ensure we don't block the calling thread if _taskSelector is synchronous
+        await Task.Yield();
+        await _taskSelector(item).ConfigureAwait(false);
+    }
 }
 #endif

EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableUnboundedParallelProcessor.cs

@@ -33,13 +33,7 @@ public async IAsyncEnumerable<TOutput> ExecuteAsync()
         await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false))
         {
             var capturedItem = item;
-            var task = Task.Run(async () =>
-            {
-                // Yield to ensure we don't block the thread if _taskSelector is synchronous
-                await Task.Yield();
-                return await _taskSelector(capturedItem).ConfigureAwait(false);
-            }, cancellationToken);
-            
+            var task = ProcessItemAsync(capturedItem, cancellationToken);
             tasks.Add(task);
         }
 
@@ -51,5 +45,12 @@ public async IAsyncEnumerable<TOutput> ExecuteAsync()
             yield return await completedTask.ConfigureAwait(false);
         }
     }
+    
+    private async Task<TOutput> ProcessItemAsync(TInput item, CancellationToken cancellationToken)
+    {
+        // Yield to ensure we don't block the thread if _taskSelector is synchronous
+        await Task.Yield();
+        return await _taskSelector(item).ConfigureAwait(false);
+    }
 }
 #endif

EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultRateLimitedParallelAsyncProcessor_2.cs

@@ -15,10 +15,11 @@ internal ResultRateLimitedParallelAsyncProcessor(IEnumerable<TInput> items, Func
     internal override Task Process()
     {
         // For rate-limited processing, we want strict parallelism control
+        // TaskWrapper.Process already includes Task.Yield to prevent thread pool blocking
         return TaskWrappers.InParallelAsync(_levelsOfParallelism, 
             async taskWrapper =>
             {
-                await Task.Run(() => taskWrapper.Process(CancellationToken)).ConfigureAwait(false);
+                await taskWrapper.Process(CancellationToken).ConfigureAwait(false);
             }, CancellationToken.None);
     }
 }

EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultTimedRateLimitedParallelAsyncProcessor_1.cs

@@ -17,11 +17,12 @@ internal ResultTimedRateLimitedParallelAsyncProcessor(int count, Func<Task<TOutp
     internal override Task Process()
     {
         // For timed rate-limited processing, we want strict parallelism control
+        // TaskWrapper.Process already includes Task.Yield to prevent thread pool blocking
         return TaskWrappers.InParallelAsync(_levelsOfParallelism, 
             async taskWrapper =>
             {
                 await Task.WhenAll(
-                    Task.Run(() => taskWrapper.Process(CancellationToken)),
+                    taskWrapper.Process(CancellationToken),
                     Task.Delay(_timeSpan, CancellationToken)).ConfigureAwait(false);
             }, CancellationToken.None);
     }

EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultTimedRateLimitedParallelAsyncProcessor_2.cs

@@ -17,11 +17,12 @@ internal ResultTimedRateLimitedParallelAsyncProcessor(IEnumerable<TInput> items,
     internal override Task Process()
     {
         // For timed rate-limited processing, we want strict parallelism control
+        // TaskWrapper.Process already includes Task.Yield to prevent thread pool blocking
         return TaskWrappers.InParallelAsync(_levelsOfParallelism, 
             async taskWrapper =>
             {
                 await Task.WhenAll(
-                    Task.Run(() => taskWrapper.Process(CancellationToken)),
+                    taskWrapper.Process(CancellationToken),
                     Task.Delay(_timeSpan, CancellationToken)).ConfigureAwait(false);
             }, CancellationToken.None);
     }

EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultUnboundedParallelAsyncProcessor_1.cs

@@ -4,8 +4,24 @@ 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.
+/// WARNING: Use with extreme caution - this WILL overwhelm system resources with large task counts.
+/// 
+/// RISKS:
+/// - No concurrency throttling - all tasks start immediately
+/// - Can cause thread pool starvation with thousands of tasks
+/// - May exhaust system memory with very large collections
+/// - Can lead to degraded system performance
+/// 
+/// RECOMMENDATIONS:
+/// - For collections > 1000 items, use ResultParallelAsyncProcessor with appropriate concurrency limits
+/// - Monitor system resources when using this processor
+/// - Consider the bounded alternatives for production use
+/// 
+/// Ideal only for scenarios with:
+/// - Small task counts (< 100)
+/// - Very lightweight operations
+/// - Sufficient system resources
+/// - Controlled environments where task count is known
 /// </summary>
 public class ResultUnboundedParallelAsyncProcessor<TOutput> : ResultAbstractAsyncProcessor<TOutput>
 {
@@ -17,15 +33,7 @@ internal override async 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;
-        });
+        var tasks = TaskWrappers.Select(taskWrapper => taskWrapper.Process(CancellationToken));
 
         await Task.WhenAll(tasks).ConfigureAwait(false);
     }

EnumerableAsyncProcessor/RunnableProcessors/UnboundedParallelAsyncProcessor.cs

@@ -4,8 +4,24 @@ 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.
+/// WARNING: Use with extreme caution - this WILL overwhelm system resources with large task counts.
+/// 
+/// RISKS:
+/// - No concurrency throttling - all tasks start immediately
+/// - Can cause thread pool starvation with thousands of tasks
+/// - May exhaust system memory with very large collections
+/// - Can lead to degraded system performance
+/// 
+/// RECOMMENDATIONS:
+/// - For collections > 1000 items, use ParallelAsyncProcessor with appropriate concurrency limits
+/// - Monitor system resources when using this processor
+/// - Consider the bounded alternatives for production use
+/// 
+/// Ideal only for scenarios with:
+/// - Small task counts (< 100)
+/// - Very lightweight operations
+/// - Sufficient system resources
+/// - Controlled environments where task count is known
 /// </summary>
 public class UnboundedParallelAsyncProcessor : AbstractAsyncProcessor
 {
@@ -17,15 +33,7 @@ internal override async 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;
-        });
+        var tasks = TaskWrappers.Select(taskWrapper => taskWrapper.Process(CancellationToken));
 
         await Task.WhenAll(tasks).ConfigureAwait(false);
     }