Parallel Execution

Prerequisites: Defining Pipelines, Custom Nodes

By default, NPipeline processes items sequentially through each node. The NPipeline.Extensions.Parallelism package lets you run transform nodes concurrently with bounded parallelism and backpressure control.

Installation

dotnet add package NPipeline.Extensions.Parallelism

Quick Start: Workload Presets

The simplest approach uses workload presets that set sensible defaults:

var transform = builder.AddTransform<CallApi, Request, Response>("api-call");
transform.RunParallel(builder, ParallelWorkloadType.IoBound);

Preset	DOP	Queue Length	Best For
General	CPU × 2	CPU × 4	Mixed workloads
CpuBound	CPU	CPU × 2	CPU-intensive transforms
IoBound	CPU × 4	CPU × 8	Database, HTTP, file I/O
NetworkBound	min(CPU × 8, 100)	200	High-latency network calls

Custom Configuration

Use the fluent builder for fine-grained control:

transform.RunParallel(builder, opt => opt
    .MaxDegreeOfParallelism(8)
    .MaxQueueLength(50)
    .BlockOnBackpressure()
    .OutputBufferCapacity(200)
    .AllowUnorderedOutput());

Key Settings

Setting	Default	Description
MaxDegreeOfParallelism	varies by preset	Max concurrent items being processed
MaxQueueLength	varies by preset	Max items waiting in the input queue
PreserveOrdering	true	Output items in the same order as input
OutputBufferCapacity	unset	Buffer size for output items
MetricsInterval	unset	How often to report throughput metrics
EnableInputWaitTiming	false	Per-item input-wait timing attribution (adds overhead)

Backpressure Policies

When the input queue is full, three policies control what happens:

// Block upstream until queue has space (default)
opt.BlockOnBackpressure();

// Drop the oldest queued item to make room
opt.DropOldestOnBackpressure();

// Drop the newest incoming item
opt.DropNewestOnBackpressure();

Policy	Data Loss	Use When
Block (default)	None	Every item must be processed
DropOldest	Oldest items	Real-time systems where stale data is less valuable
DropNewest	Newest items	Rate-limiting with preference for older items

⚠️ Warning: DropOldest and DropNewest silently discard items. Use them only when data loss is acceptable, and always monitor dropped-item counts.

Explicit Strategy Methods

For maximum control, use the strategy-specific extension methods:

// Blocking with explicit parameters
transform.WithBlockingParallelism(builder,
    maxDegreeOfParallelism: 4,
    maxQueueLength: 100,
    outputBufferCapacity: 200);

// Blocking without ordering guarantees (highest throughput, no data loss)
transform.WithUnorderedParallelism(builder,
    maxDegreeOfParallelism: 4,
    maxQueueLength: 100);

// Drop-oldest
transform.WithDropOldestParallelism(builder,
    maxDegreeOfParallelism: 4,
    maxQueueLength: 100);

// Drop-newest
transform.WithDropNewestParallelism(builder,
    maxDegreeOfParallelism: 4,
    maxQueueLength: 100);

Thread Safety

When using parallel execution, your transform node's TransformAsync method is called concurrently from multiple threads. You must ensure:

1. No shared mutable state - or protect it with locks/concurrent collections
2. No direct writes to context dictionaries - see the safety matrix below
3. All dependencies are thread-safe - HttpClient (via IHttpClientFactory), connection pools, etc.

Context Dictionary Thread Safety

Thread safety of context dictionaries depends on the optimization profile:

Default profile - dictionaries are ConcurrentDictionary (thread-safe):

Dictionary	Read	Write	Notes
context.Parameters	Safe	Safe	Concurrent reads and writes supported
context.Items	Safe	Safe	Concurrent reads and writes supported
context.Properties	Safe	Safe	Framework-managed; avoid writing from nodes

HighThroughput profile - dictionaries are plain Dictionary (not thread-safe):

