C# task parallel library

ProcessItemAsync(item, ct); progress.Report(result); } });

await Task.WhenAll(consumers);

}

### Step 4: Handle Exceptions and Cancellation Correctly
TPL aggregates exceptions into `AggregateException`. Always observe tasks. Use `CancellationTokenSource` linked to external timeouts.

```csharp
try
{
    await ProcessPipelineAsync(source, progress, cts.Token);
}
catch (OperationCanceledException)
{
    // Graceful shutdown path
}
catch (Exception ex) when (ex is not OperationCanceledException)
{
    // Log, alert, or fallback
}

Architecture Decisions and Rationale

• Channel<T> over ConcurrentQueue<T>: Channels provide async wait semantics, bounded capacity, and completion signaling without busy-waiting.
• Parallel.ForEachAsync over Parallel.ForEach: The async variant integrates with the async state machine, avoids blocking threads, and respects CancellationToken natively.
• Bounded over Unbounded: Unbounded queues shift memory pressure to Gen 2 GC. Bounding forces flow control, which stabilizes latency under load.
• Explicit CancellationToken propagation: Cancellation is cooperative. Without token flow, stalled I/O or locked resources cannot be reclaimed.

Pitfall Guide

1. Fire-and-forget without observation
   Task.Run(() => DoWork()); without awaiting or storing the task reference means exceptions are swallowed until the finalizer thread observes them, often crashing the process. Always capture the task or use Task.WhenAll/Task.WhenAny.

2. Sync-over-async blocking
   Calling .Result or .Wait() on async tasks blocks ThreadPool threads, preventing scaling. Use await consistently. If forced into a sync context, use ConfigureAwait(false) in libraries and consider Task.Run(() => asyncMethod().GetAwaiter().GetResult()) only as a last resort with explicit timeout bounds.

3. Using Parallel for I/O-bound work
   Parallel.ForEach assumes CPU-bound execution. Applying it to HTTP calls or database queries saturates threads waiting on sockets, triggering ThreadPool starvation. Replace with SemaphoreSlim or Channel<T> + async consumers.

4. Ignoring CancellationToken propagation
   TPL respects cancellation only when tokens are explicitly passed. Omitting tokens means work continues after shutdown signals, consuming resources and delaying process termination. Always thread CancellationToken through method signatures and TPL constructors.

5. ThreadPool starvation from lock contention
   Mixing lock or Monitor with async code creates deadlocks when threads block waiting for async continuations. Prefer SemaphoreSlim, AsyncLock, or lock-free structures (ConcurrentDictionary, Interlocked) in async paths.

6. Capturing SynchronizationContext unnecessarily
   In UI or ASP.NET contexts, await captures the context by default, causing continuations to marshal back to the original thread. In libraries and background services, use ConfigureAwait(false) to avoid thread pinning and reduce context switch overhead.

7. Leaving MaxDegreeOfParallelism at -1
   The default value removes concurrency limits, allowing unbounded task creation. This defeats ThreadPool scaling algorithms and causes latency spikes. Always set explicit bounds aligned with workload capacity.

Best Practices from Production:

• Use structured concurrency: parent tasks should await all children.
• Implement backpressure explicitly; never trust unbounded queues.
• Aggregate exceptions at the boundary; do not swallow AggregateException.
• Prefer ValueTask for hot-path async methods that frequently complete synchronously.
• Profile with dotnet-counters and dotnet-trace to validate ThreadPool behavior under load.

Production Bundle

Action Checklist

• Classify workload: CPU-bound, I/O-bound, or mixed before selecting TPL primitive
• Set explicit MaxDegreeOfParallelism aligned with external capacity, not CPU count
• Replace unbounded queues with Channel.CreateBounded and configure FullMode
• Thread CancellationToken through all async/TPL entry points and monitor IsCancellationRequested
• Replace .Result/.Wait() with await; enforce async-all-the-way in service layers
• Add ConfigureAwait(false) to library code and background workers
• Wrap TPL execution in try/catch for OperationCanceledException and AggregateException
• Validate ThreadPool saturation using dotnet-counters monitor --counters System.Threading

Decision Matrix

Configuration Template

// DI Registration for TPL Pipeline
public static class TplPipelineExtensions
{
    public static IServiceCollection AddTplProcessingPipeline(
        this IServiceCollection services,
        int maxDegreeOfParallelism = -1,
        int channelCapacity = 1000)
    {
        var degree = maxDegreeOfParallelism > 0 
            ? maxDegreeOfParallelism 
            : Environment.ProcessorCount;

        services.AddSingleton(new ParallelOptions
        {
            MaxDegreeOfParallelism = degree,
            TaskScheduler = TaskScheduler.Default
        });

        services.AddSingleton(Channel.CreateBounded<WorkItem>(new BoundedChannelOptions(channelCapacity)
        {
            FullMode = BoundedChannelFullMode.Wait,
            SingleWriter = false,
            SingleReader = false
        }));

        return services;
    }
}

// Usage in HostedService
public class ProcessingWorker : BackgroundService
{
    private readonly Channel<WorkItem> _channel;
    private readonly ParallelOptions _parallelOptions;
    private readonly ILogger<ProcessingWorker> _logger;

    public ProcessingWorker(
        Channel<WorkItem> channel,
        ParallelOptions parallelOptions,
        ILogger<ProcessingWorker> logger)
    {
        _channel = channel;
        _parallelOptions = parallelOptions;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await Parallel.ForEachAsync(
            _channel.Reader.ReadAllAsync(stoppingToken),
            _parallelOptions,
            async (item, ct) =>
            {
                await ProcessAsync(item, ct);
            });
    }
}

Quick Start Guide

1. Install dependencies: Ensure .NET 6+ runtime. No additional NuGet packages required for core TPL.
2. Configure bounds: Set MaxDegreeOfParallelism to match your workload capacity. For I/O, start at 50–100 and tune via load testing.
3. Implement backpressure: Replace direct task spawning with Channel.CreateBounded and pipe work through Parallel.ForEachAsync or async consumer loops.
4. Wire cancellation: Pass CancellationToken from BackgroundService or HTTP middleware through all TPL calls. Test shutdown behavior with dotnet run and Ctrl+C.
5. Validate under load: Run dotnet-counters monitor --counters System.Threading.ThreadPool.QueueLength,System.Threading.ThreadPool.CompletedWorkItemCount during stress testing. Adjust channel capacity and parallelism until queue length stabilizes below 10% of throughput.