Skip to content

Commit 449d741

Browse files
jhaltermanjhalterman
committed
Add better support for potentially blocked operations
- Add better support for potentially blocked operations, such as when scheduled executions, fallbacks, or retries are blocked due to thread limitations. Ensure that promises still complete as expected. - Add test coverage for blocked operation scenarios. - Add test for issue 260 - Add test for issue 231
1 parent 65f4bb1 commit 449d741

15 files changed

+144
-79
lines changed

‎src/main/java/net/jodah/failsafe/AbstractExecution.java

+16-8
Original file line numberDiff line numberDiff line change
@@ -34,11 +34,15 @@ public abstract class AbstractExecution extends ExecutionContext {
3434
final FailsafeExecutor<Object> executor;
3535
final List<PolicyExecutor<Policy<Object>>> policyExecutors;
3636

37+
enum Status {
38+
NOT_RUNNING, RUNNING, TIMED_OUT
39+
}
40+
3741
// Internally mutable state
42+
/* The status of an execution */
43+
volatile Status status = Status.NOT_RUNNING;
3844
/* Whether the execution attempt has been recorded */
3945
volatile boolean attemptRecorded;
40-
/* Whether the execution result has been recorded */
41-
volatile boolean executionRecorded;
4246
/* Whether a result has been post-executed */
4347
volatile boolean resultHandled;
4448
/* Whether the execution can be interrupted */
@@ -73,15 +77,19 @@ public abstract class AbstractExecution extends ExecutionContext {
7377
* @throws IllegalStateException if the execution is already complete
7478
*/
7579
void record(ExecutionResult result) {
80+
record(result, false);
81+
}
82+
83+
void record(ExecutionResult result, boolean timeout) {
7684
Assert.state(!completed, "Execution has already been completed");
7785
if (!interrupted) {
7886
recordAttempt();
79-
if (!executionRecorded) {
87+
if (Status.RUNNING.equals(status)) {
88+
lastResult = result.getResult();
89+
lastFailure = result.getFailure();
8090
executions.incrementAndGet();
81-
executionRecorded = true;
91+
status = timeout ? Status.TIMED_OUT : Status.NOT_RUNNING;
8292
}
83-
lastResult = result.getResult();
84-
lastFailure = result.getFailure();
8593
}
8694
}
8795

@@ -96,12 +104,12 @@ void recordAttempt() {
96104
}
97105
}
98106

99-
void preExecute() {
107+
synchronized void preExecute() {
100108
attemptStartTime = Duration.ofNanos(System.nanoTime());
101109
if (startTime == Duration.ZERO)
102110
startTime = attemptStartTime;
111+
status = Status.RUNNING;
103112
attemptRecorded = false;
104-
executionRecorded = false;
105113
resultHandled = false;
106114
cancelledIndex = 0;
107115
canInterrupt = true;

‎src/main/java/net/jodah/failsafe/AsyncExecution.java

+6-3
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121

2222
import java.util.concurrent.CompletableFuture;
2323
import java.util.concurrent.CompletionException;
24+
import java.util.concurrent.Future;
2425
import java.util.concurrent.TimeUnit;
2526
import java.util.function.Supplier;
2627

@@ -46,7 +47,7 @@ <T> AsyncExecution(Scheduler scheduler, FailsafeFuture<T> future, FailsafeExecut
4647

4748
void inject(Supplier<CompletableFuture<ExecutionResult>> syncSupplier, boolean asyncExecution) {
4849
if (!asyncExecution) {
49-
outerExecutionSupplier = Functions.getPromiseAsync(syncSupplier, scheduler, future);
50+
outerExecutionSupplier = Functions.getPromiseAsync(syncSupplier, scheduler, this);
5051
} else {
5152
outerExecutionSupplier = innerExecutionSupplier = Functions.toSettableSupplier(syncSupplier);
5253
}
@@ -181,8 +182,10 @@ ExecutionResult postExecute(ExecutionResult result) {
181182
void executeAsync(boolean asyncExecution) {
182183
if (!asyncExecution)
183184
outerExecutionSupplier.get().whenComplete(this::complete);
184-
else
185-
future.injectPolicy(scheduler.schedule(innerExecutionSupplier::get, 0, TimeUnit.NANOSECONDS));
185+
else {
186+
Future<?> scheduledSupply = scheduler.schedule(innerExecutionSupplier::get, 0, TimeUnit.NANOSECONDS);
187+
future.injectCancelFn((mayInterrupt, result) -> scheduledSupply.cancel(mayInterrupt));
188+
}
186189
}
187190

188191
/**

‎src/main/java/net/jodah/failsafe/FailsafeExecutor.java

+3-2
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,8 @@
3131
/**
3232
* <p>
3333
* An executor that handles failures according to configured {@link FailurePolicy policies}. Can be created via {@link
34-
* Failsafe#with(Policy[])}.
34+
* Failsafe#with(Policy, Policy[])} to support policy based execution failure handling, or {@link Failsafe#none()} to
35+
* support execution with no failure handling.
3536
* <p>
3637
* Async executions are run by default on the {@link ForkJoinPool#commonPool()}. Alternative executors can be configured
3738
* via {@link #with(ScheduledExecutorService)} and similar methods. All async executions are cancellable and
@@ -402,7 +403,7 @@ private <T> T call(Function<Execution, CheckedSupplier<?>> supplierFn) {
402403
* @throws NullPointerException if the {@code supplierFn} is null
403404
* @throws RejectedExecutionException if the {@code supplierFn} cannot be scheduled for execution
404405
*/
405-
@SuppressWarnings("unchecked")
406+
@SuppressWarnings({ "unchecked", "rawtypes" })
406407
private <T> CompletableFuture<T> callAsync(
407408
Function<AsyncExecution, Supplier<CompletableFuture<ExecutionResult>>> supplierFn, boolean asyncExecution) {
408409
FailsafeFuture<T> future = new FailsafeFuture(this);

‎src/main/java/net/jodah/failsafe/FailsafeFuture.java

+10-25
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import java.util.concurrent.CancellationException;
2121
import java.util.concurrent.CompletableFuture;
2222
import java.util.concurrent.Future;
23+
import java.util.function.BiConsumer;
2324

2425
/**
2526
* A CompletableFuture implementation that propagates cancellations and calls completion handlers.
@@ -34,9 +35,8 @@ public class FailsafeFuture<T> extends CompletableFuture<T> {
3435
private AbstractExecution execution;
3536

3637
// Mutable state, guarded by "this"
37-
private Future<T> policyExecFuture;
3838
private Future<?> dependentStageFuture;
39-
private Runnable cancelFn;
39+
private BiConsumer<Boolean, ExecutionResult> cancelFn;
4040
private List<Future<T>> timeoutFutures;
4141
private boolean cancelWithInterrupt;
4242

@@ -72,7 +72,7 @@ public synchronized boolean cancel(boolean mayInterruptIfRunning) {
7272
this.cancelWithInterrupt = mayInterruptIfRunning;
7373
execution.cancelledIndex = Integer.MAX_VALUE;
7474
boolean cancelResult = super.cancel(mayInterruptIfRunning);
75-
cancelResult = cancelDependencies(mayInterruptIfRunning, cancelResult);
75+
cancelDependencies(mayInterruptIfRunning, null);
7676
ExecutionResult result = ExecutionResult.failure(new CancellationException());
7777
super.completeExceptionally(result.getFailure());
7878
executor.handleComplete(result, execution);
@@ -98,46 +98,31 @@ synchronized boolean completeResult(ExecutionResult result) {
9898
return completed;
9999
}
100100

101-
synchronized Future<T> getDependency() {
102-
return policyExecFuture;
103-
}
104-
105101
synchronized List<Future<T>> getTimeoutDelegates() {
106102
return timeoutFutures;
107103
}
108104

109105
/**
110-
* Cancels the dependency passing in the {@code interruptDelegate} flag, applies the retry cancel fn, and cancels all
106+
* Cancels the dependency passing in the {@code mayInterrupt} flag, applies the retry cancel fn, and cancels all
111107
* timeout dependencies.
112108
*/
113-
synchronized boolean cancelDependencies(boolean interruptDelegate, boolean result) {
114-
execution.interrupted = interruptDelegate;
115-
if (policyExecFuture != null)
116-
result = policyExecFuture.cancel(interruptDelegate);
109+
synchronized void cancelDependencies(boolean mayInterrupt, ExecutionResult cancelResult) {
110+
execution.interrupted = mayInterrupt;
117111
if (dependentStageFuture != null)
118-
dependentStageFuture.cancel(interruptDelegate);
119-
if (cancelFn != null)
120-
cancelFn.run();
112+
dependentStageFuture.cancel(mayInterrupt);
121113
if (timeoutFutures != null) {
122114
for (Future<T> timeoutDelegate : timeoutFutures)
123115
timeoutDelegate.cancel(false);
124116
timeoutFutures.clear();
125117
}
126-
return result;
118+
if (cancelFn != null)
119+
cancelFn.accept(mayInterrupt, cancelResult);
127120
}
128121

129122
void inject(AbstractExecution execution) {
130123
this.execution = execution;
131124
}
132125

133-
/**
134-
* Injects a {@code policyExecFuture} to be cancelled when this future is cancelled.
135-
*/
136-
@SuppressWarnings({ "unchecked", "rawtypes" })
137-
synchronized void injectPolicy(Future<?> policyExecFuture) {
138-
this.policyExecFuture = (Future) policyExecFuture;
139-
}
140-
141126
/**
142127
* Injects a {@code dependentStageFuture} to be cancelled when this future is cancelled.
143128
*/
@@ -152,7 +137,7 @@ synchronized void injectStage(Future<?> dependentStageFuture) {
152137
/**
153138
* Injects a {@code cancelFn} to be called when this future is cancelled.
154139
*/
155-
synchronized void injectCancelFn(Runnable cancelFn) {
140+
synchronized void injectCancelFn(BiConsumer<Boolean, ExecutionResult> cancelFn) {
156141
this.cancelFn = cancelFn;
157142
}
158143

‎src/main/java/net/jodah/failsafe/FallbackExecutor.java

+4-5
Original file line numberDiff line numberDiff line change
@@ -96,12 +96,11 @@ protected Supplier<CompletableFuture<ExecutionResult>> supplyAsync(
9696
else {
9797
Future<?> scheduledFallback = scheduler.schedule(callable, 0, TimeUnit.NANOSECONDS);
9898

99-
// Propagate cancellation to the scheduled retry and promise
100-
future.injectCancelFn(() -> {
101-
System.out.println("cancelling scheduled fallback isdone: " + scheduledFallback.isDone());
102-
scheduledFallback.cancel(false);
99+
// Propagate cancellation to the scheduled fallback and its promise
100+
future.injectCancelFn((mayInterrupt, promiseResult) -> {
101+
scheduledFallback.cancel(mayInterrupt);
103102
if (executionCancelled())
104-
promise.complete(null);
103+
promise.complete(promiseResult);
105104
});
106105
}
107106
} catch (Throwable t) {

‎src/main/java/net/jodah/failsafe/Functions.java

+13-2
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
*/
1616
package net.jodah.failsafe;
1717

18+
import net.jodah.failsafe.AbstractExecution.Status;
1819
import net.jodah.failsafe.function.*;
1920
import net.jodah.failsafe.internal.util.Assert;
2021
import net.jodah.failsafe.util.concurrent.Scheduler;
@@ -90,7 +91,8 @@ static <T> Supplier<CompletableFuture<ExecutionResult>> getPromise(ContextualSup
9091
* calls, and returns a promise containing the result.
9192
*/
9293
static Supplier<CompletableFuture<ExecutionResult>> getPromiseAsync(
93-
Supplier<CompletableFuture<ExecutionResult>> supplier, Scheduler scheduler, FailsafeFuture<Object> future) {
94+
Supplier<CompletableFuture<ExecutionResult>> supplier, Scheduler scheduler, AsyncExecution execution) {
95+
9496
AtomicBoolean scheduled = new AtomicBoolean();
9597
return () -> {
9698
if (scheduled.get()) {
@@ -106,7 +108,16 @@ static Supplier<CompletableFuture<ExecutionResult>> getPromiseAsync(
106108

107109
try {
108110
scheduled.set(true);
109-
future.injectPolicy(scheduler.schedule(callable, 0, TimeUnit.NANOSECONDS));
111+
Future<?> scheduledSupply = scheduler.schedule(callable, 0, TimeUnit.NANOSECONDS);
112+
113+
// Propagate cancellation to the scheduled supplier and its promise
114+
execution.future.injectCancelFn((mayInterrupt, cancelResult) -> {
115+
scheduledSupply.cancel(mayInterrupt);
116+
117+
// Cancel a pending promise if the execution is not yet running
118+
if (Status.NOT_RUNNING.equals(execution.status))
119+
promise.complete(cancelResult);
120+
});
110121
} catch (Throwable t) {
111122
promise.completeExceptionally(t);
112123
}

‎src/main/java/net/jodah/failsafe/PolicyExecutor.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,8 @@
3131
public abstract class PolicyExecutor<P extends Policy> {
3232
protected final P policy;
3333
protected final AbstractExecution execution;
34-
/* Index of the policy relative to other policies in a composition, inner-most first */ int policyIndex;
34+
// Index of the policy relative to other policies in a composition, inner-most first
35+
int policyIndex;
3536

3637
protected PolicyExecutor(P policy, AbstractExecution execution) {
3738
this.policy = policy;

‎src/main/java/net/jodah/failsafe/RetryPolicyExecutor.java

+14-6
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import java.time.Duration;
2222
import java.util.concurrent.Callable;
2323
import java.util.concurrent.CompletableFuture;
24+
import java.util.concurrent.Future;
2425
import java.util.concurrent.TimeUnit;
2526
import java.util.function.Supplier;
2627

@@ -114,14 +115,18 @@ public Object call() {
114115
supplier.get().whenComplete((result, error) -> {
115116
if (error != null)
116117
promise.completeExceptionally(error);
117-
else if (result != null) {
118+
else if (result == null)
119+
promise.complete(null);
120+
else {
118121
if (retriesExceeded || executionCancelled()) {
119122
promise.complete(result);
120123
} else {
121124
postExecuteAsync(result, scheduler, future).whenComplete((postResult, postError) -> {
122125
if (postError != null)
123126
promise.completeExceptionally(postError);
124-
else if (postResult != null) {
127+
else if (postResult == null)
128+
promise.complete(null);
129+
else {
125130
if (postResult.isComplete() || executionCancelled()) {
126131
promise.complete(postResult);
127132
} else {
@@ -133,11 +138,14 @@ else if (postResult != null) {
133138
retryScheduledListener.handle(postResult, execution);
134139

135140
previousResult = postResult;
136-
future.injectPolicy(scheduler.schedule(this, postResult.getWaitNanos(), TimeUnit.NANOSECONDS));
137-
future.injectCancelFn(() -> {
138-
// Ensure that the promise completes if a scheduled retry is cancelled
141+
Future<?> scheduledRetry = scheduler.schedule(this, postResult.getWaitNanos(),
142+
TimeUnit.NANOSECONDS);
143+
144+
// Propagate cancellation to the scheduled retry and its promise
145+
future.injectCancelFn((mayInterrupt, cancelResult) -> {
146+
scheduledRetry.cancel(mayInterrupt);
139147
if (executionCancelled())
140-
promise.complete(null);
148+
promise.complete(cancelResult);
141149
});
142150
} catch (Throwable t) {
143151
// Hard scheduling failure

‎src/main/java/net/jodah/failsafe/TimeoutExecutor.java

+6-6
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ protected Supplier<ExecutionResult> supply(Supplier<ExecutionResult> supplier, S
7373
// Guard against race with the execution completing
7474
synchronized (execution) {
7575
if (execution.canInterrupt) {
76-
execution.record(result.get());
76+
execution.record(result.get(), true);
7777
execution.interrupted = true;
7878
executionThread.interrupt();
7979
}
@@ -115,17 +115,17 @@ protected Supplier<CompletableFuture<ExecutionResult>> supplyAsync(
115115
try {
116116
// Schedule timeout check
117117
timeoutFuture.set(Scheduler.DEFAULT.schedule(() -> {
118-
// Guard against race with execution completion
119-
if (executionResult.compareAndSet(null,
120-
ExecutionResult.failure(new TimeoutExceededException(policy)))) {
118+
ExecutionResult cancelResult = ExecutionResult.failure(new TimeoutExceededException(policy));
121119

120+
// Guard against race with execution completion
121+
if (executionResult.compareAndSet(null, cancelResult)) {
122122
boolean canInterrupt = policy.canInterrupt();
123123
if (canInterrupt)
124-
execution.record(executionResult.get());
124+
execution.record(executionResult.get(), true);
125125

126126
// Cancel and interrupt
127127
execution.cancelledIndex = policyIndex;
128-
future.cancelDependencies(canInterrupt, false);
128+
future.cancelDependencies(canInterrupt, cancelResult);
129129
}
130130
return null;
131131
}, policy.getTimeout().toNanos(), TimeUnit.NANOSECONDS));

‎src/test/java/net/jodah/failsafe/AbstractFailsafeTest.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -347,7 +347,7 @@ public void shouldTimeout() throws Throwable {
347347
};
348348

349349
// When / Then
350-
FailsafeExecutor<Object> failsafe = Failsafe.with(rp, timeout).onSuccess(e -> {
350+
FailsafeExecutor<Object> failsafe = Failsafe.with(rp, timeout).onComplete(e -> {
351351
waiter.assertEquals(e.getAttemptCount(), 3);
352352
waiter.assertEquals(e.getExecutionCount(), 3);
353353
waiter.assertEquals("foo2", e.getResult());

0 commit comments

Comments
 (0)