From 6e8d01c20ecfb98458cc887aa48cc103ef4a7f12 Mon Sep 17 00:00:00 2001 From: Shih-Hao Yeh Date: Mon, 26 Oct 2026 13:09:31 -0700 Subject: [PATCH 1/2] Mark timed-out mutates and reset the timeout counter on answers When the mutate latch times out, call op.timeOut() so an op still in the write queue is dropped instead of being applied later, and feed MemcachedConnection.opTimedOut so the node's continuous-timeout counter counts it. The timer is tagged ipc.status=timeout with the failed host. When the node answers (any status except timed out or cancelled), call MemcachedConnection.opSucceeded so the counter stays consecutive on clients that mostly incr/decr. The latch result decides the outcome, so a reply racing in after the deadline cannot flip it. The status holder is now an AtomicReference because op.timeOut() runs the callback on the caller thread. --- .../spy/memcached/EVCacheMemcachedClient.java | 16 +- .../EVCacheMemcachedClientMutateTest.java | 406 ++++++++++++++++++ evcache-core/src/test/java/test-suite.xml | 1 + 3 files changed, 420 insertions(+), 3 deletions(-) create mode 100644 evcache-core/src/test/java/net/spy/memcached/EVCacheMemcachedClientMutateTest.java diff --git a/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java b/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java index 8272f0cb..809f29d7 100644 --- a/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java +++ b/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java @@ -68,6 +68,7 @@ public class EVCacheMemcachedClient extends MemcachedClient { private static final Logger log = LoggerFactory.getLogger(EVCacheMemcachedClient.class); + private static final OperationStatus MUTATE_TIMED_OUT = new OperationStatus(false, "timed out", StatusCode.TIMEDOUT); private final String appName; private final Property readTimeout; @@ -801,11 +802,11 @@ public long mutate(final Mutator m, String key, long by, long def, int exp) { final long start = System.currentTimeMillis(); final AtomicLong rv = new AtomicLong(); final CountDownLatch latch = new CountDownLatch(1); - final List statusList = new ArrayList(1); + final AtomicReference statusRef = new AtomicReference(); final Operation op = opFact.mutate(m, key, by, def, exp, new OperationCallback() { @Override public void receivedStatus(OperationStatus s) { - statusList.add(s); + statusRef.compareAndSet(null, s); rv.set(new Long(s.isSuccess() ? s.getMessage() : "-1")); } @@ -816,6 +817,7 @@ public void complete() { }); mconn.enqueueOperation(key, op); long retVal = def; + OperationStatus status = null; try { if(mutateOperationTimeout == null) { mutateOperationTimeout = EVCacheConfig.getInstance().getPropertyRepository().get(appName + ".mutate.timeout", Long.class).orElse(connectionFactory.getOperationTimeout()); @@ -824,9 +826,18 @@ public void complete() { if (!latch.await(mutateOperationTimeout.get(), TimeUnit.MILLISECONDS)) { if (log.isDebugEnabled()) log.debug("Mutation operation timeout. Will return -1"); + // the latch decides the outcome, even if a reply races in before timeOut() + op.timeOut(); + MemcachedConnection.opTimedOut(op); retVal = -1; + status = MUTATE_TIMED_OUT; } else { retVal = rv.get(); + status = statusRef.get(); + if (status != null && status.getStatusCode() != StatusCode.TIMEDOUT && status.getStatusCode() != StatusCode.CANCELLED) { + // the node answered (even with an error), so reset the continuous timeout counter + MemcachedConnection.opSucceeded(op); + } } } catch (Exception e) { @@ -834,7 +845,6 @@ public void complete() { + "; val : " + retVal + "; Elapsed Time - " + (System.currentTimeMillis() - start), e); } - final OperationStatus status = statusList.size() > 0 ? statusList.get(0) : null; final String host = ((status != null && status.getStatusCode().equals(StatusCode.TIMEDOUT) && op != null) ? getHostName(op.getHandlingNode().getSocketAddress()) : null); getTimer(operationStr, EVCacheMetricsFactory.WRITE, status, null, host, getWriteMetricMaxValue()).record((System.currentTimeMillis() - start), TimeUnit.MILLISECONDS); if (log.isDebugEnabled() && client.getPool().getEVCacheClientPoolManager().shouldLog(appName)) log.debug(operationStr + " Key : " + key + "; by : " + by + "; default : " + def + "; exp : " + exp diff --git a/evcache-core/src/test/java/net/spy/memcached/EVCacheMemcachedClientMutateTest.java b/evcache-core/src/test/java/net/spy/memcached/EVCacheMemcachedClientMutateTest.java new file mode 100644 index 00000000..b55fe786 --- /dev/null +++ b/evcache-core/src/test/java/net/spy/memcached/EVCacheMemcachedClientMutateTest.java @@ -0,0 +1,406 @@ +package net.spy.memcached; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; + +import java.io.Closeable; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.channels.SocketChannel; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BooleanSupplier; +import java.util.stream.Collectors; + +import org.testng.annotations.AfterClass; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import com.netflix.archaius.DefaultPropertyFactory; +import com.netflix.archaius.config.DefaultSettableConfig; +import com.netflix.evcache.metrics.EVCacheMetricsFactory; +import com.netflix.evcache.operation.EVCacheAsciiOperationFactory; +import com.netflix.evcache.pool.EVCacheClient; +import com.netflix.evcache.pool.EVCacheClientPool; +import com.netflix.evcache.pool.EVCacheClientPoolManager; +import com.netflix.evcache.util.EVCacheConfig; +import com.netflix.spectator.api.BasicTag; +import com.netflix.spectator.api.Counter; +import com.netflix.spectator.api.DefaultRegistry; +import com.netflix.spectator.api.Spectator; +import com.netflix.spectator.api.Tag; +import com.netflix.spectator.api.Timer; + +import net.spy.memcached.ops.Mutator; +import net.spy.memcached.ops.MutatorOperation; +import net.spy.memcached.ops.Operation; +import net.spy.memcached.ops.OperationCallback; +import net.spy.memcached.protocol.ascii.EVCacheAsciiNodeImpl; + +/** + * Runs a real {@link EVCacheMemcachedClient} (ascii protocol) against a fake memcached on 127.0.0.1 that + * stays silent until told to answer. + */ +public class EVCacheMemcachedClientMutateTest { + + private static final long MUTATE_TIMEOUT_MS = 400; + private static final String KEY = "mutate-test-key"; + private static final String NOT_FOUND = "NF\r\n"; + + private final DefaultSettableConfig config = new DefaultSettableConfig(); + private final DefaultRegistry registry = new DefaultRegistry(); + + private String appName; + private TestConnectionFactory connectionFactory; + + private FakeMemcached server; + private EVCacheMemcachedClient memcachedClient; + private EVCacheNode node; + + @BeforeClass + public void setUpConfig() { + new EVCacheConfig(new DefaultPropertyFactory(config)); + Spectator.globalRegistry().add(registry); + } + + @AfterClass(alwaysRun = true) + public void tearDownConfig() { + Spectator.globalRegistry().remove(registry); + } + + @BeforeMethod + public void setUp() throws Exception { + // a new app name per test keeps the (static, cached) timers of different tests apart + appName = "MUTATETEST" + System.nanoTime(); + config.setProperty(appName + ".mutate.timeout", MUTATE_TIMEOUT_MS); + server = new FakeMemcached(0); + startClient(server.address(), true); + } + + private void startClient(InetSocketAddress address, boolean awaitConnected) throws Exception { + final EVCacheClient client = mock(EVCacheClient.class); + final EVCacheClientPool pool = mock(EVCacheClientPool.class); + final EVCacheClientPoolManager poolManager = mock(EVCacheClientPoolManager.class); + final List tags = new ArrayList(); + tags.add(new BasicTag(EVCacheMetricsFactory.CACHE, appName)); + when(client.getAppName()).thenReturn(appName); + when(client.getTagList()).thenReturn(tags); + when(client.getServerGroupName()).thenReturn("sg"); + when(client.getId()).thenReturn(1); + when(client.getOperationCounter()).thenReturn(mock(Counter.class)); + when(client.getPool()).thenReturn(pool); + when(pool.getEVCacheClientPoolManager()).thenReturn(poolManager); + + final List addrs = Collections.singletonList(address); + connectionFactory = new TestConnectionFactory(client); + memcachedClient = new EVCacheMemcachedClient(connectionFactory, addrs, null, client); + node = (EVCacheNode) memcachedClient.getEVCacheNode(KEY); + // the node must be connected so that the op is actually written and the server sees it + if (awaitConnected) awaitTrue("node connected", () -> node.isActive()); + } + + @AfterMethod(alwaysRun = true) + public void tearDown() throws Exception { + if (memcachedClient != null) memcachedClient.shutdown(1, TimeUnit.SECONDS); + if (server != null) server.close(); + } + + @Test + public void silentServerTimesOutAndRecordsTimeoutTimer() throws Exception { + final long start = System.nanoTime(); + final long result = memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0); + final long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + + assertEquals(result, -1); + assertTrue(elapsedMs >= MUTATE_TIMEOUT_MS - 50, "returned too early: " + elapsedMs); + assertTrue(elapsedMs < MUTATE_TIMEOUT_MS + 2000, "returned too late: " + elapsedMs); + awaitTrue("server received the command", () -> server.received() == 1); + assertTrue(connectionFactory.lastMutateOp().isTimedOut()); + assertEquals(node.getContinuousTimeout(), 1); + + final Timer timer = findCallTimer("timeout", "failure"); + assertNotNull(timer, "no ipc.client.call timer with ipc.status=timeout, ipc.result=failure"); + assertEquals(timer.count(), 1); + assertTrue(tagsOf(timer).contains(new BasicTag(EVCacheMetricsFactory.FAILED_HOST, server.address().getHostName()))); + } + + @Test + public void mutateThatTimedOutBeforeBeingSentIsNeverApplied() throws Exception { + // nothing listens on the port yet, so the op cannot be written and times out in the queue + final int port; + try (ServerSocket probe = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) { + port = probe.getLocalPort(); + } + memcachedClient.shutdown(1, TimeUnit.SECONDS); + server.close(); + startClient(new InetSocketAddress("127.0.0.1", port), false); + + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); + assertTrue(connectionFactory.lastMutateOp().isTimedOut()); + // the node was not connected, so opTimedOut does not count it (setContinuousTimeout counts only active nodes) + assertEquals(node.getContinuousTimeout(), 0); + + // bring the server up on that port: once the node reconnects the IO thread drops the op instead of writing it + server = new FakeMemcached(port); + awaitTrue("node reconnected", () -> node.isActive()); + awaitTrue("timed out op dropped unsent", () -> connectionFactory.lastMutateOp().isTimedOutUnsent()); + assertEquals(server.received(), 0); + + // the connection is usable and the dropped op was the only one that never reached the server + server.startReplying(valueReply(5)); + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), 5); + assertEquals(server.received(), 1); + } + + @Test + public void everyTimedOutMutateIsCountedAndAnAnswerResetsTheCounter() throws Exception { + for (int i = 1; i <= 3; i++) { + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); + assertEquals(node.getContinuousTimeout(), i); + } + + server.startReplying(valueReply(42)); + + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), 42); + assertEquals(node.getContinuousTimeout(), 0); + } + + @Test + public void serverErrorAnswerResetsTheCounterAndIsNotATimeout() throws Exception { + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); + assertEquals(node.getContinuousTimeout(), 2); + + server.startReplying(NOT_FOUND); + + assertEquals(memcachedClient.mutate(Mutator.decr, KEY, 1, 0, 0), -1); + assertFalse(connectionFactory.lastMutateOp().isTimedOut()); + assertEquals(node.getContinuousTimeout(), 0); + } + + @Test + public void cancelledOpLeavesTheCounterUnchangedAndIsNotATimeout() throws Exception { + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); + assertEquals(node.getContinuousTimeout(), 2); + + // cancel the op once it is in flight, the way a reconnect or a failure mode would + final Thread canceller = new Thread(() -> { + try { + awaitTrue("mutate op", () -> connectionFactory.lastMutateOp() != null && connectionFactory.mutateOps() == 3); + connectionFactory.lastMutateOp().cancel(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + canceller.start(); + final long result = memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0); + canceller.join(); + + assertEquals(result, -1); + assertFalse(connectionFactory.lastMutateOp().isTimedOut()); + assertEquals(node.getContinuousTimeout(), 2); + assertNotNull(findCallTimer("cancelled", "failure")); + } + + @Test + public void answeredMutateReturnsTheServerValue() throws Exception { + server.startReplying(valueReply(42)); + + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), 42); + assertFalse(connectionFactory.lastMutateOp().isTimedOut()); + assertEquals(node.getContinuousTimeout(), 0); + + server.startReplying(valueReply(7)); + assertEquals(memcachedClient.incr(KEY, 1, 0, 0), 7); + assertEquals(memcachedClient.decr(KEY, 1, 0, 0), 7); + } + + private Timer findCallTimer(String ipcStatus, String ipcResult) { + for (Timer t : registry.timers().collect(Collectors.toList())) { + final List tags = tagsOf(t); + if (t.id().name().equals(EVCacheMetricsFactory.IPC_CALL) + && tags.contains(new BasicTag(EVCacheMetricsFactory.CACHE, appName)) + && tags.contains(new BasicTag(EVCacheMetricsFactory.CALL_TAG, "incr")) + && tags.contains(new BasicTag(EVCacheMetricsFactory.IPC_STATUS, ipcStatus)) + && tags.contains(new BasicTag(EVCacheMetricsFactory.IPC_RESULT, ipcResult))) { + return t; + } + } + return null; + } + + private static List tagsOf(Timer timer) { + final List tags = new ArrayList(); + timer.id().tags().forEach(tags::add); + return tags; + } + + private static String valueReply(long value) { + final String s = Long.toString(value); + return "VA " + s.length() + "\r\n" + s + "\r\n"; + } + + private static void awaitTrue(String what, BooleanSupplier condition) throws InterruptedException { + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (!condition.getAsBoolean()) { + if (System.nanoTime() > deadline) throw new AssertionError("Timed out waiting for " + what); + Thread.sleep(5); + } + } + + /** Ascii connection factory that builds the same node, connection and operation factory as BaseAsciiConnectionFactory. */ + private static class TestConnectionFactory extends DefaultConnectionFactory { + private final EVCacheClient client; + private final long startTime = System.currentTimeMillis(); + private final AtomicInteger mutateOps = new AtomicInteger(); + private volatile Operation lastMutateOp; + private final OperationFactory operationFactory = new EVCacheAsciiOperationFactory() { + @Override + public MutatorOperation mutate(Mutator m, String key, long by, long def, int exp, OperationCallback cb) { + final MutatorOperation op = super.mutate(m, key, by, def, exp, cb); + lastMutateOp = op; + mutateOps.incrementAndGet(); + return op; + } + }; + + TestConnectionFactory(EVCacheClient client) { + super(); + this.client = client; + } + + @Override + public MemcachedConnection createConnection(List addrs) throws IOException { + return new EVCacheConnection("mutate-test", getReadBufSize(), this, addrs, getInitialObservers(), + getFailureMode(), getOperationFactory()); + } + + @Override + public OperationFactory getOperationFactory() { + return operationFactory; + } + + Operation lastMutateOp() { + return lastMutateOp; + } + + int mutateOps() { + return mutateOps.get(); + } + + @Override + public MemcachedNode createMemcachedNode(java.net.SocketAddress sa, SocketChannel c, int bufSize) { + return new EVCacheAsciiNodeImpl(sa, c, bufSize, createReadOperationQueue(), createWriteOperationQueue(), + new java.util.concurrent.ArrayBlockingQueue(getOpQueueLen()), getOpQueueMaxBlockTime(), + false, getOperationTimeout(), getAuthWaitTime(), this, client, startTime); + } + + @Override + public boolean isDaemon() { + return true; + } + } + + /** + * Minimal fake memcached: counts the meta-arithmetic ("ma ...") request lines it receives and keeps them + * unanswered until {@link #startReplying}, which also answers the backlog in order. + */ + private static class FakeMemcached implements Closeable { + private final ServerSocket serverSocket; + private final List sockets = new CopyOnWriteArrayList(); + private final List unanswered = new ArrayList(); + private int received; + private String reply; + + FakeMemcached(int port) throws IOException { + serverSocket = new ServerSocket(port, 50, InetAddress.getByName("127.0.0.1")); + final Thread acceptor = new Thread(this::acceptLoop, "fake-memcached-accept"); + acceptor.setDaemon(true); + acceptor.start(); + } + + InetSocketAddress address() { + return new InetSocketAddress("127.0.0.1", serverSocket.getLocalPort()); + } + + synchronized int received() { + return received; + } + + synchronized void startReplying(String replyBytes) throws IOException { + reply = replyBytes; + for (OutputStream out : unanswered) write(out); + unanswered.clear(); + } + + private void write(OutputStream out) throws IOException { + out.write(reply.getBytes(StandardCharsets.ISO_8859_1)); + out.flush(); + } + + private void acceptLoop() { + try { + while (!serverSocket.isClosed()) { + final Socket s = serverSocket.accept(); + sockets.add(s); + final Thread t = new Thread(() -> serve(s), "fake-memcached-conn"); + t.setDaemon(true); + t.start(); + } + } catch (IOException e) { + // closed + } + } + + private void serve(Socket s) { + try { + final InputStream in = s.getInputStream(); + final OutputStream out = s.getOutputStream(); + final StringBuilder line = new StringBuilder(); + int b; + while ((b = in.read()) != -1) { + if (b == '\n') { + if (line.toString().startsWith("ma ")) onCommand(out); + line.setLength(0); + } else if (b != '\r') { + line.append((char) b); + } + } + } catch (IOException e) { + // closed + } + } + + private synchronized void onCommand(OutputStream out) throws IOException { + received++; + if (reply != null) { + write(out); + } else { + unanswered.add(out); + } + } + + @Override + public void close() throws IOException { + serverSocket.close(); + for (Socket s : sockets) s.close(); + } + } +} diff --git a/evcache-core/src/test/java/test-suite.xml b/evcache-core/src/test/java/test-suite.xml index 6924a679..b92925cc 100644 --- a/evcache-core/src/test/java/test-suite.xml +++ b/evcache-core/src/test/java/test-suite.xml @@ -6,6 +6,7 @@ + From 17d8cd2d4d1d56df913a7a9427b54c8d1268d5bd Mon Sep 17 00:00:00 2001 From: Shih-Hao Yeh Date: Mon, 26 Oct 2026 15:57:56 -0700 Subject: [PATCH 2/2] Address review: tighten mutate tests and null-guard the timeout host tag - main: use BaseOperationImpl.TIMED_OUT instead of a private duplicate, move the "latch decides" comment next to the status it explains, and null-guard the handling node when building the evc.failed.host tag - tests: rewrite the unsent-op test so it fails without op.timeOut() (node operation timeout 60 s, assert the server saw only the second mutate) - tests: add late-reply-races-the-timeout, first-status-wins and no-handling-node tests - tests: rename the NF test, fix the TestConnectionFactory javadoc, drop a redundant null check, import SocketAddress and ArrayBlockingQueue, and give the canceller test a 2 s mutate timeout --- .../spy/memcached/EVCacheMemcachedClient.java | 9 +- .../EVCacheMemcachedClientMutateTest.java | 128 +++++++++++++++--- 2 files changed, 115 insertions(+), 22 deletions(-) diff --git a/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java b/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java index 809f29d7..d71befae 100644 --- a/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java +++ b/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java @@ -55,6 +55,7 @@ import net.spy.memcached.ops.StatusCode; import net.spy.memcached.ops.StoreOperation; import net.spy.memcached.ops.StoreType; +import net.spy.memcached.protocol.BaseOperationImpl; import net.spy.memcached.protocol.ascii.ExecCmdOperation; import net.spy.memcached.protocol.ascii.MetaDebugOperation; import net.spy.memcached.protocol.ascii.MetaGetOperation; @@ -68,7 +69,6 @@ public class EVCacheMemcachedClient extends MemcachedClient { private static final Logger log = LoggerFactory.getLogger(EVCacheMemcachedClient.class); - private static final OperationStatus MUTATE_TIMED_OUT = new OperationStatus(false, "timed out", StatusCode.TIMEDOUT); private final String appName; private final Property readTimeout; @@ -826,11 +826,11 @@ public void complete() { if (!latch.await(mutateOperationTimeout.get(), TimeUnit.MILLISECONDS)) { if (log.isDebugEnabled()) log.debug("Mutation operation timeout. Will return -1"); - // the latch decides the outcome, even if a reply races in before timeOut() op.timeOut(); MemcachedConnection.opTimedOut(op); retVal = -1; - status = MUTATE_TIMED_OUT; + // the latch decides the outcome, even if a reply races in before timeOut() + status = BaseOperationImpl.TIMED_OUT; } else { retVal = rv.get(); status = statusRef.get(); @@ -845,7 +845,8 @@ public void complete() { + "; val : " + retVal + "; Elapsed Time - " + (System.currentTimeMillis() - start), e); } - final String host = ((status != null && status.getStatusCode().equals(StatusCode.TIMEDOUT) && op != null) ? getHostName(op.getHandlingNode().getSocketAddress()) : null); + final MemcachedNode handlingNode = op.getHandlingNode(); + final String host = (status != null && status.getStatusCode() == StatusCode.TIMEDOUT && handlingNode != null) ? getHostName(handlingNode.getSocketAddress()) : null; getTimer(operationStr, EVCacheMetricsFactory.WRITE, status, null, host, getWriteMetricMaxValue()).record((System.currentTimeMillis() - start), TimeUnit.MILLISECONDS); if (log.isDebugEnabled() && client.getPool().getEVCacheClientPoolManager().shouldLog(appName)) log.debug(operationStr + " Key : " + key + "; by : " + by + "; default : " + def + "; exp : " + exp + "; val : " + retVal + "; Elapsed Time - " + (System.currentTimeMillis() - start)); diff --git a/evcache-core/src/test/java/net/spy/memcached/EVCacheMemcachedClientMutateTest.java b/evcache-core/src/test/java/net/spy/memcached/EVCacheMemcachedClientMutateTest.java index b55fe786..dd89b481 100644 --- a/evcache-core/src/test/java/net/spy/memcached/EVCacheMemcachedClientMutateTest.java +++ b/evcache-core/src/test/java/net/spy/memcached/EVCacheMemcachedClientMutateTest.java @@ -5,6 +5,7 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNull; import static org.testng.Assert.assertTrue; import java.io.Closeable; @@ -15,11 +16,13 @@ import java.net.InetSocketAddress; import java.net.ServerSocket; import java.net.Socket; +import java.net.SocketAddress; import java.nio.channels.SocketChannel; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -51,7 +54,10 @@ import net.spy.memcached.ops.MutatorOperation; import net.spy.memcached.ops.Operation; import net.spy.memcached.ops.OperationCallback; +import net.spy.memcached.ops.OperationStatus; +import net.spy.memcached.protocol.BaseOperationImpl; import net.spy.memcached.protocol.ascii.EVCacheAsciiNodeImpl; +import net.spy.memcached.protocol.ascii.MetaArithmeticOperationImpl; /** * Runs a real {@link EVCacheMemcachedClient} (ascii protocol) against a fake memcached on 127.0.0.1 that @@ -72,6 +78,7 @@ public class EVCacheMemcachedClientMutateTest { private FakeMemcached server; private EVCacheMemcachedClient memcachedClient; private EVCacheNode node; + private boolean unauthenticatedNodes; @BeforeClass public void setUpConfig() { @@ -89,11 +96,16 @@ public void setUp() throws Exception { // a new app name per test keeps the (static, cached) timers of different tests apart appName = "MUTATETEST" + System.nanoTime(); config.setProperty(appName + ".mutate.timeout", MUTATE_TIMEOUT_MS); + unauthenticatedNodes = false; server = new FakeMemcached(0); startClient(server.address(), true); } private void startClient(InetSocketAddress address, boolean awaitConnected) throws Exception { + startClient(address, awaitConnected, DefaultConnectionFactory.DEFAULT_OPERATION_TIMEOUT); + } + + private void startClient(InetSocketAddress address, boolean awaitConnected, long operationTimeoutMs) throws Exception { final EVCacheClient client = mock(EVCacheClient.class); final EVCacheClientPool pool = mock(EVCacheClientPool.class); final EVCacheClientPoolManager poolManager = mock(EVCacheClientPoolManager.class); @@ -108,7 +120,7 @@ private void startClient(InetSocketAddress address, boolean awaitConnected) thro when(pool.getEVCacheClientPoolManager()).thenReturn(poolManager); final List addrs = Collections.singletonList(address); - connectionFactory = new TestConnectionFactory(client); + connectionFactory = new TestConnectionFactory(client, operationTimeoutMs, unauthenticatedNodes); memcachedClient = new EVCacheMemcachedClient(connectionFactory, addrs, null, client); node = (EVCacheNode) memcachedClient.getEVCacheNode(KEY); // the node must be connected so that the op is actually written and the server sees it @@ -149,20 +161,15 @@ public void mutateThatTimedOutBeforeBeingSentIsNeverApplied() throws Exception { } memcachedClient.shutdown(1, TimeUnit.SECONDS); server.close(); - startClient(new InetSocketAddress("127.0.0.1", port), false); + // the node's operation timeout is far above the mutate timeout, so only op.timeOut() keeps the op off the wire + startClient(new InetSocketAddress("127.0.0.1", port), false, TimeUnit.SECONDS.toMillis(60)); assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); assertTrue(connectionFactory.lastMutateOp().isTimedOut()); - // the node was not connected, so opTimedOut does not count it (setContinuousTimeout counts only active nodes) - assertEquals(node.getContinuousTimeout(), 0); - // bring the server up on that port: once the node reconnects the IO thread drops the op instead of writing it + // once the node connects, a second mutate is answered; the first one must not have been written before it server = new FakeMemcached(port); awaitTrue("node reconnected", () -> node.isActive()); - awaitTrue("timed out op dropped unsent", () -> connectionFactory.lastMutateOp().isTimedOutUnsent()); - assertEquals(server.received(), 0); - - // the connection is usable and the dropped op was the only one that never reached the server server.startReplying(valueReply(5)); assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), 5); assertEquals(server.received(), 1); @@ -182,7 +189,7 @@ public void everyTimedOutMutateIsCountedAndAnAnswerResetsTheCounter() throws Exc } @Test - public void serverErrorAnswerResetsTheCounterAndIsNotATimeout() throws Exception { + public void notFoundAnswerResetsTheCounterAndIsNotATimeout() throws Exception { assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); assertEquals(node.getContinuousTimeout(), 2); @@ -200,10 +207,12 @@ public void cancelledOpLeavesTheCounterUnchangedAndIsNotATimeout() throws Except assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); assertEquals(node.getContinuousTimeout(), 2); + // a stalled canceller thread must not turn the cancel into a timeout + config.setProperty(appName + ".mutate.timeout", 2000L); // cancel the op once it is in flight, the way a reconnect or a failure mode would final Thread canceller = new Thread(() -> { try { - awaitTrue("mutate op", () -> connectionFactory.lastMutateOp() != null && connectionFactory.mutateOps() == 3); + awaitTrue("mutate op", () -> connectionFactory.mutateOps() == 3); connectionFactory.lastMutateOp().cancel(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -219,6 +228,48 @@ public void cancelledOpLeavesTheCounterUnchangedAndIsNotATimeout() throws Except assertNotNull(findCallTimer("cancelled", "failure")); } + @Test + public void replyThatRacesTheTimeoutDoesNotTurnItIntoASuccess() throws Exception { + connectionFactory.replyBeforeTimeOut = true; + + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); + assertEquals(node.getContinuousTimeout(), 1); + + final Timer timer = findCallTimer("timeout", "failure"); + assertNotNull(timer, "no ipc.client.call timer with ipc.status=timeout, ipc.result=failure"); + assertEquals(timer.count(), 1); + assertTrue(tagsOf(timer).contains(new BasicTag(EVCacheMetricsFactory.FAILED_HOST, server.address().getHostName()))); + } + + @Test + public void firstStatusWins() throws Exception { + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); + assertEquals(node.getContinuousTimeout(), 2); + + // the op reports CANCELLED right after the answer, but the answer came first + connectionFactory.cancelAfterStatus = true; + server.startReplying(valueReply(9)); + + memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0); + assertEquals(node.getContinuousTimeout(), 0); + } + + @Test + public void timedOutMutateWithoutHandlingNodeStillReturns() throws Exception { + // an unauthenticated node defers the op without assigning it a handling node + memcachedClient.shutdown(1, TimeUnit.SECONDS); + unauthenticatedNodes = true; + startClient(server.address(), false); + + assertEquals(memcachedClient.mutate(Mutator.incr, KEY, 1, 0, 0), -1); + assertNull(connectionFactory.lastMutateOp().getHandlingNode()); + + final Timer timer = findCallTimer("timeout", "failure"); + assertNotNull(timer, "no ipc.client.call timer with ipc.status=timeout, ipc.result=failure"); + assertTrue(tagsOf(timer).stream().noneMatch(t -> t.key().equals(EVCacheMetricsFactory.FAILED_HOST))); + } + @Test public void answeredMutateReturnsTheServerValue() throws Exception { server.startReplying(valueReply(42)); @@ -265,25 +316,61 @@ private static void awaitTrue(String what, BooleanSupplier condition) throws Int } } - /** Ascii connection factory that builds the same node, connection and operation factory as BaseAsciiConnectionFactory. */ + /** + * Ascii connection factory like BaseAsciiConnectionFactory, except that it skips registerMonitors() and wraps the + * operation factory to record the mutate ops. Per-test switches: replyBeforeTimeOut, cancelAfterStatus, unauthenticated. + */ private static class TestConnectionFactory extends DefaultConnectionFactory { private final EVCacheClient client; private final long startTime = System.currentTimeMillis(); + private final long operationTimeoutMs; + private final boolean unauthenticated; private final AtomicInteger mutateOps = new AtomicInteger(); + // the op delivers the server's answer from timeOut(), just before it reports the timeout + volatile boolean replyBeforeTimeOut; + // the op's callback gets CANCELLED right after every status + volatile boolean cancelAfterStatus; private volatile Operation lastMutateOp; private final OperationFactory operationFactory = new EVCacheAsciiOperationFactory() { @Override - public MutatorOperation mutate(Mutator m, String key, long by, long def, int exp, OperationCallback cb) { - final MutatorOperation op = super.mutate(m, key, by, def, exp, cb); + public MutatorOperation mutate(Mutator m, String key, long by, long def, int exp, final OperationCallback cb) { + final OperationCallback callback = !cancelAfterStatus ? cb : new OperationCallback() { + @Override + public void receivedStatus(OperationStatus s) { + cb.receivedStatus(s); + cb.receivedStatus(BaseOperationImpl.CANCELLED); + } + + @Override + public void complete() { + cb.complete(); + } + }; + final MutatorOperation op = !replyBeforeTimeOut ? super.mutate(m, key, by, def, exp, callback) + : new MetaArithmeticOperationImpl(m, key, by, def, exp, callback) { + @Override + public synchronized void timeOut() { + handleLine("VA 2"); + handleLine("42"); + super.timeOut(); + } + }; lastMutateOp = op; mutateOps.incrementAndGet(); return op; } }; - TestConnectionFactory(EVCacheClient client) { + TestConnectionFactory(EVCacheClient client, long operationTimeoutMs, boolean unauthenticated) { super(); this.client = client; + this.operationTimeoutMs = operationTimeoutMs; + this.unauthenticated = unauthenticated; + } + + @Override + public long getOperationTimeout() { + return operationTimeoutMs; } @Override @@ -306,10 +393,15 @@ int mutateOps() { } @Override - public MemcachedNode createMemcachedNode(java.net.SocketAddress sa, SocketChannel c, int bufSize) { + public MemcachedNode createMemcachedNode(SocketAddress sa, SocketChannel c, int bufSize) { return new EVCacheAsciiNodeImpl(sa, c, bufSize, createReadOperationQueue(), createWriteOperationQueue(), - new java.util.concurrent.ArrayBlockingQueue(getOpQueueLen()), getOpQueueMaxBlockTime(), - false, getOperationTimeout(), getAuthWaitTime(), this, client, startTime); + new ArrayBlockingQueue(getOpQueueLen()), getOpQueueMaxBlockTime(), + false, getOperationTimeout(), getAuthWaitTime(), this, client, startTime) { + @Override + public boolean isAuthenticated() { + return !unauthenticated && super.isAuthenticated(); + } + }; } @Override