FEATURE: Add operation timeouts to the v2 API

src/main/java/net/spy/memcached/MemcachedClient.java

@@ -279,6 +279,15 @@ public Transcoder<Object> getTranscoder() {
     return transcoder;
   }
 
+  /**
+   * Get the operation timeout in milliseconds
+   *
+   * @return this instance's operation timeout
+   */
+  public long getOperationTimeout() {
+    return operationTimeout;
+  }
+
   public OperationFactory getOpFact() {
     return opFact;
   }

src/main/java/net/spy/memcached/v2/ArcusExecutors.java

@@ -19,6 +19,7 @@
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 
 /**
  * Provides shared thread pool executors for Arcus v2 async operations.
@@ -41,6 +42,28 @@ final class ArcusExecutors {
             return t;
           });
 
+  /**
+   * Shared scheduler for operation timeouts.
+   *
+   * <p>
+   * The timeout task cancels the operation, so the future may be completed and
+   * its non-async dependent stages may run on this scheduler thread.
+   * </p>
+   *
+   * <p>The remove-on-cancel policy removes cancelled timeout tasks immediately.</p>
+   */
+  static final ScheduledThreadPoolExecutor TIMEOUT_SCHEDULER =
+      new ScheduledThreadPoolExecutor(1, r -> {
+        Thread t = new Thread(r);
+        t.setName("arcus-timeout");
+        t.setDaemon(true);
+        return t;
+      });
+
+  static {
+    TIMEOUT_SCHEDULER.setRemoveOnCancelPolicy(true);
+  }
+
   private ArcusExecutors() {
   }
 }

src/main/java/net/spy/memcached/v2/ArcusFuture.java

@@ -20,8 +20,30 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Future;
 
+/**
+ * Result of an asynchronous Arcus operation.
+ *
+ * <p>Timeout and cancellation apply from submission until the complete response arrives
+ * from the cache server; decoding and dependent stages are not included. On timeout, the
+ * operation is cancelled and this future completes exceptionally with
+ * {@link net.spy.memcached.OperationTimeoutException}. Once the response has arrived,
+ * {@link #cancel(boolean)} returns {@code false} and the result is delivered.
+ * For a future that combines several operations, each operation is timed out and
+ * cancelled independently.</p>
+ *
+ * <p>Non-async dependent stages may run on Arcus internal threads, so they must be short
+ * and non-blocking. Use async methods with your own executor for blocking work.</p>
+ *
+ * <p><b>Do not complete the future returned by {@link #toCompletableFuture()} directly;</b>
+ * it does not cancel the Arcus operation.</p>
+ *
+ * @param <T> operation result type
+ */
 public interface ArcusFuture<T> extends CompletionStage<T>, Future<T> {
 
+  /**
+   * Internal completion callback. Not for application use.
+   */
   void complete();
 
 }

src/main/java/net/spy/memcached/v2/ArcusFutureImpl.java

@@ -20,17 +20,29 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
+import net.spy.memcached.MemcachedConnection;
+import net.spy.memcached.OperationTimeoutException;
 import net.spy.memcached.internal.CompositeException;
 import net.spy.memcached.ops.Operation;
 
