Added an incremental, branch-oriented parallel API · aws/aws-lambda-dotnet@6ca28fc · GitHub
Skip to content

Commit 6ca28fc

Browse files
GarrettBeattyGarrettBeatty
authored
Added an incremental, branch-oriented parallel API
1 parent 52b1966 commit 6ca28fc

24 files changed

Lines changed: 3184 additions & 6 deletions

File tree

‎.autover/changes/add-incremental-heterogeneous-parallel.json‎

Lines changed: 12 additions & 0 deletions

‎Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -195,6 +195,38 @@ private Task<ICallback<T>> RunCallback<T>(
195195
return op.ExecuteAsync(cancellationToken);
196196
}
197197

198+
public IDurableParallel CreateParallel(
199+
string? name = null,
200+
ParallelConfig? config = null)
201+
{
202+
var effectiveConfig = config ?? new ParallelConfig();
203+
// Operation-level default for per-branch result serialization: the config's
204+
// ItemSerializer if set, else the globally-registered serializer. Individual
205+
// branches may still override this via Branch's serializer parameter.
206+
//
207+
// Resolved LAZILY: a workflow that overrides the serializer on every Branch
208+
// call (the AOT/per-branch scenario) must not be forced to register a global
209+
// serializer. GetRequired is deferred to the factory below and invoked only
210+
// when a branch actually falls back to this operation-level default.
211+
var lambdaContext = LambdaContext;
212+
Func<Amazon.Lambda.Core.ILambdaSerializer> defaultSerializerFactory = () =>
213+
{
214+
// Match the other serializer-resolving dispatchers (ParallelAsync, MapAsync,
215+
// etc.): bind the globally-registered serializer as the inner of an
216+
// IDefaultInnerSerializer (e.g. the inner-less FileSystemSerializer) so
217+
// ParallelConfig.ItemSerializer behaves the same on CreateParallel as on
218+
// batch ParallelAsync.
219+
var defaultSerializer = LambdaSerializerHelper.GetRequired(lambdaContext);
220+
return LambdaSerializerHelper.WithDefaultInner(
221+
effectiveConfig.ItemSerializer ?? defaultSerializer, defaultSerializer);
222+
};
223+
224+
var operationId = _idGenerator.NextId();
225+
return new Internal.IncrementalParallelOperation(
226+
operationId, name, _idGenerator.ParentId, effectiveConfig, defaultSerializerFactory, MakeChildFactory(),
227+
_state, _terminationManager, _workflowCancellation, _durableExecutionArn, _batcher);
228+
}
229+
198230
public Task<IBatchResult<T>> ParallelAsync<T>(
199231
IReadOnlyList<Func<IDurableContext, CancellationToken, Task<T>>> branches,
200232
string? name = null,

‎Libraries/src/Amazon.Lambda.DurableExecution/IDurableContext.cs‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -367,6 +367,50 @@ Task<TState> WaitForConditionAsync<TState>(
367367
string? name = null,
368368
CancellationToken cancellationToken = default);
369369

370+
/// <summary>
371+
/// Create an incremental, branch-oriented parallel operation. Unlike the
372+
/// <see cref="ParallelAsync{T}(IReadOnlyList{Func{IDurableContext, CancellationToken, Task{T}}}, string?, ParallelConfig?, CancellationToken)"/>
373+
/// overloads — which take a complete branch list up front and share one result
374+
/// type — the returned <see cref="IDurableParallel"/> lets you register branches
375+
/// one at a time via
376+
/// <see cref="IDurableParallel.Branch{T}"/>,
377+
/// each with its own result type (heterogeneous), starting each branch as it is
378+
/// registered. Call <see cref="IDurableParallel.CompleteAsync(CancellationToken)"/>
379+
/// to seal registration and obtain the aggregate <see cref="IBatchResult"/>.
380+
/// </summary>
381+
/// <remarks>
382+
/// Use <c>await using</c> so the operation is sealed and its terminal checkpoint
383+
/// written even if <see cref="IDurableParallel.CompleteAsync(CancellationToken)"/>
384+
/// is not called. Branch identity is positional and deterministic across
385+
/// replays, so register the same branches in the same order every invocation —
386+
/// derive any dynamic branch set from a checkpointed step. Per-branch results
387+
/// are serialized via the <see cref="ILambdaSerializer"/> registered on
388+
/// <see cref="ILambdaContext.Serializer"/>. Honors the same
389+
/// <see cref="ParallelConfig.MaxConcurrency"/>,
390+
/// <see cref="ParallelConfig.CompletionConfig"/>, and
391+
/// <see cref="ParallelConfig.NestingType"/> as the homogeneous API.
392+
/// </remarks>
393+
/// <param name="name">
394+
/// Optional human-readable name for the parallel operation. It surfaces on the
395+
/// wire <c>OperationUpdate.Name</c> field and in execution traces. The
396+
/// deterministic operation ID is positional (derived from the call order, not
397+
/// from this name); however, when provided, the name becomes part of the
398+
/// operation's deterministic definition and is validated on replay — changing it
399+
/// across deployments for an in-flight execution throws
400+
/// <see cref="NonDeterministicExecutionException"/>, so keep it stable (or leave
401+
/// it <c>null</c>) for the life of an execution. Defaults to <c>null</c>.
402+
/// </param>
403+
/// <param name="config">
404+
/// Optional parallel configuration. Defaults are used when null.
405+
/// </param>
406+
/// <returns>
407+
/// An <see cref="IDurableParallel"/> for registering branches and awaiting the
408+
/// aggregate result.
409+
/// </returns>
410+
IDurableParallel CreateParallel(
411+
string? name = null,
412+
ParallelConfig? config = null);
413+
370414
/// <summary>
371415
/// Execute multiple branches concurrently. Each branch runs inside its own
372416
/// child context; per-branch results are aggregated into an

‎Libraries/src/Amazon.Lambda.DurableExecution/IDurableParallel.cs‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
namespace Amazon.Lambda.DurableExecution;
5+
6+
/// <summary>
7+
/// An incremental, branch-oriented parallel operation created by
8+
/// <see cref="IDurableContext.CreateParallel(string?, ParallelConfig?)"/>. Branches
9+
/// are registered one at a time via
10+
/// <see cref="Branch{T}"/>,
11+
/// each with its own result type (heterogeneous), and each begins executing
12+
/// immediately (subject to <see cref="ParallelConfig.MaxConcurrency"/>). Call
13+
/// <see cref="CompleteAsync(System.Threading.CancellationToken)"/> to seal
14+
/// registration, await the branches according to the
15+
/// <see cref="ParallelConfig.CompletionConfig"/>, and obtain the aggregate result.
16+
/// </summary>
17+
/// <remarks>
18+
/// This is an additive alternative to the homogeneous
19+
/// <see cref="IDurableContext.ParallelAsync{T}(System.Collections.Generic.IReadOnlyList{System.Func{IDurableContext, System.Threading.CancellationToken, System.Threading.Tasks.Task{T}}}, string?, ParallelConfig?, System.Threading.CancellationToken)"/>
20+
/// overloads, which accept a complete branch list up front and share one result
21+
/// type. Use <c>CreateParallel</c> when branches return unrelated types, or when
22+
/// branches are discovered incrementally (for example, tool calls derived from a
23+
/// checkpointed plan) and earlier branches should start before later ones are known.
24+
/// <para>
25+
/// <b>Deterministic replay.</b> Branch identity is positional: the n-th
26+
/// <see cref="Branch{T}"/>
27+
/// call reuses the n-th deterministic operation ID. Workflow code must therefore
28+
/// register the same branches in the same order across invocations — produce any
29+
/// dynamic branch list inside a checkpointed <see cref="IDurableContext.StepAsync{T}(System.Func{IStepContext, System.Threading.CancellationToken, System.Threading.Tasks.Task{T}}, string?, StepConfig?, System.Threading.CancellationToken)"/>
30+
/// so replay sees the same set.
31+
/// </para>
32+
/// <para>
33+
/// <b>Disposal.</b> <see cref="System.IAsyncDisposable.DisposeAsync"/> seals and
34+
/// completes the operation if
35+
/// <see cref="CompleteAsync(System.Threading.CancellationToken)"/> was not called,
36+
/// so an <c>await using</c> block always writes the parallel's terminal checkpoint.
37+
/// Calling <see cref="CompleteAsync(System.Threading.CancellationToken)"/>
38+
/// explicitly is recommended so you can capture the aggregate result.
39+
/// </para>
40+
/// </remarks>
41+
public interface IDurableParallel : IAsyncDisposable
42+
{
43+
/// <summary>
44+
/// Registers a branch and immediately begins executing it (respecting
45+
/// <see cref="ParallelConfig.MaxConcurrency"/>). Returns a typed
46+
/// <see cref="IParallelBranch{T}"/> handle for retrieving the branch's result.
47+
/// </summary>
48+
/// <remarks>
49+
/// The branch runs inside its own child context with a deterministic
50+
/// operation-ID space; its result is serialized to a checkpoint via the
51+
/// <see cref="Amazon.Lambda.Core.ILambdaSerializer"/> registered on
52+
/// <see cref="Amazon.Lambda.Core.ILambdaContext.Serializer"/>. Per-branch
53+
/// failures are captured on the handle and aggregated into the
54+
/// <see cref="CompleteAsync(System.Threading.CancellationToken)"/> result — a
55+
/// branch failure never throws out of this method.
56+
/// </remarks>
57+
/// <typeparam name="T">The branch's result type.</typeparam>
58+
/// <param name="name">
59+
/// Human-readable branch name. Required; surfaces on
60+
/// <c>OperationUpdate.Name</c> and must remain stable at a given branch index
61+
/// across deployments (a drift is a non-deterministic-execution error).
62+
/// </param>
63+
/// <param name="func">
64+
/// The branch body. Receives its own <see cref="IDurableContext"/> and a
65+
/// <see cref="System.Threading.CancellationToken"/> linking the SDK's
66+
/// workflow-shutdown signal with the operation's completion-policy
67+
/// short-circuit, and returns the branch's result.
68+
/// </param>
69+
/// <param name="serializer">
70+
/// Optional serializer for <em>this branch's</em> result payload. When
71+
/// <c>null</c> (default), the branch uses the operation-level serializer —
72+
/// <see cref="ParallelConfig.ItemSerializer"/> if set on
73+
/// <see cref="IDurableContext.CreateParallel(string?, ParallelConfig?)"/>, otherwise
74+
/// the globally-registered <see cref="Amazon.Lambda.Core.ILambdaSerializer"/> on
75+
/// <see cref="Amazon.Lambda.Core.ILambdaContext.Serializer"/>. Because each branch
76+
/// declares its own result type, a per-branch serializer lets one branch use a
77+
/// bespoke serializer (for example a source-generated context for AOT) without
78+
/// affecting sibling branches. It is part of the deterministic definition: the same
79+
/// branch must be able to deserialize a result it previously serialized, so pass the
80+
/// same serializer at a given branch index across replays.
81+
/// </param>
82+
/// <returns>A typed handle for awaiting the branch's result.</returns>
83+
/// <exception cref="System.ArgumentNullException">
84+
/// <paramref name="name"/> or <paramref name="func"/> is <c>null</c>.
85+
/// </exception>
86+
/// <exception cref="System.InvalidOperationException">
87+
/// The operation has already been sealed by
88+
/// <see cref="CompleteAsync(System.Threading.CancellationToken)"/>.
89+
/// </exception>
90+
/// <exception cref="System.ObjectDisposedException">
91+
/// The operation has already been disposed.
92+
/// </exception>
93+
/// <exception cref="NonDeterministicExecutionException">
94+
/// On replay, the <paramref name="name"/> at this branch index differs from the
95+
/// name recorded in the checkpoint for a previous invocation (branch name drift).
96+
/// </exception>
97+
IParallelBranch<T> Branch<T>(
98+
string name,
99+
Func<IDurableContext, CancellationToken, Task<T>> func,
100+
Amazon.Lambda.Core.ILambdaSerializer? serializer = null);
101+
102+
/// <summary>
103+
/// Seals registration (no further branches may be added), awaits the
104+
/// registered branches according to the
105+
/// <see cref="ParallelConfig.CompletionConfig"/>, checkpoints the aggregate
106+
/// outcome, and returns it. Idempotent — repeated calls return the same result.
107+
/// </summary>
108+
/// <remarks>
109+
/// Like the homogeneous parallel API, this never throws on per-branch failure:
110+
/// inspect <see cref="IBatchResult.HasFailure"/> /
111+
/// <see cref="IBatchResult.CompletionReason"/>, or await individual branch
112+
/// handles, to observe failures. It does propagate workflow-level errors (for
113+
/// example <see cref="NonDeterministicExecutionException"/>) and cancellation.
114+
/// <para>
115+
/// The <paramref name="cancellationToken"/> does not interrupt in-flight branch
116+
/// settlement. Because branches begin executing when they are registered (before
117+
/// <c>CompleteAsync</c> is called), this token is neither retroactively linked
118+
/// into already-running branch bodies nor into the wait for them to settle —
119+
/// those observe the SDK's workflow-shutdown signal (and the completion-policy
120+
/// short-circuit) instead. This call blocks until every dispatched branch reaches
121+
/// its terminal checkpoint; the token is observed only after that, so cancellation
122+
/// is surfaced just before the aggregate result would be returned rather than
123+
/// cutting the wait short. Dispatched branches always run to a terminal checkpoint
124+
/// so replay stays deterministic, matching <see cref="IDurableContext.ParallelAsync{T}(System.Collections.Generic.IReadOnlyList{System.Func{IDurableContext, System.Threading.CancellationToken, System.Threading.Tasks.Task{T}}}, string?, ParallelConfig?, System.Threading.CancellationToken)"/>.
125+
/// </para>
126+
/// </remarks>
127+
/// <param name="cancellationToken">A token to observe for cancellation.</param>
128+
/// <returns>The aggregate <see cref="IBatchResult"/> summarizing branch outcomes.</returns>
129+
Task<IBatchResult> CompleteAsync(CancellationToken cancellationToken = default);
130+
}

‎Libraries/src/Amazon.Lambda.DurableExecution/IParallelBranch.cs‎

Lines changed: 65 additions & 0 deletions

0 commit comments

Comments
 (0)