GitHub user da-daken added a comment to the discussion: Parallel Tool Call
Execution
I'm glad that we've converged on the API structure and are now discussing the
detailed recovery flow in depth.
Based on our current consensus, the batch recovery flow should be: **reserve
slots → execute in parallel → finalize in order**.
### Limitations of the current implementation
The existing primitives are designed for **serial, single‑index progression**:
- `appendPendingCall` appends only one PENDING record at the tail and requires
`currentCallIndex == recoveryCallResults.size()`. It cannot reserve multiple
slots upfront.
- `finalizeCurrentCall` finalises only the slot pointed to by the current
cursor. It does not support writing back results by absolute index after the
whole batch has completed.
So the two requirements – “reserve N slots ahead of time” and “fill them back
in order after all tools finish” – cannot be satisfied with the current APIs.
**Using explicit indexes** is a clean way to address this.
### Three new methods
I suggest adding the following methods to `RunnerContext` /
`DurableExecutionContext`:
| New method | Purpose |
|------------|---------|
| `reservePendingBatch(List<String> ids, String digest)` | Write N PENDING
records at the tail in one go; cursor does **not** move. |
| `finalizeCallAt(int index, ...)` | Write a terminal state (SUCCESS/FAILURE)
at the given absolute index. |
| `advanceCallIndexBy(int n)` | Advance the cursor by `n` steps after the
entire batch is finalised. |
### Demo implementation
```java
@Override
public <T> List<T> durableExecuteAllAsync(List<DurableCallable<T>> callables)
throws Exception {
Preconditions.checkState(durableExecutionContext != null, "...");
if (callables.isEmpty()) return List.of();
String argsDigest = "";
int base = durableExecutionContext.getCurrentCallIndex();
int n = callables.size();
// ---- Phase 1: scan each slot in [base, base+n) ----
List<Plan<T>> plans = new ArrayList<>(n);
for (int i = 0; i < n; i++) {
CallResult slot = durableExecutionContext.getCallResultAt(base + i);
DurableCallable<T> c = callables.get(i);
if (slot == null) {
plans.add(Plan.submit(c)); // no record →
need to run
} else if (!slot.matches(c.getId(), argsDigest)) {
// clear from the mismatch point onward (not all)
durableExecutionContext.clearCallResultsFrom(base + i);
durableExecutionContext.reservePendingBatch(idsFrom(callables, i),
argsDigest);
for (int j = i; j < n; j++)
plans.add(Plan.submit(callables.get(j)));
break;
} else if (slot.isSuccess() || slot.isFailure()) {
plans.add(Plan.cached(slot, c.getResultClass())); // hit → use
directly
} else { // PENDING
plans.add(c.reconciler() != null ? Plan.reconcile(c) :
Plan.submit(c));
}
}
// ---- Phase 2: reserve only on a fresh run (on recovery, slots already
exist) ----
if (isFreshRun(plans, base, n)) {
durableExecutionContext.reservePendingBatch(allIds(callables),
argsDigest);
}
// ---- Phase 3: fan‑out – submit only submit/reconcile plans, yield until
all finish ----
Map<Integer, T> asyncResults =
continuationExecutor.executeAllAsync(continuationContext,
toSuppliers(plans));
// ---- Phase 4: fan‑in – finalise strictly in i=0..n-1 order (mailbox
thread) ----
List<T> results = new ArrayList<>(n);
for (int i = 0; i < n; i++) {
Outcome<T> o = plans.get(i).materialize(asyncResults.get(i));
durableExecutionContext.finalizeCallAt(base + i,
callables.get(i).getId(), argsDigest,
serializeDurableResult(o.result()),
serializeDurableException(o.exception()));
results.add(o.result()); // collect‑all: store result; exception
handling is left to ToolCallAction
}
durableExecutionContext.advanceCallIndexBy(n);
return results;
}
```
### TimeOut
I fully agree with the timeout mechanism. I think two aspects are needed: one
is the timeout for a single tool, which can be added via a configurable timeout
method on DurableCallable. However, this has a broader scope, and I think we
can leave it out of the current discussion for now. The other is the batch
timeout, which can be added as a user‑configurable parameter – I believe this
can be done in the current work. Below is the revised API for
ContinuationActionExecutor based on the first design:
```java
public <T> List<T> executeAllAsync(
ContinuationContext ctx,
List<Supplier<T>> suppliers,
Duration timeout);
```
At the same time, ContinuationContext will be supplemented to support List
```java
// core waiting logic
while (true) {
boolean allDone = futures.stream().allMatch(Future::isDone);
if (allDone) break;
if (hasTimeout && System.nanoTime() >= deadline) {
futures.stream().filter(f -> !f.isDone()).forEach(f -> {
f.cancel(true);
});
break;
}
Continuation.yield(SCOPE);
}
```
GitHub link:
https://github.com/apache/flink-agents/discussions/855#discussioncomment-17575200
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