+/**
+ * {@link ArcusFuture} for a single Arcus operation.
+ *
+ * @param <T> result type
+ */
 public class ArcusFutureImpl<T> extends CompletableFuture<T> implements ArcusFuture<T> {
 
+  private static final String TIMEOUT_CANCEL_CAUSE = "by operation timeout";
+
   private Operation op;
 
   private final ArcusResult<?> arcusResult;
   private final Function<Object, T> decoder;
+  private volatile ScheduledFuture<?> timeoutFuture;
 
   /**
    * Use only when the result needs to be decoded.
@@ -50,16 +62,25 @@ public ArcusFutureImpl(ArcusResult<T> arcusResult) {
   }
 
   /**
-   * Called by the IO(Internal) Thread when all Operations are completed.
-   * Submits the decoding task to a dedicated thread pool and completes the future.
-   * If there is an error in the response, completes the CompletableFuture with an exception.
+   * Called by the operation callback when the operation is completed.
+   *
+   * <p>
+   * Stops the operation timeout and resets the node timeout count,
+   * then completes this future. An error, or a result that needs no decoding,
+   * completes it on the calling thread.
+   * Otherwise, the result is decoded and completed on the completion executor.
+   * Does nothing if this future is already done by cancellation or timeout.
+   * </p>
    */
   @Override
   public void complete() {
     if (this.isDone()) {
       return;
     }
 
+    cancelTimeoutTask();
+    MemcachedConnection.opSucceeded(op);
+
     Exception exception = getError();
     if (exception != null) {
       this.completeExceptionally(exception);
@@ -119,14 +140,12 @@ private Exception getError() {
   }
 
   /**
-   * Cancel this future and the related operations.
-   * This method is thread-safe and prevents multiple concurrent cancellation attempts.
+   * Cancel this future and the related operation.
+   * The operation can no longer be cancelled once its complete response has arrived.
    *
-   * @param mayInterruptIfRunning this value has no effect in this
-   *                              implementation because interrupts are not used to control
-   *                              processing.
-   * @return {@code true} if this future and all the operations were cancelled,
-   * {@code false} if this future was already cancelled or completed.
+   * @param mayInterruptIfRunning this value has no effect in this implementation
+   * @return {@code true} if the operation was cancelled,
+   * {@code false} if this future is already done or the response has already arrived.
    */
   @Override
   public boolean cancel(boolean mayInterruptIfRunning) {
@@ -137,14 +156,72 @@ public boolean cancel(boolean mayInterruptIfRunning) {
     return op.cancel("by application.");
   }
 
+  /**
+   * Called by the operation callback when the operation is cancelled by the application,
+   * the operation timeout, or the connection, for example on connection loss.
+   * Completes this future with {@link OperationTimeoutException} if cancelled by the
+   * operation timeout, or cancels this future otherwise.
+   */
   void internalCancel() {
-    super.cancel(true);
+    cancelTimeoutTask();
+
+    String cause = op.getCancelCause();
+    if (cause != null && cause.contains(TIMEOUT_CANCEL_CAUSE)) {
+      OperationTimeoutException exception = new OperationTimeoutException(
+          op.getAPIType() + " operation timed out. " + cause);
+      super.completeExceptionally(exception);
+    } else {
+      super.cancel(true);
+    }
   }
 
   /**
-   * For internal use only.
+   * Sets the operation backing this future. Must be called before the operation is submitted.
    */
   void setOp(Operation op) {
     this.op = op;
   }
+
+  /**
+   * Starts the operation timeout. Called once, right after the operation is submitted.
+   * Does nothing if the operation was already cancelled while being submitted.
+   *
+   * @param timeoutMillis operation timeout in milliseconds
+   */
+  void scheduleTimeout(long timeoutMillis) {
+    if (this.isDone()) {
+      return;
+    }
+
+    timeoutFuture = ArcusExecutors.TIMEOUT_SCHEDULER.schedule(
+        () -> timeout(timeoutMillis),
+        timeoutMillis,
+        TimeUnit.MILLISECONDS);
+  }
+
+  /**
+   * Cancels the operation if its response has not arrived yet.
+   * The node timeout count is updated only when the cancellation succeeds,
+   * because this future is not done yet while the result is being decoded.
+   */
+  private void timeout(long timeoutMillis) {
+    if (this.isDone()) {
+      return;
+    }
+
+    String timeoutCause = TIMEOUT_CANCEL_CAUSE + " after " + timeoutMillis + " milliseconds";
+    if (op.cancel(timeoutCause)) {
+      MemcachedConnection.opTimedOut(op);
+    }
+  }
+
+  /**
+   * Cancels the timeout task if it exists.
+   */
+  private void cancelTimeoutTask() {
+    ScheduledFuture<?> task = timeoutFuture;
+    if (task != null) {
+      task.cancel(false);
+    }
+  }
 }