5.5.2.1.17. Parallel Activity Node

Parallel Activity Node

A Parallel Activity Node runs a nested Activity concurrently for partitions of its incoming tokens. AddParallelActivity<TToken>(..., chunkSize, maxConcurrency) selects tokens of type TToken and creates one nested execution per chunk. Use it for independent work; coordinate any shared external state in application code.

flowchart LR
    load(Load images) -->|Image tokens| parallel{{Parallel Activity}}
    subgraph body["Concurrent nested executions"]
        input["Input pin"] --> resize(Resize image)
        resize --> output["Output pin"]
    end
    parallel -->|Resized images| save(Save results)

Loading

Mermaid approximation: this diagram shows the model's routing, not a complete UML Activity notation. Rounded boxes represent actions; edge labels identify token types or guards. Labeled boxes represent Stateflows boundary pins, not UML activity-parameter notation. Mermaid does not model pins, activity parameters, or expansion regions natively.

Definition

.AddInitial(initial => initial.AddControlFlow("LoadImages"))
.AddAction(
    "LoadImages",
    async c => c.OutputRange(await images.LoadAsync(c.CancellationToken)),
    node => node.AddFlow<Image>("Resize"))
.AddParallelActivity<Image>("Resize", nested => nested
    .AddInputPin<Image>(pin => pin.AddFlow("ResizeOne"))
    .AddAction(
        "ResizeOne",
        async c => c.Output(await imageProcessor.ResizeAsync(
            c.GetTokensOfType<Image>().Single(), c.CancellationToken)),
        node => node.AddFlow<Image, OutputPin<Image>>())
    .AddOutputPin<Image>()
    .AddFlow<Image>("Save"),
    chunkSize: 1)
.AddAction("Save", async c =>
    await images.SaveAsync(c.GetTokensOfType<Image>(), c.CancellationToken))

The default chunk size is one token. A larger chunk gives one nested execution multiple selected tokens. Each concurrent execution receives a distinct token partition; capture shared mutable state safely and ensure downstream services support concurrent calls. Aggregate boundary output is collected through inner output pins declared with AddOutputPin<TToken>(). Without boundary output, the node can still perform side effects or use the explicit outgoing crossings below.

Concurrency limits and interleaving

Every AddParallelActivity overload, including typed/named/inherited/reactive overloads, accepts int maxConcurrency = -1 immediately after chunkSize. -1 means unlimited active partitions; positive values impose a per-invocation bound. 0 and values below -1 fail definition validation. “Concurrency” counts active Activity partitions, not operating-system or CPU threads. Queued chunks wait asynchronously for slots, without rejection or dropping; unrelated invocations do not share the bound.

Completed interleaved partitions and their disposed scopes are released while the invocation remains active. Checkpoints retain unfinished partitions, not a growing list of completed work. Aggregate boundary output still accumulates until completion. The pending partition queue has no capacity limit; maxConcurrency bounds active partitions, not queued tokens or buffered output.

.AddParallelActivity<Image>("Resize", nested => nested
    .Interleave()
    .AddInputPin<Image>(pin => pin.AddFlow("ResizeOne"))
    .AddAction("ResizeOne", async context =>
        context.Output(await imageProcessor.ResizeAsync(
            context.GetTokensOfType<Image>().Single(), context.CancellationToken)),
        action => action.AddFlow<Image, OutputPin<Image>>())
    .AddOutputPin<Image>(),
    chunkSize: 1,
    maxConcurrency: 4)

Parameterless .Interleave() sets NodeOptions.Interleave. First arrival initializes once; later arrivals add chunks to that same invocation while it is open. The limit covers partitions from all arrivals of that invocation. Each arrival is chunked separately and its final partial chunk is included immediately. A partition receives non-selection/common input only from its own arrival; existing partitions keep their inputs.

When the executable/running network drains, finalize once and aggregate ordinary boundary output/control. Unsatisfied join/weight tokens are discarded. There is no stay-open-while-idle behavior, close API or graceful-finalization node. Admission and drain are atomic: arrivals after finalization starts wait for OnFinalize and completion-output routing, then start a fresh invocation. Checkpoints preserve queued chunks, input provenance and progress without reinitialization. Existing cancellation, failures, ActivityFinal and restrictions on event waits in expansion bodies remain unchanged. .Interleave() and .Reentrant() together fail registration.

Interleaving does not publish aggregate boundary output early. For live results, publish notifications from the body. JSON uses "interleave" in options and "maxConcurrency": 4; PlantUML shows <<interleave>> [maxConcurrency=4].

Outgoing crossings

An internal producer can use AddFlow<Result>("record") or AddControlFlow("notify") to route to a globally named node in the ParallelActivity's immediately enclosing scope. Destination names are registered as-is, not prefixed by their scope. Destinations in a grandparent scope or inside a sibling structure are invalid.

Incoming crossings across a ParallelActivity boundary are rejected, including targets nested more deeply inside it. Local partition flows and ordinary structured crossings entirely within one partition remain available.

Outgoing tokens enter the enclosing invocation, not the emitting partition's marking. The destination uses its own cancellation, selection, and checkpoint ownership, preserving streamed token provenance. Inner termination does not cancel work accepted outside; arrival order between partitions is unspecified. Output pins still aggregate partition output before the container's outgoing flows run. Direct crossings do not change aggregation or enable interleaving.

See Flows for crossing-flow activation and ownership rules.

Unlike Iterative Activity Node, this node runs chunks concurrently. See also Structured Activity Node.