Allow interruption with Call cancellation, and add some tests

Author: jhalterman

core/src/main/java/dev/failsafe/Call.java

@@ -38,10 +38,19 @@ public interface Call<R> {
   R execute();
 
   /**
-   * Cancels a synchronous execution by calling the most recent {@link ExecutionContext#onCancel(CheckedRunnable)
-   * cancelCallback} that was registered during the execution and marking the execution as cancelled. The execution is
-   * still allowed to complete and return a result. In addition to using a {@link ExecutionContext#onCancel(CheckedRunnable)
-   * cancelCallback}, executions can cooperate with cancellation by checking {@link ExecutionContext#isCancelled()}.
+   * Cancels a synchronous execution and calls the most recent {@link ExecutionContext#onCancel(CheckedRunnable)
+   * cancelCallback} that was registered. The execution is still allowed to complete and return a result. In addition to
+   * using a {@link ExecutionContext#onCancel(CheckedRunnable) cancelCallback}, executions can cooperate with
+   * cancellation by checking {@link ExecutionContext#isCancelled()}.
+   *
+   * @param mayInterruptIfRunning whether the execution should be interrupted
+   * @return whether cancellation was successful or not. Returns {@code false} if the execution was already cancelled or
+   * completed.
+   */
+  boolean cancel(boolean mayInterruptIfRunning);
+
+  /**
+   * Returns whether the call has been cancelled.
    */
-  void cancel();
+  boolean isCancelled();
 }

core/src/main/java/dev/failsafe/CallImpl.java

@@ -34,7 +34,15 @@ public R execute() {
   }
 
   @Override
-  public void cancel() {
-    execution.cancel();
+  public boolean cancel(boolean mayInterruptIfRunning) {
+    boolean result = execution.cancel();
+    if (mayInterruptIfRunning)
+      execution.interrupt();
+    return result;
+  }
+
+  @Override
+  public boolean isCancelled() {
+    return execution.isCancelled();
   }
 }

core/src/main/java/dev/failsafe/ExecutionImpl.java

@@ -161,10 +161,11 @@ synchronized ExecutionResult<R> postExecute(ExecutionResult<R> result) {
     return result;
   }
 
-  /** Called by users. */
+  /** Called indorectly by users. */
   @Override
