From b128c9f5fc7ce4c0ca2b5866a1203e95107411fb Mon Sep 17 00:00:00 2001 From: f1v3-dev Date: Thu, 1 Oct 2026 17:54:20 +0900 Subject: [PATCH] FEATURE: Add operation timeouts to the v2 API --- .../net/spy/memcached/MemcachedClient.java | 9 + .../net/spy/memcached/v2/ArcusExecutors.java | 23 + .../net/spy/memcached/v2/ArcusFuture.java | 22 + .../net/spy/memcached/v2/ArcusFutureImpl.java | 101 ++++- .../spy/memcached/v2/AsyncArcusCommands.java | 147 ++++--- .../memcached/v2/AsyncArcusCommandsIF.java | 24 + .../memcached/v2/ArcusFutureTimeoutTest.java | 411 ++++++++++++++++++ .../memcached/v2/OperationTimeoutTest.java | 162 +++++++ 8 files changed, 815 insertions(+), 84 deletions(-) create mode 100644 src/test/java/net/spy/memcached/v2/ArcusFutureTimeoutTest.java create mode 100644 src/test/java/net/spy/memcached/v2/OperationTimeoutTest.java diff --git a/src/main/java/net/spy/memcached/MemcachedClient.java b/src/main/java/net/spy/memcached/MemcachedClient.java index 8ba9466cc..93af05cdb 100644 --- a/src/main/java/net/spy/memcached/MemcachedClient.java +++ b/src/main/java/net/spy/memcached/MemcachedClient.java @@ -279,6 +279,15 @@ public Transcoder 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; } diff --git a/src/main/java/net/spy/memcached/v2/ArcusExecutors.java b/src/main/java/net/spy/memcached/v2/ArcusExecutors.java index 0232b4609..b934b31db 100644 --- a/src/main/java/net/spy/memcached/v2/ArcusExecutors.java +++ b/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. + * + *

+ * The timeout task cancels the operation, so the future may be completed and + * its non-async dependent stages may run on this scheduler thread. + *

+ * + *

The remove-on-cancel policy removes cancelled timeout tasks immediately.

+ */ + 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() { } } diff --git a/src/main/java/net/spy/memcached/v2/ArcusFuture.java b/src/main/java/net/spy/memcached/v2/ArcusFuture.java index fa6a7c299..c2374281c 100644 --- a/src/main/java/net/spy/memcached/v2/ArcusFuture.java +++ b/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. + * + *

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.

+ * + *

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.

+ * + *

