# 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](https://www.kanject.com/docs/core-recurring/): cadence and parallelism both belong on the method.

## Install

```bash
dotnet add package Kanject.Core
dotnet 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

```csharp
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);
}
```

```csharp
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.

```csharp
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);
}
```

```csharp
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.

```csharp
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);
}
```

```csharp
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](https://www.kanject.com/docs/core-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.

> **Keep the [Parallel] method public:** The generated overloads are extension methods in their own static class, so they cannot call a `private` method. Put the work on a public method of its own `partial` class, as above, and inject it where you need it.

## When a partition fails

```csharp
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.

```csharp
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.

```csharp
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

> **In-process, and no hosted mode yet:** Parallel fans out inside one process. It is not a distributed job runner: across instances, use [Queue](https://www.kanject.com/docs/core-queue/) or [Channels](https://www.kanject.com/docs/core-channels/) to spread the work and `[Parallel]` inside each consumer. `[ParallelHosted]` exists, but this release generates no hosting code for it (`KANPAR014` warns); to run a fan-out in the background, call `{Method}ParallelAsync` from your own `BackgroundService`.

---
_Source: https://www.kanject.com/docs/core-parallel/ · Kanject Docs_