Dictionary	Read	Write	Notes
context.Parameters	Safe	Unsafe	Populated before execution, treat as read-only
context.Items	Unsafe	Unsafe	Use IPipelineStateManager for shared state
context.Properties	Safe	Unsafe	Framework-managed; do not write from nodes

Dictionary Implementation by Profile

In the Default profile, context dictionaries are thread-safe: ConcurrentDictionary<string, object> when the context creates them, and a locking wrapper around dictionaries you supply yourself. This eliminates the most common source of bugs when developers first enable parallel execution - concurrent writes to context.Items no longer throw or corrupt data.

In the HighThroughput profile, context dictionaries use pooled Dictionary<string, object> instances for zero locking overhead. This avoids memory barriers on every dictionary access, which matters at millions of operations per second. The trade-off is that concurrent writes are unsafe - use IPipelineStateManager for shared state in parallel scenarios.

Safe and Unsafe Patterns

// ✓ Safe: no shared state
public override ValueTask<Result> TransformAsync(
    Input item, PipelineContext context, CancellationToken ct)
{
    return ValueTask.FromResult(new Result(item.Value * 2));
}

// ✓ Safe: atomic operations for simple counters
private int _count;
public override ValueTask<Result> TransformAsync(
    Input item, PipelineContext context, CancellationToken ct)
{
    Interlocked.Increment(ref _count);
    return ValueTask.FromResult(new Result(item.Value));
}

// ✗ Unsafe: shared mutable state without synchronization
private int _count;
public override ValueTask<Result> TransformAsync(
    Input item, PipelineContext context, CancellationToken ct)
{
    _count++; // Race condition!
    return ValueTask.FromResult(new Result(item.Value));
}

IPipelineStateManager

For complex shared state across parallel nodes, implement IPipelineStateManager:

public interface IPipelineStateManager
{
    ValueTask CreateSnapshotAsync(PipelineContext context, CancellationToken ct, bool forceFullSnapshot = false);
    ValueTask<bool> TryRestoreAsync(PipelineContext context, CancellationToken ct);
    void MarkNodeCompleted(string nodeId, PipelineContext context);
    void MarkNodeError(string nodeId, PipelineContext context);
}

This interface provides checkpoint/restore semantics for pipeline state, enabling safe state management with parallel execution and resilience features. The framework calls CreateSnapshotAsync after each node completes. It doesn't call TryRestoreAsync for you: node restart and node retry don't roll back shared state, so call it from your own recovery logic when you need a rollback.

📝 Note: Ordering guarantees come with a performance cost. Set AllowUnorderedOutput() if downstream nodes don't depend on item order.

Combining with Resilience

Parallel strategies transform items with the same item executor as the sequential strategy, so item retry, backoff, the circuit breaker, and OnItemFailure behave the same way. Parallel strategies also implement IResumableExecutionStrategy, so node restart works with them. You don't wrap the strategy yourself; when node restart is enabled for the node, the builder wraps it when you build the pipeline:

var transform = builder.AddTransform<MyTransform, In, Out>("transform");
transform.RunParallel(builder, ParallelWorkloadType.IoBound);

builder.WithResilience(transform, options => options with
{
    ItemRetry = ItemRetryOptions.Default with { MaxRetries = 5 },
    NodeRestart = new NodeRestartOptions { MaxRestarts = 3 },
});

A restart resumes the node from its checkpoint. What a restart delivers depends on output ordering:

• With ordered output, the default, each output is delivered exactly once across restarts.
• With unordered output (AllowUnorderedOutput()) or a dropping queue policy, delivery is at least once. After a restart, outputs that were delivered ahead of the checkpoint are delivered again, at most as many as were in flight when the node failed. Make downstream nodes idempotent if that matters.

For more information, see Node restart and the replay window.

Next Steps

• Parallelism Extension Reference - strategies, options, metrics, and queue policies
• Streaming Large Datasets - memory management with parallel processing
• Pipeline Validation - analyzer rules for parallel configuration
• Error Handling - resilience with parallel nodes