-  public void cancel() {
-    if (!isCancelled()) {
+  public boolean cancel() {
+    boolean cancelled = isCancelled();
+    if (!cancelled) {
       cancelledIndex = Integer.MAX_VALUE;
       if (cancelCallback != null) {
         try {
@@ -173,6 +174,7 @@ public void cancel() {
         }
       }
     }
+    return !cancelled && !completed;
   }
 
   /** Called by policies. */

core/src/main/java/dev/failsafe/SyncExecutionImpl.java

@@ -35,10 +35,12 @@ final class SyncExecutionImpl<R> extends ExecutionImpl<R> implements SyncExecuti
 
   // An optional Failsafe executor
   private final FailsafeExecutor<R> executor;
-  // An optoinal Failsafe call
+  // An optional Failsafe call
   private final CallImpl<R> call;
   // The outer-most function that executions begin with
   private Function<SyncExecutionInternal<R>, ExecutionResult<R>> outerFn;
+  // The interruptable execution thread
+  private final Thread executionThread;
   // Whether the execution is currently interruptable
   private volatile boolean interruptable;
   // Whether the execution has been internally interrupted
@@ -59,6 +61,7 @@ final class SyncExecutionImpl<R> extends ExecutionImpl<R> implements SyncExecuti
     executor = null;
     call = null;
     initial = this;
+    executionThread = Thread.currentThread();
     preExecute();
   }
 
@@ -71,6 +74,7 @@ final class SyncExecutionImpl<R> extends ExecutionImpl<R> implements SyncExecuti
     this.executor = executor;
     this.call = call;
     initial = this;
+    executionThread = Thread.currentThread();
     if (call != null)
       call.setExecution(this);
 
@@ -89,6 +93,7 @@ private SyncExecutionImpl(SyncExecutionImpl<R> execution) {
     interruptable = execution.interruptable;
     interrupted = execution.interrupted;
     initial = execution.initial;
+    executionThread = execution.executionThread;
     if (call != null)
       call.setExecution(this);
   }
@@ -150,11 +155,6 @@ synchronized ExecutionResult<R> postExecute(ExecutionResult<R> result) {
     return result;
   }
 
-  @Override
-  public boolean isInterruptable() {
-    return interruptable;
-  }
-
   @Override
   public boolean isInterrupted() {
     return interrupted;
@@ -166,8 +166,14 @@ public void setInterruptable(boolean interruptable) {
   }
 
   @Override
-  public void setInterrupted(boolean interrupted) {
-    this.interrupted = interrupted;
+  public void interrupt() {
+    // Guard against race with the execution completing
+    synchronized (getInitial()) {
+      if (interruptable) {
+        interrupted = true;
+        executionThread.interrupt();
+      }
+    }
   }
 
   @Override

core/src/main/java/dev/failsafe/internal/TimeoutExecutor.java

@@ -65,7 +65,6 @@ public Function<SyncExecutionInternal<R>, ExecutionResult<R>> apply(
       // Coordinates a result between the timeout and execution threads
       AtomicReference<ExecutionResult<R>> result = new AtomicReference<>();
       Future<?> timeoutFuture;
-      Thread executionThread = Thread.currentThread();
 
       try {
         // Schedule timeout check
@@ -76,15 +75,8 @@ public Function<SyncExecutionInternal<R>, ExecutionResult<R>> apply(
             // Cancel and interrupt
             execution.record(cancelResult);
             execution.cancel(this);
-            if (config.canInterrupt()) {
-              // Guard against race with the execution completing
-              synchronized (execution.getInitial()) {
-                if (execution.isInterruptable()) {
-                  execution.setInterrupted(true);
-                  executionThread.interrupt();
-                }
-              }
-            }
+            if (config.canInterrupt())
+              execution.interrupt();
           }
           return null;
         }, config.getTimeout().toNanos(), TimeUnit.NANOSECONDS);

core/src/main/java/dev/failsafe/spi/ExecutionInternal.java

@@ -53,8 +53,11 @@ public interface ExecutionInternal<R> extends ExecutionContext<R> {
 
   /**
    * Marks the execution as having been cancelled externally, which will cancel pending executions of all policies.
+   *
+   * @return whether cancellation was successful or not. Returns {@code false} if the execution was already cancelled or
+   * completed.
    */
-  void cancel();
+  boolean cancel();
 
   /**
    * Marks the execution as having been cancelled by the {@code policyExecutor}, which will also cancel pending

core/src/main/java/dev/failsafe/spi/SyncExecutionInternal.java

@@ -30,11 +30,6 @@ public interface SyncExecutionInternal<R> extends ExecutionInternal<R>, Executio
    */
   SyncExecutionInternal<R> getInitial();
 
-  /**
-   * Returns whether the execution is currently interruptable.
-   */
-  boolean isInterruptable();
-
   /**
    * Returns whether the execution is currently interrupted.
    */
@@ -46,9 +41,9 @@ public interface SyncExecutionInternal<R> extends ExecutionInternal<R>, Executio
   void setInterruptable(boolean interruptable);
 
   /**
-   * Sets whether the execution has been internally {@code interrupted}.
+   * Interrupts the execution.
    */
-  void setInterrupted(boolean interrupted);
+  void interrupt();
 
   /**
    * Returns a new copy of the SyncExecutionInternal if it is not standalone, else returns {@code this} since standalone

core/src/test/java/dev/failsafe/functional/CallCancellationTest.java

@@ -0,0 +1,97 @@
+package dev.failsafe.functional;
+
+import dev.failsafe.*;
+import dev.failsafe.function.CheckedConsumer;
+import dev.failsafe.function.ContextualRunnable;
+import dev.failsafe.testing.Testing;
+import net.jodah.concurrentunit.Waiter;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.testng.Assert.assertTrue;
+
+@Test
+public class CallCancellationTest extends Testing {
+  Waiter waiter;
+
+  @BeforeMethod
+  void beforeMethod() {
+    waiter = new Waiter();
+  }
+
+  private void assertCancel(FailsafeExecutor<Void> executor, ContextualRunnable<Void> runnable,
+    boolean testWithoutInterrupt) throws Throwable {
+    CheckedConsumer<Boolean> test = interrupt -> {
+      // Given
+      Call<Void> call = executor.onComplete(e -> {
+        waiter.assertNull(e.getResult());
+        if (!interrupt)
+          waiter.assertNull(e.getException());
+        else
+          waiter.assertTrue(e.getException() instanceof InterruptedException);
+        waiter.resume();
+      }).newCall(runnable);
+
+      // When
+      runInThread(() -> {
+        Testing.sleep(300);
+        waiter.assertTrue(call.cancel(interrupt));
+      });
+      if (!interrupt)
+        call.execute();
+      else
+        assertThrows(call::execute, FailsafeException.class, InterruptedException.class);
+
+      waiter.await(1000);
+
+      // Then
+      assertTrue(call.isCancelled());
+    };
+
+    // Test without interrupt
+    if (testWithoutInterrupt)
+      test.accept(false);
+
+    // Test with interrupt
+    test.accept(true);
+  }
+
+  public void shouldCancelRetriesWithBlockedExecution() throws Throwable {
+    assertCancel(Failsafe.with(RetryPolicy.ofDefaults()), ctx -> {
+      try {
+        waiter.assertFalse(ctx.isCancelled());
+        Thread.sleep(1000);
+      } catch (InterruptedException e) {
+        waiter.assertTrue(ctx.isCancelled());
+        throw e;
+      }
+    }, true);
+  }
+
+  public void shouldCancelRetriesWithPendingDelay() throws Throwable {
+    RetryPolicy<Void> retryPolicy = RetryPolicy.<Void>builder().withDelay(Duration.ofMinutes(1)).build();
+    assertCancel(Failsafe.with(retryPolicy), ctx -> {
+      throw new IllegalStateException();
+    }, false);
+  }
+
+  public void shouldPropagateCancelToCallback() throws Throwable {
+    AtomicBoolean callbackCalled = new AtomicBoolean();
+    assertCancel(Failsafe.with(RetryPolicy.ofDefaults()), ctx -> {
+      ctx.onCancel(() -> callbackCalled.set(true));
+
+      try {
+        waiter.assertFalse(ctx.isCancelled());
+        Thread.sleep(1000);
+      } catch (InterruptedException e) {
+        waiter.assertTrue(ctx.isCancelled());
+        waiter.assertTrue(callbackCalled.get());
+        throw e;
+      }
+    }, true);
+  }
+}
+

core/src/test/java/dev/failsafe/functional/FailsafeFutureCancellationTest.java renamed to core/src/test/java/dev/failsafe/functional/FutureCancellationTest.java

@@ -20,7 +20,6 @@
 import dev.failsafe.testing.Testing;
 import net.jodah.concurrentunit.Waiter;
 import dev.failsafe.event.ExecutionCompletedEvent;
-import dev.failsafe.function.ContextualSupplier;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
@@ -34,7 +33,7 @@
  * Tests behavior when a FailsafeFuture is explicitly cancelled.
  */
 @Test
-public class FailsafeFutureCancellationTest extends Testing {
+public class FutureCancellationTest extends Testing {
   Waiter waiter;
 
   @BeforeMethod

core/src/test/java/dev/failsafe/functional/FailsafeFutureCompletionTest.java renamed to core/src/test/java/dev/failsafe/functional/FutureCompletionTest.java

@@ -28,7 +28,7 @@
  * Tests behavior when a FailsafeFuture is explicitly completed.
  */
 @Test
-public class FailsafeFutureCompletionTest extends Testing {
+public class FutureCompletionTest extends Testing {
   /**
    * Asserts that an externally completed FailsafeFuture works as expected.
    */