http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/ReadCallback.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/ReadCallback.java b/src/java/org/apache/cassandra/service/ReadCallback.java deleted file mode 100644 index e7f30b4..0000000 --- a/src/java/org/apache/cassandra/service/ReadCallback.java +++ /dev/null @@ -1,266 +0,0 @@ -/* - * 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.service; - -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; - -import org.apache.commons.lang3.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.cassandra.concurrent.Stage; -import org.apache.cassandra.concurrent.StageManager; -import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.db.*; -import org.apache.cassandra.db.partitions.PartitionIterator; -import org.apache.cassandra.exceptions.RequestFailureReason; -import org.apache.cassandra.exceptions.ReadFailureException; -import org.apache.cassandra.exceptions.ReadTimeoutException; -import org.apache.cassandra.exceptions.UnavailableException; -import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.metrics.ReadRepairMetrics; -import org.apache.cassandra.net.IAsyncCallbackWithFailure; -import org.apache.cassandra.net.MessageIn; -import org.apache.cassandra.net.MessagingService; -import org.apache.cassandra.db.ConsistencyLevel; -import org.apache.cassandra.tracing.TraceState; -import org.apache.cassandra.tracing.Tracing; -import org.apache.cassandra.utils.FBUtilities; -import org.apache.cassandra.utils.concurrent.SimpleCondition; - -public class ReadCallback implements IAsyncCallbackWithFailure<ReadResponse> -{ - protected static final Logger logger = LoggerFactory.getLogger( ReadCallback.class ); - - public final ResponseResolver resolver; - final SimpleCondition condition = new SimpleCondition(); - private final long queryStartNanoTime; - final int blockfor; - final List<InetAddressAndPort> endpoints; - private final ReadCommand command; - private final ConsistencyLevel consistencyLevel; - private static final AtomicIntegerFieldUpdater<ReadCallback> recievedUpdater - = AtomicIntegerFieldUpdater.newUpdater(ReadCallback.class, "received"); - private volatile int received = 0; - private static final AtomicIntegerFieldUpdater<ReadCallback> failuresUpdater - = AtomicIntegerFieldUpdater.newUpdater(ReadCallback.class, "failures"); - private volatile int failures = 0; - private final Map<InetAddressAndPort, RequestFailureReason> failureReasonByEndpoint; - - private final Keyspace keyspace; // TODO push this into ConsistencyLevel? - - /** - * Constructor when response count has to be calculated and blocked for. - */ - public ReadCallback(ResponseResolver resolver, ConsistencyLevel consistencyLevel, ReadCommand command, List<InetAddressAndPort> filteredEndpoints, long queryStartNanoTime) - { - this(resolver, - consistencyLevel, - consistencyLevel.blockFor(Keyspace.open(command.metadata().keyspace)), - command, - Keyspace.open(command.metadata().keyspace), - filteredEndpoints, - queryStartNanoTime); - } - - public ReadCallback(ResponseResolver resolver, ConsistencyLevel consistencyLevel, int blockfor, ReadCommand command, Keyspace keyspace, List<InetAddressAndPort> endpoints, long queryStartNanoTime) - { - this.command = command; - this.keyspace = keyspace; - this.blockfor = blockfor; - this.consistencyLevel = consistencyLevel; - this.resolver = resolver; - this.queryStartNanoTime = queryStartNanoTime; - this.endpoints = endpoints; - this.failureReasonByEndpoint = new ConcurrentHashMap<>(); - // we don't support read repair (or rapid read protection) for range scans yet (CASSANDRA-6897) - assert !(command instanceof PartitionRangeReadCommand) || blockfor >= endpoints.size(); - - if (logger.isTraceEnabled()) - logger.trace("Blockfor is {}; setting up requests to {}", blockfor, StringUtils.join(this.endpoints, ",")); - } - - public boolean await(long timePastStart, TimeUnit unit) - { - long time = unit.toNanos(timePastStart) - (System.nanoTime() - queryStartNanoTime); - try - { - return condition.await(time, TimeUnit.NANOSECONDS); - } - catch (InterruptedException ex) - { - throw new AssertionError(ex); - } - } - - public void awaitResults() throws ReadFailureException, ReadTimeoutException - { - boolean signaled = await(command.getTimeout(), TimeUnit.MILLISECONDS); - boolean failed = blockfor + failures > endpoints.size(); - if (signaled && !failed) - return; - - if (Tracing.isTracing()) - { - String gotData = received > 0 ? (resolver.isDataPresent() ? " (including data)" : " (only digests)") : ""; - Tracing.trace("{}; received {} of {} responses{}", new Object[]{ (failed ? "Failed" : "Timed out"), received, blockfor, gotData }); - } - else if (logger.isDebugEnabled()) - { - String gotData = received > 0 ? (resolver.isDataPresent() ? " (including data)" : " (only digests)") : ""; - logger.debug("{}; received {} of {} responses{}", new Object[]{ (failed ? "Failed" : "Timed out"), received, blockfor, gotData }); - } - - // Same as for writes, see AbstractWriteResponseHandler - throw failed - ? new ReadFailureException(consistencyLevel, received, blockfor, resolver.isDataPresent(), failureReasonByEndpoint) - : new ReadTimeoutException(consistencyLevel, received, blockfor, resolver.isDataPresent()); - } - - public PartitionIterator get() throws ReadFailureException, ReadTimeoutException, DigestMismatchException - { - awaitResults(); - - PartitionIterator result = blockfor == 1 ? resolver.getData() : resolver.resolve(); - if (logger.isTraceEnabled()) - logger.trace("Read: {} ms.", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - queryStartNanoTime)); - return result; - } - - public int blockFor() - { - return blockfor; - } - - public void response(MessageIn<ReadResponse> message) - { - resolver.preprocess(message); - int n = waitingFor(message.from) - ? recievedUpdater.incrementAndGet(this) - : received; - if (n >= blockfor && resolver.isDataPresent()) - { - condition.signalAll(); - // kick off a background digest comparison if this is a result that (may have) arrived after - // the original resolve that get() kicks off as soon as the condition is signaled - if (blockfor < endpoints.size() && n == endpoints.size()) - { - TraceState traceState = Tracing.instance.get(); - if (traceState != null) - traceState.trace("Initiating read-repair"); - StageManager.getStage(Stage.READ_REPAIR).execute(new AsyncRepairRunner(traceState, queryStartNanoTime)); - } - } - } - - /** - * @return true if the message counts towards the blockfor threshold - */ - private boolean waitingFor(InetAddressAndPort from) - { - return consistencyLevel.isDatacenterLocal() - ? DatabaseDescriptor.getLocalDataCenter().equals(DatabaseDescriptor.getEndpointSnitch().getDatacenter(from)) - : true; - } - - /** - * @return the current number of received responses - */ - public int getReceivedCount() - { - return received; - } - - public void response(ReadResponse result) - { - MessageIn<ReadResponse> message = MessageIn.create(FBUtilities.getBroadcastAddressAndPort(), - result, - Collections.emptyMap(), - MessagingService.Verb.INTERNAL_RESPONSE, - MessagingService.current_version); - response(message); - } - - public void assureSufficientLiveNodes() throws UnavailableException - { - consistencyLevel.assureSufficientLiveNodes(keyspace, endpoints); - } - - public boolean isLatencyForSnitch() - { - return true; - } - - private class AsyncRepairRunner implements Runnable - { - private final TraceState traceState; - private final long queryStartNanoTime; - - public AsyncRepairRunner(TraceState traceState, long queryStartNanoTime) - { - this.traceState = traceState; - this.queryStartNanoTime = queryStartNanoTime; - } - - public void run() - { - // If the resolver is a DigestResolver, we need to do a full data read if there is a mismatch. - // Otherwise, resolve will send the repairs directly if needs be (and in that case we should never - // get a digest mismatch). - try - { - resolver.compareResponses(); - } - catch (DigestMismatchException e) - { - assert resolver instanceof DigestResolver; - - if (traceState != null) - traceState.trace("Digest mismatch: {}", e.toString()); - if (logger.isDebugEnabled()) - logger.debug("Digest mismatch:", e); - - ReadRepairMetrics.repairedBackground.mark(); - - final DataResolver repairResolver = new DataResolver(keyspace, command, consistencyLevel, endpoints.size(), queryStartNanoTime); - AsyncRepairCallback repairHandler = new AsyncRepairCallback(repairResolver, endpoints.size()); - - for (InetAddressAndPort endpoint : endpoints) - MessagingService.instance().sendRR(command.createMessage(), endpoint, repairHandler); - } - } - } - - @Override - public void onFailure(InetAddressAndPort from, RequestFailureReason failureReason) - { - int n = waitingFor(from) - ? failuresUpdater.incrementAndGet(this) - : failures; - - failureReasonByEndpoint.put(from, failureReason); - - if (blockfor + n > endpoints.size()) - condition.signalAll(); - } -}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/ReadRepairDecision.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/ReadRepairDecision.java b/src/java/org/apache/cassandra/service/ReadRepairDecision.java deleted file mode 100644 index 8d2ced7..0000000 --- a/src/java/org/apache/cassandra/service/ReadRepairDecision.java +++ /dev/null @@ -1,23 +0,0 @@ -/* - * 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.service; - -public enum ReadRepairDecision -{ - NONE, GLOBAL, DC_LOCAL; -} http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/ResponseResolver.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/ResponseResolver.java b/src/java/org/apache/cassandra/service/ResponseResolver.java deleted file mode 100644 index 81b18b6..0000000 --- a/src/java/org/apache/cassandra/service/ResponseResolver.java +++ /dev/null @@ -1,73 +0,0 @@ -/* - * 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.service; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.cassandra.db.*; -import org.apache.cassandra.db.partitions.*; -import org.apache.cassandra.net.MessageIn; -import org.apache.cassandra.utils.concurrent.Accumulator; - -public abstract class ResponseResolver -{ - protected static final Logger logger = LoggerFactory.getLogger(ResponseResolver.class); - - protected final Keyspace keyspace; - protected final ReadCommand command; - protected final ConsistencyLevel consistency; - - // Accumulator gives us non-blocking thread-safety with optimal algorithmic constraints - protected final Accumulator<MessageIn<ReadResponse>> responses; - - public ResponseResolver(Keyspace keyspace, ReadCommand command, ConsistencyLevel consistency, int maxResponseCount) - { - this.keyspace = keyspace; - this.command = command; - this.consistency = consistency; - this.responses = new Accumulator<>(maxResponseCount); - } - - public abstract PartitionIterator getData(); - public abstract PartitionIterator resolve() throws DigestMismatchException; - - /** - * Compares received responses, potentially triggering a digest mismatch (for a digest resolver) and read-repairs - * (for a data resolver). - * <p> - * This is functionally equivalent to calling {@link #resolve()} and consuming the result, but can be slightly more - * efficient in some case due to the fact that we don't care about the result itself. This is used when doing - * asynchronous read-repairs. - * - * @throws DigestMismatchException if it's a digest resolver and the responses don't match. - */ - public abstract void compareResponses() throws DigestMismatchException; - - public abstract boolean isDataPresent(); - - public void preprocess(MessageIn<ReadResponse> message) - { - responses.add(message); - } - - public Iterable<MessageIn<ReadResponse>> getMessages() - { - return responses; - } -} http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/StorageProxy.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index e2125d4..7d4e34c 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -44,6 +44,10 @@ import org.apache.cassandra.batchlog.Batch; import org.apache.cassandra.batchlog.BatchlogManager; import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.concurrent.StageManager; +import org.apache.cassandra.service.reads.AbstractReadExecutor; +import org.apache.cassandra.service.reads.DataResolver; +import org.apache.cassandra.service.reads.ReadCallback; +import org.apache.cassandra.service.reads.repair.ReadRepair; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.schema.Schema; @@ -1751,139 +1755,54 @@ public class StorageProxy implements StorageProxyMBean { int cmdCount = commands.size(); - SinglePartitionReadLifecycle[] reads = new SinglePartitionReadLifecycle[cmdCount]; - for (int i = 0; i < cmdCount; i++) - reads[i] = new SinglePartitionReadLifecycle(commands.get(i), consistencyLevel, queryStartNanoTime); + AbstractReadExecutor[] reads = new AbstractReadExecutor[cmdCount]; - for (int i = 0; i < cmdCount; i++) - reads[i].doInitialQueries(); - - for (int i = 0; i < cmdCount; i++) - reads[i].maybeTryAdditionalReplicas(); - - for (int i = 0; i < cmdCount; i++) - reads[i].awaitResultsAndRetryOnDigestMismatch(); - - for (int i = 0; i < cmdCount; i++) - if (!reads[i].isDone()) - reads[i].maybeAwaitFullDataRead(); - - List<PartitionIterator> results = new ArrayList<>(cmdCount); - for (int i = 0; i < cmdCount; i++) + for (int i=0; i<cmdCount; i++) { - assert reads[i].isDone(); - results.add(reads[i].getResult()); + reads[i] = AbstractReadExecutor.getReadExecutor(commands.get(i), consistencyLevel, queryStartNanoTime); } - return PartitionIterators.concat(results); - } - - private static class SinglePartitionReadLifecycle - { - private final SinglePartitionReadCommand command; - private final AbstractReadExecutor executor; - private final ConsistencyLevel consistency; - private final long queryStartNanoTime; - - private PartitionIterator result; - private ReadCallback repairHandler; - - SinglePartitionReadLifecycle(SinglePartitionReadCommand command, ConsistencyLevel consistency, long queryStartNanoTime) + for (int i=0; i<cmdCount; i++) { - this.command = command; - this.executor = AbstractReadExecutor.getReadExecutor(command, consistency, queryStartNanoTime); - this.consistency = consistency; - this.queryStartNanoTime = queryStartNanoTime; + reads[i].executeAsync(); } - boolean isDone() + for (int i=0; i<cmdCount; i++) { - return result != null; + reads[i].maybeTryAdditionalReplicas(); } - void doInitialQueries() + for (int i=0; i<cmdCount; i++) { - executor.executeAsync(); + reads[i].awaitResponses(); } - void maybeTryAdditionalReplicas() + for (int i=0; i<cmdCount; i++) { - executor.maybeTryAdditionalReplicas(); + reads[i].maybeRepairAdditionalReplicas(); } - void awaitResultsAndRetryOnDigestMismatch() throws ReadFailureException, ReadTimeoutException + for (int i=0; i<cmdCount; i++) { - try - { - result = executor.get(); - } - catch (DigestMismatchException ex) - { - Tracing.trace("Digest mismatch: {}", ex.getMessage()); - - ReadRepairMetrics.repairedBlocking.mark(); - - // Do a full data read to resolve the correct response (and repair node that need be) - Keyspace keyspace = Keyspace.open(command.metadata().keyspace); - DataResolver resolver = new DataResolver(keyspace, command, ConsistencyLevel.ALL, executor.handler.endpoints.size(), queryStartNanoTime); - repairHandler = new ReadCallback(resolver, - ConsistencyLevel.ALL, - executor.getContactedReplicas().size(), - command, - keyspace, - executor.handler.endpoints, - queryStartNanoTime); - - for (InetAddressAndPort endpoint : executor.getContactedReplicas()) - { - Tracing.trace("Enqueuing full data read to {}", endpoint); - MessagingService.instance().sendRRWithFailure(command.createMessage(), endpoint, repairHandler); - } - } + reads[i].awaitReadRepair(); } - void maybeAwaitFullDataRead() throws ReadTimeoutException + List<PartitionIterator> results = new ArrayList<>(cmdCount); + for (int i=0; i<cmdCount; i++) { - // There wasn't a digest mismatch, we're good - if (repairHandler == null) - return; - - // Otherwise, get the result from the full-data read and check that it's not a short read - try - { - result = repairHandler.get(); - } - catch (DigestMismatchException e) - { - throw new AssertionError(e); // full data requested from each node here, no digests should be sent - } - catch (ReadTimeoutException e) - { - if (Tracing.isTracing()) - Tracing.trace("Timed out waiting on digest mismatch repair requests"); - else - logger.trace("Timed out waiting on digest mismatch repair requests"); - // the caught exception here will have CL.ALL from the repair command, - // not whatever CL the initial command was at (CASSANDRA-7947) - int blockFor = consistency.blockFor(Keyspace.open(command.metadata().keyspace)); - throw new ReadTimeoutException(consistency, blockFor-1, blockFor, true); - } + results.add(reads[i].getResult()); } - PartitionIterator getResult() - { - assert result != null; - return result; - } + return PartitionIterators.concat(results); } - static class LocalReadRunnable extends DroppableRunnable + public static class LocalReadRunnable extends DroppableRunnable { private final ReadCommand command; private final ReadCallback handler; private final long start = System.nanoTime(); - LocalReadRunnable(ReadCommand command, ReadCallback handler) + public LocalReadRunnable(ReadCommand command, ReadCallback handler) { super(MessagingService.Verb.READ); this.command = command; @@ -2081,11 +2000,13 @@ public class StorageProxy implements StorageProxyMBean private static class SingleRangeResponse extends AbstractIterator<RowIterator> implements PartitionIterator { + private final DataResolver resolver; private final ReadCallback handler; private PartitionIterator result; - private SingleRangeResponse(ReadCallback handler) + private SingleRangeResponse(DataResolver resolver, ReadCallback handler) { + this.resolver = resolver; this.handler = handler; } @@ -2094,14 +2015,8 @@ public class StorageProxy implements StorageProxyMBean if (result != null) return; - try - { - result = handler.get(); - } - catch (DigestMismatchException e) - { - throw new AssertionError(e); // no digests in range slices yet - } + handler.awaitResults(); + result = resolver.resolve(); } protected RowIterator computeNext() @@ -2223,12 +2138,13 @@ public class StorageProxy implements StorageProxyMBean { PartitionRangeReadCommand rangeCommand = command.forSubRange(toQuery.range, isFirst); - DataResolver resolver = new DataResolver(keyspace, rangeCommand, consistency, toQuery.filteredEndpoints.size(), queryStartNanoTime); + ReadRepair readRepair = ReadRepair.create(command, toQuery.filteredEndpoints, queryStartNanoTime, consistency); + DataResolver resolver = new DataResolver(keyspace, rangeCommand, consistency, toQuery.filteredEndpoints.size(), queryStartNanoTime, readRepair); int blockFor = consistency.blockFor(keyspace); int minResponses = Math.min(toQuery.filteredEndpoints.size(), blockFor); List<InetAddressAndPort> minimalEndpoints = toQuery.filteredEndpoints.subList(0, minResponses); - ReadCallback handler = new ReadCallback(resolver, consistency, rangeCommand, minimalEndpoints, queryStartNanoTime); + ReadCallback handler = new ReadCallback(resolver, consistency, rangeCommand, minimalEndpoints, queryStartNanoTime, readRepair); handler.assureSufficientLiveNodes(); @@ -2245,7 +2161,7 @@ public class StorageProxy implements StorageProxyMBean } } - return new SingleRangeResponse(handler); + return new SingleRangeResponse(resolver, handler); } private PartitionIterator sendNextRequests() http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java new file mode 100644 index 0000000..5a660ca --- /dev/null +++ b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java @@ -0,0 +1,482 @@ +/* + * 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.service.reads; + +import java.util.List; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Iterables; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.concurrent.StageManager; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.ReadCommand; +import org.apache.cassandra.db.SinglePartitionReadCommand; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.partitions.PartitionIterator; +import org.apache.cassandra.exceptions.ReadFailureException; +import org.apache.cassandra.exceptions.ReadTimeoutException; +import org.apache.cassandra.exceptions.UnavailableException; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.metrics.ReadRepairMetrics; +import org.apache.cassandra.net.MessageOut; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.service.StorageProxy; +import org.apache.cassandra.service.reads.repair.ReadRepair; +import org.apache.cassandra.schema.SpeculativeRetryParam; +import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.service.StorageProxy.LocalReadRunnable; +import org.apache.cassandra.tracing.TraceState; +import org.apache.cassandra.tracing.Tracing; + +/** + * Sends a read request to the replicas needed to satisfy a given ConsistencyLevel. + * + * Optionally, may perform additional requests to provide redundancy against replica failure: + * AlwaysSpeculatingReadExecutor will always send a request to one extra replica, while + * SpeculatingReadExecutor will wait until it looks like the original request is in danger + * of timing out before performing extra reads. + */ +public abstract class AbstractReadExecutor +{ + private static final Logger logger = LoggerFactory.getLogger(AbstractReadExecutor.class); + + protected final ReadCommand command; + protected final ConsistencyLevel consistency; + protected final List<InetAddressAndPort> targetReplicas; + protected final ReadRepair readRepair; + protected final DigestResolver digestResolver; + protected final ReadCallback handler; + protected final TraceState traceState; + protected final ColumnFamilyStore cfs; + protected final long queryStartNanoTime; + protected volatile PartitionIterator result = null; + + AbstractReadExecutor(Keyspace keyspace, ColumnFamilyStore cfs, ReadCommand command, ConsistencyLevel consistency, List<InetAddressAndPort> targetReplicas, long queryStartNanoTime) + { + this.command = command; + this.consistency = consistency; + this.targetReplicas = targetReplicas; + this.readRepair = ReadRepair.create(command, targetReplicas, queryStartNanoTime, consistency); + this.digestResolver = new DigestResolver(keyspace, command, consistency, readRepair, targetReplicas.size()); + this.handler = new ReadCallback(digestResolver, consistency, command, targetReplicas, queryStartNanoTime, readRepair); + this.cfs = cfs; + this.traceState = Tracing.instance.get(); + this.queryStartNanoTime = queryStartNanoTime; + + // Set the digest version (if we request some digests). This is the smallest version amongst all our target replicas since new nodes + // knows how to produce older digest but the reverse is not true. + // TODO: we need this when talking with pre-3.0 nodes. So if we preserve the digest format moving forward, we can get rid of this once + // we stop being compatible with pre-3.0 nodes. + int digestVersion = MessagingService.current_version; + for (InetAddressAndPort replica : targetReplicas) + digestVersion = Math.min(digestVersion, MessagingService.instance().getVersion(replica)); + command.setDigestVersion(digestVersion); + } + + private DecoratedKey getKey() + { + if (command instanceof SinglePartitionReadCommand) + { + return ((SinglePartitionReadCommand) command).partitionKey(); + } + else + { + return null; + } + } + + protected void makeDataRequests(Iterable<InetAddressAndPort> endpoints) + { + makeRequests(command, endpoints); + + } + + protected void makeDigestRequests(Iterable<InetAddressAndPort> endpoints) + { + makeRequests(command.copyAsDigestQuery(), endpoints); + } + + private void makeRequests(ReadCommand readCommand, Iterable<InetAddressAndPort> endpoints) + { + boolean hasLocalEndpoint = false; + + for (InetAddressAndPort endpoint : endpoints) + { + if (StorageProxy.canDoLocalRequest(endpoint)) + { + hasLocalEndpoint = true; + continue; + } + + if (traceState != null) + traceState.trace("reading {} from {}", readCommand.isDigestQuery() ? "digest" : "data", endpoint); + logger.trace("reading {} from {}", readCommand.isDigestQuery() ? "digest" : "data", endpoint); + MessageOut<ReadCommand> message = readCommand.createMessage(); + MessagingService.instance().sendRRWithFailure(message, endpoint, handler); + } + + // We delay the local (potentially blocking) read till the end to avoid stalling remote requests. + if (hasLocalEndpoint) + { + logger.trace("reading {} locally", readCommand.isDigestQuery() ? "digest" : "data"); + StageManager.getStage(Stage.READ).maybeExecuteImmediately(new LocalReadRunnable(command, handler)); + } + } + + /** + * Perform additional requests if it looks like the original will time out. May block while it waits + * to see if the original requests are answered first. + */ + public abstract void maybeTryAdditionalReplicas(); + + /** + * Get the replicas involved in the [finished] request. + * + * @return target replicas + the extra replica, *IF* we speculated. + */ + public abstract List<InetAddressAndPort> getContactedReplicas(); + + /** + * send the initial set of requests + */ + public abstract void executeAsync(); + + private static ReadRepairDecision newReadRepairDecision(TableMetadata metadata) + { + if (metadata.params.readRepairChance > 0d || + metadata.params.dcLocalReadRepairChance > 0) + { + double chance = ThreadLocalRandom.current().nextDouble(); + if (metadata.params.readRepairChance > chance) + return ReadRepairDecision.GLOBAL; + + if (metadata.params.dcLocalReadRepairChance > chance) + return ReadRepairDecision.DC_LOCAL; + } + + return ReadRepairDecision.NONE; + } + + /** + * @return an executor appropriate for the configured speculative read policy + */ + public static AbstractReadExecutor getReadExecutor(SinglePartitionReadCommand command, ConsistencyLevel consistencyLevel, long queryStartNanoTime) throws UnavailableException + { + Keyspace keyspace = Keyspace.open(command.metadata().keyspace); + List<InetAddressAndPort> allReplicas = StorageProxy.getLiveSortedEndpoints(keyspace, command.partitionKey()); + // 11980: Excluding EACH_QUORUM reads from potential RR, so that we do not miscount DC responses + ReadRepairDecision repairDecision = consistencyLevel == ConsistencyLevel.EACH_QUORUM + ? ReadRepairDecision.NONE + : newReadRepairDecision(command.metadata()); + List<InetAddressAndPort> targetReplicas = consistencyLevel.filterForQuery(keyspace, allReplicas, repairDecision); + + // Throw UAE early if we don't have enough replicas. + consistencyLevel.assureSufficientLiveNodes(keyspace, targetReplicas); + + if (repairDecision != ReadRepairDecision.NONE) + { + Tracing.trace("Read-repair {}", repairDecision); + ReadRepairMetrics.attempted.mark(); + } + + ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(command.metadata().id); + SpeculativeRetryParam retry = cfs.metadata().params.speculativeRetry; + + // Speculative retry is disabled *OR* + // 11980: Disable speculative retry if using EACH_QUORUM in order to prevent miscounting DC responses + if (retry.equals(SpeculativeRetryParam.NONE) + | consistencyLevel == ConsistencyLevel.EACH_QUORUM) + return new NeverSpeculatingReadExecutor(keyspace, cfs, command, consistencyLevel, targetReplicas, queryStartNanoTime, false); + + // There are simply no extra replicas to speculate. + // Handle this separately so it can log failed attempts to speculate due to lack of replicas + if (consistencyLevel.blockFor(keyspace) == allReplicas.size()) + return new NeverSpeculatingReadExecutor(keyspace, cfs, command, consistencyLevel, targetReplicas, queryStartNanoTime, true); + + if (targetReplicas.size() == allReplicas.size()) + { + // CL.ALL, RRD.GLOBAL or RRD.DC_LOCAL and a single-DC. + // We are going to contact every node anyway, so ask for 2 full data requests instead of 1, for redundancy + // (same amount of requests in total, but we turn 1 digest request into a full blown data request). + return new AlwaysSpeculatingReadExecutor(keyspace, cfs, command, consistencyLevel, targetReplicas, queryStartNanoTime); + } + + // RRD.NONE or RRD.DC_LOCAL w/ multiple DCs. + InetAddressAndPort extraReplica = allReplicas.get(targetReplicas.size()); + // With repair decision DC_LOCAL all replicas/target replicas may be in different order, so + // we might have to find a replacement that's not already in targetReplicas. + if (repairDecision == ReadRepairDecision.DC_LOCAL && targetReplicas.contains(extraReplica)) + { + for (InetAddressAndPort address : allReplicas) + { + if (!targetReplicas.contains(address)) + { + extraReplica = address; + break; + } + } + } + targetReplicas.add(extraReplica); + + if (retry.equals(SpeculativeRetryParam.ALWAYS)) + return new AlwaysSpeculatingReadExecutor(keyspace, cfs, command, consistencyLevel, targetReplicas, queryStartNanoTime); + else // PERCENTILE or CUSTOM. + return new SpeculatingReadExecutor(keyspace, cfs, command, consistencyLevel, targetReplicas, queryStartNanoTime); + } + + /** + * Returns true if speculation should occur and if it should then block until it is time to + * send the speculative reads + */ + boolean shouldSpeculateAndMaybeWait() + { + // no latency information, or we're overloaded + if (cfs.sampleLatencyNanos > TimeUnit.MILLISECONDS.toNanos(command.getTimeout())) + return false; + + return !handler.await(cfs.sampleLatencyNanos, TimeUnit.NANOSECONDS); + } + + void onReadTimeout() {} + + public static class NeverSpeculatingReadExecutor extends AbstractReadExecutor + { + /** + * If never speculating due to lack of replicas + * log it is as a failure if it should have happened + * but couldn't due to lack of replicas + */ + private final boolean logFailedSpeculation; + + public NeverSpeculatingReadExecutor(Keyspace keyspace, ColumnFamilyStore cfs, ReadCommand command, ConsistencyLevel consistencyLevel, List<InetAddressAndPort> targetReplicas, long queryStartNanoTime, boolean logFailedSpeculation) + { + super(keyspace, cfs, command, consistencyLevel, targetReplicas, queryStartNanoTime); + this.logFailedSpeculation = logFailedSpeculation; + } + + public void executeAsync() + { + makeDataRequests(targetReplicas.subList(0, 1)); + if (targetReplicas.size() > 1) + makeDigestRequests(targetReplicas.subList(1, targetReplicas.size())); + } + + public void maybeTryAdditionalReplicas() + { + if (shouldSpeculateAndMaybeWait() && logFailedSpeculation) + { + cfs.metric.speculativeInsufficientReplicas.inc(); + } + } + + public List<InetAddressAndPort> getContactedReplicas() + { + return targetReplicas; + } + } + + static class SpeculatingReadExecutor extends AbstractReadExecutor + { + private volatile boolean speculated = false; + + public SpeculatingReadExecutor(Keyspace keyspace, + ColumnFamilyStore cfs, + ReadCommand command, + ConsistencyLevel consistencyLevel, + List<InetAddressAndPort> targetReplicas, + long queryStartNanoTime) + { + super(keyspace, cfs, command, consistencyLevel, targetReplicas, queryStartNanoTime); + } + + public void executeAsync() + { + // if CL + RR result in covering all replicas, getReadExecutor forces AlwaysSpeculating. So we know + // that the last replica in our list is "extra." + List<InetAddressAndPort> initialReplicas = targetReplicas.subList(0, targetReplicas.size() - 1); + + if (handler.blockfor < initialReplicas.size()) + { + // We're hitting additional targets for read repair. Since our "extra" replica is the least- + // preferred by the snitch, we do an extra data read to start with against a replica more + // likely to reply; better to let RR fail than the entire query. + makeDataRequests(initialReplicas.subList(0, 2)); + if (initialReplicas.size() > 2) + makeDigestRequests(initialReplicas.subList(2, initialReplicas.size())); + } + else + { + // not doing read repair; all replies are important, so it doesn't matter which nodes we + // perform data reads against vs digest. + makeDataRequests(initialReplicas.subList(0, 1)); + if (initialReplicas.size() > 1) + makeDigestRequests(initialReplicas.subList(1, initialReplicas.size())); + } + } + + public void maybeTryAdditionalReplicas() + { + if (shouldSpeculateAndMaybeWait()) + { + //Handle speculation stats first in case the callback fires immediately + speculated = true; + cfs.metric.speculativeRetries.inc(); + // Could be waiting on the data, or on enough digests. + ReadCommand retryCommand = command; + if (handler.resolver.isDataPresent()) + retryCommand = command.copyAsDigestQuery(); + + InetAddressAndPort extraReplica = Iterables.getLast(targetReplicas); + if (traceState != null) + traceState.trace("speculating read retry on {}", extraReplica); + logger.trace("speculating read retry on {}", extraReplica); + MessagingService.instance().sendRRWithFailure(retryCommand.createMessage(), extraReplica, handler); + } + } + + public List<InetAddressAndPort> getContactedReplicas() + { + return speculated + ? targetReplicas + : targetReplicas.subList(0, targetReplicas.size() - 1); + } + + @Override + void onReadTimeout() + { + //Shouldn't be possible to get here without first attempting to speculate even if the + //timing is bad + assert speculated; + cfs.metric.speculativeFailedRetries.inc(); + } + } + + private static class AlwaysSpeculatingReadExecutor extends AbstractReadExecutor + { + public AlwaysSpeculatingReadExecutor(Keyspace keyspace, + ColumnFamilyStore cfs, + ReadCommand command, + ConsistencyLevel consistencyLevel, + List<InetAddressAndPort> targetReplicas, + long queryStartNanoTime) + { + super(keyspace, cfs, command, consistencyLevel, targetReplicas, queryStartNanoTime); + } + + public void maybeTryAdditionalReplicas() + { + // no-op + } + + public List<InetAddressAndPort> getContactedReplicas() + { + return targetReplicas; + } + + @Override + public void executeAsync() + { + makeDataRequests(targetReplicas.subList(0, targetReplicas.size() > 1 ? 2 : 1)); + if (targetReplicas.size() > 2) + makeDigestRequests(targetReplicas.subList(2, targetReplicas.size())); + cfs.metric.speculativeRetries.inc(); + } + + @Override + void onReadTimeout() + { + cfs.metric.speculativeFailedRetries.inc(); + } + } + + public void setResult(PartitionIterator result) + { + Preconditions.checkState(this.result == null, "Result can only be set once"); + this.result = result; + } + + /** + * Wait for the CL to be satisfied by responses + */ + public void awaitResponses() throws ReadTimeoutException + { + try + { + handler.awaitResults(); + } + catch (ReadTimeoutException e) + { + try + { + onReadTimeout(); + } + finally + { + throw e; + } + } + + // return immediately, or begin a read repair + if (digestResolver.responsesMatch()) + { + setResult(digestResolver.getData()); + } + else + { + Tracing.trace("Digest mismatch: Mismatch for key {}", getKey()); + readRepair.startForegroundRepair(digestResolver, handler.endpoints, getContactedReplicas(), this::setResult); + } + } + + public void awaitReadRepair() throws ReadTimeoutException + { + try + { + readRepair.awaitForegroundRepairFinish(); + } + catch (ReadTimeoutException e) + { + if (Tracing.isTracing()) + Tracing.trace("Timed out waiting on digest mismatch repair requests"); + else + logger.trace("Timed out waiting on digest mismatch repair requests"); + // the caught exception here will have CL.ALL from the repair command, + // not whatever CL the initial command was at (CASSANDRA-7947) + int blockFor = consistency.blockFor(Keyspace.open(command.metadata().keyspace)); + throw new ReadTimeoutException(consistency, blockFor-1, blockFor, true); + } + } + + public void maybeRepairAdditionalReplicas() + { + // TODO: this + } + + public PartitionIterator getResult() throws ReadFailureException, ReadTimeoutException + { + Preconditions.checkState(result != null, "Result must be set first"); + return result; + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/reads/AsyncRepairCallback.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/AsyncRepairCallback.java b/src/java/org/apache/cassandra/service/reads/AsyncRepairCallback.java new file mode 100644 index 0000000..b7e7435 --- /dev/null +++ b/src/java/org/apache/cassandra/service/reads/AsyncRepairCallback.java @@ -0,0 +1,61 @@ +/* + * 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.service.reads; + +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.concurrent.StageManager; +import org.apache.cassandra.db.ReadResponse; +import org.apache.cassandra.net.IAsyncCallback; +import org.apache.cassandra.net.MessageIn; +import org.apache.cassandra.service.reads.DataResolver; +import org.apache.cassandra.utils.WrappedRunnable; + +public class AsyncRepairCallback implements IAsyncCallback<ReadResponse> +{ + private final DataResolver repairResolver; + private final int blockfor; + protected final AtomicInteger received = new AtomicInteger(0); + + public AsyncRepairCallback(DataResolver repairResolver, int blockfor) + { + this.repairResolver = repairResolver; + this.blockfor = blockfor; + } + + public void response(MessageIn<ReadResponse> message) + { + repairResolver.preprocess(message); + if (received.incrementAndGet() == blockfor) + { + StageManager.getStage(Stage.READ_REPAIR).execute(new WrappedRunnable() + { + protected void runMayThrow() + { + repairResolver.evaluateAllResponses(); + } + }); + } + } + + public boolean isLatencyForSnitch() + { + return true; + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/reads/DataResolver.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/DataResolver.java b/src/java/org/apache/cassandra/service/reads/DataResolver.java new file mode 100644 index 0000000..11dd083 --- /dev/null +++ b/src/java/org/apache/cassandra/service/reads/DataResolver.java @@ -0,0 +1,133 @@ +/* + * 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.service.reads; + +import java.net.InetAddress; +import java.util.*; + +import com.google.common.annotations.VisibleForTesting; + +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.service.reads.repair.ReadRepair; +import org.apache.cassandra.db.*; +import org.apache.cassandra.db.filter.*; +import org.apache.cassandra.db.partitions.*; +import org.apache.cassandra.db.transform.*; +import org.apache.cassandra.net.*; +import org.apache.cassandra.tracing.TraceState; + +public class DataResolver extends ResponseResolver +{ + private static final boolean DROP_OVERSIZED_READ_REPAIR_MUTATIONS = + Boolean.getBoolean("cassandra.drop_oversized_readrepair_mutations"); + + @VisibleForTesting + final List<AsyncOneResponse> repairResults = Collections.synchronizedList(new ArrayList<>()); + private final long queryStartNanoTime; + private final boolean enforceStrictLiveness; + + public DataResolver(Keyspace keyspace, ReadCommand command, ConsistencyLevel consistency, int maxResponseCount, long queryStartNanoTime, ReadRepair readRepair) + { + super(keyspace, command, consistency, readRepair, maxResponseCount); + this.queryStartNanoTime = queryStartNanoTime; + this.enforceStrictLiveness = command.metadata().enforceStrictLiveness(); + } + + public PartitionIterator getData() + { + ReadResponse response = responses.iterator().next().payload; + return UnfilteredPartitionIterators.filter(response.makeIterator(command), command.nowInSec()); + } + + public boolean isDataPresent() + { + return !responses.isEmpty(); + } + + public PartitionIterator resolve() + { + // We could get more responses while this method runs, which is ok (we're happy to ignore any response not here + // at the beginning of this method), so grab the response count once and use that through the method. + int count = responses.size(); + List<UnfilteredPartitionIterator> iters = new ArrayList<>(count); + InetAddressAndPort[] sources = new InetAddressAndPort[count]; + for (int i = 0; i < count; i++) + { + MessageIn<ReadResponse> msg = responses.get(i); + iters.add(msg.payload.makeIterator(command)); + sources[i] = msg.from; + } + + /* + * Even though every response, individually, will honor the limit, it is possible that we will, after the merge, + * have more rows than the client requested. To make sure that we still conform to the original limit, + * we apply a top-level post-reconciliation counter to the merged partition iterator. + * + * Short read protection logic (ShortReadRowsProtection.moreContents()) relies on this counter to be applied + * to the current partition to work. For this reason we have to apply the counter transformation before + * empty partition discard logic kicks in - for it will eagerly consume the iterator. + * + * That's why the order here is: 1) merge; 2) filter rows; 3) count; 4) discard empty partitions + * + * See CASSANDRA-13747 for more details. + */ + + DataLimits.Counter mergedResultCounter = + command.limits().newCounter(command.nowInSec(), true, command.selectsFullPartition(), enforceStrictLiveness); + + UnfilteredPartitionIterator merged = mergeWithShortReadProtection(iters, sources, mergedResultCounter); + FilteredPartitions filtered = FilteredPartitions.filter(merged, new Filter(command.nowInSec(), command.metadata().enforceStrictLiveness())); + PartitionIterator counted = Transformation.apply(filtered, mergedResultCounter); + return Transformation.apply(counted, new EmptyPartitionsDiscarder()); + } + + private UnfilteredPartitionIterator mergeWithShortReadProtection(List<UnfilteredPartitionIterator> results, + InetAddressAndPort[] sources, + DataLimits.Counter mergedResultCounter) + { + // If we have only one results, there is no read repair to do and we can't get short reads + if (results.size() == 1) + return results.get(0); + + /* + * So-called short reads stems from nodes returning only a subset of the results they have due to the limit, + * but that subset not being enough post-reconciliation. So if we don't have a limit, don't bother. + */ + if (!command.limits().isUnlimited()) + for (int i = 0; i < results.size(); i++) + { + results.set(i, ShortReadProtection.extend(sources[i], results.get(i), command, mergedResultCounter, queryStartNanoTime, enforceStrictLiveness)); + } + + return UnfilteredPartitionIterators.merge(results, command.nowInSec(), readRepair.getMergeListener(sources)); + } + + public void evaluateAllResponses() + { + // We need to fully consume the results to trigger read repairs if appropriate + try (PartitionIterator iterator = resolve()) + { + PartitionIterators.consume(iterator); + } + } + + public void evaluateAllResponses(TraceState traceState) + { + evaluateAllResponses(); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/reads/DigestResolver.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/DigestResolver.java b/src/java/org/apache/cassandra/service/reads/DigestResolver.java new file mode 100644 index 0000000..828a65e --- /dev/null +++ b/src/java/org/apache/cassandra/service/reads/DigestResolver.java @@ -0,0 +1,93 @@ +/* + * 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.service.reads; + +import java.nio.ByteBuffer; +import java.util.concurrent.TimeUnit; + +import com.google.common.base.Preconditions; + +import org.apache.cassandra.db.*; +import org.apache.cassandra.db.partitions.PartitionIterator; +import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators; +import org.apache.cassandra.net.MessageIn; +import org.apache.cassandra.service.reads.repair.ReadRepair; +import org.apache.cassandra.tracing.TraceState; + +public class DigestResolver extends ResponseResolver +{ + private volatile ReadResponse dataResponse; + + public DigestResolver(Keyspace keyspace, ReadCommand command, ConsistencyLevel consistency, ReadRepair readRepair, int maxResponseCount) + { + super(keyspace, command, consistency, readRepair, maxResponseCount); + Preconditions.checkArgument(command instanceof SinglePartitionReadCommand, + "DigestResolver can only be used with SinglePartitionReadCommand commands"); + } + + @Override + public void preprocess(MessageIn<ReadResponse> message) + { + super.preprocess(message); + if (dataResponse == null && !message.payload.isDigestResponse()) + dataResponse = message.payload; + } + + public PartitionIterator getData() + { + assert isDataPresent(); + return UnfilteredPartitionIterators.filter(dataResponse.makeIterator(command), command.nowInSec()); + } + + public boolean responsesMatch() + { + long start = System.nanoTime(); + + // validate digests against each other; return false immediately on mismatch. + ByteBuffer digest = null; + for (MessageIn<ReadResponse> message : responses) + { + ReadResponse response = message.payload; + + ByteBuffer newDigest = response.digest(command); + if (digest == null) + digest = newDigest; + else if (!digest.equals(newDigest)) + // rely on the fact that only single partition queries use digests + return false; + } + + if (logger.isTraceEnabled()) + logger.trace("responsesMatch: {} ms.", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)); + + return true; + } + + public void evaluateAllResponses(TraceState traceState) + { + if (!responsesMatch()) + { + readRepair.backgroundDigestRepair(traceState); + } + } + + public boolean isDataPresent() + { + return dataResponse != null; + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/reads/ReadCallback.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/ReadCallback.java b/src/java/org/apache/cassandra/service/reads/ReadCallback.java new file mode 100644 index 0000000..62fdfaa --- /dev/null +++ b/src/java/org/apache/cassandra/service/reads/ReadCallback.java @@ -0,0 +1,213 @@ +/* + * 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.service.reads; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; + +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.*; +import org.apache.cassandra.exceptions.RequestFailureReason; +import org.apache.cassandra.exceptions.ReadFailureException; +import org.apache.cassandra.exceptions.ReadTimeoutException; +import org.apache.cassandra.exceptions.UnavailableException; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.metrics.ReadRepairMetrics; +import org.apache.cassandra.net.IAsyncCallbackWithFailure; +import org.apache.cassandra.net.MessageIn; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.service.reads.repair.ReadRepair; +import org.apache.cassandra.tracing.Tracing; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.concurrent.SimpleCondition; + +public class ReadCallback implements IAsyncCallbackWithFailure<ReadResponse> +{ + protected static final Logger logger = LoggerFactory.getLogger( ReadCallback.class ); + + public final ResponseResolver resolver; + final SimpleCondition condition = new SimpleCondition(); + private final long queryStartNanoTime; + final int blockfor; + final List<InetAddressAndPort> endpoints; + private final ReadCommand command; + private final ConsistencyLevel consistencyLevel; + private static final AtomicIntegerFieldUpdater<ReadCallback> recievedUpdater + = AtomicIntegerFieldUpdater.newUpdater(ReadCallback.class, "received"); + private volatile int received = 0; + private static final AtomicIntegerFieldUpdater<ReadCallback> failuresUpdater + = AtomicIntegerFieldUpdater.newUpdater(ReadCallback.class, "failures"); + private volatile int failures = 0; + private final Map<InetAddressAndPort, RequestFailureReason> failureReasonByEndpoint; + + private final Keyspace keyspace; // TODO push this into ConsistencyLevel? + + private final ReadRepair readRepair; + + /** + * Constructor when response count has to be calculated and blocked for. + */ + public ReadCallback(ResponseResolver resolver, ConsistencyLevel consistencyLevel, ReadCommand command, List<InetAddressAndPort> filteredEndpoints, long queryStartNanoTime, ReadRepair readRepair) + { + this(resolver, + consistencyLevel, + consistencyLevel.blockFor(Keyspace.open(command.metadata().keyspace)), + command, + Keyspace.open(command.metadata().keyspace), + filteredEndpoints, + queryStartNanoTime, readRepair); + } + + public ReadCallback(ResponseResolver resolver, ConsistencyLevel consistencyLevel, int blockfor, ReadCommand command, Keyspace keyspace, List<InetAddressAndPort> endpoints, long queryStartNanoTime, ReadRepair readRepair) + { + this.command = command; + this.keyspace = keyspace; + this.blockfor = blockfor; + this.consistencyLevel = consistencyLevel; + this.resolver = resolver; + this.queryStartNanoTime = queryStartNanoTime; + this.endpoints = endpoints; + this.readRepair = readRepair; + this.failureReasonByEndpoint = new ConcurrentHashMap<>(); + // we don't support read repair (or rapid read protection) for range scans yet (CASSANDRA-6897) + assert !(command instanceof PartitionRangeReadCommand) || blockfor >= endpoints.size(); + + if (logger.isTraceEnabled()) + logger.trace("Blockfor is {}; setting up requests to {}", blockfor, StringUtils.join(this.endpoints, ",")); + } + + public boolean await(long timePastStart, TimeUnit unit) + { + long time = unit.toNanos(timePastStart) - (System.nanoTime() - queryStartNanoTime); + try + { + return condition.await(time, TimeUnit.NANOSECONDS); + } + catch (InterruptedException ex) + { + throw new AssertionError(ex); + } + } + + public void awaitResults() throws ReadFailureException, ReadTimeoutException + { + boolean signaled = await(command.getTimeout(), TimeUnit.MILLISECONDS); + boolean failed = blockfor + failures > endpoints.size(); + if (signaled && !failed) + return; + + if (Tracing.isTracing()) + { + String gotData = received > 0 ? (resolver.isDataPresent() ? " (including data)" : " (only digests)") : ""; + Tracing.trace("{}; received {} of {} responses{}", new Object[]{ (failed ? "Failed" : "Timed out"), received, blockfor, gotData }); + } + else if (logger.isDebugEnabled()) + { + String gotData = received > 0 ? (resolver.isDataPresent() ? " (including data)" : " (only digests)") : ""; + logger.debug("{}; received {} of {} responses{}", new Object[]{ (failed ? "Failed" : "Timed out"), received, blockfor, gotData }); + } + + // Same as for writes, see AbstractWriteResponseHandler + throw failed + ? new ReadFailureException(consistencyLevel, received, blockfor, resolver.isDataPresent(), failureReasonByEndpoint) + : new ReadTimeoutException(consistencyLevel, received, blockfor, resolver.isDataPresent()); + } + + public int blockFor() + { + return blockfor; + } + + public void response(MessageIn<ReadResponse> message) + { + resolver.preprocess(message); + int n = waitingFor(message.from) + ? recievedUpdater.incrementAndGet(this) + : received; + if (n >= blockfor && resolver.isDataPresent()) + { + condition.signalAll(); + // kick off a background digest comparison if this is a result that (may have) arrived after + // the original resolve that get() kicks off as soon as the condition is signaled + if (blockfor < endpoints.size() && n == endpoints.size()) + { + readRepair.maybeStartBackgroundRepair(resolver); + } + } + } + + /** + * @return true if the message counts towards the blockfor threshold + */ + private boolean waitingFor(InetAddressAndPort from) + { + return consistencyLevel.isDatacenterLocal() + ? DatabaseDescriptor.getLocalDataCenter().equals(DatabaseDescriptor.getEndpointSnitch().getDatacenter(from)) + : true; + } + + /** + * @return the current number of received responses + */ + public int getReceivedCount() + { + return received; + } + + public void response(ReadResponse result) + { + MessageIn<ReadResponse> message = MessageIn.create(FBUtilities.getBroadcastAddressAndPort(), + result, + Collections.emptyMap(), + MessagingService.Verb.INTERNAL_RESPONSE, + MessagingService.current_version); + response(message); + } + + public void assureSufficientLiveNodes() throws UnavailableException + { + consistencyLevel.assureSufficientLiveNodes(keyspace, endpoints); + } + + public boolean isLatencyForSnitch() + { + return true; + } + + @Override + public void onFailure(InetAddressAndPort from, RequestFailureReason failureReason) + { + int n = waitingFor(from) + ? failuresUpdater.incrementAndGet(this) + : failures; + + failureReasonByEndpoint.put(from, failureReason); + + if (blockfor + n > endpoints.size()) + condition.signalAll(); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/reads/ReadRepairDecision.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/ReadRepairDecision.java b/src/java/org/apache/cassandra/service/reads/ReadRepairDecision.java new file mode 100644 index 0000000..f434c88 --- /dev/null +++ b/src/java/org/apache/cassandra/service/reads/ReadRepairDecision.java @@ -0,0 +1,23 @@ +/* + * 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.service.reads; + +public enum ReadRepairDecision +{ + NONE, GLOBAL, DC_LOCAL; +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/reads/ResponseResolver.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/ResponseResolver.java b/src/java/org/apache/cassandra/service/reads/ResponseResolver.java new file mode 100644 index 0000000..69ec063 --- /dev/null +++ b/src/java/org/apache/cassandra/service/reads/ResponseResolver.java @@ -0,0 +1,66 @@ +/* + * 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.service.reads; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.db.*; +import org.apache.cassandra.net.MessageIn; +import org.apache.cassandra.service.reads.repair.ReadRepair; +import org.apache.cassandra.tracing.TraceState; +import org.apache.cassandra.utils.concurrent.Accumulator; + +public abstract class ResponseResolver +{ + protected static final Logger logger = LoggerFactory.getLogger(ResponseResolver.class); + + protected final Keyspace keyspace; + protected final ReadCommand command; + protected final ConsistencyLevel consistency; + protected final ReadRepair readRepair; + + // Accumulator gives us non-blocking thread-safety with optimal algorithmic constraints + protected final Accumulator<MessageIn<ReadResponse>> responses; + + public ResponseResolver(Keyspace keyspace, ReadCommand command, ConsistencyLevel consistency, ReadRepair readRepair, int maxResponseCount) + { + this.keyspace = keyspace; + this.command = command; + this.consistency = consistency; + this.readRepair = readRepair; + this.responses = new Accumulator<>(maxResponseCount); + } + + /** + * Consume the accumulated responses, starting a read repair if neccesary + */ + public abstract void evaluateAllResponses(TraceState traceState); + + public abstract boolean isDataPresent(); + + public void preprocess(MessageIn<ReadResponse> message) + { + responses.add(message); + } + + public Accumulator<MessageIn<ReadResponse>> getMessages() + { + return responses; + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/reads/ShortReadPartitionsProtection.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/ShortReadPartitionsProtection.java b/src/java/org/apache/cassandra/service/reads/ShortReadPartitionsProtection.java new file mode 100644 index 0000000..fb5f48e --- /dev/null +++ b/src/java/org/apache/cassandra/service/reads/ShortReadPartitionsProtection.java @@ -0,0 +1,187 @@ +/* + * 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.service.reads; + +import java.util.Collections; + +import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.concurrent.StageManager; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.DataRange; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.PartitionPosition; +import org.apache.cassandra.db.PartitionRangeReadCommand; +import org.apache.cassandra.db.ReadCommand; +import org.apache.cassandra.db.filter.DataLimits; +import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator; +import org.apache.cassandra.db.rows.UnfilteredRowIterator; +import org.apache.cassandra.db.transform.MorePartitions; +import org.apache.cassandra.db.transform.MoreRows; +import org.apache.cassandra.db.transform.Transformation; +import org.apache.cassandra.dht.AbstractBounds; +import org.apache.cassandra.dht.ExcludingBounds; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.service.reads.repair.NoopReadRepair; +import org.apache.cassandra.service.StorageProxy; +import org.apache.cassandra.tracing.Tracing; + +public class ShortReadPartitionsProtection extends Transformation<UnfilteredRowIterator> implements MorePartitions<UnfilteredPartitionIterator> +{ + private final ReadCommand command; + private final InetAddressAndPort source; + + private final DataLimits.Counter singleResultCounter; // unmerged per-source counter + private final DataLimits.Counter mergedResultCounter; // merged end-result counter + + private DecoratedKey lastPartitionKey; // key of the last observed partition + + private boolean partitionsFetched; // whether we've seen any new partitions since iteration start or last moreContents() call + + private final long queryStartNanoTime; + + public ShortReadPartitionsProtection(ReadCommand command, InetAddressAndPort source, + DataLimits.Counter singleResultCounter, + DataLimits.Counter mergedResultCounter, + long queryStartNanoTime) + { + this.command = command; + this.source = source; + this.singleResultCounter = singleResultCounter; + this.mergedResultCounter = mergedResultCounter; + this.queryStartNanoTime = queryStartNanoTime; + } + + @Override + public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition) + { + partitionsFetched = true; + + lastPartitionKey = partition.partitionKey(); + + /* + * Extend for moreContents() then apply protection to track lastClustering by applyToRow(). + * + * If we don't apply the transformation *after* extending the partition with MoreRows, + * applyToRow() method of protection will not be called on the first row of the new extension iterator. + */ + ShortReadRowsProtection protection = new ShortReadRowsProtection(partition.partitionKey(), + command, source, + this::executeReadCommand, + singleResultCounter, + mergedResultCounter); + return Transformation.apply(MoreRows.extend(partition, protection), protection); + } + + /* + * We only get here once all the rows and partitions in this iterator have been iterated over, and so + * if the node had returned the requested number of rows but we still get here, then some results were + * skipped during reconciliation. + */ + public UnfilteredPartitionIterator moreContents() + { + // never try to request additional partitions from replicas if our reconciled partitions are already filled to the limit + assert !mergedResultCounter.isDone(); + + // we do not apply short read protection when we have no limits at all + assert !command.limits().isUnlimited(); + + /* + * If this is a single partition read command or an (indexed) partition range read command with + * a partition key specified, then we can't and shouldn't try fetch more partitions. + */ + assert !command.isLimitedToOnePartition(); + + /* + * If the returned result doesn't have enough rows/partitions to satisfy even the original limit, don't ask for more. + * + * Can only take the short cut if there is no per partition limit set. Otherwise it's possible to hit false + * positives due to some rows being uncounted for in certain scenarios (see CASSANDRA-13911). + */ + if (!singleResultCounter.isDone() && command.limits().perPartitionCount() == DataLimits.NO_LIMIT) + return null; + + /* + * Either we had an empty iterator as the initial response, or our moreContents() call got us an empty iterator. + * There is no point to ask the replica for more rows - it has no more in the requested range. + */ + if (!partitionsFetched) + return null; + partitionsFetched = false; + + /* + * We are going to fetch one partition at a time for thrift and potentially more for CQL. + * The row limit will either be set to the per partition limit - if the command has no total row limit set, or + * the total # of rows remaining - if it has some. If we don't grab enough rows in some of the partitions, + * then future ShortReadRowsProtection.moreContents() calls will fetch the missing ones. + */ + int toQuery = command.limits().count() != DataLimits.NO_LIMIT + ? command.limits().count() - counted(mergedResultCounter) + : command.limits().perPartitionCount(); + + ColumnFamilyStore.metricsFor(command.metadata().id).shortReadProtectionRequests.mark(); + Tracing.trace("Requesting {} extra rows from {} for short read protection", toQuery, source); + + PartitionRangeReadCommand cmd = makeFetchAdditionalPartitionReadCommand(toQuery); + return executeReadCommand(cmd); + } + + // Counts the number of rows for regular queries and the number of groups for GROUP BY queries + private int counted(DataLimits.Counter counter) + { + return command.limits().isGroupByLimit() + ? counter.rowCounted() + : counter.counted(); + } + + private PartitionRangeReadCommand makeFetchAdditionalPartitionReadCommand(int toQuery) + { + PartitionRangeReadCommand cmd = (PartitionRangeReadCommand) command; + + DataLimits newLimits = cmd.limits().forShortReadRetry(toQuery); + + AbstractBounds<PartitionPosition> bounds = cmd.dataRange().keyRange(); + AbstractBounds<PartitionPosition> newBounds = bounds.inclusiveRight() + ? new Range<>(lastPartitionKey, bounds.right) + : new ExcludingBounds<>(lastPartitionKey, bounds.right); + DataRange newDataRange = cmd.dataRange().forSubRange(newBounds); + + return cmd.withUpdatedLimitsAndDataRange(newLimits, newDataRange); + } + + private UnfilteredPartitionIterator executeReadCommand(ReadCommand cmd) + { + Keyspace keyspace = Keyspace.open(command.metadata().keyspace); + DataResolver resolver = new DataResolver(keyspace, cmd, ConsistencyLevel.ONE, 1, queryStartNanoTime, NoopReadRepair.instance); + ReadCallback handler = new ReadCallback(resolver, ConsistencyLevel.ONE, cmd, Collections.singletonList(source), queryStartNanoTime, NoopReadRepair.instance); + + if (StorageProxy.canDoLocalRequest(source)) + StageManager.getStage(Stage.READ).maybeExecuteImmediately(new StorageProxy.LocalReadRunnable(cmd, handler)); + else + MessagingService.instance().sendRRWithFailure(cmd.createMessage(), source, handler); + + // We don't call handler.get() because we want to preserve tombstones since we're still in the middle of merging node results. + handler.awaitResults(); + assert resolver.getMessages().size() == 1; + return resolver.getMessages().get(0).payload.makeIterator(command); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/39807ba4/src/java/org/apache/cassandra/service/reads/ShortReadProtection.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/ShortReadProtection.java b/src/java/org/apache/cassandra/service/reads/ShortReadProtection.java new file mode 100644 index 0000000..93c02e0 --- /dev/null +++ b/src/java/org/apache/cassandra/service/reads/ShortReadProtection.java @@ -0,0 +1,74 @@ +/* + * 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.service.reads; + +import java.net.InetAddress; + +import org.apache.cassandra.db.ReadCommand; +import org.apache.cassandra.db.filter.DataLimits; +import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator; +import org.apache.cassandra.db.transform.MorePartitions; +import org.apache.cassandra.db.transform.Transformation; +import org.apache.cassandra.locator.InetAddressAndPort; + +/** + * We have a potential short read if the result from a given node contains the requested number of rows + * (i.e. it has stopped returning results due to the limit), but some of them haven't + * made it into the final post-reconciliation result due to other nodes' row, range, and/or partition tombstones. + * + * If that is the case, then that node may have more rows that we should fetch, as otherwise we could + * ultimately return fewer rows than required. Also, those additional rows may contain tombstones which + * which we also need to fetch as they may shadow rows or partitions from other replicas' results, which we would + * otherwise return incorrectly. + */ +public class ShortReadProtection +{ + public static UnfilteredPartitionIterator extend(InetAddressAndPort source, UnfilteredPartitionIterator partitions, + ReadCommand command, DataLimits.Counter mergedResultCounter, + long queryStartNanoTime, boolean enforceStrictLiveness) + { + DataLimits.Counter singleResultCounter = command.limits().newCounter(command.nowInSec(), + false, + command.selectsFullPartition(), + enforceStrictLiveness).onlyCount(); + + ShortReadPartitionsProtection protection = new ShortReadPartitionsProtection(command, source, singleResultCounter, mergedResultCounter, queryStartNanoTime); + + /* + * The order of extention and transformations is important here. Extending with more partitions has to happen + * first due to the way BaseIterator.hasMoreContents() works: only transformations applied after extension will + * be called on the first partition of the extended iterator. + * + * Additionally, we want singleResultCounter to be applied after SRPP, so that its applyToPartition() method will + * be called last, after the extension done by SRRP.applyToPartition() call. That way we preserve the same order + * when it comes to calling SRRP.moreContents() and applyToRow() callbacks. + * + * See ShortReadPartitionsProtection.applyToPartition() for more details. + */ + + // extend with moreContents() only if it's a range read command with no partition key specified + if (!command.isLimitedToOnePartition()) + partitions = MorePartitions.extend(partitions, protection); // register SRPP.moreContents() + + partitions = Transformation.apply(partitions, protection); // register SRPP.applyToPartition() + partitions = Transformation.apply(partitions, singleResultCounter); // register the per-source counter + + return partitions; + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
