diff --git a/CHANGES.txt b/CHANGES.txt index 95a940b2fa..725da54290 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.2 + * Abort in-progress queries that time out (CASSANDRA-7392) * Add transparent data encryption core classes (CASSANDRA-9945) diff --git a/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java b/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java index 5935669e39..35469cc295 100644 --- a/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java +++ b/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java @@ -22,6 +22,11 @@ package org.apache.cassandra.concurrent; */ public class ScheduledExecutors { + /** + * This pool is used for periodic fast (sub-microsecond) tasks. + */ + public static final DebuggableScheduledThreadPoolExecutor scheduledFastTasks = new DebuggableScheduledThreadPoolExecutor("ScheduledFastTasks"); + /** * This pool is used for periodic short (sub-second) tasks. */ diff --git a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java index 136e2fd82d..a6d8e93bf0 100644 --- a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java +++ b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java @@ -187,7 +187,8 @@ public abstract class UntypedResultSet implements Iterable if (pager.isExhausted()) return endOfData(); - try (ReadOrderGroup orderGroup = pager.startOrderGroup(); PartitionIterator iter = pager.fetchPageInternal(pageSize, orderGroup)) + try (ReadExecutionController executionController = pager.executionController(); + PartitionIterator iter = pager.fetchPageInternal(pageSize, executionController)) { currentPage = select.process(iter, nowInSec).rows.iterator(); } diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index 7eefd8e275..02872997c5 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -360,7 +360,8 @@ public abstract class ModificationStatement implements CQLStatement if (local) { - try (ReadOrderGroup orderGroup = group.startOrderGroup(); PartitionIterator iter = group.executeInternal(orderGroup)) + try (ReadExecutionController executionController = group.executionController(); + PartitionIterator iter = group.executeInternal(executionController)) { return asMaterializedMap(iter); } @@ -575,7 +576,8 @@ public abstract class ModificationStatement implements CQLStatement SinglePartitionReadCommand readCommand = request.readCommand(FBUtilities.nowInSeconds()); FilteredPartition current; - try (ReadOrderGroup orderGroup = readCommand.startOrderGroup(); PartitionIterator iter = readCommand.executeInternal(orderGroup)) + try (ReadExecutionController executionController = readCommand.executionController(); + PartitionIterator iter = readCommand.executeInternal(executionController)) { current = FilteredPartition.create(PartitionIterators.getOnlyElement(iter, readCommand)); } diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index 21bb257714..67d46228c9 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -253,9 +253,9 @@ public class SelectStatement implements CQLStatement this.pager = pager; } - public static Pager forInternalQuery(QueryPager pager, ReadOrderGroup orderGroup) + public static Pager forInternalQuery(QueryPager pager, ReadExecutionController executionController) { - return new InternalPager(pager, orderGroup); + return new InternalPager(pager, executionController); } public static Pager forDistributedQuery(QueryPager pager, ConsistencyLevel consistency, ClientState clientState) @@ -295,17 +295,17 @@ public class SelectStatement implements CQLStatement public static class InternalPager extends Pager { - private final ReadOrderGroup orderGroup; + private final ReadExecutionController executionController; - private InternalPager(QueryPager pager, ReadOrderGroup orderGroup) + private InternalPager(QueryPager pager, ReadExecutionController executionController) { super(pager); - this.orderGroup = orderGroup; + this.executionController = executionController; } public PartitionIterator fetchPage(int pageSize) { - return pager.fetchPageInternal(pageSize, orderGroup); + return pager.fetchPageInternal(pageSize, executionController); } } } @@ -378,11 +378,11 @@ public class SelectStatement implements CQLStatement ReadQuery query = getQuery(options, nowInSec); int pageSize = getPageSize(options); - try (ReadOrderGroup orderGroup = query.startOrderGroup()) + try (ReadExecutionController executionController = query.executionController()) { if (pageSize <= 0 || query.limits().count() <= pageSize) { - try (PartitionIterator data = query.executeInternal(orderGroup)) + try (PartitionIterator data = query.executeInternal(executionController)) { return processResults(data, options, nowInSec); } @@ -390,7 +390,7 @@ public class SelectStatement implements CQLStatement else { QueryPager pager = query.getPager(options.getPagingState(), options.getProtocolVersion()); - return execute(Pager.forInternalQuery(pager, orderGroup), options, pageSize, nowInSec); + return execute(Pager.forInternalQuery(pager, executionController), options, pageSize, nowInSec); } } } diff --git a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java index f17f3e3591..8fd53a7236 100644 --- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java +++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java @@ -175,7 +175,7 @@ public class PartitionRangeReadCommand extends ReadCommand metric.rangeLatency.addNano(latencyNanos); } - protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadOrderGroup orderGroup) + protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadExecutionController executionController) { ColumnFamilyStore.ViewFragment view = cfs.select(View.select(SSTableSet.LIVE, dataRange().keyRange())); Tracing.trace("Executing seq scan across {} sstables for {}", view.sstables.size(), dataRange().keyRange().getString(metadata().getKeyValidator())); diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index 91227cfc2b..222f81b58e 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.config.*; import org.apache.cassandra.cql3.Operator; import org.apache.cassandra.db.filter.*; +import org.apache.cassandra.db.monitoring.MonitorableImpl; import org.apache.cassandra.db.partitions.*; import org.apache.cassandra.db.rows.*; import org.apache.cassandra.dht.AbstractBounds; @@ -43,6 +44,7 @@ import org.apache.cassandra.schema.UnknownIndexException; import org.apache.cassandra.service.ClientWarn; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.Pair; /** @@ -51,10 +53,10 @@ import org.apache.cassandra.utils.Pair; *

* This contains all the informations needed to do a local read. */ -public abstract class ReadCommand implements ReadQuery +public abstract class ReadCommand extends MonitorableImpl implements ReadQuery { + private static final int TEST_ITERATION_DELAY_MILLIS = Integer.valueOf(System.getProperty("cassandra.test.read_iteration_delay_ms", "0")); protected static final Logger logger = LoggerFactory.getLogger(ReadCommand.class); - public static final IVersionedSerializer serializer = new Serializer(); // For RANGE_SLICE verb: will either dispatch on 'serializer' for 3.0 or 'legacyRangeSliceCommandSerializer' for earlier version. // Can be removed (and replaced by 'serializer') once we drop pre-3.0 backward compatibility. @@ -276,7 +278,7 @@ public abstract class ReadCommand implements ReadQuery */ public abstract ReadCommand copy(); - protected abstract UnfilteredPartitionIterator queryStorage(ColumnFamilyStore cfs, ReadOrderGroup orderGroup); + protected abstract UnfilteredPartitionIterator queryStorage(ColumnFamilyStore cfs, ReadExecutionController executionController); protected abstract int oldestUnrepairedTombstone(); @@ -321,13 +323,13 @@ public abstract class ReadCommand implements ReadQuery /** * Executes this command on the local host. * - * @param orderGroup the operation group spanning this command + * @param executionController the execution controller spanning this command * * @return an iterator over the result of executing this command locally. */ @SuppressWarnings("resource") // The result iterator is closed upon exceptions (we know it's fine to potentially not close the intermediary // iterators created inside the try as long as we do close the original resultIterator), or by closing the result. - public UnfilteredPartitionIterator executeLocally(ReadOrderGroup orderGroup) + public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController) { long startTimeNanos = System.nanoTime(); @@ -339,11 +341,12 @@ public abstract class ReadCommand implements ReadQuery Tracing.trace("Executing read on {}.{} using index {}", cfs.metadata.ksName, cfs.metadata.cfName, index.getIndexName()); UnfilteredPartitionIterator resultIterator = searcher == null - ? queryStorage(cfs, orderGroup) - : searcher.search(orderGroup); + ? queryStorage(cfs, executionController) + : searcher.search(executionController); try { + resultIterator = withStateTracking(resultIterator); resultIterator = withMetricsRecording(withoutPurgeableTombstones(resultIterator, cfs), cfs.metric, startTimeNanos); // If we've used a 2ndary index, we know the result already satisfy the primary expression used, so @@ -367,14 +370,14 @@ public abstract class ReadCommand implements ReadQuery protected abstract void recordLatency(TableMetrics metric, long latencyNanos); - public PartitionIterator executeInternal(ReadOrderGroup orderGroup) + public PartitionIterator executeInternal(ReadExecutionController controller) { - return UnfilteredPartitionIterators.filter(executeLocally(orderGroup), nowInSec()); + return UnfilteredPartitionIterators.filter(executeLocally(controller), nowInSec()); } - public ReadOrderGroup startOrderGroup() + public ReadExecutionController executionController() { - return ReadOrderGroup.forCommand(this); + return ReadExecutionController.forCommand(this); } /** @@ -470,6 +473,48 @@ public abstract class ReadCommand implements ReadQuery }; } + protected UnfilteredPartitionIterator withStateTracking(UnfilteredPartitionIterator iter) + { + return new WrappingUnfilteredPartitionIterator(iter) + { + @Override + public UnfilteredRowIterator computeNext(UnfilteredRowIterator iter) + { + if (isAborted()) + return null; + + if (TEST_ITERATION_DELAY_MILLIS > 0) + maybeDelayForTesting(); + + return iter; + } + }; + } + + protected UnfilteredRowIterator withStateTracking(UnfilteredRowIterator iter) + { + return new WrappingUnfilteredRowIterator(iter) + { + @Override + public boolean hasNext() + { + if (isAborted()) + return false; + + if (TEST_ITERATION_DELAY_MILLIS > 0) + maybeDelayForTesting(); + + return super.hasNext(); + } + }; + } + + private void maybeDelayForTesting() + { + if (!metadata.ksName.startsWith("system")) + FBUtilities.sleepQuietly(TEST_ITERATION_DELAY_MILLIS); + } + /** * Creates a message for this command. */ @@ -512,6 +557,12 @@ public abstract class ReadCommand implements ReadQuery return sb.toString(); } + // Monitorable interface + public String name() + { + return toCQLString(); + } + private static class Serializer implements IVersionedSerializer { private static int digestFlag(boolean isDigest) diff --git a/src/java/org/apache/cassandra/db/ReadCommandVerbHandler.java b/src/java/org/apache/cassandra/db/ReadCommandVerbHandler.java index 72a6fa8690..9eaa8fae72 100644 --- a/src/java/org/apache/cassandra/db/ReadCommandVerbHandler.java +++ b/src/java/org/apache/cassandra/db/ReadCommandVerbHandler.java @@ -41,15 +41,24 @@ public class ReadCommandVerbHandler implements IVerbHandler } ReadCommand command = message.payload; + command.setMonitoringTime(message.constructionTime, message.getTimeout()); + ReadResponse response; - try (ReadOrderGroup opGroup = command.startOrderGroup(); UnfilteredPartitionIterator iterator = command.executeLocally(opGroup)) + try (ReadExecutionController executionController = command.executionController(); + UnfilteredPartitionIterator iterator = command.executeLocally(executionController)) { response = command.createResponse(iterator, command.columnFilter()); } - MessageOut reply = new MessageOut<>(MessagingService.Verb.REQUEST_RESPONSE, response, serializer()); + if (!command.complete()) + { + Tracing.trace("Discarding partial response to {} (timed out)", message.from); + MessagingService.instance().incrementDroppedMessages(message); + return; + } Tracing.trace("Enqueuing response to {}", message.from); + MessageOut reply = new MessageOut<>(MessagingService.Verb.REQUEST_RESPONSE, response, ReadResponse.serializer); MessagingService.instance().sendReply(reply, id, message.from); } } diff --git a/src/java/org/apache/cassandra/db/ReadOrderGroup.java b/src/java/org/apache/cassandra/db/ReadExecutionController.java similarity index 84% rename from src/java/org/apache/cassandra/db/ReadOrderGroup.java rename to src/java/org/apache/cassandra/db/ReadExecutionController.java index 0720d79d70..0bb40f8f27 100644 --- a/src/java/org/apache/cassandra/db/ReadOrderGroup.java +++ b/src/java/org/apache/cassandra/db/ReadExecutionController.java @@ -20,7 +20,7 @@ package org.apache.cassandra.db; import org.apache.cassandra.index.Index; import org.apache.cassandra.utils.concurrent.OpOrder; -public class ReadOrderGroup implements AutoCloseable +public class ReadExecutionController implements AutoCloseable { // For every reads private final OpOrder.Group baseOp; @@ -29,7 +29,7 @@ public class ReadOrderGroup implements AutoCloseable private final OpOrder.Group indexOp; private final OpOrder.Group writeOp; - private ReadOrderGroup(OpOrder.Group baseOp, OpOrder.Group indexOp, OpOrder.Group writeOp) + private ReadExecutionController(OpOrder.Group baseOp, OpOrder.Group indexOp, OpOrder.Group writeOp) { this.baseOp = baseOp; this.indexOp = indexOp; @@ -51,19 +51,24 @@ public class ReadOrderGroup implements AutoCloseable return writeOp; } - public static ReadOrderGroup emptyGroup() + public static ReadExecutionController empty() { - return new ReadOrderGroup(null, null, null); + return new ReadExecutionController(null, null, null); } - public static ReadOrderGroup forCommand(ReadCommand command) + public static ReadExecutionController forReadOp(OpOrder.Group readOp) + { + return new ReadExecutionController(readOp, null, null); + } + + public static ReadExecutionController forCommand(ReadCommand command) { ColumnFamilyStore baseCfs = Keyspace.openAndGetStore(command.metadata()); ColumnFamilyStore indexCfs = maybeGetIndexCfs(baseCfs, command); if (indexCfs == null) { - return new ReadOrderGroup(baseCfs.readOrdering.start(), null, null); + return new ReadExecutionController(baseCfs.readOrdering.start(), null, null); } else { @@ -76,7 +81,7 @@ public class ReadOrderGroup implements AutoCloseable // TODO: this should perhaps not open and maintain a writeOp for the full duration, but instead only *try* to delete stale entries, without blocking if there's no room // as it stands, we open a writeOp and keep it open for the duration to ensure that should this CF get flushed to make room we don't block the reclamation of any room being made writeOp = baseCfs.keyspace.writeOrder.start(); - return new ReadOrderGroup(baseOp, indexOp, writeOp); + return new ReadExecutionController(baseOp, indexOp, writeOp); } catch (RuntimeException e) { diff --git a/src/java/org/apache/cassandra/db/ReadQuery.java b/src/java/org/apache/cassandra/db/ReadQuery.java index d1f52727cc..2b5c09c0f9 100644 --- a/src/java/org/apache/cassandra/db/ReadQuery.java +++ b/src/java/org/apache/cassandra/db/ReadQuery.java @@ -35,9 +35,9 @@ public interface ReadQuery { ReadQuery EMPTY = new ReadQuery() { - public ReadOrderGroup startOrderGroup() + public ReadExecutionController executionController() { - return ReadOrderGroup.emptyGroup(); + return ReadExecutionController.empty(); } public PartitionIterator execute(ConsistencyLevel consistency, ClientState clientState) throws RequestExecutionException @@ -45,7 +45,7 @@ public interface ReadQuery return PartitionIterators.EMPTY; } - public PartitionIterator executeInternal(ReadOrderGroup orderGroup) + public PartitionIterator executeInternal(ReadExecutionController controller) { return PartitionIterators.EMPTY; } @@ -86,9 +86,9 @@ public interface ReadQuery * The returned object must be closed on all path and it is thus strongly advised to * use it in a try-with-ressource construction. * - * @return a newly started order group for this {@code ReadQuery}. + * @return a newly started execution controller for this {@code ReadQuery}. */ - public ReadOrderGroup startOrderGroup(); + public ReadExecutionController executionController(); /** * Executes the query at the provided consistency level. @@ -104,10 +104,10 @@ public interface ReadQuery /** * Execute the query for internal queries (that is, it basically executes the query locally). * - * @param orderGroup the {@code ReadOrderGroup} protecting the read. + * @param controller the {@code ReadExecutionController} protecting the read. * @return the result of the query. */ - public PartitionIterator executeInternal(ReadOrderGroup orderGroup); + public PartitionIterator executeInternal(ReadExecutionController controller); /** * Returns a pager for the query. diff --git a/src/java/org/apache/cassandra/db/SinglePartitionNamesCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionNamesCommand.java index 430e4a1de0..1181485c6b 100644 --- a/src/java/org/apache/cassandra/db/SinglePartitionNamesCommand.java +++ b/src/java/org/apache/cassandra/db/SinglePartitionNamesCommand.java @@ -181,7 +181,7 @@ public class SinglePartitionNamesCommand extends SinglePartitionReadCommand partitions = new ArrayList<>(commands.size()); for (SinglePartitionReadCommand cmd : commands) - partitions.add(cmd.executeInternal(orderGroup)); + partitions.add(cmd.executeInternal(controller)); // Because we only have enforce the limit per command, we need to enforce it globally. return limits.filter(PartitionIterators.concat(partitions), nowInSec); diff --git a/src/java/org/apache/cassandra/db/SinglePartitionSliceCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionSliceCommand.java index 27aab6232e..9fabdc2395 100644 --- a/src/java/org/apache/cassandra/db/SinglePartitionSliceCommand.java +++ b/src/java/org/apache/cassandra/db/SinglePartitionSliceCommand.java @@ -247,7 +247,7 @@ public class SinglePartitionSliceCommand extends SinglePartitionReadCommand public boolean isEQ() { - return startValue.equals(endValue); + return Objects.equals(startValue, endValue); } } } diff --git a/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java b/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java index d3a289a3e1..be6d7e2322 100644 --- a/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java +++ b/src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java @@ -217,7 +217,7 @@ public class ClusteringIndexNamesFilter extends AbstractClusteringIndexFilter public String toCQLString(CFMetaData metadata) { - if (clusterings.isEmpty()) + if (metadata.clusteringColumns().isEmpty() || clusterings.size() <= 1) return ""; StringBuilder sb = new StringBuilder(); diff --git a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java index 1a4573e8f1..98b3600e31 100644 --- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java +++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java @@ -317,9 +317,12 @@ public class ColumnFilter return ""; StringBuilder sb = new StringBuilder(); - appendColumnDef(sb, defs.next()); while (defs.hasNext()) - appendColumnDef(sb.append(", "), defs.next()); + { + appendColumnDef(sb, defs.next()); + if (defs.hasNext()) + sb.append(", "); + } return sb.toString(); } diff --git a/src/java/org/apache/cassandra/db/monitoring/ApproximateTime.java b/src/java/org/apache/cassandra/db/monitoring/ApproximateTime.java new file mode 100644 index 0000000000..1d5739847b --- /dev/null +++ b/src/java/org/apache/cassandra/db/monitoring/ApproximateTime.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.db.monitoring; + +import java.util.concurrent.TimeUnit; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.concurrent.ScheduledExecutors; +import org.apache.cassandra.config.Config; + +/** + * This is an approximation of System.currentTimeInMillis(). It updates its + * time value at periodic intervals of CHECK_INTERVAL_MS milliseconds + * (currently 10 milliseconds by default). It can be used as a faster alternative + * to System.currentTimeInMillis() every time an imprecision of a few milliseconds + * can be accepted. + */ +public class ApproximateTime +{ + private static final Logger logger = LoggerFactory.getLogger(ApproximateTime.class); + private static final int CHECK_INTERVAL_MS = Math.max(5, Integer.valueOf(System.getProperty(Config.PROPERTY_PREFIX + "approximate_time_precision_ms", "10"))); + + private static volatile long time = System.currentTimeMillis(); + static + { + logger.info("Scheduling approximate time-check task with a precision of {} milliseconds", CHECK_INTERVAL_MS); + ScheduledExecutors.scheduledFastTasks.scheduleWithFixedDelay(() -> time = System.currentTimeMillis(), + CHECK_INTERVAL_MS, + CHECK_INTERVAL_MS, + TimeUnit.MILLISECONDS); + } + + public static long currentTimeMillis() + { + return time; + } + + public static long precision() + { + return 2 * CHECK_INTERVAL_MS; + } + +} diff --git a/src/java/org/apache/cassandra/db/monitoring/ConstructionTime.java b/src/java/org/apache/cassandra/db/monitoring/ConstructionTime.java new file mode 100644 index 0000000000..d6b607848c --- /dev/null +++ b/src/java/org/apache/cassandra/db/monitoring/ConstructionTime.java @@ -0,0 +1,41 @@ +/* + * 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.db.monitoring; + +public final class ConstructionTime +{ + public final long timestamp; + public final boolean isCrossNode; + + public ConstructionTime() + { + this(ApproximateTime.currentTimeMillis()); + } + + public ConstructionTime(long timestamp) + { + this(timestamp, false); + } + + public ConstructionTime(long timestamp, boolean isCrossNode) + { + this.timestamp = timestamp; + this.isCrossNode = isCrossNode; + } +} diff --git a/src/java/org/apache/cassandra/db/monitoring/Monitorable.java b/src/java/org/apache/cassandra/db/monitoring/Monitorable.java new file mode 100644 index 0000000000..202ac87850 --- /dev/null +++ b/src/java/org/apache/cassandra/db/monitoring/Monitorable.java @@ -0,0 +1,33 @@ +/* + * 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.db.monitoring; + +public interface Monitorable +{ + String name(); + ConstructionTime constructionTime(); + long timeout(); + + boolean isInProgress(); + boolean isAborted(); + boolean isCompleted(); + + boolean abort(); + boolean complete(); +} diff --git a/src/java/org/apache/cassandra/db/monitoring/MonitorableImpl.java b/src/java/org/apache/cassandra/db/monitoring/MonitorableImpl.java new file mode 100644 index 0000000000..f89f8ad539 --- /dev/null +++ b/src/java/org/apache/cassandra/db/monitoring/MonitorableImpl.java @@ -0,0 +1,104 @@ +/* + * 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.db.monitoring; + +public abstract class MonitorableImpl implements Monitorable +{ + private MonitoringState state; + private ConstructionTime constructionTime; + private long timeout; + + protected MonitorableImpl() + { + this.state = MonitoringState.IN_PROGRESS; + } + + /** + * This setter is ugly but the construction chain to ReadCommand + * is too complex, it would require passing new parameters to all serializers + * or specializing the serializers to accept these message properties. + */ + public void setMonitoringTime(ConstructionTime constructionTime, long timeout) + { + this.constructionTime = constructionTime; + this.timeout = timeout; + } + + public ConstructionTime constructionTime() + { + return constructionTime; + } + + public long timeout() + { + return timeout; + } + + public boolean isInProgress() + { + check(); + return state == MonitoringState.IN_PROGRESS; + } + + public boolean isAborted() + { + check(); + return state == MonitoringState.ABORTED; + } + + public boolean isCompleted() + { + check(); + return state == MonitoringState.COMPLETED; + } + + public boolean abort() + { + if (state == MonitoringState.IN_PROGRESS) + { + if (constructionTime != null) + MonitoringTask.addFailedOperation(this, ApproximateTime.currentTimeMillis()); + state = MonitoringState.ABORTED; + return true; + } + + return state == MonitoringState.ABORTED; + } + + public boolean complete() + { + if (state == MonitoringState.IN_PROGRESS) + { + state = MonitoringState.COMPLETED; + return true; + } + + return state == MonitoringState.COMPLETED; + } + + private void check() + { + if (constructionTime == null || state != MonitoringState.IN_PROGRESS) + return; + + long elapsed = ApproximateTime.currentTimeMillis() - constructionTime.timestamp; + if (elapsed >= timeout) + abort(); + } +} diff --git a/src/java/org/apache/cassandra/db/monitoring/MonitoringState.java b/src/java/org/apache/cassandra/db/monitoring/MonitoringState.java new file mode 100644 index 0000000000..4fe3cf8bef --- /dev/null +++ b/src/java/org/apache/cassandra/db/monitoring/MonitoringState.java @@ -0,0 +1,26 @@ +/* + * 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.db.monitoring; + +public enum MonitoringState +{ + IN_PROGRESS, + ABORTED, + COMPLETED +} diff --git a/src/java/org/apache/cassandra/db/monitoring/MonitoringTask.java b/src/java/org/apache/cassandra/db/monitoring/MonitoringTask.java new file mode 100644 index 0000000000..6d28078c4d --- /dev/null +++ b/src/java/org/apache/cassandra/db/monitoring/MonitoringTask.java @@ -0,0 +1,264 @@ +/* + * 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.db.monitoring; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import com.google.common.annotations.VisibleForTesting; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.concurrent.ScheduledExecutors; +import org.apache.cassandra.config.Config; + +import static java.lang.System.getProperty; + +/** + * A task for monitoring in progress operations, currently only read queries, and aborting them if they time out. + * We also log timed out operations, see CASSANDRA-7392. + */ +public class MonitoringTask +{ + private static final String LINE_SEPARATOR = getProperty("line.separator"); + private static final Logger logger = LoggerFactory.getLogger(MonitoringTask.class); + + /** + * Defines the interval for reporting any operations that have timed out. + */ + private static final int REPORT_INTERVAL_MS = Math.max(0, Integer.valueOf(System.getProperty(Config.PROPERTY_PREFIX + "monitoring_report_interval_ms", "5000"))); + + /** + * Defines the maximum number of unique timed out queries that will be reported in the logs. + * Use a negative number to remove any limit. + */ + private static final int MAX_OPERATIONS = Integer.valueOf(System.getProperty(Config.PROPERTY_PREFIX + "monitoring_max_operations", "50")); + + @VisibleForTesting + static MonitoringTask instance = make(REPORT_INTERVAL_MS, MAX_OPERATIONS); + + private final int maxOperations; + private final ScheduledFuture reportingTask; + private final BlockingQueue operationsQueue; + private final AtomicLong numDroppedOperations; + private long lastLogTime; + + @VisibleForTesting + static MonitoringTask make(int reportIntervalMillis, int maxTimedoutOperations) + { + if (instance != null) + { + instance.cancel(); + instance = null; + } + + return new MonitoringTask(reportIntervalMillis, maxTimedoutOperations); + } + + private MonitoringTask(int reportIntervalMillis, int maxOperations) + { + this.maxOperations = maxOperations; + this.operationsQueue = maxOperations > 0 ? new ArrayBlockingQueue<>(maxOperations) : new LinkedBlockingQueue<>(); + this.numDroppedOperations = new AtomicLong(); + this.lastLogTime = ApproximateTime.currentTimeMillis(); + + logger.info("Scheduling monitoring task with report interval of {} ms, max operations {}", reportIntervalMillis, maxOperations); + this.reportingTask = ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(() -> logFailedOperations(ApproximateTime.currentTimeMillis()), + reportIntervalMillis, + reportIntervalMillis, + TimeUnit.MILLISECONDS); + } + + public void cancel() + { + reportingTask.cancel(false); + } + + public static void addFailedOperation(Monitorable operation, long now) + { + instance.innerAddFailedOperation(operation, now); + } + + private void innerAddFailedOperation(Monitorable operation, long now) + { + if (maxOperations == 0) + return; // logging of failed operations disabled + + if (!operationsQueue.offer(new FailedOperation(operation, now))) + numDroppedOperations.incrementAndGet(); + } + + @VisibleForTesting + FailedOperations aggregateFailedOperations() + { + Map operations = new HashMap<>(); + + FailedOperation failedOperation; + while((failedOperation = operationsQueue.poll()) != null) + { + FailedOperation existing = operations.get(failedOperation.name()); + if (existing != null) + existing.addTimeout(failedOperation); + else + operations.put(failedOperation.name(), failedOperation); + } + + return new FailedOperations(operations, numDroppedOperations.getAndSet(0L)); + } + + @VisibleForTesting + List getFailedOperations() + { + FailedOperations failedOperations = aggregateFailedOperations(); + String ret = failedOperations.getLogMessage(); + lastLogTime = ApproximateTime.currentTimeMillis(); + return ret.isEmpty() ? Collections.emptyList() : Arrays.asList(ret.split("\n")); + } + + @VisibleForTesting + void logFailedOperations(long now) + { + FailedOperations failedOperations = aggregateFailedOperations(); + if (!failedOperations.isEmpty()) + { + long elapsed = now - lastLogTime; + logger.warn("{} operations timed out in the last {} msecs, operation list available at debug log level", + failedOperations.num(), + elapsed); + + if (logger.isDebugEnabled()) + logger.debug("{} operations timed out in the last {} msecs:{}{}", + failedOperations.num(), + elapsed, + LINE_SEPARATOR, + failedOperations.getLogMessage()); + } + + lastLogTime = now; + } + + private static final class FailedOperations + { + public final Map operations; + public final long numDropped; + + FailedOperations(Map operations, long numDropped) + { + this.operations = operations; + this.numDropped = numDropped; + } + + public boolean isEmpty() + { + return operations.isEmpty() && numDropped == 0; + } + + public long num() + { + return operations.size() + numDropped; + } + + public String getLogMessage() + { + if (isEmpty()) + return ""; + + final StringBuilder ret = new StringBuilder(); + operations.values().forEach(o -> addOperation(ret, o)); + + if (numDropped > 0) + ret.append(LINE_SEPARATOR) + .append("... (") + .append(numDropped) + .append(" were dropped)"); + + return ret.toString(); + } + + private static void addOperation(StringBuilder ret, FailedOperation operation) + { + if (ret.length() > 0) + ret.append(LINE_SEPARATOR); + + ret.append(operation.getLogMessage()); + } + } + + private final static class FailedOperation + { + public final Monitorable operation; + public int numTimeouts; + public long totalTime; + public long maxTime; + public long minTime; + private String name; + + FailedOperation(Monitorable operation, long failedAt) + { + this.operation = operation; + numTimeouts = 1; + totalTime = failedAt - operation.constructionTime().timestamp; + minTime = totalTime; + maxTime = totalTime; + } + + public String name() + { + if (name == null) + name = operation.name(); + return name; + } + + void addTimeout(FailedOperation operation) + { + numTimeouts++; + totalTime += operation.totalTime; + maxTime = Math.max(maxTime, operation.maxTime); + minTime = Math.min(minTime, operation.minTime); + } + + public String getLogMessage() + { + if (numTimeouts == 1) + return String.format("%s: total time %d msec - timeout %d %s", + name(), + totalTime, + operation.timeout(), + operation.constructionTime().isCrossNode ? "msec/cross-node" : "msec"); + else + return String.format("%s (timed out %d times): total time avg/min/max %d/%d/%d msec - timeout %d %s", + name(), + numTimeouts, + totalTime / numTimeouts, + minTime, + maxTime, + operation.timeout(), + operation.constructionTime().isCrossNode ? "msec/cross-node" : "msec"); + } + } +} diff --git a/src/java/org/apache/cassandra/db/view/View.java b/src/java/org/apache/cassandra/db/view/View.java index 87ea2ec8d6..7ce790403d 100644 --- a/src/java/org/apache/cassandra/db/view/View.java +++ b/src/java/org/apache/cassandra/db/view/View.java @@ -480,8 +480,8 @@ public class View // Add all of the rows which were recovered from the query to the row set while (!pager.isExhausted()) { - try (ReadOrderGroup orderGroup = pager.startOrderGroup(); - PartitionIterator iter = pager.fetchPageInternal(128, orderGroup)) + try (ReadExecutionController executionController = pager.executionController(); + PartitionIterator iter = pager.fetchPageInternal(128, executionController)) { if (!iter.hasNext()) break; @@ -538,8 +538,8 @@ public class View while (!pager.isExhausted()) { - try (ReadOrderGroup orderGroup = pager.startOrderGroup(); - PartitionIterator iter = pager.fetchPageInternal(128, orderGroup)) + try (ReadExecutionController executionController = pager.executionController(); + PartitionIterator iter = pager.fetchPageInternal(128, executionController)) { while (iter.hasNext()) { diff --git a/src/java/org/apache/cassandra/db/view/ViewBuilder.java b/src/java/org/apache/cassandra/db/view/ViewBuilder.java index 35b023bc6b..8146211018 100644 --- a/src/java/org/apache/cassandra/db/view/ViewBuilder.java +++ b/src/java/org/apache/cassandra/db/view/ViewBuilder.java @@ -80,8 +80,8 @@ public class ViewBuilder extends CompactionInfo.Holder while (!pager.isExhausted()) { - try (ReadOrderGroup orderGroup = pager.startOrderGroup(); - PartitionIterator partitionIterator = pager.fetchPageInternal(128, orderGroup)) + try (ReadExecutionController executionController = pager.executionController(); + PartitionIterator partitionIterator = pager.fetchPageInternal(128, executionController)) { if (!partitionIterator.hasNext()) return; diff --git a/src/java/org/apache/cassandra/index/Index.java b/src/java/org/apache/cassandra/index/Index.java index 3ceec13f24..bd48063f1f 100644 --- a/src/java/org/apache/cassandra/index/Index.java +++ b/src/java/org/apache/cassandra/index/Index.java @@ -427,6 +427,6 @@ public interface Index * @param orderGroup the collection of OpOrder.Groups which the ReadCommand is being performed under. * @return partitions from the base table matching the criteria of the search. */ - public UnfilteredPartitionIterator search(ReadOrderGroup orderGroup); + public UnfilteredPartitionIterator search(ReadExecutionController executionController); } } diff --git a/src/java/org/apache/cassandra/index/internal/CassandraIndexSearcher.java b/src/java/org/apache/cassandra/index/internal/CassandraIndexSearcher.java index 72d252827a..1e2857910d 100644 --- a/src/java/org/apache/cassandra/index/internal/CassandraIndexSearcher.java +++ b/src/java/org/apache/cassandra/index/internal/CassandraIndexSearcher.java @@ -36,14 +36,14 @@ public abstract class CassandraIndexSearcher implements Index.Searcher @SuppressWarnings("resource") // Both the OpOrder and 'indexIter' are closed on exception, or through the closing of the result // of this method. - public UnfilteredPartitionIterator search(ReadOrderGroup orderGroup) + public UnfilteredPartitionIterator search(ReadExecutionController executionController) { // the value of the index expression is the partition key in the index table DecoratedKey indexKey = index.getBackingTable().get().decorateKey(expression.getIndexValue()); - UnfilteredRowIterator indexIter = queryIndex(indexKey, command, orderGroup); + UnfilteredRowIterator indexIter = queryIndex(indexKey, command, executionController); try { - return queryDataFromIndex(indexKey, UnfilteredRowIterators.filter(indexIter, command.nowInSec()), command, orderGroup); + return queryDataFromIndex(indexKey, UnfilteredRowIterators.filter(indexIter, command.nowInSec()), command, executionController); } catch (RuntimeException | Error e) { @@ -52,13 +52,13 @@ public abstract class CassandraIndexSearcher implements Index.Searcher } } - private UnfilteredRowIterator queryIndex(DecoratedKey indexKey, ReadCommand command, ReadOrderGroup orderGroup) + private UnfilteredRowIterator queryIndex(DecoratedKey indexKey, ReadCommand command, ReadExecutionController executionController) { ClusteringIndexFilter filter = makeIndexFilter(command); ColumnFamilyStore indexCfs = index.getBackingTable().get(); CFMetaData indexCfm = indexCfs.metadata; return SinglePartitionReadCommand.create(indexCfm, command.nowInSec(), indexKey, ColumnFilter.all(indexCfm), filter) - .queryMemtableAndDisk(indexCfs, orderGroup.indexReadOpOrderGroup()); + .queryMemtableAndDisk(indexCfs, executionController.indexReadOpOrderGroup()); } private ClusteringIndexFilter makeIndexFilter(ReadCommand command) @@ -168,5 +168,5 @@ public abstract class CassandraIndexSearcher implements Index.Searcher protected abstract UnfilteredPartitionIterator queryDataFromIndex(DecoratedKey indexKey, RowIterator indexHits, ReadCommand command, - ReadOrderGroup orderGroup); + ReadExecutionController executionController); } diff --git a/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java b/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java index f1751f5007..d77b889215 100644 --- a/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java +++ b/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java @@ -57,7 +57,7 @@ public class CompositesSearcher extends CassandraIndexSearcher protected UnfilteredPartitionIterator queryDataFromIndex(final DecoratedKey indexKey, final RowIterator indexHits, final ReadCommand command, - final ReadOrderGroup orderGroup) + final ReadExecutionController executionController) { assert indexHits.staticRow() == Rows.EMPTY_STATIC_ROW; @@ -143,10 +143,10 @@ public class CompositesSearcher extends CassandraIndexSearcher @SuppressWarnings("resource") // We close right away if empty, and if it's assign to next it will be called either // by the next caller of next, or through closing this iterator is this come before. UnfilteredRowIterator dataIter = filterStaleEntries(dataCmd.queryMemtableAndDisk(index.baseCfs, - orderGroup.baseReadOpOrderGroup()), + executionController.baseReadOpOrderGroup()), indexKey.getKey(), entries, - orderGroup.writeOpOrderGroup(), + executionController.writeOpOrderGroup(), command.nowInSec()); diff --git a/src/java/org/apache/cassandra/index/internal/keys/KeysSearcher.java b/src/java/org/apache/cassandra/index/internal/keys/KeysSearcher.java index b60d2d98fa..0ad08918a0 100644 --- a/src/java/org/apache/cassandra/index/internal/keys/KeysSearcher.java +++ b/src/java/org/apache/cassandra/index/internal/keys/KeysSearcher.java @@ -47,7 +47,7 @@ public class KeysSearcher extends CassandraIndexSearcher protected UnfilteredPartitionIterator queryDataFromIndex(final DecoratedKey indexKey, final RowIterator indexHits, final ReadCommand command, - final ReadOrderGroup orderGroup) + final ReadExecutionController executionController) { assert indexHits.staticRow() == Rows.EMPTY_STATIC_ROW; @@ -100,10 +100,10 @@ public class KeysSearcher extends CassandraIndexSearcher // Otherwise, we close right away if empty, and if it's assigned to next it will be called either // by the next caller of next, or through closing this iterator is this come before. UnfilteredRowIterator dataIter = filterIfStale(dataCmd.queryMemtableAndDisk(index.baseCfs, - orderGroup.baseReadOpOrderGroup()), + executionController.baseReadOpOrderGroup()), hit, indexKey.getKey(), - orderGroup.writeOpOrderGroup(), + executionController.writeOpOrderGroup(), isForThrift(), command.nowInSec()); diff --git a/src/java/org/apache/cassandra/net/IAsyncCallback.java b/src/java/org/apache/cassandra/net/IAsyncCallback.java index a29260c253..d159e0c41b 100644 --- a/src/java/org/apache/cassandra/net/IAsyncCallback.java +++ b/src/java/org/apache/cassandra/net/IAsyncCallback.java @@ -31,7 +31,7 @@ import org.apache.cassandra.gms.FailureDetector; */ public interface IAsyncCallback { - public static Predicate isAlive = new Predicate() + Predicate isAlive = new Predicate() { public boolean apply(InetAddress endpoint) { @@ -42,7 +42,7 @@ public interface IAsyncCallback /** * @param msg response received. */ - public void response(MessageIn msg); + void response(MessageIn msg); /** * @return true if this callback is on the read path and its latency should be diff --git a/src/java/org/apache/cassandra/net/IAsyncCallbackWithFailure.java b/src/java/org/apache/cassandra/net/IAsyncCallbackWithFailure.java index 744bb621b4..546a416262 100644 --- a/src/java/org/apache/cassandra/net/IAsyncCallbackWithFailure.java +++ b/src/java/org/apache/cassandra/net/IAsyncCallbackWithFailure.java @@ -25,5 +25,5 @@ public interface IAsyncCallbackWithFailure extends IAsyncCallback /** * Called when there is an exception on the remote node or timeout happens */ - public void onFailure(InetAddress from); + void onFailure(InetAddress from); } diff --git a/src/java/org/apache/cassandra/net/IVerbHandler.java b/src/java/org/apache/cassandra/net/IVerbHandler.java index 574f30f8fb..b9f1a54f1b 100644 --- a/src/java/org/apache/cassandra/net/IVerbHandler.java +++ b/src/java/org/apache/cassandra/net/IVerbHandler.java @@ -19,6 +19,8 @@ package org.apache.cassandra.net; import java.io.IOException; +import org.apache.cassandra.db.ReadCommand; + /** * IVerbHandler provides the method that all verb handlers need to implement. * The concrete implementation of this interface would provide the functionality @@ -36,5 +38,5 @@ public interface IVerbHandler * @param message - incoming message that needs handling. * @param id */ - public void doVerb(MessageIn message, int id) throws IOException; + void doVerb(MessageIn message, int id) throws IOException; } diff --git a/src/java/org/apache/cassandra/net/IncomingTcpConnection.java b/src/java/org/apache/cassandra/net/IncomingTcpConnection.java index 7054bcc9a4..a5e86fad03 100644 --- a/src/java/org/apache/cassandra/net/IncomingTcpConnection.java +++ b/src/java/org/apache/cassandra/net/IncomingTcpConnection.java @@ -184,18 +184,7 @@ public class IncomingTcpConnection extends Thread implements Closeable else id = input.readInt(); - long timestamp = System.currentTimeMillis(); - boolean isCrossNodeTimestamp = false; - // make sure to readInt, even if cross_node_to is not enabled - int partial = input.readInt(); - if (DatabaseDescriptor.hasCrossNodeTimeout()) - { - long crossNodeTimestamp = (timestamp & 0xFFFFFFFF00000000L) | (((partial & 0xFFFFFFFFL) << 2) >> 2); - isCrossNodeTimestamp = (timestamp != crossNodeTimestamp); - timestamp = crossNodeTimestamp; - } - - MessageIn message = MessageIn.read(input, version, id); + MessageIn message = MessageIn.read(input, version, id, MessageIn.readTimestamp(input)); if (message == null) { // callback expired; nothing to do @@ -203,7 +192,7 @@ public class IncomingTcpConnection extends Thread implements Closeable } if (version <= MessagingService.current_version) { - MessagingService.instance().receive(message, id, timestamp, isCrossNodeTimestamp); + MessagingService.instance().receive(message, id); } else { diff --git a/src/java/org/apache/cassandra/net/MessageDeliveryTask.java b/src/java/org/apache/cassandra/net/MessageDeliveryTask.java index a46366c444..46376d0531 100644 --- a/src/java/org/apache/cassandra/net/MessageDeliveryTask.java +++ b/src/java/org/apache/cassandra/net/MessageDeliveryTask.java @@ -32,25 +32,21 @@ public class MessageDeliveryTask implements Runnable private final MessageIn message; private final int id; - private final long constructionTime; - private final boolean isCrossNodeTimestamp; - public MessageDeliveryTask(MessageIn message, int id, long timestamp, boolean isCrossNodeTimestamp) + public MessageDeliveryTask(MessageIn message, int id) { assert message != null; this.message = message; this.id = id; - this.constructionTime = timestamp; - this.isCrossNodeTimestamp = isCrossNodeTimestamp; } public void run() { MessagingService.Verb verb = message.verb; if (MessagingService.DROPPABLE_VERBS.contains(verb) - && System.currentTimeMillis() > constructionTime + message.getTimeout()) + && System.currentTimeMillis() > message.constructionTime.timestamp + message.getTimeout()) { - MessagingService.instance().incrementDroppedMessages(verb, isCrossNodeTimestamp); + MessagingService.instance().incrementDroppedMessages(message); return; } @@ -82,7 +78,7 @@ public class MessageDeliveryTask implements Runnable } if (GOSSIP_VERBS.contains(message.verb)) - Gossiper.instance.setLastProcessedMessageAt(constructionTime); + Gossiper.instance.setLastProcessedMessageAt(message.constructionTime.timestamp); } private void handleFailure(Throwable t) diff --git a/src/java/org/apache/cassandra/net/MessageIn.java b/src/java/org/apache/cassandra/net/MessageIn.java index 64b8e81cab..c6e4d89787 100644 --- a/src/java/org/apache/cassandra/net/MessageIn.java +++ b/src/java/org/apache/cassandra/net/MessageIn.java @@ -24,39 +24,54 @@ import java.util.Map; import com.google.common.collect.ImmutableMap; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.monitoring.ConstructionTime; +import org.apache.cassandra.db.monitoring.MonitorableImpl; import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.FileUtils; public class MessageIn { - private static final Logger logger = LoggerFactory.getLogger(MessageIn.class); - public final InetAddress from; public final T payload; public final Map parameters; public final MessagingService.Verb verb; public final int version; + public final ConstructionTime constructionTime; - private MessageIn(InetAddress from, T payload, Map parameters, MessagingService.Verb verb, int version) + private MessageIn(InetAddress from, + T payload, + Map parameters, + MessagingService.Verb verb, + int version, + ConstructionTime constructionTime) { this.from = from; this.payload = payload; this.parameters = parameters; this.verb = verb; this.version = version; + this.constructionTime = constructionTime; } - public static MessageIn create(InetAddress from, T payload, Map parameters, MessagingService.Verb verb, int version) + public static MessageIn create(InetAddress from, + T payload, + Map parameters, + MessagingService.Verb verb, + int version, + ConstructionTime constructionTime) { - return new MessageIn(from, payload, parameters, verb, version); + return new MessageIn<>(from, payload, parameters, verb, version, constructionTime); } public static MessageIn read(DataInputPlus in, int version, int id) throws IOException + { + return read(in, version, id, new ConstructionTime()); + } + + public static MessageIn read(DataInputPlus in, int version, int id, ConstructionTime constructionTime) throws IOException { InetAddress from = CompactEndpointSerializationHelper.deserialize(in); @@ -94,9 +109,31 @@ public class MessageIn serializer = (IVersionedSerializer) callback.serializer; } if (payloadSize == 0 || serializer == null) - return create(from, null, parameters, verb, version); + return create(from, null, parameters, verb, version, constructionTime); + T2 payload = serializer.deserialize(in, version); - return MessageIn.create(from, payload, parameters, verb, version); + return MessageIn.create(from, payload, parameters, verb, version, constructionTime); + } + + public static ConstructionTime createTimestamp() + { + return new ConstructionTime(); + } + + public static ConstructionTime readTimestamp(DataInputPlus input) throws IOException + { + // make sure to readInt, even if cross_node_to is not enabled + int partial = input.readInt(); + if(DatabaseDescriptor.hasCrossNodeTimeout()) + { + long timestamp = System.currentTimeMillis(); + long crossNodeTimestamp = (timestamp & 0xFFFFFFFF00000000L) | (((partial & 0xFFFFFFFFL) << 2) >> 2); + return new ConstructionTime(crossNodeTimestamp, timestamp != crossNodeTimestamp); + } + else + { + return new ConstructionTime(); + } } public Stage getMessageType() diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index 810d08648d..1ef35f4236 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -789,7 +789,7 @@ public final class MessagingService implements MessagingServiceMBean } } - public void receive(MessageIn message, int id, long timestamp, boolean isCrossNodeTimestamp) + public void receive(MessageIn message, int id) { TraceState state = Tracing.instance.initializeFromMessage(message); if (state != null) @@ -800,7 +800,7 @@ public final class MessagingService implements MessagingServiceMBean if (!ms.allowIncomingMessage(message, id)) return; - Runnable runnable = new MessageDeliveryTask(message, id, timestamp, isCrossNodeTimestamp); + Runnable runnable = new MessageDeliveryTask(message, id); TracingAwareExecutorService stage = StageManager.getStage(message.getMessageType()); assert stage != null : "No stage for message type " + message.verb; @@ -937,6 +937,11 @@ public final class MessagingService implements MessagingServiceMBean incrementDroppedMessages(verb, false); } + public void incrementDroppedMessages(MessageIn message) + { + incrementDroppedMessages(message.verb, message.constructionTime.isCrossNode); + } + public void incrementDroppedMessages(Verb verb, boolean isCrossNodeTimeout) { assert DROPPABLE_VERBS.contains(verb) : "Verb " + verb + " should not legally be dropped"; diff --git a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java index f0bdd142a5..8377ead629 100644 --- a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java +++ b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java @@ -270,7 +270,8 @@ public final class SchemaKeyspace public static List readSchemaFromSystemTables() { ReadCommand cmd = getReadCommandForTableSchema(KEYSPACES); - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); PartitionIterator schema = cmd.executeInternal(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); + PartitionIterator schema = cmd.executeInternal(executionController)) { List keyspaces = new ArrayList<>(); @@ -326,8 +327,8 @@ public final class SchemaKeyspace for (String table : ALL) { ReadCommand cmd = getReadCommandForTableSchema(table); - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); - PartitionIterator schema = cmd.executeInternal(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); + PartitionIterator schema = cmd.executeInternal(executionController)) { while (schema.hasNext()) { @@ -374,7 +375,8 @@ public final class SchemaKeyspace private static void convertSchemaToMutations(Map mutationMap, String schemaTableName) { ReadCommand cmd = getReadCommandForTableSchema(schemaTableName); - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iter = cmd.executeLocally(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); + UnfilteredPartitionIterator iter = cmd.executeLocally(executionController)) { while (iter.hasNext()) { diff --git a/src/java/org/apache/cassandra/serializers/ByteSerializer.java b/src/java/org/apache/cassandra/serializers/ByteSerializer.java index 8c736cb157..9d34fbc414 100644 --- a/src/java/org/apache/cassandra/serializers/ByteSerializer.java +++ b/src/java/org/apache/cassandra/serializers/ByteSerializer.java @@ -28,7 +28,7 @@ public class ByteSerializer implements TypeSerializer public Byte deserialize(ByteBuffer bytes) { - return bytes.remaining() == 0 ? null : bytes.get(bytes.position()); + return bytes == null || bytes.remaining() == 0 ? null : bytes.get(bytes.position()); } public ByteBuffer serialize(Byte value) diff --git a/src/java/org/apache/cassandra/service/ReadCallback.java b/src/java/org/apache/cassandra/service/ReadCallback.java index 8747004d02..47eacdf5a4 100644 --- a/src/java/org/apache/cassandra/service/ReadCallback.java +++ b/src/java/org/apache/cassandra/service/ReadCallback.java @@ -192,7 +192,8 @@ public class ReadCallback implements IAsyncCallbackWithFailure result, Collections.emptyMap(), MessagingService.Verb.INTERNAL_RESPONSE, - MessagingService.current_version); + MessagingService.current_version, + MessageIn.createTimestamp()); response(message); } diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index d1142fc83e..86ecb10fa8 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -47,6 +47,7 @@ import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.*; import org.apache.cassandra.db.filter.DataLimits; import org.apache.cassandra.db.filter.TombstoneOverwhelmingException; +import org.apache.cassandra.db.monitoring.ConstructionTime; import org.apache.cassandra.db.partitions.*; import org.apache.cassandra.db.rows.RowIterator; import org.apache.cassandra.db.view.ViewUtils; @@ -1678,10 +1679,25 @@ public class StorageProxy implements StorageProxyMBean { try { - try (ReadOrderGroup orderGroup = command.startOrderGroup(); UnfilteredPartitionIterator iterator = command.executeLocally(orderGroup)) + command.setMonitoringTime(new ConstructionTime(constructionTime), timeout); + + ReadResponse response; + try (ReadExecutionController executionController = command.executionController(); + UnfilteredPartitionIterator iterator = command.executeLocally(executionController)) { - handler.response(command.createResponse(iterator, command.columnFilter())); + response = command.createResponse(iterator, command.columnFilter()); } + + if (command.complete()) + { + handler.response(response); + } + else + { + MessagingService.instance().incrementDroppedMessages(verb); + handler.onFailure(FBUtilities.getBroadcastAddress()); + } + MessagingService.instance().addLatency(FBUtilities.getBroadcastAddress(), TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)); } catch (Throwable t) @@ -2314,18 +2330,20 @@ public class StorageProxy implements StorageProxyMBean */ private static abstract class DroppableRunnable implements Runnable { - private final long constructionTime = System.nanoTime(); - private final MessagingService.Verb verb; + final long constructionTime; + final MessagingService.Verb verb; + final long timeout; public DroppableRunnable(MessagingService.Verb verb) { + this.constructionTime = System.currentTimeMillis(); this.verb = verb; + this.timeout = DatabaseDescriptor.getTimeout(verb); } public final void run() { - - if (TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - constructionTime) > DatabaseDescriptor.getTimeout(verb)) + if (System.currentTimeMillis() > constructionTime + timeout) { MessagingService.instance().incrementDroppedMessages(verb); return; diff --git a/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java b/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java index b92d1e19f8..c6e447a700 100644 --- a/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java +++ b/src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java @@ -54,9 +54,9 @@ abstract class AbstractQueryPager implements QueryPager this.remainingInPartition = limits.perPartitionCount(); } - public ReadOrderGroup startOrderGroup() + public ReadExecutionController executionController() { - return command.startOrderGroup(); + return command.executionController(); } public PartitionIterator fetchPage(int pageSize, ConsistencyLevel consistency, ClientState clientState) throws RequestValidationException, RequestExecutionException @@ -68,13 +68,13 @@ abstract class AbstractQueryPager implements QueryPager return new PagerIterator(nextPageReadCommand(pageSize).execute(consistency, clientState), limits.forPaging(pageSize), command.nowInSec()); } - public PartitionIterator fetchPageInternal(int pageSize, ReadOrderGroup orderGroup) throws RequestValidationException, RequestExecutionException + public PartitionIterator fetchPageInternal(int pageSize, ReadExecutionController executionController) throws RequestValidationException, RequestExecutionException { if (isExhausted()) return PartitionIterators.EMPTY; pageSize = Math.min(pageSize, remaining); - return new PagerIterator(nextPageReadCommand(pageSize).executeInternal(orderGroup), limits.forPaging(pageSize), command.nowInSec()); + return new PagerIterator(nextPageReadCommand(pageSize).executeInternal(executionController), limits.forPaging(pageSize), command.nowInSec()); } private class PagerIterator extends CountingPartitionIterator diff --git a/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java b/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java index e826be6ff1..09eabb894f 100644 --- a/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java +++ b/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java @@ -108,14 +108,14 @@ public class MultiPartitionPager implements QueryPager return true; } - public ReadOrderGroup startOrderGroup() + public ReadExecutionController executionController() { // Note that for all pagers, the only difference is the partition key to which it applies, so in practice we // can use any of the sub-pager ReadOrderGroup group to protect the whole pager for (int i = current; i < pagers.length; i++) { if (pagers[i] != null) - return pagers[i].startOrderGroup(); + return pagers[i].executionController(); } throw new AssertionError("Shouldn't be called on an exhausted pager"); } @@ -130,10 +130,10 @@ public class MultiPartitionPager implements QueryPager return countingIter; } - public PartitionIterator fetchPageInternal(int pageSize, ReadOrderGroup orderGroup) throws RequestValidationException, RequestExecutionException + public PartitionIterator fetchPageInternal(int pageSize, ReadExecutionController executionController) throws RequestValidationException, RequestExecutionException { int toQuery = Math.min(remaining, pageSize); - PagersIterator iter = new PagersIterator(toQuery, null, null, orderGroup); + PagersIterator iter = new PagersIterator(toQuery, null, null, executionController); CountingPartitionIterator countingIter = new CountingPartitionIterator(iter, limit.forPaging(toQuery), nowInSec); iter.setCounter(countingIter.counter()); return countingIter; @@ -150,14 +150,14 @@ public class MultiPartitionPager implements QueryPager private final ClientState clientState; // For internal queries - private final ReadOrderGroup orderGroup; + private final ReadExecutionController executionController; - public PagersIterator(int pageSize, ConsistencyLevel consistency, ClientState clientState, ReadOrderGroup orderGroup) + public PagersIterator(int pageSize, ConsistencyLevel consistency, ClientState clientState, ReadExecutionController executionController) { this.pageSize = pageSize; this.consistency = consistency; this.clientState = clientState; - this.orderGroup = orderGroup; + this.executionController = executionController; } public void setCounter(DataLimits.Counter counter) @@ -178,7 +178,7 @@ public class MultiPartitionPager implements QueryPager int toQuery = pageSize - counter.counted(); result = consistency == null - ? pagers[current].fetchPageInternal(toQuery, orderGroup) + ? pagers[current].fetchPageInternal(toQuery, executionController) : pagers[current].fetchPage(toQuery, consistency, clientState); } return result.next(); diff --git a/src/java/org/apache/cassandra/service/pager/QueryPager.java b/src/java/org/apache/cassandra/service/pager/QueryPager.java index a69335df16..1d5a739e07 100644 --- a/src/java/org/apache/cassandra/service/pager/QueryPager.java +++ b/src/java/org/apache/cassandra/service/pager/QueryPager.java @@ -18,7 +18,7 @@ package org.apache.cassandra.service.pager; import org.apache.cassandra.db.ConsistencyLevel; -import org.apache.cassandra.db.ReadOrderGroup; +import org.apache.cassandra.db.ReadExecutionController; import org.apache.cassandra.db.partitions.PartitionIterator; import org.apache.cassandra.db.partitions.PartitionIterators; import org.apache.cassandra.exceptions.RequestExecutionException; @@ -46,11 +46,11 @@ import org.apache.cassandra.service.ClientState; */ public interface QueryPager { - public static final QueryPager EMPTY = new QueryPager() + QueryPager EMPTY = new QueryPager() { - public ReadOrderGroup startOrderGroup() + public ReadExecutionController executionController() { - return ReadOrderGroup.emptyGroup(); + return ReadExecutionController.empty(); } public PartitionIterator fetchPage(int pageSize, ConsistencyLevel consistency, ClientState clientState) throws RequestValidationException, RequestExecutionException @@ -58,7 +58,7 @@ public interface QueryPager return PartitionIterators.EMPTY; } - public PartitionIterator fetchPageInternal(int pageSize, ReadOrderGroup orderGroup) throws RequestValidationException, RequestExecutionException + public PartitionIterator fetchPageInternal(int pageSize, ReadExecutionController executionController) throws RequestValidationException, RequestExecutionException { return PartitionIterators.EMPTY; } @@ -99,16 +99,16 @@ public interface QueryPager * * @return a newly started order group for this {@code QueryPager}. */ - public ReadOrderGroup startOrderGroup(); + public ReadExecutionController executionController(); /** * Fetches the next page internally (in other, this does a local query). * * @param pageSize the maximum number of elements to return in the next page. - * @param orderGroup the {@code ReadOrderGroup} protecting the read. + * @param executionController the {@code ReadExecutionController} protecting the read. * @return the page of result. */ - public PartitionIterator fetchPageInternal(int pageSize, ReadOrderGroup orderGroup) throws RequestValidationException, RequestExecutionException; + public PartitionIterator fetchPageInternal(int pageSize, ReadExecutionController executionController) throws RequestValidationException, RequestExecutionException; /** * Whether or not this pager is exhausted, i.e. whether or not a call to diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java index 83b07e06a8..933ea512d4 100644 --- a/src/java/org/apache/cassandra/utils/FBUtilities.java +++ b/src/java/org/apache/cassandra/utils/FBUtilities.java @@ -816,4 +816,16 @@ public class FBUtilities throw new RuntimeException(e); } } + + public static void sleepQuietly(long millis) + { + try + { + Thread.sleep(millis); + } + catch (InterruptedException e) + { + throw new RuntimeException(e); + } + } } diff --git a/src/java/org/apache/cassandra/utils/NanoTimeToCurrentTimeMillis.java b/src/java/org/apache/cassandra/utils/NanoTimeToCurrentTimeMillis.java index f124383cdd..69b7a47b43 100644 --- a/src/java/org/apache/cassandra/utils/NanoTimeToCurrentTimeMillis.java +++ b/src/java/org/apache/cassandra/utils/NanoTimeToCurrentTimeMillis.java @@ -19,6 +19,7 @@ package org.apache.cassandra.utils; import java.util.concurrent.TimeUnit; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.Config; import com.google.common.annotations.VisibleForTesting; @@ -36,9 +37,6 @@ public class NanoTimeToCurrentTimeMillis private static volatile long TIMESTAMP_BASE[] = new long[] { System.currentTimeMillis(), System.nanoTime() }; - @VisibleForTesting - public static final Object TIMESTAMP_UPDATE = new Object(); - /* * System.currentTimeMillis() is 25 nanoseconds. This is 2 nanoseconds (maybe) according to JMH. * Faster than calling both currentTimeMillis() and nanoTime(). @@ -48,41 +46,29 @@ public class NanoTimeToCurrentTimeMillis * These timestamps don't order with System.currentTimeMillis() because currentTimeMillis() can tick over * before this one does. I have seen it behind by as much as 2ms on Linux and 25ms on Windows. */ - public static final long convert(long nanoTime) + public static long convert(long nanoTime) { final long timestampBase[] = TIMESTAMP_BASE; return timestampBase[0] + TimeUnit.NANOSECONDS.toMillis(nanoTime - timestampBase[1]); } + public static void updateNow() + { + ScheduledExecutors.scheduledFastTasks.submit(NanoTimeToCurrentTimeMillis::updateTimestampBase); + } + static { - //Pick up updates from NTP periodically - Thread t = new Thread("NanoTimeToCurrentTimeMillis updater") - { - @Override - public void run() - { - while (true) - { - try - { - synchronized (TIMESTAMP_UPDATE) - { - TIMESTAMP_UPDATE.wait(TIMESTAMP_UPDATE_INTERVAL); - } - } - catch (InterruptedException e) - { - return; - } + ScheduledExecutors.scheduledFastTasks.scheduleWithFixedDelay(NanoTimeToCurrentTimeMillis::updateTimestampBase, + TIMESTAMP_UPDATE_INTERVAL, + TIMESTAMP_UPDATE_INTERVAL, + TimeUnit.MILLISECONDS); + } - TIMESTAMP_BASE = new long[] { - Math.max(TIMESTAMP_BASE[0], System.currentTimeMillis()), - Math.max(TIMESTAMP_BASE[1], System.nanoTime()) }; - } - } - }; - t.setDaemon(true); - t.start(); + private static void updateTimestampBase() + { + TIMESTAMP_BASE = new long[] { + Math.max(TIMESTAMP_BASE[0], System.currentTimeMillis()), + Math.max(TIMESTAMP_BASE[1], System.nanoTime()) }; } } diff --git a/test/conf/logback-test.xml b/test/conf/logback-test.xml index 0e1bb76230..2591890131 100644 --- a/test/conf/logback-test.xml +++ b/test/conf/logback-test.xml @@ -42,13 +42,13 @@ false - + %-5level %date{HH:mm:ss,SSS} %msg%n - INFO + DEBUG @@ -59,6 +59,8 @@ + + diff --git a/test/unit/org/apache/cassandra/Util.java b/test/unit/org/apache/cassandra/Util.java index 5e19d5e2b0..4e122ed36e 100644 --- a/test/unit/org/apache/cassandra/Util.java +++ b/test/unit/org/apache/cassandra/Util.java @@ -272,7 +272,8 @@ public class Util public static void assertEmptyUnfiltered(ReadCommand command) { - try (ReadOrderGroup orderGroup = command.startOrderGroup(); UnfilteredPartitionIterator iterator = command.executeLocally(orderGroup)) + try (ReadExecutionController executionController = command.executionController(); + UnfilteredPartitionIterator iterator = command.executeLocally(executionController)) { if (iterator.hasNext()) { @@ -286,7 +287,8 @@ public class Util public static void assertEmpty(ReadCommand command) { - try (ReadOrderGroup orderGroup = command.startOrderGroup(); PartitionIterator iterator = command.executeInternal(orderGroup)) + try (ReadExecutionController executionController = command.executionController(); + PartitionIterator iterator = command.executeInternal(executionController)) { if (iterator.hasNext()) { @@ -301,7 +303,8 @@ public class Util public static List getAllUnfiltered(ReadCommand command) { List results = new ArrayList<>(); - try (ReadOrderGroup orderGroup = command.startOrderGroup(); UnfilteredPartitionIterator iterator = command.executeLocally(orderGroup)) + try (ReadExecutionController executionController = command.executionController(); + UnfilteredPartitionIterator iterator = command.executeLocally(executionController)) { while (iterator.hasNext()) { @@ -317,7 +320,8 @@ public class Util public static List getAll(ReadCommand command) { List results = new ArrayList<>(); - try (ReadOrderGroup orderGroup = command.startOrderGroup(); PartitionIterator iterator = command.executeInternal(orderGroup)) + try (ReadExecutionController executionController = command.executionController(); + PartitionIterator iterator = command.executeInternal(executionController)) { while (iterator.hasNext()) { @@ -332,7 +336,8 @@ public class Util public static Row getOnlyRowUnfiltered(ReadCommand cmd) { - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); + UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController)) { assert iterator.hasNext() : "Expecting one row in one partition but got nothing"; try (UnfilteredRowIterator partition = iterator.next()) @@ -349,7 +354,8 @@ public class Util public static Row getOnlyRow(ReadCommand cmd) { - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); PartitionIterator iterator = cmd.executeInternal(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); + PartitionIterator iterator = cmd.executeInternal(executionController)) { assert iterator.hasNext() : "Expecting one row in one partition but got nothing"; try (RowIterator partition = iterator.next()) @@ -365,7 +371,8 @@ public class Util public static ImmutableBTreePartition getOnlyPartitionUnfiltered(ReadCommand cmd) { - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); + UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController)) { assert iterator.hasNext() : "Expecting a single partition but got nothing"; try (UnfilteredRowIterator partition = iterator.next()) @@ -378,7 +385,8 @@ public class Util public static FilteredPartition getOnlyPartition(ReadCommand cmd) { - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); PartitionIterator iterator = cmd.executeInternal(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); + PartitionIterator iterator = cmd.executeInternal(executionController)) { assert iterator.hasNext() : "Expecting a single partition but got nothing"; try (RowIterator partition = iterator.next()) diff --git a/test/unit/org/apache/cassandra/db/KeyspaceTest.java b/test/unit/org/apache/cassandra/db/KeyspaceTest.java index 2e79dfea4e..e3243c74f4 100644 --- a/test/unit/org/apache/cassandra/db/KeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/KeyspaceTest.java @@ -135,7 +135,8 @@ public class KeyspaceTest extends CQLTester ClusteringIndexSliceFilter filter = new ClusteringIndexSliceFilter(slices, reversed); SinglePartitionSliceCommand command = singlePartitionSlice(cfs, key, filter, limit); - try (ReadOrderGroup orderGroup = command.startOrderGroup(); PartitionIterator iterator = command.executeInternal(orderGroup)) + try (ReadExecutionController executionController = command.executionController(); + PartitionIterator iterator = command.executeInternal(executionController)) { try (RowIterator rowIterator = iterator.next()) { @@ -208,7 +209,8 @@ public class KeyspaceTest extends CQLTester PartitionColumns columns = PartitionColumns.of(cfs.metadata.getColumnDefinition(new ColumnIdentifier("c", false))); ClusteringIndexSliceFilter filter = new ClusteringIndexSliceFilter(Slices.ALL, false); SinglePartitionSliceCommand command = singlePartitionSlice(cfs, "0", filter, null); - try (ReadOrderGroup orderGroup = command.startOrderGroup(); PartitionIterator iterator = command.executeInternal(orderGroup)) + try (ReadExecutionController executionController = command.executionController(); + PartitionIterator iterator = command.executeInternal(executionController)) { try (RowIterator rowIterator = iterator.next()) { @@ -222,7 +224,8 @@ public class KeyspaceTest extends CQLTester private static void assertRowsInResult(ColumnFamilyStore cfs, SinglePartitionReadCommand command, int ... columnValues) { - try (ReadOrderGroup orderGroup = command.startOrderGroup(); PartitionIterator iterator = command.executeInternal(orderGroup)) + try (ReadExecutionController executionController = command.executionController(); + PartitionIterator iterator = command.executeInternal(executionController)) { if (columnValues.length == 0) { diff --git a/test/unit/org/apache/cassandra/db/ReadCommandTest.java b/test/unit/org/apache/cassandra/db/ReadCommandTest.java new file mode 100644 index 0000000000..fcaa92f910 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/ReadCommandTest.java @@ -0,0 +1,145 @@ +/* + * 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.db; + +import java.util.List; + +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.Util; +import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.db.marshal.AsciiType; +import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.db.partitions.FilteredPartition; +import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.schema.KeyspaceParams; +import org.apache.cassandra.utils.ByteBufferUtil; + +import static org.junit.Assert.assertEquals; + +public class ReadCommandTest +{ + private static final String KEYSPACE = "ReadCommandTest"; + private static final String CF1 = "Standard1"; + private static final String CF2 = "Standard2"; + + @BeforeClass + public static void defineSchema() throws ConfigurationException + { + CFMetaData metadata1 = SchemaLoader.standardCFMD(KEYSPACE, CF1); + + CFMetaData metadata2 = CFMetaData.Builder.create(KEYSPACE, CF2) + .addPartitionKey("key", BytesType.instance) + .addClusteringColumn("col", AsciiType.instance) + .addRegularColumn("a", AsciiType.instance) + .addRegularColumn("b", AsciiType.instance).build(); + + SchemaLoader.prepareServer(); + SchemaLoader.createKeyspace(KEYSPACE, + KeyspaceParams.simple(1), + metadata1, + metadata2); + } + + @Test + public void testPartitionRangeAbort() throws Exception + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF1); + + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key1")) + .clustering("Column1") + .add("val", ByteBufferUtil.bytes("abcd")) + .build() + .apply(); + + cfs.forceBlockingFlush(); + + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key2")) + .clustering("Column1") + .add("val", ByteBufferUtil.bytes("abcd")) + .build() + .apply(); + + ReadCommand readCommand = Util.cmd(cfs).build(); + assertEquals(2, Util.getAll(readCommand).size()); + + readCommand.abort(); + assertEquals(0, Util.getAll(readCommand).size()); + } + + @Test + public void testSinglePartitionSliceAbort() throws Exception + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2); + + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key")) + .clustering("cc") + .add("a", ByteBufferUtil.bytes("abcd")) + .build() + .apply(); + + cfs.forceBlockingFlush(); + + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key")) + .clustering("dd") + .add("a", ByteBufferUtil.bytes("abcd")) + .build() + .apply(); + + ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).build(); + + List partitions = Util.getAll(readCommand); + assertEquals(1, partitions.size()); + assertEquals(2, partitions.get(0).rowCount()); + + readCommand.abort(); + assertEquals(0, Util.getAll(readCommand).size()); + } + + @Test + public void testSinglePartitionNamesAbort() throws Exception + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2); + + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key")) + .clustering("cc") + .add("a", ByteBufferUtil.bytes("abcd")) + .build() + .apply(); + + cfs.forceBlockingFlush(); + + new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key")) + .clustering("cdd") + .add("a", ByteBufferUtil.bytes("abcd")) + .build() + .apply(); + + ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).includeRow("cc").includeRow("dd").build(); + + List partitions = Util.getAll(readCommand); + assertEquals(1, partitions.size()); + assertEquals(2, partitions.get(0).rowCount()); + + readCommand.abort(); + assertEquals(0, Util.getAll(readCommand).size()); + } +} diff --git a/test/unit/org/apache/cassandra/db/RepairedDataTombstonesTest.java b/test/unit/org/apache/cassandra/db/RepairedDataTombstonesTest.java index 3a74029a10..ad009a47e1 100644 --- a/test/unit/org/apache/cassandra/db/RepairedDataTombstonesTest.java +++ b/test/unit/org/apache/cassandra/db/RepairedDataTombstonesTest.java @@ -177,7 +177,8 @@ public class RepairedDataTombstonesTest extends CQLTester Thread.sleep(1000); ReadCommand cmd = Util.cmd(getCurrentColumnFamilyStore()).build(); int partitionsFound = 0; - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); + UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController)) { while (iterator.hasNext()) { @@ -232,7 +233,8 @@ public class RepairedDataTombstonesTest extends CQLTester { ReadCommand cmd = Util.cmd(getCurrentColumnFamilyStore()).build(); int foundRows = 0; - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); + UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController)) { while (iterator.hasNext()) { @@ -263,7 +265,8 @@ public class RepairedDataTombstonesTest extends CQLTester { ReadCommand cmd = Util.cmd(getCurrentColumnFamilyStore(), Util.dk(ByteBufferUtil.bytes(key))).build(); int foundRows = 0; - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); + UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController)) { while (iterator.hasNext()) { diff --git a/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java b/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java index 55ab5742a9..cfa0e9e94b 100644 --- a/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java +++ b/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java @@ -117,7 +117,8 @@ public class SecondaryIndexTest .build(); Index.Searcher searcher = cfs.indexManager.getBestIndexFor(rc).searcherFor(rc); - try (ReadOrderGroup orderGroup = rc.startOrderGroup(); UnfilteredPartitionIterator pi = searcher.search(orderGroup)) + try (ReadExecutionController executionController = rc.executionController(); + UnfilteredPartitionIterator pi = searcher.search(executionController)) { assertTrue(pi.hasNext()); pi.next().close(); @@ -501,8 +502,8 @@ public class SecondaryIndexTest if (count != 0) assertNotNull(searcher); - try (ReadOrderGroup orderGroup = rc.startOrderGroup(); - PartitionIterator iter = UnfilteredPartitionIterators.filter(searcher.search(orderGroup), + try (ReadExecutionController executionController = rc.executionController(); + PartitionIterator iter = UnfilteredPartitionIterators.filter(searcher.search(executionController), FBUtilities.nowInSeconds())) { assertEquals(count, Util.size(iter)); diff --git a/test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java b/test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java index 9f80023733..1a93fec655 100644 --- a/test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java +++ b/test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java @@ -103,7 +103,7 @@ public class SinglePartitionSliceCommandTest cmd = ReadCommand.legacyReadCommandSerializer.deserialize(in, MessagingService.VERSION_21); logger.debug("ReadCommand: {}", cmd); - UnfilteredPartitionIterator partitionIterator = cmd.executeLocally(ReadOrderGroup.emptyGroup()); + UnfilteredPartitionIterator partitionIterator = cmd.executeLocally(ReadExecutionController.empty()); ReadResponse response = ReadResponse.createDataResponse(partitionIterator, cmd.columnFilter()); logger.debug("creating response: {}", response); @@ -148,7 +148,7 @@ public class SinglePartitionSliceCommandTest sliceFilter); // check raw iterator for static cell - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator pi = cmd.executeLocally(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); UnfilteredPartitionIterator pi = cmd.executeLocally(executionController)) { checkForS(pi); } @@ -159,7 +159,7 @@ public class SinglePartitionSliceCommandTest ReadResponse dst; // check (de)serialized iterator for memtable static cell - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator pi = cmd.executeLocally(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); UnfilteredPartitionIterator pi = cmd.executeLocally(executionController)) { response = ReadResponse.createDataResponse(pi, cmd.columnFilter()); } @@ -175,7 +175,7 @@ public class SinglePartitionSliceCommandTest // check (de)serialized iterator for sstable static cell Schema.instance.getColumnFamilyStoreInstance(cfm.cfId).forceBlockingFlush(); - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator pi = cmd.executeLocally(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); UnfilteredPartitionIterator pi = cmd.executeLocally(executionController)) { response = ReadResponse.createDataResponse(pi, cmd.columnFilter()); } diff --git a/test/unit/org/apache/cassandra/db/monitoring/MonitoringTaskTest.java b/test/unit/org/apache/cassandra/db/monitoring/MonitoringTaskTest.java new file mode 100644 index 0000000000..4490519630 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/monitoring/MonitoringTaskTest.java @@ -0,0 +1,341 @@ +/* + * 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.db.monitoring; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.junit.BeforeClass; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +public class MonitoringTaskTest +{ + private static final long timeout = 100; + private static final long MAX_SPIN_TIME_NANOS = TimeUnit.SECONDS.toNanos(5); + + public static final int REPORT_INTERVAL_MS = 600000; // long enough so that it won't check unless told to do so + public static final int MAX_TIMEDOUT_OPERATIONS = -1; // unlimited + + @BeforeClass + public static void setup() + { + MonitoringTask.instance = MonitoringTask.make(REPORT_INTERVAL_MS, MAX_TIMEDOUT_OPERATIONS); + } + + private static final class TestMonitor extends MonitorableImpl + { + private final String name; + + TestMonitor(String name, ConstructionTime constructionTime, long timeout) + { + this.name = name; + setMonitoringTime(constructionTime, timeout); + } + + public String name() + { + return name; + } + + @Override + public String toString() + { + return name(); + } + } + + private static void waitForOperationsToComplete(Monitorable... operations) throws InterruptedException + { + waitForOperationsToComplete(Arrays.asList(operations)); + } + + private static void waitForOperationsToComplete(List operations) throws InterruptedException + { + long timeout = operations.stream().map(Monitorable::timeout).reduce(0L, Long::max); + Thread.sleep(timeout * 2 + ApproximateTime.precision()); + + long start = System.nanoTime(); + while(System.nanoTime() - start <= MAX_SPIN_TIME_NANOS) + { + long numInProgress = operations.stream().filter(Monitorable::isInProgress).count(); + if (numInProgress == 0) + return; + + Thread.yield(); + } + } + + @Test + public void testAbort() throws InterruptedException + { + Monitorable operation = new TestMonitor("Test abort", new ConstructionTime(System.currentTimeMillis()), timeout); + waitForOperationsToComplete(operation); + + assertTrue(operation.isAborted()); + assertFalse(operation.isCompleted()); + assertEquals(1, MonitoringTask.instance.getFailedOperations().size()); + } + + @Test + public void testAbortIdemPotent() throws InterruptedException + { + Monitorable operation = new TestMonitor("Test abort", new ConstructionTime(System.currentTimeMillis()), timeout); + waitForOperationsToComplete(operation); + + assertTrue(operation.abort()); + + assertTrue(operation.isAborted()); + assertFalse(operation.isCompleted()); + assertEquals(1, MonitoringTask.instance.getFailedOperations().size()); + } + + @Test + public void testAbortCrossNode() throws InterruptedException + { + Monitorable operation = new TestMonitor("Test for cross node", new ConstructionTime(System.currentTimeMillis(), true), timeout); + waitForOperationsToComplete(operation); + + assertTrue(operation.isAborted()); + assertFalse(operation.isCompleted()); + assertEquals(1, MonitoringTask.instance.getFailedOperations().size()); + } + + @Test + public void testComplete() throws InterruptedException + { + Monitorable operation = new TestMonitor("Test complete", new ConstructionTime(System.currentTimeMillis()), timeout); + operation.complete(); + waitForOperationsToComplete(operation); + + assertFalse(operation.isAborted()); + assertTrue(operation.isCompleted()); + assertEquals(0, MonitoringTask.instance.getFailedOperations().size()); + } + + @Test + public void testCompleteIdemPotent() throws InterruptedException + { + Monitorable operation = new TestMonitor("Test complete", new ConstructionTime(System.currentTimeMillis()), timeout); + operation.complete(); + waitForOperationsToComplete(operation); + + assertTrue(operation.complete()); + + assertFalse(operation.isAborted()); + assertTrue(operation.isCompleted()); + assertEquals(0, MonitoringTask.instance.getFailedOperations().size()); + } + + @Test + public void testReport() throws InterruptedException + { + Monitorable operation = new TestMonitor("Test report", new ConstructionTime(System.currentTimeMillis()), timeout); + waitForOperationsToComplete(operation); + + assertTrue(operation.isAborted()); + assertFalse(operation.isCompleted()); + MonitoringTask.instance.logFailedOperations(ApproximateTime.currentTimeMillis()); + assertEquals(0, MonitoringTask.instance.getFailedOperations().size()); + } + + @Test + public void testRealScheduling() throws InterruptedException + { + MonitoringTask.instance = MonitoringTask.make(10, -1); + try + { + Monitorable operation = new TestMonitor("Test report", new ConstructionTime(System.currentTimeMillis()), timeout); + waitForOperationsToComplete(operation); + + assertTrue(operation.isAborted()); + assertFalse(operation.isCompleted()); + + Thread.sleep(ApproximateTime.precision() + 500); + assertEquals(0, MonitoringTask.instance.getFailedOperations().size()); + } + finally + { + MonitoringTask.instance = MonitoringTask.make(REPORT_INTERVAL_MS, MAX_TIMEDOUT_OPERATIONS); + } + } + + @Test + public void testMultipleThreads() throws InterruptedException + { + final int opCount = 50; + final ExecutorService executorService = Executors.newFixedThreadPool(20); + final List operations = Collections.synchronizedList(new ArrayList<>(opCount)); + + for (int i = 0; i < opCount; i++) + { + executorService.submit(() -> + operations.add(new TestMonitor(UUID.randomUUID().toString(), new ConstructionTime(), timeout)) + ); + } + + executorService.shutdown(); + assertTrue(executorService.awaitTermination(30, TimeUnit.SECONDS)); + assertEquals(opCount, operations.size()); + + waitForOperationsToComplete(operations); + assertEquals(opCount, MonitoringTask.instance.getFailedOperations().size()); + } + + @Test + public void testZeroMaxTimedoutOperations() throws InterruptedException + { + doTestMaxTimedoutOperations(0, 1, 0); + } + + @Test + public void testMaxTimedoutOperationsExceeded() throws InterruptedException + { + doTestMaxTimedoutOperations(5, 10, 6); + } + + private static void doTestMaxTimedoutOperations(int maxTimedoutOperations, + int numThreads, + int numExpectedOperations) throws InterruptedException + { + MonitoringTask.instance = MonitoringTask.make(REPORT_INTERVAL_MS, maxTimedoutOperations); + try + { + final int threadCount = numThreads; + ExecutorService executorService = Executors.newFixedThreadPool(threadCount); + final CountDownLatch finished = new CountDownLatch(threadCount); + + for (int i = 0; i < threadCount; i++) + { + final String operationName = "Operation " + Integer.toString(i+1); + final int numTimes = i + 1; + executorService.submit(() -> { + try + { + for (int j = 0; j < numTimes; j++) + { + Monitorable operation = new TestMonitor(operationName, + new ConstructionTime(System.currentTimeMillis()), + timeout); + waitForOperationsToComplete(operation); + } + } + catch (InterruptedException e) + { + e.printStackTrace(); + fail("Unexpected exception"); + } + finally + { + finished.countDown(); + } + }); + } + + finished.await(); + assertEquals(0, executorService.shutdownNow().size()); + + List failedOperations = MonitoringTask.instance.getFailedOperations(); + assertEquals(numExpectedOperations, failedOperations.size()); + if (numExpectedOperations > 0) + assertTrue(failedOperations.get(numExpectedOperations - 1).startsWith("...")); + } + finally + { + MonitoringTask.instance = MonitoringTask.make(REPORT_INTERVAL_MS, MAX_TIMEDOUT_OPERATIONS); + } + } + + @Test + public void testMultipleThreadsSameName() throws InterruptedException + { + final int threadCount = 50; + final List operations = new ArrayList<>(threadCount); + ExecutorService executorService = Executors.newFixedThreadPool(threadCount); + final CountDownLatch finished = new CountDownLatch(threadCount); + + for (int i = 0; i < threadCount; i++) + { + executorService.submit(() -> { + try + { + Monitorable operation = new TestMonitor("Test testMultipleThreadsSameName", + new ConstructionTime(System.currentTimeMillis()), + timeout); + operations.add(operation); + } + finally + { + finished.countDown(); + } + }); + } + + finished.await(); + assertEquals(0, executorService.shutdownNow().size()); + + waitForOperationsToComplete(operations); + //MonitoringTask.instance.checkFailedOperations(ApproximateTime.currentTimeMillis()); + assertEquals(1, MonitoringTask.instance.getFailedOperations().size()); + } + + @Test + public void testMultipleThreadsNoFailedOps() throws InterruptedException + { + final int threadCount = 50; + final List operations = new ArrayList<>(threadCount); + ExecutorService executorService = Executors.newFixedThreadPool(threadCount); + final CountDownLatch finished = new CountDownLatch(threadCount); + + for (int i = 0; i < threadCount; i++) + { + executorService.submit(() -> { + try + { + Monitorable operation = new TestMonitor("Test thread " + Thread.currentThread().getName(), + new ConstructionTime(System.currentTimeMillis()), + timeout); + operations.add(operation); + operation.complete(); + } + finally + { + finished.countDown(); + } + }); + } + + finished.await(); + assertEquals(0, executorService.shutdownNow().size()); + + waitForOperationsToComplete(operations); + assertEquals(0, MonitoringTask.instance.getFailedOperations().size()); + } +} diff --git a/test/unit/org/apache/cassandra/hints/HintTest.java b/test/unit/org/apache/cassandra/hints/HintTest.java index 4c7ec70fc7..4d6ed15887 100644 --- a/test/unit/org/apache/cassandra/hints/HintTest.java +++ b/test/unit/org/apache/cassandra/hints/HintTest.java @@ -222,8 +222,8 @@ public class HintTest { ReadCommand cmd = cmd(key, table); - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); - PartitionIterator iterator = cmd.executeInternal(orderGroup)) + try (ReadExecutionController executionController = cmd.executionController(); + PartitionIterator iterator = cmd.executeInternal(executionController)) { assertFalse(iterator.hasNext()); } diff --git a/test/unit/org/apache/cassandra/index/StubIndex.java b/test/unit/org/apache/cassandra/index/StubIndex.java index c8a3241d57..3d24d6e014 100644 --- a/test/unit/org/apache/cassandra/index/StubIndex.java +++ b/test/unit/org/apache/cassandra/index/StubIndex.java @@ -227,9 +227,9 @@ public class StubIndex implements Index Optional.empty()); } - private UnfilteredPartitionIterator queryStorageInternal(ColumnFamilyStore cfs, ReadOrderGroup orderGroup) + private UnfilteredPartitionIterator queryStorageInternal(ColumnFamilyStore cfs, ReadExecutionController executionController) { - return queryStorage(cfs, orderGroup); + return queryStorage(cfs, executionController); } } } diff --git a/test/unit/org/apache/cassandra/index/internal/CassandraIndexTest.java b/test/unit/org/apache/cassandra/index/internal/CassandraIndexTest.java index 73ce6c0d22..529f463762 100644 --- a/test/unit/org/apache/cassandra/index/internal/CassandraIndexTest.java +++ b/test/unit/org/apache/cassandra/index/internal/CassandraIndexTest.java @@ -429,8 +429,8 @@ public class CassandraIndexTest extends CQLTester indexKey, ColumnFilter.all(indexCfs.metadata), filter); - try (ReadOrderGroup orderGroup = ReadOrderGroup.forCommand(command); - UnfilteredRowIterator iter = command.queryMemtableAndDisk(indexCfs, orderGroup.indexReadOpOrderGroup())) + try (ReadExecutionController executionController = ReadExecutionController.forCommand(command); + UnfilteredRowIterator iter = command.queryMemtableAndDisk(indexCfs, executionController.indexReadOpOrderGroup())) { while( iter.hasNext()) { diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java index c7f3c361b4..4546867ad7 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java @@ -47,7 +47,6 @@ import org.apache.cassandra.index.Index; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.FileDataInput; import org.apache.cassandra.io.util.MmappedRegions; -import org.apache.cassandra.io.util.MmappedSegmentedFile; import org.apache.cassandra.io.util.SegmentedFile; import org.apache.cassandra.schema.CachingParams; import org.apache.cassandra.schema.KeyspaceParams; @@ -549,9 +548,9 @@ public class SSTableReaderTest .build(); Index.Searcher searcher = indexedCFS.indexManager.getBestIndexFor(rc).searcherFor(rc); assertNotNull(searcher); - try (ReadOrderGroup orderGroup = ReadOrderGroup.forCommand(rc)) + try (ReadExecutionController executionController = ReadExecutionController.forCommand(rc)) { - assertEquals(1, Util.size(UnfilteredPartitionIterators.filter(searcher.search(orderGroup), rc.nowInSec()))); + assertEquals(1, Util.size(UnfilteredPartitionIterators.filter(searcher.search(executionController), rc.nowInSec()))); } } diff --git a/test/unit/org/apache/cassandra/service/DataResolverTest.java b/test/unit/org/apache/cassandra/service/DataResolverTest.java index b60a039787..6048b9cc5b 100644 --- a/test/unit/org/apache/cassandra/service/DataResolverTest.java +++ b/test/unit/org/apache/cassandra/service/DataResolverTest.java @@ -715,7 +715,8 @@ public class DataResolverTest ReadResponse.createRemoteDataResponse(partitionIterator, cmd.columnFilter()), Collections.EMPTY_MAP, MessagingService.Verb.REQUEST_RESPONSE, - MessagingService.current_version); + MessagingService.current_version, + MessageIn.createTimestamp()); } private RangeTombstone tombstone(Object start, Object end, long markedForDeleteAt, int localDeletionTime) diff --git a/test/unit/org/apache/cassandra/service/QueryPagerTest.java b/test/unit/org/apache/cassandra/service/QueryPagerTest.java index 0f79e841e3..4daf2e50a8 100644 --- a/test/unit/org/apache/cassandra/service/QueryPagerTest.java +++ b/test/unit/org/apache/cassandra/service/QueryPagerTest.java @@ -121,7 +121,8 @@ public class QueryPagerTest StringBuilder sb = new StringBuilder(); List partitionList = new ArrayList<>(); int rows = 0; - try (ReadOrderGroup orderGroup = pager.startOrderGroup(); PartitionIterator iterator = pager.fetchPageInternal(toQuery, orderGroup)) + try (ReadExecutionController executionController = pager.executionController(); + PartitionIterator iterator = pager.fetchPageInternal(toQuery, executionController)) { while (iterator.hasNext()) { diff --git a/test/unit/org/apache/cassandra/utils/NanoTimeToCurrentTimeMillisTest.java b/test/unit/org/apache/cassandra/utils/NanoTimeToCurrentTimeMillisTest.java index 1662e7731c..b3dfad3a0f 100644 --- a/test/unit/org/apache/cassandra/utils/NanoTimeToCurrentTimeMillisTest.java +++ b/test/unit/org/apache/cassandra/utils/NanoTimeToCurrentTimeMillisTest.java @@ -34,10 +34,7 @@ public class NanoTimeToCurrentTimeMillisTest now = Math.max(now, System.currentTimeMillis()); if (ii % 10000 == 0) { - synchronized (NanoTimeToCurrentTimeMillis.TIMESTAMP_UPDATE) - { - NanoTimeToCurrentTimeMillis.TIMESTAMP_UPDATE.notify(); - } + NanoTimeToCurrentTimeMillis.updateNow(); Thread.sleep(1); }