Kanject.Core.Parallel

Bounded, attribute-driven fan-out. Put [Parallel] on a method that handles one item or one chunk, and the generator writes {Method}ParallelAsync: it splits a collection into partitions, runs them concurrently up to a limit, retries and times out each one, and hands back a typed result instead of a thrown exception. It is the sibling of Recurring: cadence and parallelism both belong on the method.

Install

dotnet add package Kanject.Coredotnet add package Kanject.Core.Annotations

The engine ships in Kanject.Core and the generator in Kanject.Core.Annotations; there is no separate Parallel package. Keep the two on the same version and mark the generator PrivateAssets="all". Both are on public NuGet and free under the Community licence.

Why not Parallel.ForEachAsync?

Task.WhenAll has no limit. Parallel.ForEachAsync bounds concurrency, but its body sees one item at a time, one exception ends the whole run, and there is no retry or per-call timeout. A batched write to DynamoDB or SQS needs all of that at once, so teams end up hand-writing the same worker pool. [Parallel] is that worker pool, generated.

One item at a time

using Kanject.Core.Annotations.Attributes.Parallel;​public sealed partial class OrderProcessor(IOrderClient client){    // The defaults for every call. A call site can override them.    [Parallel(MaxDegreeOfParallelism = 8, RetryCount = 1, RetryBackoffMs = 250)]    public ValueTask ProcessAsync(Order order, CancellationToken ct)        => client.ProcessAsync(order, ct);}

ParallelResult result = await processor.ProcessParallelAsync(orders, cancellationToken: ct);​Console.WriteLine($"{result.ItemCount} orders, {result.FailedCount} failed partitions, {result.StopReason}");

The containing type must be partial. A trailing Async is dropped before the suffix is added, so ProcessAsync becomes ProcessParallelAsync. The aggregated overload accepts ReadOnlyMemory<T>, T[], IReadOnlyList<T>, IEnumerable<T> and IAsyncEnumerable<T> sources.

A chunk at a time, for batched writes

Take a ReadOnlyMemory<T> instead of a single item and the body receives a whole partition. The slice is a view over your array, not a copy.

using Kanject.Core.Annotations.Attributes.Parallel;using Kanject.Core.Annotations.Attributes.Parallel.Enums;​public sealed partial class OrderWriter(IOrderStore store){    // 25 items per chunk: DynamoDB's BatchWriteItem limit.    [Parallel(ChunkingMode = ParallelChunkingMode.FixedChunkSize, ChunkSize = 25,              MaxDegreeOfParallelism = 4, ChunkTimeoutMs = 10_000,              StopOnException = false)]    public ValueTask<int> SaveAsync(ReadOnlyMemory<Order> chunk, CancellationToken ct)        => store.BatchInsertAsync(chunk, ct);}

ParallelResult<int> saved = await writer.SaveParallelAsync(orders, cancellationToken: ct);​// One value per partition. A failed partition leaves null.var written = saved.Values.Sum();Console.WriteLine($"{written} rows written, {saved.TimedOutCount} chunks timed out");

A chunk method that returns a value also gets {Method}ParallelStream, an IAsyncEnumerable that yields each partition as it finishes, with its PartitionIndex so you can put results back in order.

Per-item outcomes and SQS partial batches

A per-item method that returns a value also gets {Method}ParallelOutcomesAsync. Every entry in its Outcomes list carries the item's Index, Input, Value, Exception, TimedOut, Cancelled and RetryCount, in input order. That is the shape a Lambda needs to acknowledge part of a batch.

using Kanject.Core.Annotations.Attributes.Parallel;​public sealed partial class OrderHandler(IOrderClient client){    [Parallel(MaxDegreeOfParallelism = 8, RetryCount = 1, RetryBackoffMs = 250)]    public ValueTask<bool> HandleAsync(Order order, CancellationToken ct)        => client.HandleAsync(order, ct);}

using Kanject.Core.Queue.Abstractions.Models;using Kanject.Core.Queue.Provider.AwsSqs.Abstractions.Extensions;using Kanject.Core.Queue.Provider.AwsSqs.Abstractions.Interfaces;using Kanject.Core.Queue.Provider.AwsSqs.Consumers;​public sealed class OrderQueueConsumer(IQueueRegistry registry, OrderHandler handler)    : AbstractQueueConsumer<Order>(registry){    protected override async Task ConsumeAsync(        List<MessageContext<Order>> messages, CancellationToken ct)    {        var orders = messages.Select(m => m.Message).ToArray();​        // One outcome per message, in input order.        var result = await handler.HandleParallelOutcomesAsync(orders, cancellationToken: ct);​        // Successes are acknowledged. Failures go back to SQS as BatchItemFailures.        await result.AcknowledgeOrFailAsync(messages, Response, AcknowledgeAsync);    }}

AcknowledgeOrFailAsync comes from the Queue SQS provider. The message list must be index-aligned with the inputs. By default an outcome counts as a success when it succeeded and returned a non-null value; pass a successPredicate to change that.

When a partition fails

var result = await processor.ProcessParallelAsync(orders, cancellationToken: ct);​if (result.StopReason == ParallelStopReason.Cancelled)    ct.ThrowIfCancellationRequested();​// The fault that stopped the run, never a sibling's cancellation.if (result.LastException is not null)    ExceptionDispatchInfo.Capture(result.LastException).Throw();

By default the first fault cancels the partitions still running. LastException is the fault that stopped the run, never a sibling's cancellation, and FailedCount counts every partition that did not finish. Set StopOnException = false to keep going: the result then carries an AggregateException of every fault in partition order. Rethrow through ExceptionDispatchInfo so the original stack trace survives.

Override at the call site

The attribute holds defaults. Pass options to change them for one call.

var result = await processor.ProcessParallelAsync(orders,    options: ParallelLoopOptions.Default with { MaxDegreeOfParallelism = 2 },    cancellationToken: ct);

The engine without the attribute

The generator is a convenience. ParallelLoop is public and takes a body directly.

using Kanject.Core.Parallel.Engine;using Kanject.Core.Parallel.Enums;using Kanject.Core.Parallel.Models;​var result = await ParallelLoop.ForEachChunkAsync<Order, int>(    orders,    (chunk, partitionIndex, token) => store.BatchInsertAsync(chunk, token),    ParallelLoopOptions.Default with    {        ChunkingStrategy = ParallelChunkingStrategy.FixedChunkSize,        ChunkSize = 25,        MaxDegreeOfParallelism = 4,        ExceptionPolicy = ParallelExceptionPolicy.ContinueAndCollect,    },    ct);

What ships with it

• [Parallel] options: MaxDegreeOfParallelism, ChunkingMode, ChunkSize, ChunkTimeoutMs, StopOnException, FailOnTimeout, RetryCount, RetryBackoffMs
• Generated {Method}ParallelAsync, plus {Method}ParallelStream for value-returning chunk methods and {Method}ParallelOutcomesAsync for value-returning per-item methods
• A generated I{Type}Parallel interface on the containing type, so the parallel surface can be registered and mocked through DI
• Typed results: ParallelResult reports ItemCount, PartitionCount, FailedCount, TimedOutCount, RetryAttempts, Elapsed, StopReason and LastException
• Static ParallelHookProvider hooks (OnPartition, OnException, OnStop) for process-wide telemetry
• Roslyn analyzer rules KANPAR001–KANPAR014 that catch an unsupported signature, a non-partial type or a contradictory option while you type