diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 42630eeae2e3..86c68bd2fe4a 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -1116,6 +1116,16 @@ uuid_sstable_identifiers_enabled: false # Lowest acceptable value is 10 ms. # Min unit: ms read_request_timeout: 5000ms + +# What the coordinator does when a read message is dropped because the send queue for a +# replica is full. This happens while a replica is unreachable and gossip has not yet marked +# it down. +# If true, the coordinator sends the read to another candidate replica, as a speculative +# retry. The table must permit speculative retry, so a table with speculative_retry: NONE +# keeps the behaviour below. +# If false, the coordinator fails the read. +# read_fallback_on_overloaded_connection: false + # How long the coordinator should wait for seq or index scans to complete. # Lowest acceptable value is 10 ms. # Min unit: ms diff --git a/doc/modules/cassandra/pages/operating/metrics.adoc b/doc/modules/cassandra/pages/operating/metrics.adoc index 11a204f740d9..3c89bc4f69fc 100644 --- a/doc/modules/cassandra/pages/operating/metrics.adoc +++ b/doc/modules/cassandra/pages/operating/metrics.adoc @@ -223,6 +223,10 @@ ongoing incremental repair |SpeculativeRetries |Counter |Number of times speculative retries were sent for this table. +|OverloadSpeculativeRetries |Counter |Number of speculative retries that +a read message dropped on an overloaded connection triggered. A subset of +SpeculativeRetries. + |SpeculativeFailedRetries |Counter |Number of speculative retries that failed to prevent a timeout diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 21ca1b595cd1..29859b9cf888 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -132,6 +132,8 @@ public static Set splitCommaDelimited(String src) @Replaces(oldName = "read_request_timeout_in_ms", converter = Converters.MILLIS_DURATION_LONG, deprecated = true) public volatile DurationSpec.LongMillisecondsBound read_request_timeout = new DurationSpec.LongMillisecondsBound("5000ms"); + public volatile boolean read_fallback_on_overloaded_connection = false; + @Replaces(oldName = "range_request_timeout_in_ms", converter = Converters.MILLIS_DURATION_LONG, deprecated = true) public volatile DurationSpec.LongMillisecondsBound range_request_timeout = new DurationSpec.LongMillisecondsBound("10000ms"); diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 82ef40f6ee09..172f5b178c3f 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -1836,6 +1836,16 @@ public static void setReadRpcTimeout(long timeOutInMillis) conf.read_request_timeout = new DurationSpec.LongMillisecondsBound(timeOutInMillis); } + public static boolean getReadFallbackOnOverloadedConnection() + { + return conf.read_fallback_on_overloaded_connection; + } + + public static void setReadFallbackOnOverloadedConnection(boolean enabled) + { + conf.read_fallback_on_overloaded_connection = enabled; + } + public static long getRangeRpcTimeout(TimeUnit unit) { return conf.range_request_timeout.to(unit); diff --git a/src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java b/src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java index 71feb48508b7..4028f39ffb77 100644 --- a/src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java +++ b/src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java @@ -101,6 +101,8 @@ public class KeyspaceMetrics public final LatencyMetrics idealCLWriteLatency; /** Speculative retries **/ public final Counter speculativeRetries; + /** The subset of speculative retries that a dropped read message triggered **/ + public final Counter overloadSpeculativeRetries; /** Speculative retry occured but still timed out **/ public final Counter speculativeFailedRetries; /** Needed to speculate, but didn't have enough replicas **/ @@ -238,6 +240,7 @@ public KeyspaceMetrics(final Keyspace ks) idealCLWriteLatency = createLatencyMetrics("IdealCLWrite"); speculativeRetries = createKeyspaceCounter("SpeculativeRetries", metric -> metric.speculativeRetries.getCount()); + overloadSpeculativeRetries = createKeyspaceCounter("OverloadSpeculativeRetries", metric -> metric.overloadSpeculativeRetries.getCount()); speculativeFailedRetries = createKeyspaceCounter("SpeculativeFailedRetries", metric -> metric.speculativeFailedRetries.getCount()); speculativeInsufficientReplicas = createKeyspaceCounter("SpeculativeInsufficientReplicas", metric -> metric.speculativeInsufficientReplicas.getCount()); additionalWrites = createKeyspaceCounter("AdditionalWrites", metric -> metric.additionalWrites.getCount()); diff --git a/src/java/org/apache/cassandra/metrics/TableMetrics.java b/src/java/org/apache/cassandra/metrics/TableMetrics.java index 24c4b16153e6..5aabfdef44d5 100644 --- a/src/java/org/apache/cassandra/metrics/TableMetrics.java +++ b/src/java/org/apache/cassandra/metrics/TableMetrics.java @@ -221,6 +221,7 @@ public class TableMetrics private final MetricNameFactory aliasFactory; public final Counter speculativeRetries; + public final Counter overloadSpeculativeRetries; public final Counter speculativeFailedRetries; public final Counter speculativeInsufficientReplicas; public final Gauge speculativeSampleLatencyNanos; @@ -823,6 +824,7 @@ public Long getValue() } }); speculativeRetries = createTableCounter("SpeculativeRetries"); + overloadSpeculativeRetries = createTableCounter("OverloadSpeculativeRetries"); speculativeFailedRetries = createTableCounter("SpeculativeFailedRetries"); speculativeInsufficientReplicas = createTableCounter("SpeculativeInsufficientReplicas"); speculativeSampleLatencyNanos = createTableGauge("SpeculativeSampleLatencyNanos", () -> MICROSECONDS.toNanos(cfs.sampleReadLatencyMicros)); diff --git a/src/java/org/apache/cassandra/net/OutboundConnection.java b/src/java/org/apache/cassandra/net/OutboundConnection.java index 821521bfb932..40edb17c996b 100644 --- a/src/java/org/apache/cassandra/net/OutboundConnection.java +++ b/src/java/org/apache/cassandra/net/OutboundConnection.java @@ -339,8 +339,10 @@ public void enqueue(Message message) throws ClosedChannelException // this is an optimisation only; messages will be expired on ~100ms cycle, and by Delivery when it runs if (queue.maybePruneExpired() && SUCCESS == acquireCapacity(canonicalSize)) break; + onOverloaded(message, INSUFFICIENT_ENDPOINT); + return; case INSUFFICIENT_GLOBAL: - onOverloaded(message); + onOverloaded(message, INSUFFICIENT_GLOBAL); return; } @@ -452,7 +454,7 @@ private void releaseCapacity(long count, long bytes) } } - private void onOverloaded(Message message) + private void onOverloaded(Message message, Outcome outcome) { overloadedCountUpdater.incrementAndGet(this); @@ -463,7 +465,7 @@ private void onOverloaded(Message message) this, FBUtilities.prettyPrintMemory(canonicalSize), readablePendingBytes, readableReserveEndpointUsing, readableReserveGlobalUsing); - callbacks.onOverloaded(message, template.to); + callbacks.onOverloaded(message, template.to, outcome); } /** diff --git a/src/java/org/apache/cassandra/net/OutboundMessageCallbacks.java b/src/java/org/apache/cassandra/net/OutboundMessageCallbacks.java index abf3f4117d0e..f367365c3c43 100644 --- a/src/java/org/apache/cassandra/net/OutboundMessageCallbacks.java +++ b/src/java/org/apache/cassandra/net/OutboundMessageCallbacks.java @@ -18,11 +18,17 @@ package org.apache.cassandra.net; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.ResourceLimits.Outcome; interface OutboundMessageCallbacks { - /** A message was not enqueued to the link because too many messages are already waiting to send */ - void onOverloaded(Message message, InetAddressAndPort peer); + /** + * A message was not enqueued to the link because too many messages are already waiting to send. + * + * {@code outcome} is {@link Outcome#INSUFFICIENT_ENDPOINT} when only this peer is out of capacity, + * and {@link Outcome#INSUFFICIENT_GLOBAL} when the node-wide reserve is exhausted. + */ + void onOverloaded(Message message, InetAddressAndPort peer, Outcome outcome); /** A message was not serialized to a frame because it had expired */ void onExpired(Message message, InetAddressAndPort peer); diff --git a/src/java/org/apache/cassandra/net/RequestCallback.java b/src/java/org/apache/cassandra/net/RequestCallback.java index bd14cae1d04d..189bfbadd2c8 100644 --- a/src/java/org/apache/cassandra/net/RequestCallback.java +++ b/src/java/org/apache/cassandra/net/RequestCallback.java @@ -19,6 +19,7 @@ import org.apache.cassandra.exceptions.RequestFailureReason; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.ResourceLimits.Outcome; /** * implementors of {@link RequestCallback} need to make sure that any public methods @@ -40,6 +41,31 @@ default void onFailure(InetAddressAndPort from, RequestFailureReason failureReas { } + /** + * Called when the outbound connection to the peer was overloaded i.e., the request was dropped + * before it was sent. + * + * This method runs on the internal response stage, unless the callback asks for it to run inline + * with {@link #invokeOnOverloadedInline()}. + * + * @param outcome which capacity limit the request was refused by + */ + default void onOverloaded(InetAddressAndPort from, Outcome outcome) + { + onFailure(from, RequestFailureReason.TIMEOUT); + } + + /** + * Dictates whether the overloaded callback should run on the INTERNAL_RESPONSE threadpool or on the + * calling thread. + * + * @return true if {@link #onOverloaded} must run on the thread that dropped the request + */ + default boolean invokeOnOverloadedInline() + { + return false; + } + /** * Returns true if the callback handles failure reporting - in which case the remove host will be asked to * report failures to us in the event of a problem processing the request. diff --git a/src/java/org/apache/cassandra/net/RequestCallbacks.java b/src/java/org/apache/cassandra/net/RequestCallbacks.java index 663126ff02c7..2c63345c0212 100644 --- a/src/java/org/apache/cassandra/net/RequestCallbacks.java +++ b/src/java/org/apache/cassandra/net/RequestCallbacks.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.net; +import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -36,6 +37,7 @@ import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.Replica; import org.apache.cassandra.metrics.InternodeOutboundMetrics; +import org.apache.cassandra.net.ResourceLimits.Outcome; import org.apache.cassandra.service.AbstractWriteResponseHandler; import static java.lang.String.format; @@ -128,6 +130,12 @@ private void removeAndExpire(long id, InetAddressAndPort peer) if (null != ci) onExpired(ci); } + private void removeAndOverload(long id, InetAddressAndPort peer, Outcome outcome) + { + CallbackInfo ci = remove(id, peer); + if (null != ci) onOverloaded(ci, outcome); + } + private void expire() { long start = preciseTime.now(); @@ -164,6 +172,18 @@ private void onExpired(CallbackInfo info) INTERNAL_RESPONSE.submit(() -> info.callback.onFailure(info.peer, RequestFailureReason.TIMEOUT)); } + + private void onOverloaded(CallbackInfo info, Outcome outcome) + { + if (!info.invokeOnFailure()) + return; + + if (info.callback.invokeOnOverloadedInline()) + info.callback.onOverloaded(info.peer, outcome); + else + INTERNAL_RESPONSE.submit(() -> info.callback.onOverloaded(info.peer, outcome)); + } + void shutdownNow(boolean expireCallbacks) { executor.shutdownNow(); @@ -277,9 +297,9 @@ public String toString() } @Override - public void onOverloaded(Message message, InetAddressAndPort peer) + public void onOverloaded(Message message, InetAddressAndPort peer, Outcome outcome) { - removeAndExpire(message, peer); + removeAndOverload(message, peer, outcome); } @Override @@ -310,6 +330,16 @@ private void removeAndExpire(Message message, InetAddressAndPort peer) forwardTo.forEach(this::removeAndExpire); } + private void removeAndOverload(Message message, InetAddressAndPort peer, Outcome outcome) + { + removeAndOverload(message.id(), peer, outcome); + + /* in case of a write sent to a different DC, also fail all forwarding targets */ + ForwardingInfo forwardTo = message.forwardTo(); + if (null != forwardTo) + forwardTo.forEach((id, target) -> removeAndOverload(id, target, outcome)); + } + public static long defaultExpirationInterval() { return DatabaseDescriptor.getMinRpcTimeout(NANOSECONDS) / 2; diff --git a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java index a812b606ad51..b777fa480588 100644 --- a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java +++ b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java @@ -22,6 +22,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.DecoratedKey; @@ -172,11 +173,76 @@ private void makeRequests(ReadCommand readCommand, Iterable replicas) */ public abstract void maybeTryAdditionalReplicas(); + /** + * Perform an additional request because the outbound connection to a contacted replica was overloaded and + * dropped our request before it was sent. Unlike {@link #maybeTryAdditionalReplicas()} this never blocks, + * since we already know the request will not be answered. + * + * @return true if an additional replica was contacted + */ + abstract boolean maybeTryAdditionalReplicasOnOverload(); + + /** + * Send an extra read to the next uncontacted candidate. + * + * @return true if an additional replica was contacted + */ + boolean trySpeculativeRetry() + { + //Handle speculation stats first in case the callback fires immediately + cfs.metric.speculativeRetries.inc(); + + ReplicaPlan.ForTokenRead replicaPlan = replicaPlan(); + ReadCommand retryCommand; + Replica extraReplica; + if (handler.resolver.isDataPresent()) + { + extraReplica = replicaPlan.firstUncontactedCandidate(replica -> true); + if (extraReplica == null) + { + cfs.metric.speculativeInsufficientReplicas.inc(); + return false; + } + + retryCommand = extraReplica.isTransient() + ? command.copyAsTransientQuery(extraReplica) + : command.copyAsDigestQuery(extraReplica); + } + else + { + extraReplica = replicaPlan.firstUncontactedCandidate(Replica::isFull); + retryCommand = command; + if (extraReplica == null) + { + cfs.metric.speculativeInsufficientReplicas.inc(); + // cannot safely speculate a new data request, without more work - requests assumed to be + // unique per endpoint, and we have no full nodes left to speculate against + return false; + } + } + + // we must update the plan to include this new node, else when we come to read-repair, we may not include this + // speculated response in the data requests we make again, and we will not be able to 'speculate' an extra repair read, + // nor would we be able to speculate a new 'write' if the repair writes are insufficient + this.replicaPlan.addToContacts(extraReplica); + + if (traceState != null) + traceState.trace("speculating read retry on {}", extraReplica); + logger.trace("speculating read retry on {}", extraReplica); + + MessagingService.instance().sendWithCallback(retryCommand.createMessage(false, requestTime), extraReplica.endpoint(), handler); + + return true; + } + /** * send the initial set of requests */ public void executeAsync() { + // the handler needs us before the first send, because a send can drop its message and call back inline + handler.setExecutor(this); + EndpointsForToken selected = replicaPlan().contacts(); EndpointsForToken fullDataRequests = selected.filter(Replica::isFull, initialDataRequestCount); makeFullDataRequests(fullDataRequests); @@ -281,6 +347,16 @@ public void maybeTryAdditionalReplicas() cfs.metric.speculativeInsufficientReplicas.inc(); } } + + boolean maybeTryAdditionalReplicasOnOverload() + { + if (DatabaseDescriptor.getReadFallbackOnOverloadedConnection() && logFailedSpeculation) + { + cfs.metric.speculativeInsufficientReplicas.inc(); + } + + return false; + } } static class SpeculatingReadExecutor extends AbstractReadExecutor @@ -302,50 +378,24 @@ public void maybeTryAdditionalReplicas() { if (shouldSpeculateAndMaybeWait()) { - //Handle speculation stats first in case the callback fires immediately - cfs.metric.speculativeRetries.inc(); speculated = true; - - ReplicaPlan.ForTokenRead replicaPlan = replicaPlan(); - ReadCommand retryCommand; - Replica extraReplica; - if (handler.resolver.isDataPresent()) - { - extraReplica = replicaPlan.firstUncontactedCandidate(replica -> true); - - // we should only use a SpeculatingReadExecutor if we have an extra replica to speculate against - assert extraReplica != null; - - retryCommand = extraReplica.isTransient() - ? command.copyAsTransientQuery(extraReplica) - : command.copyAsDigestQuery(extraReplica); - } - else - { - extraReplica = replicaPlan.firstUncontactedCandidate(Replica::isFull); - retryCommand = command; - if (extraReplica == null) - { - cfs.metric.speculativeInsufficientReplicas.inc(); - // cannot safely speculate a new data request, without more work - requests assumed to be - // unique per endpoint, and we have no full nodes left to speculate against - return; - } - } - - // we must update the plan to include this new node, else when we come to read-repair, we may not include this - // speculated response in the data requests we make again, and we will not be able to 'speculate' an extra repair read, - // nor would we be able to speculate a new 'write' if the repair writes are insufficient - super.replicaPlan.addToContacts(extraReplica); - - if (traceState != null) - traceState.trace("speculating read retry on {}", extraReplica); - logger.trace("speculating read retry on {}", extraReplica); - - MessagingService.instance().sendWithCallback(retryCommand.createMessage(false, requestTime), extraReplica.endpoint(), handler); + trySpeculativeRetry(); } } + boolean maybeTryAdditionalReplicasOnOverload() + { + if (!DatabaseDescriptor.getReadFallbackOnOverloadedConnection()) + return false; + + speculated = true; + if (!trySpeculativeRetry()) + return false; + + cfs.metric.overloadSpeculativeRetries.inc(); + return true; + } + @Override void onReadTimeout() { @@ -373,6 +423,15 @@ public void maybeTryAdditionalReplicas() // no-op } + boolean maybeTryAdditionalReplicasOnOverload() + { + if (!DatabaseDescriptor.getReadFallbackOnOverloadedConnection() || !trySpeculativeRetry()) + return false; + + cfs.metric.overloadSpeculativeRetries.inc(); + return true; + } + @Override public void executeAsync() { diff --git a/src/java/org/apache/cassandra/service/reads/ReadCallback.java b/src/java/org/apache/cassandra/service/reads/ReadCallback.java index 26bd52973972..ab573f8fdd8f 100644 --- a/src/java/org/apache/cassandra/service/reads/ReadCallback.java +++ b/src/java/org/apache/cassandra/service/reads/ReadCallback.java @@ -42,6 +42,7 @@ import org.apache.cassandra.net.Message; import org.apache.cassandra.net.ParamType; import org.apache.cassandra.net.RequestCallback; +import org.apache.cassandra.net.ResourceLimits.Outcome; import org.apache.cassandra.net.Verb; import org.apache.cassandra.service.reads.thresholds.CoordinatorWarnings; import org.apache.cassandra.service.reads.thresholds.WarningContext; @@ -51,6 +52,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.atomic.AtomicIntegerFieldUpdater.newUpdater; +import static org.apache.cassandra.net.ResourceLimits.Outcome.INSUFFICIENT_ENDPOINT; import static org.apache.cassandra.tracing.Tracing.isTracing; import static org.apache.cassandra.utils.concurrent.Condition.newOneTimeCondition; @@ -73,6 +75,9 @@ public class ReadCallback, P extends ReplicaPlan.ForRead< private volatile WarningContext warningContext; private static final AtomicReferenceFieldUpdater warningsUpdater = AtomicReferenceFieldUpdater.newUpdater(ReadCallback.class, WarningContext.class, "warningContext"); + // set by AbstractReadExecutor before its first send. A range read, a read repair and the short read + // and replica filtering protections have no executor, and leave this null + private volatile AbstractReadExecutor executor; public ReadCallback(ResponseResolver resolver, ReadCommand command, ReplicaPlan.Shared replicaPlan, Dispatcher.RequestTime requestTime) { @@ -94,6 +99,11 @@ protected P replicaPlan() return replicaPlan.get(); } + void setExecutor(AbstractReadExecutor executor) + { + this.executor = executor; + } + public boolean await(long commandTimeout, TimeUnit unit) { return awaitUntil(requestTime.computeDeadline(unit.toNanos(commandTimeout))); @@ -241,6 +251,27 @@ public void onFailure(InetAddressAndPort from, RequestFailureReason failureReaso condition.signalAll(); } + /** + * Try an additional replica when the connection to one peer is overloaded. Fails when we run out of possible + * candidates, and fails without trying at all once the node-wide reserve is exhausted, since no connection to + * any peer can be allocated from it. + */ + @Override + public void onOverloaded(InetAddressAndPort from, Outcome outcome) + { + AbstractReadExecutor executor = this.executor; + if (executor != null && INSUFFICIENT_ENDPOINT == outcome) + executor.maybeTryAdditionalReplicasOnOverload(); + + onFailure(from, RequestFailureReason.TIMEOUT); + } + + @Override + public boolean invokeOnOverloadedInline() + { + return DatabaseDescriptor.getReadFallbackOnOverloadedConnection(); + } + @Override public boolean invokeOnFailure() { diff --git a/test/burn/org/apache/cassandra/net/Connection.java b/test/burn/org/apache/cassandra/net/Connection.java index de5df6b65de5..7da17659213d 100644 --- a/test/burn/org/apache/cassandra/net/Connection.java +++ b/test/burn/org/apache/cassandra/net/Connection.java @@ -364,7 +364,7 @@ public void onConnectInbound(int messagingVersion, InboundMessageHandler handler verifier.onConnectInbound(messagingVersion, handler); } - public void onOverloaded(Message message, InetAddressAndPort peer) + public void onOverloaded(Message message, InetAddressAndPort peer, ResourceLimits.Outcome outcome) { controller.fail(message.serializedSize(current_version)); verifier.onOverloaded(message.id()); diff --git a/test/distributed/org/apache/cassandra/distributed/test/OverloadedConnectionReadFallbackTest.java b/test/distributed/org/apache/cassandra/distributed/test/OverloadedConnectionReadFallbackTest.java new file mode 100644 index 000000000000..5d6897d0700b --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/OverloadedConnectionReadFallbackTest.java @@ -0,0 +1,290 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Test; + +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.shared.ClusterUtils; +import org.apache.cassandra.locator.EndpointsForToken; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.locator.Replica; +import org.apache.cassandra.locator.ReplicaLayout; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.net.OutboundConnections; + +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.assertj.core.api.Assertions.assertThat; + +public class OverloadedConnectionReadFallbackTest extends TestBaseImpl +{ + private static final String TABLE = "t"; + private static final int SEED_ROWS = 200; + + private static final int COORDINATOR = 1; + private static final String COORDINATOR_ADDRESS = "127.0.0.1"; + + private static final String SMALL_SEND_QUEUE = "4KiB"; + + private static final int READ_THREADS = 128; + + private static final int READ_SECONDS = 3; + + private static final String UNAVAILABLE = "Cannot achieve consistency level"; + + private static final String TIMED_OUT = "Operation timed out - received only"; + + @Test + public void oneOverloadedConnectionFallsBackToTheSpareReplica() throws Throwable + { + assertOverloadedConnectionsDoNotFailReads(3, new int[]{ 3 }); + } + + @Test + public void twoOverloadedConnectionsFallBackTwice() throws Throwable + { + assertOverloadedConnectionsDoNotFailReads(5, new int[]{ 3, 4 }); + } + + private void assertOverloadedConnectionsDoNotFailReads(int rf, int[] downNodes) throws Throwable + { + int quorum = rf / 2 + 1; + List downAddresses = addresses(downNodes); + + try (Cluster cluster = startCluster(rf + 1, rf)) + { + seed(cluster); + + List keys = keysContactingAllOf(cluster, quorum, downAddresses); + assertThat(keys) + .as("expected keys whose replicas exclude the coordinator and contact " + downAddresses) + .isNotEmpty(); + + for (int node : downNodes) + ClusterUtils.stopAbrupt(cluster, cluster.get(node)); + + ReadOutcome outcome = hammerReads(cluster, keys, ConsistencyLevel.QUORUM); + + for (String downAddress : downAddresses) + assertThat(overloadedMessages(cluster, downAddress)) + .as("the send queue for " + downAddress + " should overload") + .isGreaterThan(0L); + + assertThat(overloadSpeculativeRetries(cluster)) + .as("a dropped read message should move the read to another replica") + .isGreaterThan(0L); + + assertThat(outcome.firstFailure) + .as("a QUORUM read with enough live replicas should not fail because a connection is overloaded" + + " (" + outcome.timeouts + " reads timed out on the expiry route)") + .isNull(); + + assertThat(outcome.successes) + .as("reads should continue to be served from the remaining replicas") + .isGreaterThan(0); + + assertThat(outcome.wrongResults) + .as("the fallback replicas should return the seeded row") + .isEqualTo(0); + } + } + + private Cluster startCluster(int nodes, int rf) throws Throwable + { + return init(builder().withNodes(nodes) + .withConfig(config -> { + config.with(NETWORK, GOSSIP); + config.set("read_request_timeout", "5s"); + config.set("write_request_timeout", "5s"); + config.set("internode_application_send_queue_capacity", SMALL_SEND_QUEUE); + config.set("dynamic_snitch", false); + config.set("read_fallback_on_overloaded_connection", true); + }) + .start(), rf); + } + + private void seed(Cluster cluster) + { + cluster.schemaChange(withKeyspace("CREATE TABLE %s." + TABLE + " (pk int PRIMARY KEY, v int)")); + for (int pk = 0; pk < SEED_ROWS; pk++) + cluster.coordinator(COORDINATOR) + .execute(withKeyspace("INSERT INTO %s." + TABLE + " (pk, v) VALUES (" + pk + ", 1)"), + ConsistencyLevel.ALL); + } + + private List keysContactingAllOf(Cluster cluster, int quorum, List required) + { + String keyspace = KEYSPACE; + String table = TABLE; + int keyCount = SEED_ROWS; + String coordinator = COORDINATOR_ADDRESS; + ArrayList mustContact = new ArrayList<>(required); + + return cluster.get(COORDINATOR).callOnInstance(() -> { + Keyspace ks = Keyspace.open(keyspace); + List selected = new ArrayList<>(); + + for (int pk = 0; pk < keyCount; pk++) + { + ByteBuffer key = Int32Type.instance.decompose(pk); + Token token = ks.getColumnFamilyStore(table).metadata().partitioner.decorateKey(key).getToken(); + EndpointsForToken sorted = ReplicaLayout.forTokenReadLiveSorted(ks.getReplicationStrategy(), token) + .natural(); + + List ordered = new ArrayList<>(); + for (Replica replica : sorted) + ordered.add(replica.endpoint().getHostAddress(false)); + + if (ordered.contains(coordinator)) + continue; + if (ordered.subList(0, Math.min(quorum, ordered.size())).containsAll(mustContact)) + selected.add(pk); + } + return selected; + }); + } + + private long overloadSpeculativeRetries(Cluster cluster) + { + String keyspace = KEYSPACE; + String table = TABLE; + + return cluster.get(COORDINATOR).callOnInstance( + () -> Keyspace.open(keyspace).getColumnFamilyStore(table).metric.overloadSpeculativeRetries.getCount()); + } + + private long overloadedMessages(Cluster cluster, String peerAddress) + { + return cluster.get(COORDINATOR).callOnInstance(() -> { + long overloaded = 0; + for (Map.Entry entry : + MessagingService.instance().channelManagers.entrySet()) + { + if (!entry.getKey().getHostAddress(false).equals(peerAddress)) + continue; + + OutboundConnections connections = entry.getValue(); + overloaded += connections.small.overloadedCount() + + connections.large.overloadedCount() + + connections.urgent.overloadedCount(); + } + return overloaded; + }); + } + + private ReadOutcome hammerReads(Cluster cluster, List keys, ConsistencyLevel cl) + throws InterruptedException + { + String query = withKeyspace("SELECT pk, v FROM %s." + TABLE + " WHERE pk = ?"); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(READ_SECONDS); + + AtomicReference firstFailure = new AtomicReference<>(); + AtomicInteger successes = new AtomicInteger(); + AtomicInteger timeouts = new AtomicInteger(); + AtomicInteger wrongResults = new AtomicInteger(); + + ExecutorService pool = Executors.newFixedThreadPool(READ_THREADS); + try + { + for (int i = 0; i < READ_THREADS; i++) + { + final int offset = i; + pool.submit(() -> { + int index = offset; + while (System.nanoTime() < deadline && firstFailure.get() == null) + { + int pk = keys.get(Math.floorMod(index, keys.size())); + try + { + Object[][] rows = cluster.coordinator(COORDINATOR).execute(query, cl, pk); + if (rows.length != 1 || !Integer.valueOf(pk).equals(rows[0][0]) + || !Integer.valueOf(1).equals(rows[0][1])) + wrongResults.incrementAndGet(); + else + successes.incrementAndGet(); + } + catch (Throwable t) + { + String rendered = t.toString(); + + if (rendered.contains(UNAVAILABLE)) + return; + + // A message that reached the send queue before it filled cannot be detected + // at send time, so it expires and the read times out. The fallback cannot + // prevent that, so a timeout is not a failure of the route under test. + if (rendered.contains(TIMED_OUT)) + timeouts.incrementAndGet(); + else + firstFailure.compareAndSet(null, rendered); + } + index += READ_THREADS; + } + }); + } + pool.shutdown(); + pool.awaitTermination(READ_SECONDS * 3L, TimeUnit.SECONDS); + } + finally + { + pool.shutdownNow(); + } + + return new ReadOutcome(firstFailure.get(), successes.get(), timeouts.get(), wrongResults.get()); + } + + private static List addresses(int[] nodes) + { + List addresses = new ArrayList<>(nodes.length); + for (int node : nodes) + addresses.add("127.0.0." + node); + return addresses; + } + + private static final class ReadOutcome + { + final String firstFailure; + final int successes; + final int timeouts; + final int wrongResults; + + ReadOutcome(String firstFailure, int successes, int timeouts, int wrongResults) + { + this.firstFailure = firstFailure; + this.successes = successes; + this.timeouts = timeouts; + this.wrongResults = wrongResults; + } + } +} diff --git a/test/unit/org/apache/cassandra/service/reads/ReadExecutorTest.java b/test/unit/org/apache/cassandra/service/reads/ReadExecutorTest.java index 0c4b2f685ef1..9bab83cd7f5f 100644 --- a/test/unit/org/apache/cassandra/service/reads/ReadExecutorTest.java +++ b/test/unit/org/apache/cassandra/service/reads/ReadExecutorTest.java @@ -31,6 +31,7 @@ import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.Keyspace; @@ -50,7 +51,10 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static org.apache.cassandra.db.ConsistencyLevel.LOCAL_QUORUM; import static org.apache.cassandra.locator.ReplicaUtils.full; +import static org.apache.cassandra.net.ResourceLimits.Outcome.INSUFFICIENT_ENDPOINT; +import static org.apache.cassandra.net.ResourceLimits.Outcome.INSUFFICIENT_GLOBAL; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -83,7 +87,9 @@ public void resetCounters() throws Throwable { cfs.metric.speculativeInsufficientReplicas.dec(cfs.metric.speculativeInsufficientReplicas.getCount()); cfs.metric.speculativeRetries.dec(cfs.metric.speculativeRetries.getCount()); + cfs.metric.overloadSpeculativeRetries.dec(cfs.metric.overloadSpeculativeRetries.getCount()); cfs.metric.speculativeFailedRetries.dec(cfs.metric.speculativeFailedRetries.getCount()); + DatabaseDescriptor.setReadFallbackOnOverloadedConnection(false); } /** @@ -238,6 +244,116 @@ public void testRaceWithNonSpeculativeFailure() } } + @Test + public void testOverloadWhenCandidateAvailableShouldContactItAndNotFailRead() + { + DatabaseDescriptor.setReadFallbackOnOverloadedConnection(true); + + AbstractReadExecutor executor = speculatingExecutor(targets.subList(0, 1)); + executor.executeAsync(); + + executor.handler.onOverloaded(targets.get(0).endpoint(), INSUFFICIENT_ENDPOINT); + + assertEquals(2, executor.replicaPlan().contacts().size()); + assertEquals(1, cfs.metric.overloadSpeculativeRetries.getCount()); + assertEquals(1, ks.metric.overloadSpeculativeRetries.getCount()); + assertEquals(1, cfs.metric.speculativeRetries.getCount()); + assertFalse(executor.handler.condition.isSignalled()); + } + + @Test + public void testOverloadOfGlobalReserveShouldFailReadWithoutContactingACandidate() + { + DatabaseDescriptor.setReadFallbackOnOverloadedConnection(true); + + AbstractReadExecutor executor = speculatingExecutor(targets.subList(0, 1)); + executor.executeAsync(); + + executor.handler.onOverloaded(targets.get(0).endpoint(), INSUFFICIENT_GLOBAL); + + assertEquals(1, executor.replicaPlan().contacts().size()); + assertEquals(0, cfs.metric.overloadSpeculativeRetries.getCount()); + assertEquals(0, cfs.metric.speculativeRetries.getCount()); + assertTrue(executor.handler.condition.isSignalled()); + } + + @Test + public void testOverloadWhenTwoConnectionsDropShouldContactTwoCandidates() + { + DatabaseDescriptor.setReadFallbackOnOverloadedConnection(true); + + AbstractReadExecutor executor = speculatingExecutor(targets.subList(0, 1)); + executor.executeAsync(); + + executor.handler.onOverloaded(targets.get(0).endpoint(), INSUFFICIENT_ENDPOINT); + executor.handler.onOverloaded(targets.get(1).endpoint(), INSUFFICIENT_ENDPOINT); + + assertEquals(3, executor.replicaPlan().contacts().size()); + assertEquals(2, cfs.metric.overloadSpeculativeRetries.getCount()); + } + + @Test + public void testOverloadWhenFallbackDisabledShouldFailRead() + { + AbstractReadExecutor executor = speculatingExecutor(targets.subList(0, 1)); + executor.executeAsync(); + + executor.handler.onOverloaded(targets.get(0).endpoint(), INSUFFICIENT_ENDPOINT); + + assertEquals(1, executor.replicaPlan().contacts().size()); + assertEquals(0, cfs.metric.overloadSpeculativeRetries.getCount()); + assertEquals(0, cfs.metric.speculativeRetries.getCount()); + assertTrue(executor.handler.condition.isSignalled()); + } + + @Test + public void testOverloadWhenExecutorNeverSpeculatesShouldNotContactAnotherReplica() + { + DatabaseDescriptor.setReadFallbackOnOverloadedConnection(true); + + AbstractReadExecutor executor = new AbstractReadExecutor.NeverSpeculatingReadExecutor(cfs, new MockSinglePartitionReadCommand(DAYS.toMillis(365)), plan(ConsistencyLevel.LOCAL_ONE, targets, targets.subList(0, 1)), Dispatcher.RequestTime.forImmediateExecution(), false); + executor.executeAsync(); + + executor.handler.onOverloaded(targets.get(0).endpoint(), INSUFFICIENT_ENDPOINT); + + assertEquals(1, executor.replicaPlan().contacts().size()); + assertEquals(0, cfs.metric.overloadSpeculativeRetries.getCount()); + assertEquals(0, cfs.metric.speculativeRetries.getCount()); + } + + @Test + public void testReadCallbackRunsInlineOnOverloadOnlyWhenFallbackEnabled() + { + AbstractReadExecutor executor = speculatingExecutor(targets.subList(0, 1)); + + assertFalse(executor.handler.invokeOnOverloadedInline()); + + DatabaseDescriptor.setReadFallbackOnOverloadedConnection(true); + + assertTrue(executor.handler.invokeOnOverloadedInline()); + } + + @Test + public void testOverloadWhenNoCandidateLeftShouldCountInsufficientReplicas() + { + DatabaseDescriptor.setReadFallbackOnOverloadedConnection(true); + + AbstractReadExecutor executor = speculatingExecutor(targets); + + assertFalse(executor.maybeTryAdditionalReplicasOnOverload()); + assertEquals(3, executor.replicaPlan().contacts().size()); + assertEquals(0, cfs.metric.overloadSpeculativeRetries.getCount()); + assertEquals(1, cfs.metric.speculativeInsufficientReplicas.getCount()); + } + + private AbstractReadExecutor speculatingExecutor(EndpointsForToken contacts) + { + return new AbstractReadExecutor.SpeculatingReadExecutor(cfs, + new MockSinglePartitionReadCommand(DAYS.toMillis(365)), + plan(ConsistencyLevel.LOCAL_ONE, targets, contacts), + Dispatcher.RequestTime.forImmediateExecution()); + } + public static class MockSinglePartitionReadCommand extends SinglePartitionReadCommand { private final long timeout;