Skip to content

Commit 665a411

Browse files
l46kokcopybara-github
authored andcommitted
Remove cycleId tracking from completion coordinator
PiperOrigin-RevId: 975452308
1 parent 248b623 commit 665a411

2 files changed

Lines changed: 103 additions & 113 deletions

File tree

‎runtime/src/main/java/dev/cel/runtime/planner/AsyncCompletionCoordinator.java‎

Lines changed: 54 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -87,9 +87,7 @@ enum WaitResult {
8787

8888
private final ThreadLocal<Deque<Runnable>> continuationTrampoline;
8989

90-
@GuardedBy("lock")
91-
private long cycleId;
92-
90+
/** Monotonic staleness token bumped on every drain and snapshot. */
9391
@GuardedBy("lock")
9492
private long debounceGeneration;
9593

@@ -118,25 +116,25 @@ void callCompleted(CelAsyncCall call) {
118116
CompletionSnapshot snapshot = null;
119117

120118
synchronized (lock) {
119+
// Check cancellation before activeCount() so late-finishing calls do not trip the unbalanced
120+
// fail-safe on an evaluation that was already cancelled.
121+
if (isCancelled) {
122+
gate.release();
123+
return;
124+
}
121125
// Check activeCount() <= 0 BEFORE release() to detect unbalanced completion misuse.
122126
if (gate.activeCount() <= 0) {
123127
unbalanced = true;
124128
} else {
125129
gate.release();
126-
if (isCancelled) {
127-
return;
128-
}
129130
completedBatch.add(call);
130131
if (!isWaiting) {
131132
return;
132133
}
133134
// Take inFlight AFTER release() to capture remaining active calls for the drain strategy.
134135
snapshot =
135136
new CompletionSnapshot(
136-
ImmutableList.copyOf(completedBatch),
137-
gate.activeCount(),
138-
cycleId,
139-
++debounceGeneration);
137+
ImmutableList.copyOf(completedBatch), gate.activeCount(), ++debounceGeneration);
140138
}
141139
}
142140

@@ -155,7 +153,7 @@ void callCompleted(CelAsyncCall call) {
155153
failAndCancel(t);
156154
return;
157155
}
158-
applyDrainAction(action, snapshot.cycleId, snapshot.debounceGeneration);
156+
applyDrainAction(action, snapshot.debounceGeneration);
159157
}
160158