Do not complete the future returned by {@link #toCompletableFuture()} directly; + * it does not cancel the Arcus operation.

+ * + * @param operation result type + */ public interface ArcusFuture extends CompletionStage, Future { + /** + * Internal completion callback. Not for application use. + */ void complete(); } diff --git a/src/main/java/net/spy/memcached/v2/ArcusFutureImpl.java b/src/main/java/net/spy/memcached/v2/ArcusFutureImpl.java index 14e24bc70..51d28043a 100644 --- a/src/main/java/net/spy/memcached/v2/ArcusFutureImpl.java +++ b/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 result type + */ public class ArcusFutureImpl extends CompletableFuture implements ArcusFuture { + private static final String TIMEOUT_CANCEL_CAUSE = "by operation timeout"; + private Operation op; private final ArcusResult arcusResult; private final Function decoder; + private volatile ScheduledFuture timeoutFuture; /** * Use only when the result needs to be decoded. @@ -50,9 +62,15 @@ public ArcusFutureImpl(ArcusResult 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. + * + *

+ * 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. + *

*/ @Override public void complete() { @@ -60,6 +78,9 @@ public void complete() { 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); + } + } } diff --git a/src/main/java/net/spy/memcached/v2/AsyncArcusCommands.java b/src/main/java/net/spy/memcached/v2/AsyncArcusCommands.java index 00b12a475..9cd98846f 100644 --- a/src/main/java/net/spy/memcached/v2/AsyncArcusCommands.java +++ b/src/main/java/net/spy/memcached/v2/AsyncArcusCommands.java @@ -28,6 +28,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -123,6 +124,7 @@ public class AsyncArcusCommands implements AsyncArcusCommandsIF { private final Transcoder tcForCollection; private final KeyValidator keyValidator; private final Supplier arcusClientSupplier; + private final long operationTimeout; @SuppressWarnings("unchecked") public AsyncArcusCommands(Supplier arcusClientSupplier) { @@ -131,6 +133,29 @@ public AsyncArcusCommands(Supplier arcusClientSupplier) { this.tcForCollection = (Transcoder) client.getCollectionTranscoder(); this.keyValidator = client.getKeyValidator(); this.arcusClientSupplier = arcusClientSupplier; + this.operationTimeout = client.getOperationTimeout(); + } + + private AsyncArcusCommands(AsyncArcusCommands origin, long operationTimeout) { + this.tc = origin.tc; + this.tcForCollection = origin.tcForCollection; + this.keyValidator = origin.keyValidator; + this.arcusClientSupplier = origin.arcusClientSupplier; + this.operationTimeout = operationTimeout; + } + + @Override + public AsyncArcusCommands withOperationTimeout(long duration, TimeUnit unit) { + if (unit == null) { + throw new IllegalArgumentException("TimeUnit cannot be null"); + } + + long timeoutMillis = unit.toMillis(duration); + if (timeoutMillis <= 0) { + throw new IllegalArgumentException("Operation timeout must be greater than 0 milliseconds"); + } + + return new AsyncArcusCommands<>(this, timeoutMillis); } @Override @@ -182,8 +207,7 @@ public void complete() { }; Operation op = client.getOpFact() .store(type, key, co.getFlags(), exp, co.getData(), cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -269,8 +293,7 @@ public void complete() { } }; Operation op = client.getOpFact().cat(concatType, 0L, key, co.getData(), cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -311,8 +334,7 @@ public void complete() { }; Operation op = client.getOpFact() .cas(StoreType.set, key, casId, co.getFlags(), exp, co.getData(), cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -379,8 +401,7 @@ public void complete() { } }; Operation op = client.getOpFact().mutate(mutator, key, delta, initial, exp, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -420,8 +441,7 @@ public void complete() { } }; Operation op = client.getOpFact().get(key, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -462,8 +482,7 @@ public void complete() { } }; GetsOperation op = client.getOpFact().gets(key, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -544,8 +563,7 @@ public void complete() { } }; Operation op = client.getOpFact().get(keyList, cb, node.enabledMGetOp()); - future.setOp(op); - client.addOp(node, op); + submit(client, future, node, op); return future; } @@ -627,8 +645,7 @@ public void complete() { } }; Operation op = client.getOpFact().gets(keyList, cb, node.enabledMGetsOp()); - future.setOp(op); - client.addOp(node, op); + submit(client, future, node, op); return future; } @@ -666,8 +683,7 @@ public void complete() { } }; Operation op = client.getOpFact().delete(key, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -759,8 +775,7 @@ public void gotData(String subKey, int flags, byte[] data, byte[] eFlag) { } }; Operation op = client.getOpFact().collectionGet(key, get, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -805,8 +820,7 @@ public void gotData(String subKey, int flags, byte[] data, byte[] eFlag) { } }; Operation op = client.getOpFact().collectionGet(key, get, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -886,8 +900,7 @@ public void complete() { } }; Operation op = client.getOpFact().collectionGet(key, get, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -929,8 +942,7 @@ public void complete() { } }; Operation op = client.getOpFact().collectionExist(key, "", exist, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -1036,8 +1048,7 @@ public void complete() { } }; Operation op = client.getOpFact().collectionGet(key, get, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -1090,8 +1101,7 @@ public void complete() { } }; Operation op = client.getOpFact().collectionGet(key, get, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -1235,8 +1245,7 @@ public void gotData(int flags, BKeyObject bKeyObject, byte[] eFlag, byte[] data) }; Operation op = client.getOpFact() .bopInsertAndGet(key, insertAndGet, co.getData(), cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -1323,8 +1332,7 @@ public void gotData(String bKey, int flags, byte[] data, byte[] eFlag) { } }; Operation op = client.getOpFact().collectionGet(key, get, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -1375,8 +1383,7 @@ public void gotData(String bKey, int flags, byte[] data, byte[] eFlag) { } }; Operation op = client.getOpFact().collectionGet(key, get, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -1512,8 +1519,7 @@ public void gotElement(String key, int flags, Object bKey, byte[] eFlag, byte[] } }; Operation op = client.getOpFact().bopGetBulk(getBulk, cb); - future.setOp(op); - client.addOp(getBulk.getMemcachedNode(), op); + submit(client, future, getBulk.getMemcachedNode(), op); return future; } @@ -1628,8 +1634,7 @@ public void gotTrimmedKey(String key, Object bKey) { } }; Operation op = client.getOpFact().bopsmget(smGet, cb); - future.setOp(op); - client.addOp(smGet.getMemcachedNode(), op); + submit(client, future, smGet.getMemcachedNode(), op); return future; } @@ -1686,8 +1691,7 @@ public void complete() { } }; Operation op = client.getOpFact().bopFindPosition(key, findPosition, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -1734,8 +1738,7 @@ public void complete() { } }; Operation op = client.getOpFact().bopGetByPosition(key, getByPosition, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -1788,8 +1791,7 @@ public void complete() { } }; Operation op = client.getOpFact().bopGetByPosition(key, getByPosition, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -1841,8 +1843,7 @@ public void complete() { } }; Operation op = client.getOpFact().bopFindPositionWithGet(key, findPositionWithGet, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -1884,8 +1885,7 @@ public void complete() { } }; Operation op = client.getOpFact().collectionCount(key, collectionCount, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -1947,8 +1947,7 @@ public void complete() { }; CollectionCreateOperation op = client.getOpFact() .collectionCreate(key, collectionCreate, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -1994,8 +1993,7 @@ public void complete() { }; CollectionInsertOperation op = client.getOpFact() .collectionInsert(key, internalKey, collectionInsert, co.getData(), cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -2045,8 +2043,7 @@ public void complete() { Operation op = client.getOpFact() .collectionUpdate(key, internalKey, collectionUpdate, (co == null) ? null : co.getData(), cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -2086,8 +2083,7 @@ public void complete() { } }; Operation op = client.getOpFact().collectionMutate(key, internalKey, mutate, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -2127,8 +2123,7 @@ public void complete() { } }; Operation op = client.getOpFact().collectionDelete(key, delete, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -2173,8 +2168,7 @@ public void complete() { }; Operation op = client.getOpFact().flush(delay, cb); - future.setOp(op); - client.addOp(node, op); + submit(client, future, node, op); futures.add(future); } @@ -2230,8 +2224,7 @@ public void complete() { Operation op = client.getOpFact() .flush(prefix.isEmpty() ? "" : prefix, delay, false, cb); - future.setOp(op); - client.addOp(node, op); + submit(client, future, node, op); futures.add(future); } @@ -2291,8 +2284,7 @@ public void complete() { } }; Operation op = client.getOpFact().stats(arg.getArg(), cb); - future.setOp(op); - client.addOp(node, op); + submit(client, future, node, op); addressToFuture.put(address, future); } @@ -2348,8 +2340,7 @@ public void complete() { } }; Operation op = client.getOpFact().version(cb); - future.setOp(op); - client.addOp(node, op); + submit(client, future, node, op); addressToFuture.put(address, future); } @@ -2400,8 +2391,7 @@ public void complete() { } }; Operation op = client.getOpFact().setAttr(key, attributes, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -2445,8 +2435,7 @@ public void complete() { } }; Operation op = client.getOpFact().getAttr(key, cb); - future.setOp(op); - client.addOp(key, op); + submit(client, future, key, op); return future; } @@ -2613,4 +2602,18 @@ private List> decodeBTreePositionElementList( } return decoded; } + + private void submit(ArcusClient client, ArcusFutureImpl future, + String key, Operation op) { + future.setOp(op); + client.addOp(key, op); + future.scheduleTimeout(operationTimeout); + } + + private void submit(ArcusClient client, ArcusFutureImpl future, + MemcachedNode node, Operation op) { + future.setOp(op); + client.addOp(node, op); + future.scheduleTimeout(operationTimeout); + } } diff --git a/src/main/java/net/spy/memcached/v2/AsyncArcusCommandsIF.java b/src/main/java/net/spy/memcached/v2/AsyncArcusCommandsIF.java index 37c48dbc3..aecd11ec7 100644 --- a/src/main/java/net/spy/memcached/v2/AsyncArcusCommandsIF.java +++ b/src/main/java/net/spy/memcached/v2/AsyncArcusCommandsIF.java @@ -21,6 +21,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.TimeUnit; import net.spy.memcached.CASValue; import net.spy.memcached.collection.BTreeOrder; @@ -43,6 +44,29 @@ public interface AsyncArcusCommandsIF { + /** + * Returns a new command instance that uses the given operation timeout. + * + *

+ * This instance is not modified. The timeout is a relative duration, so every operation + * issued through the returned instance starts its own deadline when it is submitted. + * The returned instance holds no deadline state and can be reused or cached. + * Without this override, the operation timeout is the + * {@code ConnectionFactoryBuilder.setOpTimeout()} value of the client. + *

+ * + *

+ * The period covered by the operation timeout is described in {@link ArcusFuture}. + * The given value is truncated to milliseconds. + *

+ * + * @param duration the operation timeout duration + * @param unit the time unit of the timeout duration + * @return a new command instance using the given operation timeout + * @throws IllegalArgumentException if unit is null or the duration is less than 1 millisecond + */ + AsyncArcusCommandsIF withOperationTimeout(long duration, TimeUnit unit); + /** * Set a value for the given key. * diff --git a/src/test/java/net/spy/memcached/v2/ArcusFutureTimeoutTest.java b/src/test/java/net/spy/memcached/v2/ArcusFutureTimeoutTest.java new file mode 100644 index 000000000..382240470 --- /dev/null +++ b/src/test/java/net/spy/memcached/v2/ArcusFutureTimeoutTest.java @@ -0,0 +1,411 @@ +package net.spy.memcached.v2; + +import java.lang.reflect.Field; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import net.spy.memcached.MemcachedNode; +import net.spy.memcached.OperationTimeoutException; +import net.spy.memcached.ops.APIType; +import net.spy.memcached.ops.Operation; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.jmock.Mockery; +import org.jmock.api.Action; +import org.jmock.api.Invocation; +import org.jmock.lib.action.CustomAction; +import org.jmock.lib.concurrent.Synchroniser; + +import static net.spy.memcached.ExpectationsUtil.buildExpectations; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import static org.jmock.AbstractExpectations.returnValue; + +public class ArcusFutureTimeoutTest { + + private static final long OP_TIMEOUT_MILLIS = 300L; + private static final long TEST_TIMEOUT_SECONDS = 3L; + private static final String TIMEOUT_CAUSE = + "by operation timeout after " + OP_TIMEOUT_MILLIS + " milliseconds"; + + private Mockery context; + private Operation operation; + private MemcachedNode node; + private AbstractArcusResult result; + private ArcusFutureImpl future; + + private final CountDownLatch decodingStarted = new CountDownLatch(1); + private final CountDownLatch releaseDecoding = new CountDownLatch(1); + private final AtomicReference cancelCause = new AtomicReference<>(); + + @BeforeEach + void setUp() { + context = new Mockery(); + context.setThreadingPolicy(new Synchroniser()); + + operation = context.mock(Operation.class); + node = context.mock(MemcachedNode.class); + + context.checking(buildExpectations(e -> { + e.allowing(operation).getAPIType(); + e.will(returnValue(APIType.GET)); + + e.allowing(operation).getHandlingNode(); + e.will(returnValue(node)); + + e.allowing(operation).hasErrored(); + e.will(returnValue(false)); + + e.allowing(operation).isCancelled(); + e.will(returnValue(false)); + + e.allowing(operation).getCancelCause(); + e.will(new CustomAction("return the recorded cancel cause") { + @Override + public Object invoke(Invocation invocation) { + return cancelCause.get(); + } + }); + + e.allowing(node).getNodeName(); + e.will(returnValue("test-node")); + })); + + result = new AbstractArcusResult<>(new AtomicReference<>("value")); + future = new ArcusFutureImpl<>(result); + future.setOp(operation); + } + + @AfterEach + void tearDown() throws Exception { + releaseDecoding.countDown(); + + ScheduledFuture timeout = scheduledTimeout(); + if (timeout != null) { + timeout.cancel(true); + } + + context.assertIsSatisfied(); + } + + @Test + void timeOutWithoutWaitingFuture() throws Exception { + // given + CountDownLatch completed = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + + context.checking(buildExpectations(e -> { + e.oneOf(operation).cancel(TIMEOUT_CAUSE); + e.will(cancelOperation()); + + e.oneOf(node).setContinuousTimeout(true); + })); + + future.whenComplete((value, error) -> { + failure.set(error); + completed.countDown(); + }); + + // when + future.scheduleTimeout(OP_TIMEOUT_MILLIS); + + // then + assertTrue(completed.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertInstanceOf(OperationTimeoutException.class, failure.get()); + assertTrue(failure.get().getMessage().contains(TIMEOUT_CAUSE)); + awaitTimeoutTask(); + } + + @Test + void ignoresTimeoutWhenOperationAlreadyCompleted() throws Exception { + // given: the operation has completed, but its callback has not reached this future yet + context.checking(buildExpectations(e -> { + e.oneOf(operation).cancel(TIMEOUT_CAUSE); + e.will(returnValue(false)); + + e.oneOf(node).setContinuousTimeout(false); + })); + + future.scheduleTimeout(OP_TIMEOUT_MILLIS); + + // when: the timeout fails to cancel the completed operation + awaitTimeoutWindow(); + + // then: no node timeout update, and the response still completes this future + assertFalse(future.isDone()); + future.complete(); + assertEquals("value", future.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + } + + @Test + void cancelsTimeoutWhenResponseArrivesFirst() throws Exception { + // given + context.checking(buildExpectations(e -> { + e.oneOf(node).setContinuousTimeout(false); + })); + + future.scheduleTimeout(OP_TIMEOUT_MILLIS); + ScheduledFuture timeout = scheduledTimeout(); + assertNotNull(timeout); + + // when + future.complete(); + + // then + assertEquals("value", future.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertTrue(timeout.isCancelled()); + + awaitTimeoutWindow(); + assertEquals("value", future.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + } + + @Test + void cancelsOperationWhileWaitingForResponse() throws Exception { + // given + context.checking(buildExpectations(e -> { + e.oneOf(operation).cancel("by application."); + e.will(cancelOperation()); + })); + + future.scheduleTimeout(OP_TIMEOUT_MILLIS); + ScheduledFuture timeout = scheduledTimeout(); + assertNotNull(timeout); + + // when + boolean cancelled = future.cancel(false); + + // then + assertTrue(cancelled); + assertTrue(timeout.isCancelled()); + assertThrows(CancellationException.class, + () -> future.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + awaitTimeoutWindow(); + assertTrue(future.isCancelled()); + } + + @Test + void excludesDecodingTimeFromOperationTimeout() throws Exception { + // given + future = new ArcusFutureImpl<>(result, this::blockingDecode); + future.setOp(operation); + + context.checking(buildExpectations(e -> { + e.oneOf(node).setContinuousTimeout(false); + })); + + future.scheduleTimeout(OP_TIMEOUT_MILLIS); + + // when + future.complete(); + + assertTrue(decodingStarted.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + // block the decoding to simulate a long-running decode operation + awaitTimeoutWindow(); + + // then + assertFalse(future.isDone()); + releaseDecoding.countDown(); + assertEquals("value", future.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + } + + @Test + void doesNotCancelAfterResponseArrives() throws Exception { + // given + future = new ArcusFutureImpl<>(result, this::blockingDecode); + future.setOp(operation); + + context.checking(buildExpectations(e -> { + e.oneOf(node).setContinuousTimeout(false); + + // the completed operation can no longer be cancelled + e.oneOf(operation).cancel("by application."); + e.will(returnValue(false)); + })); + + future.scheduleTimeout(OP_TIMEOUT_MILLIS); + future.complete(); + + assertTrue(decodingStarted.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + // when: cancel the future while the decoding is still in progress + boolean cancelled = future.cancel(false); + + // then: the decoded result completes the future + assertFalse(cancelled); + assertFalse(future.isCancelled()); + + releaseDecoding.countDown(); + assertEquals("value", future.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + } + + @Test + void ignoresResponseAfterOperationTimeout() throws Exception { + // given + context.checking(buildExpectations(e -> { + e.oneOf(operation).cancel(TIMEOUT_CAUSE); + e.will(cancelOperation()); + + e.oneOf(node).setContinuousTimeout(true); + })); + + future.scheduleTimeout(OP_TIMEOUT_MILLIS); + + ExecutionException timeout = assertThrows( + ExecutionException.class, + () -> future.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS) + ); + + assertInstanceOf(OperationTimeoutException.class, timeout.getCause()); + awaitTimeoutTask(); + + // when: complete the future after the operation has timed out + future.complete(); + + // then + ExecutionException afterResponse = assertThrows( + ExecutionException.class, + () -> future.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS) + ); + + assertSame(timeout.getCause(), afterResponse.getCause()); + } + + @Test + void doesNotScheduleTimeoutAfterResponseCompletes() throws Exception { + // given + context.checking(buildExpectations(e -> { + e.oneOf(node).setContinuousTimeout(false); + })); + + future.complete(); + + // when: schedule a timeout after the response has already completed + future.scheduleTimeout(OP_TIMEOUT_MILLIS); + + // then + assertNull(scheduledTimeout()); + assertEquals("value", future.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + } + + @Test + void doesNotCancelAfterOperationTimeout() throws Exception { + // given + context.checking(buildExpectations(e -> { + e.oneOf(operation).cancel(TIMEOUT_CAUSE); + e.will(cancelOperation()); + + e.oneOf(node).setContinuousTimeout(true); + })); + + future.scheduleTimeout(OP_TIMEOUT_MILLIS); + + ExecutionException timeout = assertThrows( + ExecutionException.class, + () -> future.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS) + ); + awaitTimeoutTask(); + + // when: an unexpected op.cancel("by application.") fails the mock + boolean cancelled = future.cancel(false); + + // then + assertFalse(cancelled); + assertFalse(future.isCancelled()); + + ExecutionException afterCancel = assertThrows( + ExecutionException.class, + () -> future.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS) + ); + + assertSame(timeout.getCause(), afterCancel.getCause()); + } + + @Test + void stopsTimeoutOnInternalCancel() throws Exception { + // given + future.scheduleTimeout(OP_TIMEOUT_MILLIS); + ScheduledFuture timeout = scheduledTimeout(); + assertNotNull(timeout); + + // when + future.internalCancel(); + + // then + assertTrue(timeout.isCancelled()); + assertTrue(future.isCancelled()); + + // no timeout side effects: unexpected mock calls fail in tearDown + awaitTimeoutWindow(); + assertTrue(future.isCancelled()); + } + + /** + * Acts like {@code BaseOperationImpl.cancel()}, which records the cancel cause, + * invokes the callback with the cancelled status and then completes it. + */ + private Action cancelOperation() { + return new CustomAction("cancel the operation and invoke its callback") { + @Override + public Object invoke(Invocation invocation) { + cancelCause.set("Cancelled (" + invocation.getParameter(0) + " @ test-node)"); + future.internalCancel(); + future.complete(); + return true; + } + }; + } + + private void awaitTimeoutWindow() throws Exception { + ArcusExecutors.TIMEOUT_SCHEDULER + .schedule(() -> null, OP_TIMEOUT_MILLIS + 50L, TimeUnit.MILLISECONDS) + .get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + + /** + * Waits until the running timeout task returns. + * The node timeout count is updated after this future is completed. + */ + private void awaitTimeoutTask() throws Exception { + ArcusExecutors.TIMEOUT_SCHEDULER + .submit(() -> null) + .get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + + private ScheduledFuture scheduledTimeout() throws Exception { + Field field = ArcusFutureImpl.class.getDeclaredField("timeoutFuture"); + field.setAccessible(true); + + return (ScheduledFuture) field.get(future); + } + + private String blockingDecode(String value) { + decodingStarted.countDown(); + + try { + if (!releaseDecoding.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + throw new IllegalStateException("Decoder was not released"); + } + return value; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Decoder was interrupted", e); + } + } +} diff --git a/src/test/java/net/spy/memcached/v2/OperationTimeoutTest.java b/src/test/java/net/spy/memcached/v2/OperationTimeoutTest.java new file mode 100644 index 000000000..d0f2b18ba --- /dev/null +++ b/src/test/java/net/spy/memcached/v2/OperationTimeoutTest.java @@ -0,0 +1,162 @@ +package net.spy.memcached.v2; + +import java.net.InetSocketAddress; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import net.spy.memcached.ArcusClient; +import net.spy.memcached.ConnectionFactoryBuilder; +import net.spy.memcached.FailureMode; +import net.spy.memcached.OperationTimeoutException; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class OperationTimeoutTest { + + private ArcusClient client; + private AsyncArcusCommands commands; + + @BeforeEach + void setUp() throws Exception { + ConnectionFactoryBuilder cfb = new ConnectionFactoryBuilder() + .setOpTimeout(1L) + .setFailureMode(FailureMode.Retry); + + client = new ArcusClient( + cfb.build(), + Collections.singletonList(new InetSocketAddress("127.0.0.1", 54321)) + ); + + commands = client.asyncCommands(); + } + + @AfterEach + void tearDown() { + if (client != null) { + client.shutdown(0, TimeUnit.MILLISECONDS); + } + } + + @Test + void usesDefaultOperationTimeout() { + // given + + // when + ArcusFuture future = commands.get("key"); + + // then + ExecutionException exception = assertThrows( + ExecutionException.class, + () -> future.get(300L, TimeUnit.MILLISECONDS)); + + assertInstanceOf(OperationTimeoutException.class, exception.getCause()); + assertFalse(future.isCancelled()); + assertTrue(exception.getCause().getMessage().contains("after 1 milliseconds")); + } + + @Test + void overridesOperationTimeout() { + // given + AsyncArcusCommands scoped = + commands.withOperationTimeout(2L, TimeUnit.MILLISECONDS); + + // when + ArcusFuture scopedFuture = scoped.get("scoped-key"); + ArcusFuture originalFuture = commands.get("original-key"); + + // then + ExecutionException scopedException = assertThrows( + ExecutionException.class, + () -> scopedFuture.get(300L, TimeUnit.MILLISECONDS)); + + assertInstanceOf(OperationTimeoutException.class, scopedException.getCause()); + assertTrue(scopedException.getCause().getMessage().contains("after 2 milliseconds")); + + ExecutionException originalException = assertThrows( + ExecutionException.class, + () -> originalFuture.get(300L, TimeUnit.MILLISECONDS)); + + assertInstanceOf(OperationTimeoutException.class, originalException.getCause()); + assertTrue(originalException.getCause().getMessage().contains("after 1 milliseconds")); + } + + @Test + void timeOutWithoutWaitingFuture() throws Exception { + // given + CountDownLatch latch = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + + // when + commands.get("key") + .whenComplete((value, error) -> { + failure.set(error); + latch.countDown(); + }); + + // then + assertTrue(latch.await(700L, TimeUnit.MILLISECONDS)); + assertInstanceOf(OperationTimeoutException.class, failure.get()); + } + + @Test + void rejectsNullTimeUnit() { + IllegalArgumentException exception = assertThrows( + IllegalArgumentException.class, + () -> commands.withOperationTimeout(1L, null) + ); + + assertEquals("TimeUnit cannot be null", exception.getMessage()); + } + + @Test + void rejectsZeroTimeout() { + IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, + () -> commands.withOperationTimeout(0L, TimeUnit.MILLISECONDS)); + + assertEquals( + "Operation timeout must be greater than 0 milliseconds", exception.getMessage() + ); + } + + @Test + void rejectsNegativeTimeout() { + IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, + () -> commands.withOperationTimeout(-1, TimeUnit.MILLISECONDS) + ); + + assertEquals( + "Operation timeout must be greater than 0 milliseconds", exception.getMessage() + ); + } + + @Test + void truncatesTimeoutToMilliseconds() { + // given + AsyncArcusCommands scoped = + commands.withOperationTimeout(1999L, TimeUnit.MICROSECONDS); + + // when + ArcusFuture future = scoped.get("key"); + + // then + ExecutionException exception = assertThrows( + ExecutionException.class, + () -> future.get(300L, TimeUnit.MILLISECONDS) + ); + + assertInstanceOf(OperationTimeoutException.class, exception.getCause()); + assertTrue(exception.getCause().getMessage().contains("after 1 milliseconds")); + } + +}