da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3675774615
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java:
##########
@@ -75,6 +102,159 @@ private <T> T durableExecuteAsyncWithReconcile(
callable, reconcileCallable, () ->
executeAsyncCallable(callable));
}
+ @Override
+ public <T> List<Outcome<T>>
durableExecuteAllAsync(List<DurableCallable<T>> callables)
+ throws Exception {
+ if (callables.isEmpty()) {
+ return List.of();
+ }
+ if (durableExecutionContext == null) {
+ return executeAllWithoutDurableState(callables);
+ }
+
+ String argsDigest = "";
+ int base = durableExecutionContext.getCurrentCallIndex();
+ BatchExecutionPlan<T> plan = buildBatchExecutionPlan(callables, base,
argsDigest);
+
+ reservePendingBatchIfNeeded(callables, argsDigest, plan);
+
+ List<Outcome<T>> executed = executeOutcomeSuppliers(plan.suppliers);
+ finalizeExecutedOutcomes(callables, base, argsDigest, plan, executed);
+
+ advanceCallIndexBy(callables.size());
+ return plan.outcomes;
+ }
+
+ private <T> BatchExecutionPlan<T> buildBatchExecutionPlan(
+ List<DurableCallable<T>> callables, int base, String argsDigest)
throws Exception {
+ BatchExecutionPlan<T> plan = new
BatchExecutionPlan<>(callables.size());
+ for (int i = 0; i < callables.size(); i++) {
+ DurableCallable<T> callable = callables.get(i);
+ CallResult current = getCallResultAt(base + i);
+ if (current == null) {
+ markReservationStart(plan, i);
+ addExecutableCall(plan, i, callable::call);
+ continue;
+ }
+ if (!current.matches(callable.getId(), argsDigest)) {
+ clearCallResultsFromAndPersist(base + i);
+ plan.needsReservation = true;
+ plan.executionStart = i;
+ addExecutableCall(plan, i, callable::call);
+ appendRemainingExecutions(callables, plan, i + 1);
+ break;
+ }
+ if (current.isPending()) {
+ Callable<T> reconcileCallable = callable.reconciler();
+ Callable<T> executionCallable =
+ reconcileCallable != null ? reconcileCallable :
callable::call;
+ addExecutableCall(plan, i, executionCallable);
+ } else {
+ plan.outcomes.add(
+ readTerminalOutcomeAt(
+ base + i, callable.getId(), argsDigest,
callable.getResultClass()));
+ }
+ }
+ return plan;
+ }
+
+ private <T> void markReservationStart(BatchExecutionPlan<T> plan, int
callIndex) {
+ plan.needsReservation = true;
+ if (plan.executionStart < 0) {
+ plan.executionStart = callIndex;
+ }
+ }
+
+ private <T> void addExecutableCall(
+ BatchExecutionPlan<T> plan, int callIndex, Callable<T>
executionCallable) {
+ plan.outcomes.add(null);
+ plan.suppliers.add(executionCallable);
+ plan.executableCallIndexes.add(callIndex);
+ }
+
+ private <T> void appendRemainingExecutions(
+ List<DurableCallable<T>> callables, BatchExecutionPlan<T> plan,
int startIndex) {
+ for (int i = startIndex; i < callables.size(); i++) {
+ DurableCallable<T> remaining = callables.get(i);
+ addExecutableCall(plan, i, remaining::call);
+ }
+ }
+
+ private <T> void reservePendingBatchIfNeeded(
+ List<DurableCallable<T>> callables, String argsDigest,
BatchExecutionPlan<T> plan) {
+ if (!plan.needsReservation) {
+ return;
+ }
+ List<String> ids = new ArrayList<>();
+ for (DurableCallable<T> callable :
+ callables.subList(plan.executionStart, callables.size())) {
+ ids.add(callable.getId());
+ }
+ reservePendingBatch(ids, argsDigest);
Review Comment:
Yes, I'll mark it as 'no result read' during the read operation and include
this in the next commit. I'm a bit tight on time at the moment, so I plan to
push it over the weekend. We can revisit it then～
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