161159
/**
@@ -189,10 +187,7 @@ WaitResult waitForCompletions(Runnable continuationCallback) {
189187

190188
snapshot =
191189
new CompletionSnapshot(
192-
ImmutableList.copyOf(completedBatch),
193-
gate.activeCount(),
194-
this.cycleId,
195-
++this.debounceGeneration);
190+
ImmutableList.copyOf(completedBatch), gate.activeCount(), ++this.debounceGeneration);
196191
}
197192

198193
CelAsyncDrainAction action;
@@ -207,6 +202,7 @@ WaitResult waitForCompletions(Runnable continuationCallback) {
207202
}
208203

209204
boolean reevaluateNow = false;
205+
ScheduledFuture<?> timerToCancel = null;
210206
synchronized (lock) {
211207
if (isCancelled) {
212208
return WaitResult.CANCELLED;
@@ -218,13 +214,16 @@ WaitResult waitForCompletions(Runnable continuationCallback) {
218214
if (action.shouldReevaluate() || gate.activeCount() == 0) {
219215
// The continuation in DrainResult is intentionally not dispatched here because
220216
// WaitResult.REEVALUATE_NOW instructs the calling thread to re-evaluate synchronously.
221-
DrainResult unused = drainAndResetUnderLock();
217+
timerToCancel = drainAndResetUnderLock().timer;
222218
reevaluateNow = true;
223219
}
224220
} else {
225221
return WaitResult.REGISTERED;
226222
}
227223
}
224+
if (timerToCancel != null) {
225+
timerToCancel.cancel(false);
226+
}
228227
if (reevaluateNow) {
229228
return WaitResult.REEVALUATE_NOW;
230229
}
@@ -240,20 +239,21 @@ WaitResult waitForCompletions(Runnable continuationCallback) {
240239
failAndCancel(e);
241240
return WaitResult.CANCELLED;
242241
}
243-
return scheduleDebounce(delayNanos, snapshot.cycleId, snapshot.debounceGeneration)
242+
return scheduleDebounce(delayNanos, snapshot.debounceGeneration)
244243
? WaitResult.REGISTERED
245244
: WaitResult.CANCELLED;
246245
}
247246

248-
private void applyDrainAction(
249-
CelAsyncDrainAction action, long expectedCycleId, long expectedGen) {
250-
boolean shouldReevaluate;
247+
private void applyDrainAction(CelAsyncDrainAction action, long expectedGen) {
251248
DrainResult drainResult = null;
252249
synchronized (lock) {
250+
// Bail out if a newer generation has superseded this action.
251+
if (!isCurrentUnderLock(expectedGen)) {
252+
return;
253+
}
253254
// Live read: check gate.activeCount() == 0 under lock so we do not schedule an unnecessary
254255
// timer if all remaining calls completed while evaluating nextAction().
255-
shouldReevaluate = action.shouldReevaluate() || gate.activeCount() == 0;
256-
if (shouldReevaluate && isCurrentUnderLock(expectedCycleId, expectedGen)) {
256+
if (action.shouldReevaluate() || gate.activeCount() == 0) {
257257
drainResult = drainAndResetUnderLock();
258258
}
259259
}
@@ -266,9 +266,6 @@ private void applyDrainAction(
266266
}
267267
return;
268268
}
269-
if (shouldReevaluate) {
270-
return;
271-
}
272269

273270
Duration waitDuration = action.waitDuration();
274271
if (!waitDuration.isZero()) {
@@ -279,13 +276,13 @@ private void applyDrainAction(
279276
failAndCancel(e);
280277
return;
281278
}
282-
boolean unusedScheduled = scheduleDebounce(delayNanos, expectedCycleId, expectedGen);
279+
scheduleDebounce(delayNanos, expectedGen);
283280
return;
284281
}
285282

286283
ScheduledFuture<?> timerToCancel = null;
287284
synchronized (lock) {
288-
if (isCurrentUnderLock(expectedCycleId, expectedGen)) {
285+
if (isCurrentUnderLock(expectedGen)) {
289286
timerToCancel = cancelDebounceTimerUnderLock();
290287
}
291288
}
@@ -300,21 +297,39 @@ private void applyDrainAction(
300297
* @return false if the timer could not be scheduled, in which case the coordinator has already
301298
* been failed and cancelled.
302299
*/
303-
private boolean scheduleDebounce(long nanos, long scheduledCycleId, long scheduledGen) {
300+
private boolean scheduleDebounce(long nanos, long scheduledGen) {
301+
synchronized (lock) {
302+
if (isCancelled) {
303+
return false;
304+
}
305+
if (!isCurrentUnderLock(scheduledGen)) {
306+
return true;
307+
}
308+
}
304309
ScheduledFuture<?> future;
305310
try {
306311
future =
307312
options
308313
.resolveScheduledExecutorService()
309-
.schedule(() -> onDebounceFired(scheduledCycleId, scheduledGen), nanos, NANOSECONDS);
314+
.schedule(() -> onDebounceFired(scheduledGen), nanos, NANOSECONDS);
310315
} catch (Throwable t) {
311-
failAndCancel(t);
312-
return false;
316+
boolean shouldFail;
317+
synchronized (lock) {
318+
if (isCancelled) {
319+
return false;
320+
}
321+
shouldFail = isCurrentUnderLock(scheduledGen);
322+
}
323+
if (shouldFail) {
324+
failAndCancel(t);
325+
return false;
326+
}
327+
return true;
313328
}
314329

315330
ScheduledFuture<?> redundantFuture = null;
316331
synchronized (lock) {
317-
if (isCurrentUnderLock(scheduledCycleId, scheduledGen)) {
332+
if (isCurrentUnderLock(scheduledGen)) {
318333
if (debounceTimer != null) {
319334
redundantFuture = debounceTimer;
320335
}
@@ -330,22 +345,19 @@ private boolean scheduleDebounce(long nanos, long scheduledCycleId, long schedul
330345
}
331346

332347
/**
333-
* Returns true if the coordinator is still actively waiting on the cycle and debounce generation
334-
* that produced the in-flight action, meaning the action is not stale.
348+
* Returns true if the coordinator is still actively waiting on the debounce generation that
349+
* produced the in-flight action, meaning the action is not stale.
335350
*/
336351
@GuardedBy("lock")
337-
private boolean isCurrentUnderLock(long expectedCycleId, long expectedGen) {
338-
return !isCancelled
339-
&& isWaiting
340-
&& this.cycleId == expectedCycleId
341-
&& this.debounceGeneration == expectedGen;
352+
private boolean isCurrentUnderLock(long expectedGen) {
353+
return !isCancelled && isWaiting && this.debounceGeneration == expectedGen;
342354
}
343355

344356
@VisibleForTesting
345-
void onDebounceFired(long firedCycleId, long firedGen) {
357+
void onDebounceFired(long firedGen) {
346358
DrainResult drainResult = null;
347359
synchronized (lock) {
348-
if (isCurrentUnderLock(firedCycleId, firedGen)) {
360+
if (isCurrentUnderLock(firedGen)) {
349361
drainResult = drainAndResetUnderLock();
350362
}
351363
}
@@ -429,7 +441,6 @@ private void failAndCancel(Throwable t) {
429441
@GuardedBy("lock")
430442
@CheckReturnValue
431443
private DrainResult drainAndResetUnderLock() {
432-
cycleId++;
433444
debounceGeneration++;
434445
isWaiting = false;
435446
completedBatch.clear();
@@ -474,13 +485,6 @@ boolean hasScheduledDebounceTimer() {
474485
}
475486
}
476487

477-
@VisibleForTesting
478-
long cycleId() {
479-
synchronized (lock) {
480-
return cycleId;
481-
}
482-
}
483-
484488
@VisibleForTesting
485489
long debounceGeneration() {
486490
synchronized (lock) {
@@ -498,14 +502,12 @@ boolean isCancelled() {
498502
private static final class CompletionSnapshot {
499503
final ImmutableList<CelAsyncCall> batch;
500504
final int inFlight;
501-
final long cycleId;
502505
final long debounceGeneration;
503506

504507
private CompletionSnapshot(
505-
ImmutableList<CelAsyncCall> batch, int inFlight, long cycleId, long debounceGeneration) {
508+
ImmutableList<CelAsyncCall> batch, int inFlight, long debounceGeneration) {
506509
this.batch = checkNotNull(batch, "batch must not be null");
507510
this.inFlight = inFlight;
508-
this.cycleId = cycleId;
509511
this.debounceGeneration = debounceGeneration;
510512
}
511513
}
@@ -539,10 +541,5 @@ protected Deque<Runnable> initialValue() {
539541
return new ArrayDeque<>();
540542
}
541543
};
542-
this.isWaiting = false;
543-
this.isCancelled = false;
544-
this.failureReported = false;
545-
this.cycleId = 0;
546-
this.debounceGeneration = 0;
547544
}
548545
}

0 commit comments

Comments
 (0)