natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[CASSANDRA-16923] CEP-10 Phase 1: Mockable System Clock 

Co-authored-by: Benedict Elliott Smith <benedict@apache.org>
Co-authored-by: Aleksey Yeschenko  <aleksey@apache.org>
Co-authored-by: Sam Tunnicliffe <samt@apache.org>
This commit is contained in:
Benedict Elliott Smith 2021-01-18 13:36:58 +00:00 committed by Benedict Elliott Smith
parent 643b9b776f
commit 15a2fe00fc
195 changed files with 997 additions and 1251 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
build.xml
View File

@ -534,6 +534,7 @@
</dependency>
<dependency groupId="org.apache.cassandra" artifactId="dtest-api" version="0.0.9" scope="test"/>
<dependency groupId="org.reflections" artifactId="reflections" version="0.9.12" scope="test"/>
<dependency groupId="com.puppycrawl.tools" artifactId="checkstyle" version="8.40" scope="test"/>
<dependency groupId="org.apache.hadoop" artifactId="hadoop-core" version="1.0.3" scope="provided">
<exclusion groupId="org.mortbay.jetty" artifactId="servlet-api"/>
<exclusion groupId="commons-logging" artifactId="commons-logging"/>
@ -718,6 +719,7 @@
<dependency groupId="junit" artifactId="junit"/>
<dependency groupId="commons-io" artifactId="commons-io"/>
<dependency groupId="org.mockito" artifactId="mockito-core"/>
<dependency groupId="com.puppycrawl.tools" artifactId="checkstyle" scope="test"/>
<dependency groupId="org.quicktheories" artifactId="quicktheories"/>
<dependency groupId="org.reflections" artifactId="reflections"/>
<dependency groupId="com.google.code.java-allocation-instrumenter" artifactId="java-allocation-instrumenter" version="${allocation-instrumenter.version}"/>
@ -869,7 +871,7 @@
<!--
The build target builds all the .class files
-->
<target name="build" depends="resolver-retrieve-build,build-project" description="Compile Cassandra classes"/>
<target name="build" depends="resolver-retrieve-build,build-project,checkstyle" description="Compile Cassandra classes"/>
<target name="codecoverage" depends="jacoco-run,jacoco-report" description="Create code coverage report"/>
<target name="_build_java">
@ -2100,6 +2102,32 @@
</java>
</target>
<target name="init-checkstyle" depends="maven-ant-tasks-retrieve-build,build-project">
<path id="checkstyle.lib.path">
<fileset dir="${test.lib}/jars" includes="*.jar"/>
</path>
<!-- Sevntu custom checks are retrieved by Ivy into lib folder
and will be accessible to checkstyle-->
<taskdef resource="com/puppycrawl/tools/checkstyle/ant/checkstyle-ant-task.properties"
classpathref="checkstyle.lib.path"/>
</target>
<target name="checkstyle" depends="init-checkstyle,maven-ant-tasks-retrieve-build,build-project" description="Run custom checkstyle code analysis" if="java.version.8">
<property name="checkstyle.log.dir" value="${build.dir}/checkstyle" />
<property name="checkstyle.report.file" value="${checkstyle.log.dir}/checkstyle_report.xml"/>
<mkdir dir="${checkstyle.log.dir}" />
<property name="checkstyle.properties" value="${basedir}/checkstyle.xml" />
<property name="checkstyle.suppressions" value="${basedir}/checkstyle_suppressions.xml" />
<checkstyle config="${checkstyle.properties}"
failureProperty="checkstyle.failure"
failOnViolation="true">
<formatter type="plain"/>
<formatter type="xml" tofile="${checkstyle.report.file}"/>
<fileset dir="${build.src.java}" includes="**/*.java"/>
</checkstyle>
</target>
<!-- Installs artifacts to local Maven repository -->
<target name="mvn-install"

48
checkstyle.xml Normal file
View File

@ -0,0 +1,48 @@
<?xml version="1.0"?>
<!--
~ Licensed to the Apache Software Foundation (ASF) under one
~ or more contributor license agreements. See the NOTICE file
~ distributed with this work for additional information
~ regarding copyright ownership. The ASF licenses this file
~ to you under the Apache License, Version 2.0 (the
~ "License"); you may not use this file except in compliance
~ with the License. You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<!DOCTYPE module PUBLIC
"-//Checkstyle//DTD Checkstyle Configuration 1.3//EN"
"https://checkstyle.org/dtds/configuration_1_3.dtd">
<module name="Checker">
<property name="severity" value="error"/>
<property name="fileExtensions" value="java, properties, xml"/>
<module name="BeforeExecutionExclusionFileFilter">
<property name="fileNamePattern" value="module\-info\.java$"/>
</module>
<!-- https://checkstyle.org/config_filters.html#SuppressionFilter -->
<module name="SuppressionFilter">
<property name="file" value="${checkstyle.suppressions}"
default="checkstyle-suppressions.xml" />
<property name="optional" value="false"/>
</module>
<module name="TreeWalker">
<module name="RegexpSinglelineJava">
<!-- To prevent static imports -->
<property name="format" value="System\.(currentTimeMillis|nanoTime)"/>
<property name="ignoreComments" value="true"/>
</module>
</module>
</module>

26
checkstyle_suppressions.xml Normal file
View File

@ -0,0 +1,26 @@
<?xml version="1.0"?>
<!--
~ Licensed to the Apache Software Foundation (ASF) under one
~ or more contributor license agreements. See the NOTICE file
~ distributed with this work for additional information
~ regarding copyright ownership. The ASF licenses this file
~ to you under the Apache License, Version 2.0 (the
~ "License"); you may not use this file except in compliance
~ with the License. You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<!DOCTYPE suppressions PUBLIC
"-//Checkstyle//DTD SuppressionFilter Configuration 1.1//EN"
"https://checkstyle.org/dtds/suppressions_1_1.dtd">
<suppressions>
<suppress checks="RegexpSinglelineJava" files="Clock\.java"/>
</suppressions>

6
src/java/org/apache/cassandra/audit/AuditLogEntry.java
View File

@ -32,6 +32,8 @@ import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public class AuditLogEntry
{
private final InetAddressAndPort host = FBUtilities.getBroadcastAddressAndPort();
@ -214,7 +216,7 @@ public class AuditLogEntry
user = AuthenticatedUser.SYSTEM_USER.getName();
}
timestamp = System.currentTimeMillis();
timestamp = currentTimeMillis();
}
public Builder(AuditLogEntry entry)
@ -312,7 +314,7 @@ public class AuditLogEntry
public AuditLogEntry build()
{
timestamp = timestamp > 0 ? timestamp : System.currentTimeMillis();
timestamp = timestamp > 0 ? timestamp : currentTimeMillis();
return new AuditLogEntry(type, source, user, timestamp, batch, keyspace, scope, operation, options, state);
}
}

9
src/java/org/apache/cassandra/auth/CassandraAuthorizer.java
View File

@ -40,6 +40,11 @@ import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.cassandra.cql3.BatchQueryOptions.withoutPerStatementVariables;
import static org.apache.cassandra.cql3.QueryOptions.DEFAULT;
import static org.apache.cassandra.service.QueryState.forInternalCalls;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* CassandraAuthorizer is an IAuthorizer implementation that keeps
* user permissions internally in C* using the system_auth.role_permissions
@ -352,7 +357,7 @@ public class CassandraAuthorizer implements IAuthorizer
ResultMessage.Rows select(SelectStatement statement, QueryOptions options)
{
return statement.execute(QueryState.forInternalCalls(), options, System.nanoTime());
return statement.execute(QueryState.forInternalCalls(), options, nanoTime());
}
UntypedResultSet process(String query, ConsistencyLevel cl) throws RequestExecutionException
@ -366,7 +371,7 @@ public class CassandraAuthorizer implements IAuthorizer
QueryProcessor.instance.processBatch(statement,
QueryState.forInternalCalls(),
BatchQueryOptions.withoutPerStatementVariables(options),
System.nanoTime());
nanoTime());
}
public static ConsistencyLevel authWriteConsistencyLevel()

5
src/java/org/apache/cassandra/auth/CassandraNetworkAuthorizer.java
View File

@ -35,6 +35,9 @@ import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.cassandra.service.QueryState.forInternalCalls;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class CassandraNetworkAuthorizer implements INetworkAuthorizer
{
private SelectStatement authorizeUserStatement = null;
@ -50,7 +53,7 @@ public class CassandraNetworkAuthorizer implements INetworkAuthorizer
@VisibleForTesting
ResultMessage.Rows select(SelectStatement statement, QueryOptions options)
{
return statement.execute(QueryState.forInternalCalls(), options, System.nanoTime());
return statement.execute(forInternalCalls(), options, nanoTime());
}
@VisibleForTesting

5
src/java/org/apache/cassandra/auth/CassandraRoleManager.java
View File

@ -48,6 +48,9 @@ import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.mindrot.jbcrypt.BCrypt;
import static org.apache.cassandra.service.QueryState.forInternalCalls;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Responsible for the creation, maintenance and deletion of roles
* for the purposes of authentication and authorization.
@ -548,6 +551,6 @@ public class CassandraRoleManager implements IRoleManager
@VisibleForTesting
ResultMessage.Rows select(SelectStatement statement, QueryOptions options)
{
return statement.execute(QueryState.forInternalCalls(), options, System.nanoTime());
return statement.execute(forInternalCalls(), options, nanoTime());
}
}

3
src/java/org/apache/cassandra/auth/PasswordAuthenticator.java
View File

@ -45,6 +45,7 @@ import org.apache.cassandra.utils.ByteBufferUtil;
import org.mindrot.jbcrypt.BCrypt;
import static org.apache.cassandra.auth.CassandraRoleManager.consistencyForRoleRead;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* PasswordAuthenticator is an IAuthenticator implementation
@ -130,7 +131,7 @@ public class PasswordAuthenticator implements IAuthenticator
@VisibleForTesting
ResultMessage.Rows select(SelectStatement statement, QueryOptions options)
{
return statement.execute(QueryState.forInternalCalls(), options, System.nanoTime());
return statement.execute(QueryState.forInternalCalls(), options, nanoTime());
}
public Set<DataResource> protectedResources()

6
src/java/org/apache/cassandra/batchlog/BatchlogManager.java
View File

@ -83,6 +83,8 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
import static org.apache.cassandra.cql3.QueryProcessor.executeInternalWithPaging;
import static org.apache.cassandra.net.Verb.MUTATION_REQ;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class BatchlogManager implements BatchlogManagerMBean
{
@ -213,7 +215,7 @@ public class BatchlogManager implements BatchlogManagerMBean
}
setRate(DatabaseDescriptor.getBatchlogReplayThrottleInKB());
UUID limitUuid = UUIDGen.maxTimeUUID(System.currentTimeMillis() - getBatchlogTimeout());
UUID limitUuid = UUIDGen.maxTimeUUID(currentTimeMillis() - getBatchlogTimeout());
ColumnFamilyStore store = Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(SystemKeyspace.BATCHES);
int pageSize = calculatePageSize(store);
// There cannot be any live content where token(id) <= token(lastReplayedUuid) as every processed batch is
@ -491,7 +493,7 @@ public class BatchlogManager implements BatchlogManagerMBean
ReplicaPlan.ForTokenWrite replicaPlan = new ReplicaPlan.ForTokenWrite(keyspace, liveAndDown.replicationStrategy(),
ConsistencyLevel.ONE, liveRemoteOnly.pending(), liveRemoteOnly.all(), liveRemoteOnly.all(), liveRemoteOnly.all());
ReplayWriteResponseHandler<Mutation> handler = new ReplayWriteResponseHandler<>(replicaPlan, System.nanoTime());
ReplayWriteResponseHandler<Mutation> handler = new ReplayWriteResponseHandler<>(replicaPlan, nanoTime());
Message<Mutation> message = Message.outWithFlag(MUTATION_REQ, mutation, MessageFlag.CALL_BACK_ON_FAILURE);
for (Replica replica : liveRemoteOnly.all())
MessagingService.instance().sendWriteWithCallback(message, replica, handler, false);

14
src/java/org/apache/cassandra/cache/AutoSavingCache.java vendored
View File

@ -53,6 +53,8 @@ import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.UUIDGen;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K, V>
{
public interface IStreamFactory
@ -158,7 +160,7 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
public ListenableFuture<Integer> loadSavedAsync()
{
final ListeningExecutorService es = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
final long start = System.nanoTime();
final long start = nanoTime();
ListenableFuture<Integer> cacheLoad = es.submit(new Callable<Integer>()
{
@ -175,7 +177,7 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
{
if (size() > 0)
logger.info("Completed loading ({} ms; {} keys) {} cache",
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start),
TimeUnit.NANOSECONDS.toMillis(nanoTime() - start),
CacheService.instance.keyCache.size(),
cacheType);
es.shutdown();
@ -188,7 +190,7 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
public int loadSaved()
{
int count = 0;
long start = System.nanoTime();
long start = nanoTime();
// modern format, allows both key and value (so key cache load can be purely sequential)
File dataPath = getCacheDataPath(CURRENT_VERSION);
@ -276,7 +278,7 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
}
if (logger.isTraceEnabled())
logger.trace("completed reading ({} ms; {} keys) saved cache {}",
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start), count, dataPath);
TimeUnit.NANOSECONDS.toMillis(nanoTime() - start), count, dataPath);
return count;
}
@ -347,7 +349,7 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
return;
}
long start = System.nanoTime();
long start = nanoTime();
Pair<File, File> cacheFilePaths = tempCacheFiles();
try (WrappedDataOutputStreamPlus writer = new WrappedDataOutputStreamPlus(streamFactory.getOutputStream(cacheFilePaths.left, cacheFilePaths.right)))
@ -401,7 +403,7 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
if (!cacheFilePaths.right.renameTo(crcFile))
logger.error("Unable to rename {} to {}", cacheFilePaths.right, crcFile);
logger.info("Saved {} ({} items) in {} ms", cacheType, keysWritten, TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
logger.info("Saved {} ({} items) in {} ms", cacheType, keysWritten, TimeUnit.NANOSECONDS.toMillis(nanoTime() - start));
}
private Pair<File, File> tempCacheFiles()

5
src/java/org/apache/cassandra/concurrent/SEPWorker.java
View File

@ -30,6 +30,7 @@ import org.apache.cassandra.utils.JVMStabilityInspector;
import static org.apache.cassandra.concurrent.SEPExecutor.TakeTaskPermitResult.RETURNED_WORK_PERMIT;
import static org.apache.cassandra.concurrent.SEPExecutor.TakeTaskPermitResult.TOOK_PERMIT;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
final class SEPWorker extends AtomicReference<SEPWorker.Work> implements Runnable
{
@ -257,7 +258,7 @@ final class SEPWorker extends AtomicReference<SEPWorker.Work> implements Runnabl
sleep *= ThreadLocalRandom.current().nextDouble();
sleep = Math.max(10000, sleep);
long start = System.nanoTime();
long start = nanoTime();
// place ourselves in the spinning collection; if we clash with another thread just exit
Long target = start + sleep;
@ -269,7 +270,7 @@ final class SEPWorker extends AtomicReference<SEPWorker.Work> implements Runnabl
pool.spinning.remove(target, this);
// finish timing and grab spinningTime (before we finish timing so it is under rather than overestimated)
long end = System.nanoTime();
long end = nanoTime();
long spin = end - start;
long stopCheck = pool.stopCheck.addAndGet(spin);
maybeStop(stopCheck, end);

5
src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java
View File

@ -28,6 +28,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.LockSupport;
import static org.apache.cassandra.concurrent.SEPWorker.Work;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* A pool of worker threads that are shared between all Executors created with it. Each executor is treated as a distinct
@ -128,10 +129,10 @@ public class SharedExecutorPool
terminateWorkers();
long until = System.nanoTime() + unit.toNanos(timeout);
long until = nanoTime() + unit.toNanos(timeout);
for (SEPExecutor executor : executors)
{
executor.shutdown.await(until - System.nanoTime(), TimeUnit.NANOSECONDS);
executor.shutdown.await(until - nanoTime(), TimeUnit.NANOSECONDS);
if (!executor.isTerminated())
throw new TimeoutException(executor.name + " not terminated");
}

3
src/java/org/apache/cassandra/cql3/Lists.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.cql3;
import static org.apache.cassandra.cql3.Constants.UNSET_VALUE;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@ -536,7 +537,7 @@ public abstract class Lists
{
if (remainingInBatch == 0)
{
long time = PrecisionTime.REFERENCE_TIME - (System.currentTimeMillis() - PrecisionTime.REFERENCE_TIME);
long time = PrecisionTime.REFERENCE_TIME - (currentTimeMillis() - PrecisionTime.REFERENCE_TIME);
remainingInBatch = Math.min(PrecisionTime.MAX_NANOS, i) + 1;
pt = PrecisionTime.getNext(time, remainingInBatch);
}

5
src/java/org/apache/cassandra/cql3/QueryProcessor.java
View File

@ -58,6 +58,7 @@ import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.*;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkTrue;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class QueryProcessor implements QueryHandler
{
@ -274,7 +275,7 @@ public class QueryProcessor implements QueryHandler
QueryState queryState = QueryState.forInternalCalls();
QueryOptions options = QueryOptions.forInternalCalls(cl, values);
CQLStatement statement = instance.parse(query, queryState, options);
ResultMessage result = instance.process(statement, queryState, options, System.nanoTime());
ResultMessage result = instance.process(statement, queryState, options, nanoTime());
if (result instanceof ResultMessage.Rows)
return UntypedResultSet.create(((ResultMessage.Rows)result).result);
else
@ -339,7 +340,7 @@ public class QueryProcessor implements QueryHandler
try
{
Prepared prepared = prepareInternal(query);
ResultMessage result = prepared.statement.execute(state, makeInternalOptions(prepared.statement, values, cl), System.nanoTime());
ResultMessage result = prepared.statement.execute(state, makeInternalOptions(prepared.statement, values, cl), nanoTime());
if (result instanceof ResultMessage.Rows)
return UntypedResultSet.create(((ResultMessage.Rows)result).result);
else

4
src/java/org/apache/cassandra/cql3/UntypedResultSet.java
View File

@ -37,6 +37,8 @@ import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/** a utility for doing internal cql-based queries */
public abstract class UntypedResultSet implements Iterable<UntypedResultSet.Row>
{
@ -269,7 +271,7 @@ public abstract class UntypedResultSet implements Iterable<UntypedResultSet.Row>
if (pager.isExhausted())
return endOfData();
try (PartitionIterator iter = pager.fetchPage(pageSize, cl, clientState, System.nanoTime()))
try (PartitionIterator iter = pager.fetchPage(pageSize, cl, clientState, nanoTime()))
{
currentPage = select.process(iter, nowInSec).rows.iterator();
}

5
src/java/org/apache/cassandra/cql3/functions/UDAggregate.java
View File

@ -39,6 +39,7 @@ import org.apache.cassandra.transport.ProtocolVersion;
import static com.google.common.collect.Iterables.any;
import static com.google.common.collect.Iterables.transform;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Base class for user-defined-aggregates.
@ -182,7 +183,7 @@ public class UDAggregate extends AbstractFunction implements AggregateFunction,
{
maybeInit(protocolVersion);
long startTime = System.nanoTime();
long startTime = nanoTime();
stateFunctionCount++;
if (stateFunction instanceof UDFunction)
{
@ -194,7 +195,7 @@ public class UDAggregate extends AbstractFunction implements AggregateFunction,
{
throw new UnsupportedOperationException("UDAs only support UDFs");
}
stateFunctionDuration += (System.nanoTime() - startTime) / 1000;
stateFunctionDuration += (nanoTime() - startTime) / 1000;
}
private void maybeInit(ProtocolVersion protocolVersion)

9
src/java/org/apache/cassandra/cql3/functions/UDFunction.java
View File

@ -61,6 +61,7 @@ import org.apache.cassandra.utils.JVMStabilityInspector;
import static com.google.common.collect.Iterables.any;
import static com.google.common.collect.Iterables.transform;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Base class for User Defined Functions.
@ -356,7 +357,7 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
if (!isCallableWrtNullable(parameters))
return null;
long tStart = System.nanoTime();
long tStart = nanoTime();
parameters = makeEmptyParametersNull(parameters);
try
@ -366,7 +367,7 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
? executeAsync(protocolVersion, parameters)
: executeUserDefined(protocolVersion, parameters);
Tracing.trace("Executed UDF {} in {}\u03bcs", name(), (System.nanoTime() - tStart) / 1000);
Tracing.trace("Executed UDF {} in {}\u03bcs", name(), (nanoTime() - tStart) / 1000);
return result;
}
catch (InvalidRequestException e)
@ -395,7 +396,7 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
if (!calledOnNullInput && firstParam == null || !isCallableWrtNullable(parameters))
return null;
long tStart = System.nanoTime();
long tStart = nanoTime();
parameters = makeEmptyParametersNull(parameters);
try
@ -404,7 +405,7 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
Object result = DatabaseDescriptor.enableUserDefinedFunctionsThreads()
? executeAggregateAsync(protocolVersion, firstParam, parameters)
: executeAggregateUserDefined(protocolVersion, firstParam, parameters);
Tracing.trace("Executed UDF {} in {}\u03bcs", name(), (System.nanoTime() - tStart) / 1000);
Tracing.trace("Executed UDF {} in {}\u03bcs", name(), (nanoTime() - tStart) / 1000);
return result;
}
catch (InvalidRequestException e)

3
src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
View File

@ -51,6 +51,7 @@ import org.apache.cassandra.utils.Pair;
import static java.util.function.Predicate.isEqual;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* A <code>BATCH</code> statement parsed from a CQL query.
@ -550,7 +551,7 @@ public class BatchStatement implements CQLStatement
if (hasConditions)
return executeInternalWithConditions(batchOptions, queryState);
executeInternalWithoutCondition(queryState, batchOptions, System.nanoTime());
executeInternalWithoutCondition(queryState, batchOptions, nanoTime());
return new ResultMessage.Void();
}

3
src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
View File

@ -62,6 +62,7 @@ import org.apache.cassandra.utils.UUIDGen;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkNull;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/*
* Abstract parent class of individual modifications, i.e. INSERT, UPDATE and DELETE.
@ -622,7 +623,7 @@ public abstract class ModificationStatement implements CQLStatement
{
return hasConditions()
? executeInternalWithCondition(queryState, options)
: executeInternalWithoutCondition(queryState, options, System.nanoTime());
: executeInternalWithoutCondition(queryState, options, nanoTime());
}
public ResultMessage executeInternalWithoutCondition(QueryState queryState, QueryOptions options, long queryStartNanoTime)

3
src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
View File

@ -80,6 +80,7 @@ import static org.apache.cassandra.cql3.statements.RequestValidations.checkNotNu
import static org.apache.cassandra.cql3.statements.RequestValidations.checkNull;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkTrue;
import static org.apache.cassandra.utils.ByteBufferUtil.UNSET_BYTE_BUFFER;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Encapsulates a completely parsed SELECT query, including the target
@ -436,7 +437,7 @@ public class SelectStatement implements CQLStatement
public ResultMessage.Rows executeLocally(QueryState state, QueryOptions options) throws RequestExecutionException, RequestValidationException
{
return executeInternal(state, options, options.getNowInSeconds(state), System.nanoTime());
return executeInternal(state, options, options.getNowInSeconds(state), nanoTime());
}
public ResultMessage.Rows executeInternal(QueryState state, QueryOptions options, int nowInSec, long queryStartNanoTime) throws RequestExecutionException, RequestValidationException

4
src/java/org/apache/cassandra/cql3/statements/UseStatement.java
View File

@ -29,6 +29,8 @@ import org.apache.cassandra.service.QueryState;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class UseStatement extends CQLStatement.Raw implements CQLStatement
{
private final String keyspace;
@ -62,7 +64,7 @@ public class UseStatement extends CQLStatement.Raw implements CQLStatement
{
// In production, internal queries are exclusively on the system keyspace and 'use' is thus useless
// but for some unit tests we need to set the keyspace (e.g. for tests with DROP INDEX)
return execute(state, options, System.nanoTime());
return execute(state, options, nanoTime());
}
@Override

25
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -96,6 +96,8 @@ import org.apache.cassandra.utils.memory.MemtableAllocator;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.Throwables.maybeFail;
import static org.apache.cassandra.utils.Throwables.merge;
@ -1056,7 +1058,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
if (logger.isTraceEnabled())
logger.trace("Flush task {}@{} starts executing, waiting on barrier", hashCode(), name);
long start = System.nanoTime();
long start = nanoTime();
// mark writes older than the barrier as blocking progress, permitting them to exceed our memory limit
// if they are stuck waiting on it, then wait for them all to complete
@ -1064,7 +1066,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
writeBarrier.await();
if (logger.isTraceEnabled())
logger.trace("Flush task for task {}@{} waited {} ms at the barrier", hashCode(), name, TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
logger.trace("Flush task for task {}@{} waited {} ms at the barrier", hashCode(), name, TimeUnit.NANOSECONDS.toMillis(nanoTime() - start));
// mark all memtables as flushing, removing them from the live memtable list
for (Memtable memtable : memtables)
@ -1331,7 +1333,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public void apply(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, CommitLogPosition commitLogPosition)
{
long start = System.nanoTime();
long start = nanoTime();
try
{
Memtable mt = data.getMemtableFor(opGroup, commitLogPosition);
@ -1342,7 +1344,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
if (metric.topWritePartitionSize.isEnabled()) // dont compute datasize if not needed
metric.topWritePartitionSize.addSample(key.getKey(), update.dataSize());
StorageHook.instance.reportWrite(metadata.id, update);
metric.writeLatency.addNano(System.nanoTime() - start);
metric.writeLatency.addNano(nanoTime() - start);
// CASSANDRA-11117 - certain resolution paths on memtable put can result in very
// large time deltas, either through a variety of sentinel timestamps (used for empty values, ensuring
// a minimal write, etc). This limits the time delta to the max value the histogram
@ -1526,8 +1528,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
// skip snapshot creation during scrub, SEE JIRA 5891
if(!disableSnapshot)
{
Instant creationTime = Instant.now();
String snapshotName = "pre-scrub-" + creationTime.toEpochMilli();
long epochMilli = currentTimeMillis();
Instant creationTime = Instant.ofEpochMilli(epochMilli);
String snapshotName = "pre-scrub-" + epochMilli;
snapshotWithoutFlush(snapshotName, creationTime);
}
@ -1732,9 +1735,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return new RefViewFragment(view.sstables, view.memtables, refs);
if (failingSince <= 0)
{
failingSince = System.nanoTime();
failingSince = nanoTime();
}
else if (System.nanoTime() - failingSince > TimeUnit.MILLISECONDS.toNanos(100))
else if (nanoTime() - failingSince > TimeUnit.MILLISECONDS.toNanos(100))
{
List<SSTableReader> released = new ArrayList<>();
for (SSTableReader reader : view.sstables)
@ -1742,7 +1745,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
released.add(reader);
NoSpamLogger.log(logger, NoSpamLogger.Level.WARN, 1, TimeUnit.SECONDS,
"Spinning trying to capture readers {}, released: {}, ", view.sstables, released);
failingSince = System.nanoTime();
failingSince = nanoTime();
}
}
}
@ -2314,7 +2317,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
}
long now = System.currentTimeMillis();
long now = currentTimeMillis();
// make sure none of our sstables are somehow in the future (clock drift, perhaps)
for (ColumnFamilyStore cfs : concatWithIndexes())
for (SSTableReader sstable : cfs.getLiveSSTables())
@ -2791,7 +2794,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
double allDroppable = 0;
long allColumns = 0;
int localTime = (int)(System.currentTimeMillis()/1000);
int localTime = (int)(currentTimeMillis() / 1000);
for (SSTableReader sstable : getSSTables(SSTableSet.LIVE))
{

5
src/java/org/apache/cassandra/db/CounterMutation.java
View File

@ -50,6 +50,7 @@ import static java.util.concurrent.TimeUnit.*;
import static org.apache.cassandra.net.MessagingService.VERSION_30;
import static org.apache.cassandra.net.MessagingService.VERSION_3014;
import static org.apache.cassandra.net.MessagingService.VERSION_40;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class CounterMutation implements IMutation
{
@ -150,12 +151,12 @@ public class CounterMutation implements IMutation
private void grabCounterLocks(Keyspace keyspace, List<Lock> locks) throws WriteTimeoutException
{
long startTime = System.nanoTime();
long startTime = nanoTime();
AbstractReplicationStrategy replicationStrategy = keyspace.getReplicationStrategy();
for (Lock lock : LOCKS.bulkGet(getCounterLockKeys()))
{
long timeout = getTimeout(NANOSECONDS) - (System.nanoTime() - startTime);
long timeout = getTimeout(NANOSECONDS) - (nanoTime() - startTime);
try
{
if (!lock.tryLock(timeout, NANOSECONDS))

4
src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java
View File

@ -26,6 +26,8 @@ import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageProxy;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class CounterMutationVerbHandler implements IVerbHandler<CounterMutation>
{
public static final CounterMutationVerbHandler instance = new CounterMutationVerbHandler();
@ -34,7 +36,7 @@ public class CounterMutationVerbHandler implements IVerbHandler<CounterMutation>
public void doVerb(final Message<CounterMutation> message)
{
long queryStartNanoTime = System.nanoTime();
long queryStartNanoTime = nanoTime();
final CounterMutation cm = message.payload;
logger.trace("Applying forwarded {}", cm);

4
src/java/org/apache/cassandra/db/ExpirationDateOverflowHandling.java
View File

@ -32,6 +32,8 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientWarn;
import org.apache.cassandra.utils.NoSpamLogger;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public class ExpirationDateOverflowHandling
{
private static final Logger logger = LoggerFactory.getLogger(ExpirationDateOverflowHandling.class);
@ -75,7 +77,7 @@ public class ExpirationDateOverflowHandling
return;
// Check for localExpirationTime overflow (CASSANDRA-14092)
int nowInSecs = (int)(System.currentTimeMillis() / 1000);
int nowInSecs = (int)(currentTimeMillis() / 1000);
if (ttl + nowInSecs < 0)
{
switch (policy)

9
src/java/org/apache/cassandra/db/Keyspace.java
View File

@ -75,6 +75,7 @@ import org.apache.cassandra.utils.concurrent.OpOrder;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.MonotonicClock.approxTime;
/**
@ -285,7 +286,7 @@ public class Keyspace
*/
public static String getTimestampedSnapshotName(String clientSuppliedName)
{
String snapshotName = Long.toString(System.currentTimeMillis());
String snapshotName = Long.toString(currentTimeMillis());
if (clientSuppliedName != null && !clientSuppliedName.equals(""))
{
snapshotName = snapshotName + "-" + clientSuppliedName;
@ -547,7 +548,7 @@ public class Keyspace
if (requiresViewUpdate)
{
mutation.viewLockAcquireStart.compareAndSet(0L, System.currentTimeMillis());
mutation.viewLockAcquireStart.compareAndSet(0L, currentTimeMillis());
// the order of lock acquisition doesn't matter (from a deadlock perspective) because we only use tryLock()
Collection<TableId> tableIds = mutation.getTableIds();
@ -625,7 +626,7 @@ public class Keyspace
}
}
long acquireTime = System.currentTimeMillis() - mutation.viewLockAcquireStart.get();
long acquireTime = currentTimeMillis() - mutation.viewLockAcquireStart.get();
// Metrics are only collected for droppable write operations
// Bulk non-droppable operations (e.g. commitlog replay, hint delivery) are not measured
if (isDroppable)
@ -669,7 +670,7 @@ public class Keyspace
cfs.getWriteHandler().write(upd, ctx, indexTransaction);
if (requiresViewUpdate)
baseComplete.set(System.currentTimeMillis());
baseComplete.set(currentTimeMillis());
}
if (future != null) {

6
src/java/org/apache/cassandra/db/Memtable.java
View File

@ -75,6 +75,8 @@ import org.apache.cassandra.utils.memory.MemtablePool;
import org.apache.cassandra.utils.memory.NativePool;
import org.apache.cassandra.utils.memory.SlabPool;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class Memtable implements Comparable<Memtable>
{
private static final Logger logger = LoggerFactory.getLogger(Memtable.class);
@ -137,7 +139,7 @@ public class Memtable implements Comparable<Memtable>
// actually only store DecoratedKey.
private final ConcurrentNavigableMap<PartitionPosition, AtomicBTreePartition> partitions = new ConcurrentSkipListMap<>();
public final ColumnFamilyStore cfs;
private final long creationNano = System.nanoTime();
private final long creationNano = nanoTime();
// The smallest timestamp for all partitions stored in this memtable
private long minTimestamp = Long.MAX_VALUE;
@ -261,7 +263,7 @@ public class Memtable implements Comparable<Memtable>
public boolean isExpired()
{
int period = cfs.metadata().params.memtableFlushPeriodInMs;
return period > 0 && (System.nanoTime() - creationNano >= TimeUnit.MILLISECONDS.toNanos(period));
return period > 0 && (nanoTime() - creationNano >= TimeUnit.MILLISECONDS.toNanos(period));
}
/**

5
src/java/org/apache/cassandra/db/ReadCommand.java
View File

@ -72,6 +72,7 @@ import org.apache.cassandra.utils.ObjectSizes;
import static com.google.common.collect.Iterables.any;
import static com.google.common.collect.Iterables.filter;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.MonotonicClock.approxTime;
import static org.apache.cassandra.db.partitions.UnfilteredPartitionIterators.MergeListener.NOOP;
@ -397,7 +398,7 @@ public abstract class ReadCommand extends AbstractReadQuery
// iterators created inside the try as long as we do close the original resultIterator), or by closing the result.
public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController)
{
long startTimeNanos = System.nanoTime();
long startTimeNanos = nanoTime();
COMMAND.set(this);
try
@ -568,7 +569,7 @@ public abstract class ReadCommand extends AbstractReadQuery
@Override
public void onClose()
{
recordLatency(metric, System.nanoTime() - startTimeNanos);
recordLatency(metric, nanoTime() - startTimeNanos);
metric.tombstoneScannedHistogram.update(tombstones);
metric.liveScannedHistogram.update(liveRows);

6
src/java/org/apache/cassandra/db/RepairedDataInfo.java
View File

@ -36,6 +36,8 @@ import org.apache.cassandra.metrics.TableMetrics;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
@NotThreadSafe
class RepairedDataInfo
{
@ -281,7 +283,7 @@ class RepairedDataInfo
return null;
long countBeforeOverreads = repairedCounter.counted();
long overreadStartTime = System.nanoTime();
long overreadStartTime = nanoTime();
if (currentPartition != null)
consumePartition(currentPartition, repairedCounter);
@ -291,7 +293,7 @@ class RepairedDataInfo
// we're not actually providing any more rows, just consuming the repaired data
long rows = repairedCounter.counted() - countBeforeOverreads;
long nanos = System.nanoTime() - overreadStartTime;
long nanos = nanoTime() - overreadStartTime;
metrics.repairedDataTrackingOverreadRows.update(rows);
metrics.repairedDataTrackingOverreadTime.update(nanos, TimeUnit.NANOSECONDS);
Tracing.trace("Read {} additional rows of repaired data for tracking in {}ps", rows, TimeUnit.NANOSECONDS.toMicros(nanos));

6
src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
View File

@ -37,6 +37,8 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.Refs;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* A very simplistic/crude partition count/size estimator.
*
@ -89,7 +91,7 @@ public class SizeEstimatesRecorder extends SchemaChangeListener implements Runna
boolean rangesAreEqual = primaryRanges.equals(localPrimaryRanges);
for (ColumnFamilyStore table : keyspace.getColumnFamilyStores())
{
long start = System.nanoTime();
long start = nanoTime();
// compute estimates for primary ranges for backwards compatability
Map<Range<Token>, Pair<Long, Long>> estimates = computeSizeEstimates(table, primaryRanges);
@ -103,7 +105,7 @@ public class SizeEstimatesRecorder extends SchemaChangeListener implements Runna
}
SystemKeyspace.updateTableEstimates(table.metadata.keyspace, table.metadata.name, SystemKeyspace.TABLE_ESTIMATES_TYPE_LOCAL_PRIMARY, estimates);
long passed = System.nanoTime() - start;
long passed = nanoTime() - start;
if (logger.isTraceEnabled())
logger.trace("Spent {} milliseconds on estimating {}.{} size",
TimeUnit.NANOSECONDS.toMillis(passed),

8
src/java/org/apache/cassandra/db/SystemKeyspace.java
View File

@ -73,6 +73,8 @@ import static java.util.Collections.singletonMap;
import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
import static org.apache.cassandra.cql3.QueryProcessor.executeOnceInternal;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public final class SystemKeyspace
{
@ -1008,13 +1010,13 @@ public final class SystemKeyspace
// seconds-since-epoch isn't a foolproof new generation
// (where foolproof is "guaranteed to be larger than the last one seen at this ip address"),
// but it's as close as sanely possible
generation = (int) (System.currentTimeMillis() / 1000);
generation = (int) (currentTimeMillis() / 1000);
}
else
{
// Other nodes will ignore gossip messages about a node that have a lower generation than previously seen.
final int storedGeneration = result.one().getInt("gossip_generation") + 1;
final int now = (int) (System.currentTimeMillis() / 1000);
final int now = (int) (currentTimeMillis() / 1000);
if (storedGeneration >= now)
{
logger.warn("Using stored Gossip Generation {} as it is greater than current system time {}. See CASSANDRA-3654 if you experience problems",
@ -1177,7 +1179,7 @@ public final class SystemKeyspace
public static PaxosState loadPaxosState(DecoratedKey key, TableMetadata metadata, int nowInSec)
{
String req = "SELECT * FROM system.%s WHERE row_key = ? AND cf_id = ?";
UntypedResultSet results = QueryProcessor.executeInternalWithNow(nowInSec, System.nanoTime(), format(req, PAXOS), key.getKey(), metadata.id.asUUID());
UntypedResultSet results = QueryProcessor.executeInternalWithNow(nowInSec, nanoTime(), format(req, PAXOS), key.getKey(), metadata.id.asUUID());
if (results.isEmpty())
return new PaxosState(key, metadata);
UntypedResultSet.Row row = results.one();

7
src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
View File

@ -33,6 +33,9 @@ import org.apache.cassandra.utils.MonotonicClock;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public abstract class AbstractCommitLogService
{
/**
@ -45,7 +48,7 @@ public abstract class AbstractCommitLogService
private volatile boolean shutdown = false;
// all Allocations written before this time will be synced
protected volatile long lastSyncedAt = System.currentTimeMillis();
protected volatile long lastSyncedAt = currentTimeMillis();
// counts of total written, and pending, log messages
private final AtomicLong written = new AtomicLong(0);
@ -292,7 +295,7 @@ public abstract class AbstractCommitLogService
*/
public void syncBlocking()
{
long requestTime = System.nanoTime();
long requestTime = nanoTime();
requestExtraSync();
awaitSyncAt(requestTime, null);
}

3
src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
View File

@ -48,6 +48,7 @@ import org.apache.cassandra.utils.IntegerInterval;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
/*
@ -78,7 +79,7 @@ public abstract class CommitLogSegment
if (CommitLogDescriptor.isValid(file.getName()))
maxId = Math.max(CommitLogDescriptor.fromFileName(file.getName()).id, maxId);
}
replayLimitId = idBase = Math.max(System.currentTimeMillis(), maxId + 1);
replayLimitId = idBase = Math.max(currentTimeMillis(), maxId + 1);
}
// The commit log entry overhead in bytes (int: length + int: head checksum + int: tail checksum)

4
src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogService.java
View File

@ -21,6 +21,8 @@ import java.util.concurrent.TimeUnit;
import org.apache.cassandra.config.DatabaseDescriptor;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
class PeriodicCommitLogService extends AbstractCommitLogService
{
private static final long blockWhenSyncLagsNanos = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getPeriodicCommitLogSyncBlock());
@ -33,7 +35,7 @@ class PeriodicCommitLogService extends AbstractCommitLogService
protected void maybeWaitForSync(CommitLogSegment.Allocation alloc)
{
long expectedSyncTime = System.nanoTime() - blockWhenSyncLagsNanos;
long expectedSyncTime = nanoTime() - blockWhenSyncLagsNanos;
if (lastSyncedAt < expectedSyncTime)
{
pending.incrementAndGet();

5
src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
View File

@ -44,6 +44,9 @@ import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
import org.apache.cassandra.schema.CompactionParams;
import static org.apache.cassandra.io.sstable.Component.DATA;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Pluggable compaction strategy determines how SSTables get merged.
*
@ -388,7 +391,7 @@ public abstract class AbstractCompactionStrategy
// since we use estimations to calculate, there is a chance that compaction will not drop tombstones actually.
// if that happens we will end up in infinite compaction loop, so first we check enough if enough time has
// elapsed since SSTable created.
if (System.currentTimeMillis() < sstable.getCreationTimeFor(Component.DATA) + tombstoneCompactionInterval * 1000)
if (currentTimeMillis() < sstable.getCreationTimeFor(DATA) + tombstoneCompactionInterval * 1000)
return false;
double droppableRatio = sstable.getEstimatedDroppableTombstoneRatio(gcBefore);

4
src/java/org/apache/cassandra/db/compaction/CompactionLogger.java
View File

@ -45,6 +45,8 @@ import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.utils.NoSpamLogger;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public class CompactionLogger
{
public interface Strategy
@ -220,7 +222,7 @@ public class CompactionLogger
return;
node.put("keyspace", cfs.keyspace.getName());
node.put("table", cfs.getTableName());
node.put("time", System.currentTimeMillis());
node.put("time", currentTimeMillis());
}
private JsonNode startStrategies()

9
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -86,6 +86,7 @@ import org.apache.cassandra.utils.concurrent.Refs;
import static java.util.Collections.singleton;
import static org.apache.cassandra.service.ActiveRepairService.NO_PENDING_REPAIR;
import static org.apache.cassandra.service.ActiveRepairService.UNREPAIRED_SSTABLE;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* <p>
@ -1267,7 +1268,7 @@ public class CompactionManager implements CompactionManagerMBean
return;
}
long start = System.nanoTime();
long start = nanoTime();
long totalkeysWritten = 0;
@ -1325,7 +1326,7 @@ public class CompactionManager implements CompactionManagerMBean
if (!finished.isEmpty())
{
String format = "Cleaned up to %s. %s to %s (~%d%% of original) for %,d keys. Time: %,dms.";
long dTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
long dTime = TimeUnit.NANOSECONDS.toMillis(nanoTime() - start);
long startsize = sstable.onDiskLength();
long endsize = 0;
for (SSTableReader newSstable : finished)
@ -2247,10 +2248,10 @@ public class CompactionManager implements CompactionManagerMBean
public void waitForCessation(Iterable<ColumnFamilyStore> cfss, Predicate<SSTableReader> sstablePredicate)
{
long start = System.nanoTime();
long start = nanoTime();
long delay = TimeUnit.MINUTES.toNanos(1);
while (System.nanoTime() - start < delay)
while (nanoTime() - start < delay)
{
if (CompactionManager.instance.isCompacting(cfss, sstablePredicate))
Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS);

23
src/java/org/apache/cassandra/db/compaction/CompactionTask.java
View File

@ -47,6 +47,9 @@ import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.Refs;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class CompactionTask extends AbstractCompactionTask
{
protected static final Logger logger = LoggerFactory.getLogger(CompactionTask.class);
@ -117,11 +120,11 @@ public class CompactionTask extends AbstractCompactionTask
if (DatabaseDescriptor.isSnapshotBeforeCompaction())
{
Instant creationTime = Instant.now();
cfs.snapshotWithoutFlush(creationTime.toEpochMilli() + "-compact-" + cfs.name, creationTime);
long epochMilli = currentTimeMillis();
Instant creationTime = Instant.ofEpochMilli(epochMilli);
cfs.snapshotWithoutFlush(epochMilli + "-compact-" + cfs.name, creationTime);
}
try (CompactionController controller = getCompactionController(transaction.originals()))
{
@ -155,8 +158,8 @@ public class CompactionTask extends AbstractCompactionTask
logger.info("Compacting ({}) {}", taskId, ssTableLoggerMsg);
RateLimiter limiter = CompactionManager.instance.getRateLimiter();
long start = System.nanoTime();
long startTime = System.currentTimeMillis();
long start = nanoTime();
long startTime = currentTimeMillis();
long totalKeysWritten = 0;
long estimatedKeys = 0;
long inputSizeBytes;
@ -206,10 +209,10 @@ public class CompactionTask extends AbstractCompactionTask
lastBytesScanned = bytesScanned;
if (System.nanoTime() - lastCheckObsoletion > TimeUnit.MINUTES.toNanos(1L))
if (nanoTime() - lastCheckObsoletion > TimeUnit.MINUTES.toNanos(1L))
{
controller.maybeRefreshOverlaps();
lastCheckObsoletion = System.nanoTime();
lastCheckObsoletion = nanoTime();
}
}
@ -232,7 +235,7 @@ public class CompactionTask extends AbstractCompactionTask
{
// log a bunch of statistics about the result and save to system table compaction_history
long durationInNano = System.nanoTime() - start;
long durationInNano = nanoTime() - start;
long dTime = TimeUnit.NANOSECONDS.toMillis(durationInNano);
long startsize = inputSizeBytes;
long endsize = SSTableReader.getTotalBytes(newSStables);
@ -267,7 +270,7 @@ public class CompactionTask extends AbstractCompactionTask
logger.trace("CF Total Bytes Compacted: {}", FBUtilities.prettyPrintMemory(CompactionTask.addToTotalBytesCompacted(endsize)));
logger.trace("Actual #keys: {}, Estimated #keys:{}, Err%: {}", totalKeysWritten, estimatedKeys, ((double)(totalKeysWritten - estimatedKeys)/totalKeysWritten));
}
cfs.getCompactionStrategyManager().compactionLogger.compaction(startTime, transaction.originals(), System.currentTimeMillis(), newSStables);
cfs.getCompactionStrategyManager().compactionLogger.compaction(startTime, transaction.originals(), currentTimeMillis(), newSStables);
// update the metrics
cfs.metric.compactionBytesWritten.inc(endsize);
@ -298,7 +301,7 @@ public class CompactionTask extends AbstractCompactionTask
mergeSummary.append(String.format("%d:%d, ", rows, count));
mergedRows.put(rows, count);
}
SystemKeyspace.updateCompactionHistory(keyspaceName, columnFamilyName, System.currentTimeMillis(), startSize, endSize, mergedRows);
SystemKeyspace.updateCompactionHistory(keyspaceName, columnFamilyName, currentTimeMillis(), startSize, endSize, mergedRows);
return mergeSummary.toString();
}

5
src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
View File

@ -38,6 +38,7 @@ import org.apache.cassandra.schema.CompactionParams;
import org.apache.cassandra.utils.Pair;
import static com.google.common.collect.Iterables.filter;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* @deprecated in favour of {@link TimeWindowCompactionStrategy}
@ -116,11 +117,11 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
Set<SSTableReader> expired = Collections.emptySet();
// we only check for expired sstables every 10 minutes (by default) due to it being an expensive operation
if (System.currentTimeMillis() - lastExpiredCheck > options.expiredSSTableCheckFrequency)
if (currentTimeMillis() - lastExpiredCheck > options.expiredSSTableCheckFrequency)
{
// Find fully expired SSTables. Those will be included no matter what.
expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingLiveSSTables(uncompacting), gcBefore);
lastExpiredCheck = System.currentTimeMillis();
lastExpiredCheck = currentTimeMillis();
}
Set<SSTableReader> candidates = Sets.newHashSet(filterSuspectSSTables(uncompacting));

8
src/java/org/apache/cassandra/db/compaction/LeveledGenerations.java
View File

@ -42,6 +42,8 @@ import org.apache.cassandra.config.Config;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Handles the leveled manifest generations
*
@ -68,7 +70,7 @@ class LeveledGenerations
*/
private final Map<SSTableReader, SSTableReader> allSSTables = new HashMap<>();
private final Set<SSTableReader> l0 = new HashSet<>();
private static long lastOverlapCheck = System.nanoTime();
private static long lastOverlapCheck = nanoTime();
// note that since l0 is broken out, levels[0] represents L1:
private final TreeSet<SSTableReader> [] levels = new TreeSet[MAX_LEVEL_COUNT - 1];
@ -310,10 +312,10 @@ class LeveledGenerations
*/
private void maybeVerifyLevels()
{
if (!strictLCSChecksTest || System.nanoTime() - lastOverlapCheck <= TimeUnit.NANOSECONDS.convert(5, TimeUnit.SECONDS))
if (!strictLCSChecksTest || nanoTime() - lastOverlapCheck <= TimeUnit.NANOSECONDS.convert(5, TimeUnit.SECONDS))
return;
logger.info("LCS verifying levels");
lastOverlapCheck = System.nanoTime();
lastOverlapCheck = nanoTime();
for (int i = 1; i < levelCount(); i++)
{
SSTableReader prev = null;

5
src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
View File

@ -45,6 +45,7 @@ import org.apache.cassandra.schema.CompactionParams;
import org.apache.cassandra.utils.Pair;
import static com.google.common.collect.Iterables.filter;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public class TimeWindowCompactionStrategy extends AbstractCompactionStrategy
{
@ -117,12 +118,12 @@ public class TimeWindowCompactionStrategy extends AbstractCompactionStrategy
// Find fully expired SSTables. Those will be included no matter what.
Set<SSTableReader> expired = Collections.emptySet();
if (System.currentTimeMillis() - lastExpiredCheck > options.expiredSSTableCheckFrequency)
if (currentTimeMillis() - lastExpiredCheck > options.expiredSSTableCheckFrequency)
{
logger.debug("TWCS expired check sufficiently far in the past, checking for fully expired SSTables");
expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, options.ignoreOverlaps ? Collections.emptySet() : cfs.getOverlappingLiveSSTables(uncompacting),
gcBefore, options.ignoreOverlaps);
lastExpiredCheck = System.currentTimeMillis();
lastExpiredCheck = currentTimeMillis();
}
else
{

5
src/java/org/apache/cassandra/db/lifecycle/LogFile.java
View File

@ -41,6 +41,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.big.BigFormat;
import org.apache.cassandra.utils.Throwables;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Throwables.merge;
/**
@ -274,12 +275,12 @@ final class LogFile implements AutoCloseable
void commit()
{
addRecord(LogRecord.makeCommit(System.currentTimeMillis()));
addRecord(LogRecord.makeCommit(currentTimeMillis()));
}
void abort()
{
addRecord(LogRecord.makeAbort(System.currentTimeMillis()));
addRecord(LogRecord.makeAbort(currentTimeMillis()));
}
private boolean isLastRecordValidWithType(Type type)

4
src/java/org/apache/cassandra/db/marshal/TemporalType.java
View File

@ -21,6 +21,8 @@ import java.nio.ByteBuffer;
import org.apache.cassandra.cql3.Duration;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Base type for temporal types (timestamp, date ...).
*
@ -38,7 +40,7 @@ public abstract class TemporalType<T> extends AbstractType<T>
*/
public ByteBuffer now()
{
return fromTimeInMillis(System.currentTimeMillis());
return fromTimeInMillis(currentTimeMillis());
}
/**

4
src/java/org/apache/cassandra/db/partitions/AtomicBTreePartition.java
View File

@ -39,6 +39,8 @@ import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.memory.HeapAllocator;
import org.apache.cassandra.utils.memory.MemtableAllocator;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* A thread-safe and atomic Partition implementation.
*
@ -305,7 +307,7 @@ public final class AtomicBTreePartition extends AbstractBTreePartition
while (TRACKER_PESSIMISTIC_LOCKING != (oldTrackerValue = wasteTracker))
{
// Note this time value has an arbitrary offset, but is a constant rate 32 bit counter (that may wrap)
int time = (int) (System.nanoTime() >>> CLOCK_SHIFT);
int time = (int) (nanoTime() >>> CLOCK_SHIFT);
int delta = oldTrackerValue - time;
if (oldTrackerValue == TRACKER_NEVER_WASTED || delta >= 0 || delta < -EXCESS_WASTE_OFFSET)
delta = -EXCESS_WASTE_OFFSET;

9
src/java/org/apache/cassandra/db/repair/PendingAntiCompaction.java
View File

@ -57,6 +57,7 @@ import org.apache.cassandra.utils.concurrent.Refs;
import static org.apache.cassandra.service.ActiveRepairService.NO_PENDING_REPAIR;
import static org.apache.cassandra.service.ActiveRepairService.UNREPAIRED_SSTABLE;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Performs an anti compaction on a set of tables and token ranges, isolating the unrepaired sstables
@ -218,7 +219,7 @@ public class PendingAntiCompaction
logger.debug("acquiring sstables for pending anti compaction on session {}", sessionID);
// try to modify after cancelling running compactions. This will attempt to cancel in flight compactions including the given sstables for
// up to a minute, after which point, null will be returned
long start = System.currentTimeMillis();
long start = currentTimeMillis();
long delay = TimeUnit.SECONDS.toMillis(acquireRetrySeconds);
// Note that it is `predicate` throwing SSTableAcquisitionException if it finds a conflicting sstable
// and we only retry when runWithCompactionsDisabled throws when uses the predicate, not when acquireTuple is.
@ -238,10 +239,10 @@ public class PendingAntiCompaction
sessionID,
e.getMessage(),
acquireSleepMillis,
TimeUnit.SECONDS.convert(delay + start - System.currentTimeMillis(), TimeUnit.MILLISECONDS));
TimeUnit.SECONDS.convert(delay + start - currentTimeMillis(), TimeUnit.MILLISECONDS));
Uninterruptibles.sleepUninterruptibly(acquireSleepMillis, TimeUnit.MILLISECONDS);
if (System.currentTimeMillis() - start > delay)
if (currentTimeMillis() - start > delay)
logger.warn("{} Timed out waiting to acquire sstables", sessionID, e);
}
@ -250,7 +251,7 @@ public class PendingAntiCompaction
logger.error("Got exception disabling compactions for session {}", sessionID, t);
throw t;
}
} while (System.currentTimeMillis() - start < delay);
} while (currentTimeMillis() - start < delay);
return null;
}
}

8
src/java/org/apache/cassandra/db/view/TableViews.java
View File

@ -41,6 +41,8 @@ import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.btree.BTreeSet;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Groups all the views for a given table.
@ -146,13 +148,13 @@ public class TableViews extends AbstractCollection<View>
// Read modified rows
int nowInSec = FBUtilities.nowInSeconds();
long queryStartNanoTime = System.nanoTime();
long queryStartNanoTime = nanoTime();
SinglePartitionReadCommand command = readExistingRowsCommand(update, views, nowInSec);
if (command == null)
return;
ColumnFamilyStore cfs = Keyspace.openAndGetStore(update.metadata());
long start = System.nanoTime();
long start = nanoTime();
Collection<Mutation> mutations;
try (ReadExecutionController orderGroup = command.executionController();
UnfilteredRowIterator existings = UnfilteredPartitionIterators.getOnlyElement(command.executeLocally(orderGroup), command);
@ -160,7 +162,7 @@ public class TableViews extends AbstractCollection<View>
{
mutations = Iterators.getOnlyElement(generateViewUpdates(views, updates, existings, nowInSec, false));
}
Keyspace.openAndGetStore(update.metadata()).metric.viewReadTime.update(System.nanoTime() - start, TimeUnit.NANOSECONDS);
Keyspace.openAndGetStore(update.metadata()).metric.viewReadTime.update(nanoTime() - start, TimeUnit.NANOSECONDS);
if (!mutations.isEmpty())
StorageProxy.mutateMV(update.partitionKey().getKey(), mutations, writeCommitLog, baseComplete, queryStartNanoTime);

4
src/java/org/apache/cassandra/db/view/ViewBuilderTask.java
View File

@ -63,6 +63,8 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.cassandra.utils.concurrent.Refs;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class ViewBuilderTask extends CompactionInfo.Holder implements Callable<Long>
{
private static final Logger logger = LoggerFactory.getLogger(ViewBuilderTask.class);
@ -115,7 +117,7 @@ public class ViewBuilderTask extends CompactionInfo.Holder implements Callable<L
.generateViewUpdates(Collections.singleton(view), data, empty, nowInSec, true);
AtomicLong noBase = new AtomicLong(Long.MAX_VALUE);
mutations.forEachRemaining(m -> StorageProxy.mutateMV(key.getKey(), m, true, noBase, System.nanoTime()));
mutations.forEachRemaining(m -> StorageProxy.mutateMV(key.getKey(), m, true, noBase, nanoTime()));
}
}

6
src/java/org/apache/cassandra/db/virtual/AbstractVirtualTable.java
View File

@ -38,6 +38,8 @@ import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.schema.TableMetadata;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* An abstract virtual table implementation that builds the resultset on demand.
*/
@ -81,7 +83,7 @@ public abstract class AbstractVirtualTable implements VirtualTable
if (null == partition)
return EmptyIterators.unfilteredPartition(metadata);
long now = System.currentTimeMillis();
long now = currentTimeMillis();
UnfilteredRowIterator rowIterator = partition.toRowIterator(metadata(), clusteringIndexFilter, columnFilter, now);
return new SingletonUnfilteredPartitionIterator(rowIterator);
}
@ -96,7 +98,7 @@ public abstract class AbstractVirtualTable implements VirtualTable
Iterator<Partition> iterator = data.getPartitions(dataRange);
long now = System.currentTimeMillis();
long now = currentTimeMillis();
return new AbstractUnfilteredPartitionIterator()
{

4
src/java/org/apache/cassandra/diag/DiagnosticEvent.java
View File

@ -20,6 +20,8 @@ package org.apache.cassandra.diag;
import java.io.Serializable;
import java.util.Map;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Base class for internally emitted events used for diagnostics and testing.
*/
@ -28,7 +30,7 @@ public abstract class DiagnosticEvent
/**
* Event creation time.
*/
public final long timestamp = System.currentTimeMillis();
public final long timestamp = currentTimeMillis();
/**
* Name of allocating thread.

6
src/java/org/apache/cassandra/diag/LastEventIdBroadcaster.java
View File

@ -33,6 +33,8 @@ import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.progress.jmx.JMXBroadcastExecutor;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Broadcaster for notifying JMX clients on newly available data. Periodically sends {@link Notification}s
* containing a list of event types and greatest event IDs. Consumers may use this information to
@ -100,7 +102,7 @@ final class LastEventIdBroadcaster extends NotificationBroadcasterSupport implem
{
// ensure monotonic properties of ids
if (summary.compute(key, (k, v) -> v == null ? id : id.compareTo(v) > 0 ? id : v) == id) {
summary.put("last_updated_at", System.currentTimeMillis());
summary.put("last_updated_at", currentTimeMillis());
scheduleBroadcast();
}
}
@ -132,7 +134,7 @@ final class LastEventIdBroadcaster extends NotificationBroadcasterSupport implem
Notification notification = new Notification("event_last_id_summary",
"LastEventIdBroadcaster",
notificationSerialNumber.incrementAndGet(),
System.currentTimeMillis(),
currentTimeMillis(),
"Event last IDs summary");
notification.setUserData(summary);
sendNotification(notification);

6
src/java/org/apache/cassandra/gms/EndpointState.java
View File

@ -33,6 +33,8 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.CassandraVersion;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* This abstraction represents both the HeartBeatState and the ApplicationState in an EndpointState
* instance. Any state for a given endpoint can be retrieved from this instance.
@ -66,7 +68,7 @@ public class EndpointState
{
hbState = initialHbState;
applicationState = new AtomicReference<Map<ApplicationState, VersionedValue>>(new EnumMap<>(states));
updateTimestamp = System.nanoTime();
updateTimestamp = nanoTime();
isAlive = true;
}
@ -173,7 +175,7 @@ public class EndpointState
void updateTimestamp()
{
updateTimestamp = System.nanoTime();
updateTimestamp = nanoTime();
}
public boolean isAlive()

3
src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java
View File

@ -29,6 +29,7 @@ import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import static org.apache.cassandra.net.Verb.GOSSIP_DIGEST_ACK2;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class GossipDigestAckVerbHandler extends GossipVerbHandler<GossipDigestAck>
{
@ -68,7 +69,7 @@ public class GossipDigestAckVerbHandler extends GossipVerbHandler<GossipDigestAc
// Ignore any GossipDigestAck messages that we handle before a regular GossipDigestSyn has been send.
// This will prevent Acks from leaking over from the shadow round that are not actual part of
// the regular gossip conversation.
if ((System.nanoTime() - Gossiper.instance.firstSynSendAt) < 0 || Gossiper.instance.firstSynSendAt == 0)
if ((nanoTime() - Gossiper.instance.firstSynSendAt) < 0 || Gossiper.instance.firstSynSendAt == 0)
{
if (logger.isTraceEnabled())
logger.trace("Ignoring unrequested GossipDigestAck from {}", from);

24
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -68,6 +68,8 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.GOSSIPER_Q
import static org.apache.cassandra.net.NoPayload.noPayload;
import static org.apache.cassandra.net.Verb.ECHO_REQ;
import static org.apache.cassandra.net.Verb.GOSSIP_DIGEST_SYN;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* This module is responsible for Gossiping information for the local endpoint. This abstraction
@ -158,7 +160,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
// endpoint states as gathered during shadow round
private final Map<InetAddressAndPort, EndpointState> endpointShadowStateMap = new ConcurrentHashMap<>();
private volatile long lastProcessedMessageAt = System.currentTimeMillis();
private volatile long lastProcessedMessageAt = currentTimeMillis();
/**
* This property is initially set to {@code true} which means that we have no information about the other nodes.
@ -448,7 +450,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
{
Long downtime = unreachableEndpoints.get(ep);
if (downtime != null)
return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - downtime);
return TimeUnit.NANOSECONDS.toMillis(nanoTime() - downtime);
else
return 0L;
}
@ -620,7 +622,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
*/
private void quarantineEndpoint(InetAddressAndPort endpoint)
{
quarantineEndpoint(endpoint, System.currentTimeMillis());
quarantineEndpoint(endpoint, currentTimeMillis());
}
/**
@ -643,7 +645,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
{
// remember, quarantineEndpoint will effectively already add QUARANTINE_DELAY, so this is 2x
logger.debug("");
quarantineEndpoint(endpoint, System.currentTimeMillis() + QUARANTINE_DELAY);
quarantineEndpoint(endpoint, currentTimeMillis() + QUARANTINE_DELAY);
GossiperDiagnostics.replacementQuarantine(this, endpoint);
}
@ -777,7 +779,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
if (epState == null)
{
epState = new EndpointState(new HeartBeatState((int) ((System.currentTimeMillis() + 60000) / 1000), 9999));
epState = new EndpointState(new HeartBeatState((int) ((currentTimeMillis() + 60000) / 1000), 9999));
}
else
{
@ -851,7 +853,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
if (logger.isTraceEnabled())
logger.trace("Sending a GossipDigestSyn to {} ...", to);
if (firstSynSendAt == 0)
firstSynSendAt = System.nanoTime();
firstSynSendAt = nanoTime();
MessagingService.instance().send(message, to);
boolean isSeed = seeds.contains(to);
@ -985,8 +987,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
if (logger.isTraceEnabled())
logger.trace("Performing status check ...");
long now = System.currentTimeMillis();
long nowNano = System.nanoTime();
long now = currentTimeMillis();
long nowNano = nanoTime();
long pending = ((JMXEnabledThreadPoolExecutor) Stage.GOSSIP.executor()).metrics.pendingTasks.getValue();
if (pending > 0 && lastProcessedMessageAt < now - 1000)
@ -1307,7 +1309,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
{
localState.markDead();
liveEndpoints.remove(addr);
unreachableEndpoints.put(addr, System.nanoTime());
unreachableEndpoints.put(addr, nanoTime());
}
/**
@ -1448,7 +1450,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
{
int localGeneration = localEpStatePtr.getHeartBeatState().getGeneration();
int remoteGeneration = remoteState.getHeartBeatState().getGeneration();
long localTime = System.currentTimeMillis()/1000;
long localTime = currentTimeMillis() / 1000;
if (logger.isTraceEnabled())
logger.trace("{} local generation {}, remote generation {}", ep, localGeneration, remoteGeneration);
@ -2125,7 +2127,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
public static long computeExpireTime()
{
return System.currentTimeMillis() + Gossiper.aVeryLongTime;
return currentTimeMillis() + aVeryLongTime;
}
@Nullable

4
src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
View File

@ -54,7 +54,7 @@ import org.apache.cassandra.dht.*;
import org.apache.cassandra.hadoop.*;
import org.apache.cassandra.utils.*;
import static java.util.stream.Collectors.toMap;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Hadoop InputFormat allowing map/reduce against Cassandra rows within one ColumnFamily.
@ -242,7 +242,7 @@ public class CqlInputFormat extends org.apache.hadoop.mapreduce.InputFormat<Long
}
assert splits.size() > 0;
Collections.shuffle(splits, new Random(System.nanoTime()));
Collections.shuffle(splits, new Random(nanoTime()));
return splits;
}

3
src/java/org/apache/cassandra/hints/Hint.java
View File

@ -39,6 +39,7 @@ import org.assertj.core.util.VisibleForTesting;
import static org.apache.cassandra.db.TypeSizes.sizeof;
import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Encapsulates the hinted mutation, its creation time, and the gc grace seconds param for each table involved.
@ -135,7 +136,7 @@ public final class Hint
*/
public boolean isLive()
{
return isLive(creationTime, System.currentTimeMillis(), ttl());
return isLive(creationTime, currentTimeMillis(), ttl());
}
static boolean isLive(long creationTime, long now, int hintTTL)

4
src/java/org/apache/cassandra/hints/HintsCleanupTrigger.java
View File

@ -25,6 +25,8 @@ import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.StorageService;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Delete the expired orphaned hints files.
* An orphaned file is considered as no associating endpoint with its host ID.
@ -62,7 +64,7 @@ final class HintsCleanupTrigger implements Runnable
// Interrupt the dispatch if any. At this step, it is certain that the hintsStore is orphaned.
dispatchExecutor.interruptDispatch(hintsStore.hostId);
Runnable cleanup = () -> hintsStore.deleteExpiredHints(System.currentTimeMillis());
Runnable cleanup = () -> hintsStore.deleteExpiredHints(currentTimeMillis());
ScheduledExecutors.optionalTasks.execute(cleanup);
}
}

6
src/java/org/apache/cassandra/hints/HintsReader.java
View File

@ -36,6 +36,8 @@ import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.AbstractIterator;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* A paged non-compressed hints reader that provides two iterators:
* - a 'raw' ByteBuffer iterator that doesn't deserialize the hints, but returns the pre-encoded hints verbatim
@ -164,7 +166,7 @@ class HintsReader implements AutoCloseable, Iterable<HintsReader.Page>
final class HintsIterator extends AbstractIterator<Hint>
{
private final InputPosition offset;
private final long now = System.currentTimeMillis();
private final long now = currentTimeMillis();
HintsIterator(InputPosition offset)
{
@ -270,7 +272,7 @@ class HintsReader implements AutoCloseable, Iterable<HintsReader.Page>
final class BuffersIterator extends AbstractIterator<ByteBuffer>
{
private final InputPosition offset;
private final long now = System.currentTimeMillis();
private final long now = currentTimeMillis();
BuffersIterator(InputPosition offset)
{

4
src/java/org/apache/cassandra/hints/HintsStore.java
View File

@ -38,6 +38,8 @@ import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.SyncUtil;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Encapsulates the state of a peer's hints: the queue of hints files for dispatch, and the current writer (if any).
*
@ -247,7 +249,7 @@ final class HintsStore
private HintsWriter openWriter()
{
lastUsedTimestamp = Math.max(System.currentTimeMillis(), lastUsedTimestamp + 1);
lastUsedTimestamp = Math.max(currentTimeMillis(), lastUsedTimestamp + 1);
HintsDescriptor descriptor = new HintsDescriptor(hostId, lastUsedTimestamp, writerParams);
try

10
src/java/org/apache/cassandra/index/sasi/disk/PerSSTableIndexWriter.java
View File

@ -52,6 +52,8 @@ import com.google.common.util.concurrent.Uninterruptibles;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class PerSSTableIndexWriter implements SSTableFlushObserver
{
private static final Logger logger = LoggerFactory.getLogger(PerSSTableIndexWriter.class);
@ -251,7 +253,7 @@ public class PerSSTableIndexWriter implements SSTableFlushObserver
final String segmentFile = filename(isFinal);
return () -> {
long start = System.nanoTime();
long start = nanoTime();
try
{
@ -266,7 +268,7 @@ public class PerSSTableIndexWriter implements SSTableFlushObserver
finally
{
if (!isFinal)
logger.info("Flushed index segment {}, took {} ms.", segmentFile, TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
logger.info("Flushed index segment {}, took {} ms.", segmentFile, TimeUnit.NANOSECONDS.toMillis(nanoTime() - start));
}
};
}
@ -276,7 +278,7 @@ public class PerSSTableIndexWriter implements SSTableFlushObserver
logger.info("Scheduling index flush to {}", outputFile);
getExecutor().submit((Runnable) () -> {
long start1 = System.nanoTime();
long start1 = nanoTime();
OnDiskIndex[] parts = new OnDiskIndex[segments.size() + 1];
@ -324,7 +326,7 @@ public class PerSSTableIndexWriter implements SSTableFlushObserver
}
finally
{
logger.info("Index flush to {} took {} ms.", outputFile, TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start1));
logger.info("Index flush to {} took {} ms.", outputFile, TimeUnit.NANOSECONDS.toMillis(nanoTime() - start1));
for (int segment = 0; segment < segmentNumber; segment++)
{

6
src/java/org/apache/cassandra/index/sasi/plan/QueryController.java
View File

@ -49,6 +49,8 @@ import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.Pair;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class QueryController
{
private final long executionQuota;
@ -65,7 +67,7 @@ public class QueryController
this.command = command;
this.range = command.dataRange();
this.executionQuota = TimeUnit.MILLISECONDS.toNanos(timeQuotaMs);
this.executionStart = System.nanoTime();
this.executionStart = nanoTime();
}
public TableMetadata metadata()
@ -154,7 +156,7 @@ public class QueryController
public void checkpoint()
{
long executionTime = (System.nanoTime() - executionStart);
long executionTime = (nanoTime() - executionStart);
if (executionTime >= executionQuota)
throw new TimeQuotaExceededException(

4
src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
View File

@ -54,6 +54,8 @@ import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Utility to write SSTables.
* <p>
@ -239,7 +241,7 @@ public class CQLSSTableWriter implements Closeable
List<ByteBuffer> keys = insert.buildPartitionKeyNames(options);
SortedSet<Clustering<?>> clusterings = insert.createClustering(options);
long now = System.currentTimeMillis();
long now = currentTimeMillis();
// Note that we asks indexes to not validate values (the last 'false' arg below) because that triggers a 'Keyspace.open'
// and that forces a lot of initialization that we don't want.
UpdateParameters params = new UpdateParameters(insert.metadata,

12
src/java/org/apache/cassandra/io/sstable/format/SSTableReaderBuilder.java
View File

@ -47,6 +47,10 @@ import java.nio.file.Paths;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import static org.apache.cassandra.io.sstable.format.SSTableReader.OpenReason.NORMAL;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public abstract class SSTableReaderBuilder
{
private static final Logger logger = LoggerFactory.getLogger(SSTableReaderBuilder.class);
@ -270,7 +274,7 @@ public abstract class SSTableReaderBuilder
StatsMetadata statsMetadata,
SerializationHeader header)
{
super(descriptor, metadataRef, System.currentTimeMillis(), components, statsMetadata, SSTableReader.OpenReason.NORMAL, header);
super(descriptor, metadataRef, currentTimeMillis(), components, statsMetadata, NORMAL, header);
}
@Override
@ -338,7 +342,7 @@ public abstract class SSTableReaderBuilder
StatsMetadata statsMetadata,
SerializationHeader header)
{
super(descriptor, metadataRef, System.currentTimeMillis(), components, statsMetadata, SSTableReader.OpenReason.NORMAL, header);
super(descriptor, metadataRef, currentTimeMillis(), components, statsMetadata, NORMAL, header);
this.validationMetadata = validationMetadata;
this.isOffline = isOffline;
}
@ -353,9 +357,9 @@ public abstract class SSTableReaderBuilder
try
{
// load index and filter
long start = System.nanoTime();
long start = nanoTime();
load(validationMetadata, isOffline, components, DatabaseDescriptor.getDiskOptimizationStrategy(), statsMetadata);
logger.trace("INDEX LOAD TIME for {}: {} ms.", descriptor, TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
logger.trace("INDEX LOAD TIME for {}: {} ms.", descriptor, TimeUnit.NANOSECONDS.toMillis(nanoTime() - start));
}
catch (IOException t)
{

4
src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java
View File

@ -50,6 +50,8 @@ import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.Transactional;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public class BigTableWriter extends SSTableWriter
{
private static final Logger logger = LoggerFactory.getLogger(BigTableWriter.class);
@ -384,7 +386,7 @@ public class BigTableWriter extends SSTableWriter
private SSTableReader openFinal(SSTableReader.OpenReason openReason)
{
if (maxDataAge < 0)
maxDataAge = System.currentTimeMillis();
maxDataAge = currentTimeMillis();
StatsMetadata stats = statsMetadata();
// finalize in-memory state for the reader

5
src/java/org/apache/cassandra/locator/TokenMetadata.java
View File

@ -48,6 +48,7 @@ import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.SortedBiMultiValMap;
import static org.apache.cassandra.config.CassandraRelevantProperties.LINE_SEPARATOR;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public class TokenMetadata
{
@ -855,7 +856,7 @@ public class TokenMetadata
public void calculatePendingRanges(AbstractReplicationStrategy strategy, String keyspaceName)
{
// avoid race between both branches - do not use a lock here as this will block any other unrelated operations!
long startedAt = System.currentTimeMillis();
long startedAt = currentTimeMillis();
synchronized (pendingRanges)
{
TokenMetadataDiagnostics.pendingRangeCalculationStarted(this, keyspaceName);
@ -899,7 +900,7 @@ public class TokenMetadata
if (logger.isDebugEnabled())
logger.debug("Starting pending range calculation for {}", keyspaceName);
long took = System.currentTimeMillis() - startedAt;
long took = currentTimeMillis() - startedAt;
if (logger.isDebugEnabled())
logger.debug("Pending range calculation for {} completed (took: {}ms)", keyspaceName, took);

5
src/java/org/apache/cassandra/metrics/TableMetrics.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.metrics;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import java.nio.ByteBuffer;
import java.util.*;
@ -1291,12 +1292,12 @@ public class TableMetrics
private Context(Timer [] all)
{
this.all = all;
start = System.nanoTime();
start = nanoTime();
}
public void close()
{
long duration = System.nanoTime() - start;
long duration = nanoTime() - start;
for (Timer t : all)
t.update(duration, TimeUnit.NANOSECONDS);
}

5
src/java/org/apache/cassandra/net/AsyncPromise.java
View File

@ -36,6 +36,7 @@ import io.netty.util.internal.ThrowableUtil;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static java.util.concurrent.atomic.AtomicReferenceFieldUpdater.*;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Netty's DefaultPromise uses a mutex to coordinate notifiers AND waiters between the eventLoop and the other threads.
@ -406,7 +407,7 @@ public class AsyncPromise<V> implements Promise<V>
public boolean await(long timeout, TimeUnit unit) throws InterruptedException
{
return await(unit.toNanos(timeout),
(signal, nanos) -> signal.awaitUntil(nanos + System.nanoTime()));
(signal, nanos) -> signal.awaitUntil(nanos + nanoTime()));
}
public boolean await(long timeoutMillis) throws InterruptedException
@ -417,7 +418,7 @@ public class AsyncPromise<V> implements Promise<V>
public boolean awaitUninterruptibly(long timeout, TimeUnit unit)
{
return await(unit.toNanos(timeout),
(signal, nanos) -> signal.awaitUntilUninterruptibly(nanos + System.nanoTime()));
(signal, nanos) -> signal.awaitUntilUninterruptibly(nanos + nanoTime()));
}
public boolean awaitUninterruptibly(long timeoutMillis)

5
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -44,6 +44,7 @@ import org.apache.cassandra.utils.FBUtilities;
import static java.util.Collections.synchronizedList;
import static java.util.concurrent.TimeUnit.MINUTES;
import static org.apache.cassandra.concurrent.Stage.MUTATION;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.Throwables.maybeFail;
/**
@ -449,7 +450,7 @@ public final class MessagingService extends MessagingServiceMBeanImpl
for (OutboundConnections pool : channelManagers.values())
closing.add(pool.close(true));
long deadline = System.nanoTime() + units.toNanos(timeout);
long deadline = nanoTime() + units.toNanos(timeout);
maybeFail(() -> new FutureCombiner(closing).get(timeout, units),
() -> {
List<ExecutorService> inboundExecutors = new ArrayList<>();
@ -473,7 +474,7 @@ public final class MessagingService extends MessagingServiceMBeanImpl
for (OutboundConnections pool : channelManagers.values())
closing.add(pool.close(false));
long deadline = System.nanoTime() + units.toNanos(timeout);
long deadline = nanoTime() + units.toNanos(timeout);
maybeFail(() -> new FutureCombiner(closing).get(timeout, units),
() -> {
if (shutdownExecutors)

3
src/java/org/apache/cassandra/net/RequestCallbacks.java
View File

@ -48,6 +48,7 @@ import static java.lang.String.format;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.concurrent.Stage.INTERNAL_RESPONSE;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.MonotonicClock.preciseTime;
/**
@ -199,7 +200,7 @@ public class RequestCallbacks implements OutboundMessageCallbacks
{
if (!executor.isTerminated())
{
long wait = deadlineNanos - System.nanoTime();
long wait = deadlineNanos - nanoTime();
if (wait <= 0 || !executor.awaitTermination(wait, NANOSECONDS))
throw new TimeoutException();
}

9
src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java
View File

@ -47,6 +47,7 @@ import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.net.Verb.PING_REQ;
import static org.apache.cassandra.net.ConnectionType.LARGE_MESSAGES;
import static org.apache.cassandra.net.ConnectionType.SMALL_MESSAGES;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class StartupClusterConnectivityChecker
{
@ -125,7 +126,7 @@ public class StartupClusterConnectivityChecker
new CountDownLatch(Math.max(datacenterToPeers.get(datacenter).size() - 1, 0)));
}
long startNanos = System.nanoTime();
long startNanos = nanoTime();
// set up a listener to react to new nodes becoming alive (in gossip), and account for all the nodes that are already alive
Set<InetAddressAndPort> alivePeers = Collections.newSetFromMap(new ConcurrentHashMap<>());
@ -150,7 +151,7 @@ public class StartupClusterConnectivityChecker
boolean succeeded = true;
for (CountDownLatch countDownLatch : dcToRemainingPeers.values())
{
long remainingNanos = Math.max(1, timeoutNanos - (System.nanoTime() - startNanos));
long remainingNanos = Math.max(1, timeoutNanos - (nanoTime() - startNanos));
//noinspection UnstableApiUsage
succeeded &= Uninterruptibles.awaitUninterruptibly(countDownLatch, remainingNanos, TimeUnit.NANOSECONDS);
}
@ -164,12 +165,12 @@ public class StartupClusterConnectivityChecker
if (succeeded)
{
logger.info("Ensured sufficient healthy connections with {} after {} milliseconds",
numDown.keySet(), TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos));
numDown.keySet(), TimeUnit.NANOSECONDS.toMillis(nanoTime() - startNanos));
}
else
{
logger.warn("Timed out after {} milliseconds, was waiting for remaining peers to connect: {}",
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos), numDown);
TimeUnit.NANOSECONDS.toMillis(nanoTime() - startNanos), numDown);
}
return succeeded;

10
src/java/org/apache/cassandra/repair/RepairJob.java
View File

@ -46,6 +46,8 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MerkleTrees;
import org.apache.cassandra.utils.Pair;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* RepairJob runs repair on given ColumnFamily.
*/
@ -207,7 +209,7 @@ public class RepairJob extends AbstractFuture<RepairResult> implements Runnable
boolean pullRepair,
PreviewKind previewKind)
{
long startedAt = System.currentTimeMillis();
long startedAt = currentTimeMillis();
List<SyncTask> syncTasks = new ArrayList<>();
// We need to difference all trees one against another
for (int i = 0; i < trees.size() - 1; ++i)
@ -262,7 +264,7 @@ public class RepairJob extends AbstractFuture<RepairResult> implements Runnable
}
trees.get(trees.size() - 1).trees.release();
logger.info("Created {} sync tasks based on {} merkle tree responses for {} (took: {}ms)",
syncTasks.size(), trees.size(), desc.parentSessionId, System.currentTimeMillis() - startedAt);
syncTasks.size(), trees.size(), desc.parentSessionId, currentTimeMillis() - startedAt);
return syncTasks;
}
@ -302,7 +304,7 @@ public class RepairJob extends AbstractFuture<RepairResult> implements Runnable
boolean isIncremental,
PreviewKind previewKind)
{
long startedAt = System.currentTimeMillis();
long startedAt = currentTimeMillis();
List<SyncTask> syncTasks = new ArrayList<>();
// We need to difference all trees one against another
DifferenceHolder diffHolder = new DifferenceHolder(trees);
@ -353,7 +355,7 @@ public class RepairJob extends AbstractFuture<RepairResult> implements Runnable
}
}
logger.info("Created {} optimised sync tasks based on {} merkle tree responses for {} (took: {}ms)",
syncTasks.size(), trees.size(), desc.parentSessionId, System.currentTimeMillis() - startedAt);
syncTasks.size(), trees.size(), desc.parentSessionId, currentTimeMillis() - startedAt);
logger.trace("Optimised sync tasks for {}: {}", desc.parentSessionId, syncTasks);
return syncTasks;
}

14
src/java/org/apache/cassandra/repair/RepairRunnable.java
View File

@ -96,6 +96,10 @@ import org.apache.cassandra.utils.progress.ProgressEventNotifier;
import org.apache.cassandra.utils.progress.ProgressEventType;
import org.apache.cassandra.utils.progress.ProgressListener;
import static org.apache.cassandra.service.QueryState.forInternalCalls;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class RepairRunnable implements Runnable, ProgressEventNotifier
{
private static final Logger logger = LoggerFactory.getLogger(RepairRunnable.class);
@ -109,7 +113,7 @@ public class RepairRunnable implements Runnable, ProgressEventNotifier
private final AtomicInteger progressCounter = new AtomicInteger();
private final int totalProgress;
private final long creationTimeMillis = System.currentTimeMillis();
private final long creationTimeMillis = currentTimeMillis();
private final UUID parentSession = UUIDGen.getTimeUUID();
private final List<ProgressListener> listeners = new ArrayList<>();
@ -217,7 +221,7 @@ public class RepairRunnable implements Runnable, ProgressEventNotifier
private void complete(String msg)
{
long durationMillis = System.currentTimeMillis() - creationTimeMillis;
long durationMillis = currentTimeMillis() - creationTimeMillis;
if (msg == null)
{
String duration = DurationFormatUtils.formatDurationWords(durationMillis, true, true);
@ -804,7 +808,7 @@ public class RepairRunnable implements Runnable, ProgressEventNotifier
int si = 0;
UUID uuid;
long tlast = System.currentTimeMillis(), tcur;
long tlast = currentTimeMillis(), tcur;
TraceState.Status status;
long minWaitMillis = 125;
@ -825,11 +829,11 @@ public class RepairRunnable implements Runnable, ProgressEventNotifier
shouldDouble = false;
}
ByteBuffer tminBytes = ByteBufferUtil.bytes(UUIDGen.minTimeUUID(tlast - 1000));
ByteBuffer tmaxBytes = ByteBufferUtil.bytes(UUIDGen.maxTimeUUID(tcur = System.currentTimeMillis()));
ByteBuffer tmaxBytes = ByteBufferUtil.bytes(UUIDGen.maxTimeUUID(tcur = currentTimeMillis()));
QueryOptions options = QueryOptions.forInternalCalls(ConsistencyLevel.ONE, Lists.newArrayList(sessionIdBytes,
tminBytes,
tmaxBytes));
ResultMessage.Rows rows = statement.execute(QueryState.forInternalCalls(), options, System.nanoTime());
ResultMessage.Rows rows = statement.execute(forInternalCalls(), options, nanoTime());
UntypedResultSet result = UntypedResultSet.create(rows.result);
for (UntypedResultSet.Row r : result)

6
src/java/org/apache/cassandra/repair/SyncTask.java
View File

@ -35,6 +35,8 @@ import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.streaming.PreviewKind;
import org.apache.cassandra.tracing.Tracing;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public abstract class SyncTask extends AbstractFuture<SyncStat> implements Runnable
{
private static Logger logger = LoggerFactory.getLogger(SyncTask.class);
@ -70,7 +72,7 @@ public abstract class SyncTask extends AbstractFuture<SyncStat> implements Runna
*/
public final void run()
{
startTime = System.currentTimeMillis();
startTime = currentTimeMillis();
// choose a repair method based on the significance of the difference
@ -97,6 +99,6 @@ public abstract class SyncTask extends AbstractFuture<SyncStat> implements Runna
protected void finished()
{
if (startTime != Long.MIN_VALUE)
Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily).metric.repairSyncTime.update(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS);
Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily).metric.repairSyncTime.update(currentTimeMillis() - startTime, TimeUnit.MILLISECONDS);
}
}

6
src/java/org/apache/cassandra/repair/ValidationManager.java
View File

@ -39,6 +39,8 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTrees;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class ValidationManager
{
private static final Logger logger = LoggerFactory.getLogger(ValidationManager.class);
@ -107,7 +109,7 @@ public class ValidationManager
// Create Merkle trees suitable to hold estimated partitions for the given ranges.
// We blindly assume that a partition is evenly distributed on all sstables for now.
long start = System.nanoTime();
long start = nanoTime();
long partitionCount = 0;
long estimatedTotalBytes = 0;
try (ValidationPartitionIterator vi = getValidationIterator(cfs.getRepairManager(), validator))
@ -140,7 +142,7 @@ public class ValidationManager
}
if (logger.isDebugEnabled())
{
long duration = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
long duration = TimeUnit.NANOSECONDS.toMillis(nanoTime() - start);
logger.debug("Validation of {} partitions (~{}) finished in {} msec, for {}",
partitionCount,
FBUtilities.prettyPrintMemory(estimatedTotalBytes),

14
src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
View File

@ -53,6 +53,8 @@ import org.apache.cassandra.repair.messages.PrepareConsistentRequest;
import org.apache.cassandra.repair.messages.RepairMessage;
import org.apache.cassandra.service.ActiveRepairService;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Coordinator side logic and state of a consistent repair session. Like {@link ActiveRepairService.ParentRepairSession},
* there is only one {@code CoordinatorSession} per user repair command, regardless of the number of tables and token
@ -290,7 +292,7 @@ public class CoordinatorSession extends ConsistentSession
{
logger.info("Beginning coordination of incremental repair session {}", sessionID);
sessionStart = System.currentTimeMillis();
sessionStart = currentTimeMillis();
ListenableFuture<Boolean> prepareResult = prepare();
// run repair sessions normally
@ -300,7 +302,7 @@ public class CoordinatorSession extends ConsistentSession
{
if (success)
{
repairStart = System.currentTimeMillis();
repairStart = currentTimeMillis();
if (logger.isDebugEnabled())
{
logger.debug("Incremental repair {} prepare phase completed in {}", sessionID, formatDuration(sessionStart, repairStart));
@ -323,7 +325,7 @@ public class CoordinatorSession extends ConsistentSession
{
if (results == null || results.isEmpty() || Iterables.any(results, r -> r == null))
{
finalizeStart = System.currentTimeMillis();
finalizeStart = currentTimeMillis();
if (logger.isDebugEnabled())
{
logger.debug("Incremental repair {} validation/stream phase completed in {}", sessionID, formatDuration(repairStart, finalizeStart));
@ -352,12 +354,12 @@ public class CoordinatorSession extends ConsistentSession
{
if (logger.isDebugEnabled())
{
logger.debug("Incremental repair {} finalization phase completed in {}", sessionID, formatDuration(finalizeStart, System.currentTimeMillis()));
logger.debug("Incremental repair {} finalization phase completed in {}", sessionID, formatDuration(finalizeStart, currentTimeMillis()));
}
finalizeCommit();
if (logger.isDebugEnabled())
{
logger.debug("Incremental repair {} phase completed in {}", sessionID, formatDuration(sessionStart, System.currentTimeMillis()));
logger.debug("Incremental repair {} phase completed in {}", sessionID, formatDuration(sessionStart, currentTimeMillis()));
}
}
else
@ -379,7 +381,7 @@ public class CoordinatorSession extends ConsistentSession
{
if (logger.isDebugEnabled())
{
logger.debug("Incremental repair {} phase failed in {}", sessionID, formatDuration(sessionStart, System.currentTimeMillis()));
logger.debug("Incremental repair {} phase failed in {}", sessionID, formatDuration(sessionStart, currentTimeMillis()));
}
hasFailure.set(true);
fail();

4
src/java/org/apache/cassandra/schema/MigrationCoordinator.java
View File

@ -61,6 +61,8 @@ import org.apache.cassandra.net.Verb;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class MigrationCoordinator
{
private static final Logger logger = LoggerFactory.getLogger(MigrationCoordinator.class);
@ -581,7 +583,7 @@ public class MigrationCoordinator
signal = WaitQueue.all(signals);
}
return signal.awaitUntil(System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(waitMillis));
return signal.awaitUntil(nanoTime() + TimeUnit.MILLISECONDS.toNanos(waitMillis));
}
catch (InterruptedException e)
{

4
src/java/org/apache/cassandra/serializers/TimestampSerializer.java
View File

@ -35,6 +35,8 @@ import java.util.List;
import java.util.TimeZone;
import java.util.regex.Pattern;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public class TimestampSerializer extends TypeSerializer<Date>
{
@ -139,7 +141,7 @@ public class TimestampSerializer extends TypeSerializer<Date>
public static long dateStringToTimestamp(String source) throws MarshalException
{
if (source.equalsIgnoreCase("now"))
return System.currentTimeMillis();
return currentTimeMillis();
// Milliseconds since epoch?
if (timestampPattern.matcher(source).matches())

5
src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java
View File

@ -45,6 +45,7 @@ import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public abstract class AbstractWriteResponseHandler<T> implements RequestCallback<T>
@ -130,7 +131,7 @@ public abstract class AbstractWriteResponseHandler<T> implements RequestCallback
long requestTimeout = writeType == WriteType.COUNTER
? DatabaseDescriptor.getCounterWriteRpcTimeout(NANOSECONDS)
: DatabaseDescriptor.getWriteRpcTimeout(NANOSECONDS);
return requestTimeout - (System.nanoTime() - queryStartNanoTime);
return requestTimeout - (nanoTime() - queryStartNanoTime);
}
/**
@ -287,7 +288,7 @@ public abstract class AbstractWriteResponseHandler<T> implements RequestCallback
}
else
{
replicaPlan.keyspace().metric.idealCLWriteLatency.addNano(System.nanoTime() - queryStartNanoTime);
replicaPlan.keyspace().metric.idealCLWriteLatency.addNano(nanoTime() - queryStartNanoTime);
}
}
}

3
src/java/org/apache/cassandra/service/ActiveRepairService.java
View File

@ -96,6 +96,7 @@ import org.apache.cassandra.utils.UUIDGen;
import static com.google.common.collect.Iterables.concat;
import static com.google.common.collect.Iterables.transform;
import static org.apache.cassandra.net.Verb.PREPARE_MSG;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* ActiveRepairService is the starting point for manual "active" repairs.
@ -511,7 +512,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
// end up skipping replicas
if (options.isIncremental() && options.isGlobal() && ! force)
{
return System.currentTimeMillis();
return currentTimeMillis();
}
else
{

6
src/java/org/apache/cassandra/service/ClientState.java
View File

@ -50,6 +50,8 @@ import org.apache.cassandra.schema.SchemaKeyspace;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* State related to a client connection.
*/
@ -198,7 +200,7 @@ public class ClientState
{
while (true)
{
long current = System.currentTimeMillis() * 1000;
long current = currentTimeMillis() * 1000;
long last = lastTimestampMicros.get();
long tstamp = last >= current ? last + 1 : current;
if (lastTimestampMicros.compareAndSet(last, tstamp))
@ -252,7 +254,7 @@ public class ClientState
{
while (true)
{
long current = Math.max(System.currentTimeMillis() * 1000, minTimestampToUse);
long current = Math.max(currentTimeMillis() * 1000, minTimestampToUse);
long last = lastTimestampMicros.get();
long tstamp = last >= current ? last + 1 : current;
// Note that if we ended up picking minTimestampMicrosToUse (it was "in the future"), we don't

6
src/java/org/apache/cassandra/service/GCInspector.java
View File

@ -50,6 +50,8 @@ import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.StatusLogger;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class GCInspector implements NotificationListener, GCInspectorMXBean
{
public static final String MBEAN_NAME = "org.apache.cassandra.service:type=GCInspector";
@ -113,7 +115,7 @@ public class GCInspector implements NotificationListener, GCInspectorMXBean
State()
{
count = maxRealTimeElapsed = sumSquaresRealTimeElapsed = totalRealTimeElapsed = totalBytesReclaimed = 0;
startNanos = System.nanoTime();
startNanos = nanoTime();
}
}
@ -318,7 +320,7 @@ public class GCInspector implements NotificationListener, GCInspectorMXBean
{
State state = getTotalSinceLastCheck();
double[] r = new double[7];
r[0] = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - state.startNanos);
r[0] = TimeUnit.NANOSECONDS.toMillis(nanoTime() - state.startNanos);
r[1] = state.maxRealTimeElapsed;
r[2] = state.totalRealTimeElapsed;
r[3] = state.sumSquaresRealTimeElapsed;

6
src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java
View File

@ -34,6 +34,8 @@ import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.annotations.VisibleForTesting;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public class PendingRangeCalculatorService
{
public static final PendingRangeCalculatorService instance = new PendingRangeCalculatorService();
@ -71,12 +73,12 @@ public class PendingRangeCalculatorService
try
{
PendingRangeCalculatorServiceDiagnostics.taskStarted(instance, updateJobs);
long start = System.currentTimeMillis();
long start = currentTimeMillis();
List<String> keyspaces = Schema.instance.getNonLocalStrategyKeyspaces();
for (String keyspaceName : keyspaces)
calculatePendingRanges(Keyspace.open(keyspaceName).getReplicationStrategy(), keyspaceName);
if (logger.isTraceEnabled())
logger.trace("Finished PendingRangeTask for {} keyspaces in {}ms", keyspaces.size(), System.currentTimeMillis() - start);
logger.trace("Finished PendingRangeTask for {} keyspaces in {}ms", keyspaces.size(), currentTimeMillis() - start);
PendingRangeCalculatorServiceDiagnostics.taskFinished(instance, updateJobs);
}
finally

3
src/java/org/apache/cassandra/service/StartupChecks.java
View File

@ -60,6 +60,7 @@ import static java.lang.String.format;
import static org.apache.cassandra.config.CassandraRelevantProperties.COM_SUN_MANAGEMENT_JMXREMOTE_PORT;
import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_VERSION;
import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_VM_NAME;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Verifies that the system and environment is in a fit state to be started.
@ -169,7 +170,7 @@ public class StartupChecks
private static final long EARLIEST_LAUNCH_DATE = 1215820800000L;
public void execute() throws StartupException
{
long now = System.currentTimeMillis();
long now = currentTimeMillis();
if (now < EARLIEST_LAUNCH_DATE)
throw new StartupException(StartupException.ERR_WRONG_MACHINE_STATE,
String.format("current machine time is %s, but that is seemingly incorrect. exiting now.",

42
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -147,6 +147,8 @@ import static org.apache.cassandra.net.Verb.TRUNCATE_REQ;
import static org.apache.cassandra.service.BatchlogResponseHandler.BatchlogCleanup;
import static org.apache.cassandra.service.paxos.PrepareVerbHandler.doPrepare;
import static org.apache.cassandra.service.paxos.ProposeVerbHandler.doPropose;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class StorageProxy implements StorageProxyMBean
{
@ -288,7 +290,7 @@ public class StorageProxy implements StorageProxyMBean
long queryStartNanoTime)
throws UnavailableException, IsBootstrappingException, RequestFailureException, RequestTimeoutException, InvalidRequestException, CasWriteUnknownResultException
{
final long startTimeForMetrics = System.nanoTime();
final long startTimeForMetrics = nanoTime();
try
{
TableMetadata metadata = Schema.instance.validateTable(keyspaceName, cfName);
@ -380,7 +382,7 @@ public class StorageProxy implements StorageProxyMBean
}
finally
{
final long latency = System.nanoTime() - startTimeForMetrics;
final long latency = nanoTime() - startTimeForMetrics;
casWriteMetrics.addNano(latency);
writeMetricsMap.get(consistencyForPaxos).addNano(latency);
}
@ -448,7 +450,7 @@ public class StorageProxy implements StorageProxyMBean
consistencyForCommit.validateForCasCommit(latestRs);
long timeoutNanos = DatabaseDescriptor.getCasContentionTimeout(NANOSECONDS);
while (System.nanoTime() - queryStartNanoTime < timeoutNanos)
while (nanoTime() - queryStartNanoTime < timeoutNanos)
{
// for simplicity, we'll do a single liveness check at the start of each attempt
ReplicaPlan.ForPaxosWrite replicaPlan = ReplicaPlans.forPaxos(keyspace, key, consistencyForPaxos);
@ -535,7 +537,7 @@ public class StorageProxy implements StorageProxyMBean
PrepareCallback summary = null;
int contentions = 0;
while (System.nanoTime() - queryStartNanoTime < timeoutNanos)
while (nanoTime() - queryStartNanoTime < timeoutNanos)
{
// We want a timestamp that is guaranteed to be unique for that node (so that the ballot is globally unique), but if we've got a prepare rejected
// already we also want to make sure we pick a timestamp that has a chance to be promised, i.e. one that is greater that the most recently known
@ -806,7 +808,7 @@ public class StorageProxy implements StorageProxyMBean
*
* @param mutations the mutations to be applied across the replicas
* @param consistencyLevel the consistency level for the operation
* @param queryStartNanoTime the value of System.nanoTime() when the query started to be processed
* @param queryStartNanoTime the value of nanoTime() when the query started to be processed
*/
public static void mutate(List<? extends IMutation> mutations, ConsistencyLevel consistencyLevel, long queryStartNanoTime)
throws UnavailableException, OverloadedException, WriteTimeoutException, WriteFailureException
@ -814,7 +816,7 @@ public class StorageProxy implements StorageProxyMBean
Tracing.trace("Determining replicas for mutation");
final String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getLocalDatacenter();
long startTime = System.nanoTime();
long startTime = nanoTime();
List<AbstractWriteResponseHandler<IMutation>> responseHandlers = new ArrayList<>(mutations.size());
WriteType plainWriteType = mutations.size() <= 1 ? WriteType.SIMPLE : WriteType.UNLOGGED_BATCH;
@ -882,7 +884,7 @@ public class StorageProxy implements StorageProxyMBean
}
finally
{
long latency = System.nanoTime() - startTime;
long latency = nanoTime() - startTime;
writeMetrics.addNano(latency);
writeMetricsMap.get(consistencyLevel).addNano(latency);
updateCoordinatorWriteLatencyTableMetric(mutations, latency);
@ -938,7 +940,7 @@ public class StorageProxy implements StorageProxyMBean
* @param mutations the mutations to be applied across the replicas
* @param writeCommitLog if commitlog should be written
* @param baseComplete time from epoch in ms that the local base mutation was(or will be) completed
* @param queryStartNanoTime the value of System.nanoTime() when the query started to be processed
* @param queryStartNanoTime the value of nanoTime() when the query started to be processed
*/
public static void mutateMV(ByteBuffer dataKey, Collection<Mutation> mutations, boolean writeCommitLog, AtomicLong baseComplete, long queryStartNanoTime)
throws UnavailableException, OverloadedException, WriteTimeoutException
@ -946,7 +948,7 @@ public class StorageProxy implements StorageProxyMBean
Tracing.trace("Determining replicas for mutation");
final String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getLocalDatacenter();
long startTime = System.nanoTime();
long startTime = nanoTime();
try
@ -1041,7 +1043,7 @@ public class StorageProxy implements StorageProxyMBean
}
finally
{
viewWriteMetrics.addNano(System.nanoTime() - startTime);
viewWriteMetrics.addNano(nanoTime() - startTime);
}
}
@ -1082,7 +1084,7 @@ public class StorageProxy implements StorageProxyMBean
* @param mutations the Mutations to be applied across the replicas
* @param consistency_level the consistency level for the operation
* @param requireQuorumForRemove at least a quorum of nodes will see update before deleting batchlog
* @param queryStartNanoTime the value of System.nanoTime() when the query started to be processed
* @param queryStartNanoTime the value of nanoTime() when the query started to be processed
*/
public static void mutateAtomically(Collection<Mutation> mutations,
ConsistencyLevel consistency_level,
@ -1091,7 +1093,7 @@ public class StorageProxy implements StorageProxyMBean
throws UnavailableException, OverloadedException, WriteTimeoutException
{
Tracing.trace("Determining replicas for atomic batch");
long startTime = System.nanoTime();
long startTime = nanoTime();
List<WriteResponseHandlerWrapper> wrappers = new ArrayList<>(mutations.size());
@ -1162,7 +1164,7 @@ public class StorageProxy implements StorageProxyMBean
}
finally
{
long latency = System.nanoTime() - startTime;
long latency = nanoTime() - startTime;
writeMetrics.addNano(latency);
writeMetricsMap.get(consistency_level).addNano(latency);
updateCoordinatorWriteLatencyTableMetric(mutations, latency);
@ -1273,7 +1275,7 @@ public class StorageProxy implements StorageProxyMBean
* given the list of write endpoints (either standardWritePerformer for
* standard writes or counterWritePerformer for counter writes).
* @param callback an optional callback to be run if and when the write is
* @param queryStartNanoTime the value of System.nanoTime() when the query started to be processed
* @param queryStartNanoTime the value of nanoTime() when the query started to be processed
*/
public static AbstractWriteResponseHandler<IMutation> performWrite(IMutation mutation,
ConsistencyLevel consistencyLevel,
@ -1330,7 +1332,7 @@ public class StorageProxy implements StorageProxyMBean
ReplicaPlan.ForTokenWrite replicaPlan = ReplicaPlans.forWrite(keyspace, consistencyLevel, liveAndDown, ReplicaPlans.writeAll);
AbstractReplicationStrategy replicationStrategy = replicaPlan.replicationStrategy();
AbstractWriteResponseHandler<IMutation> writeHandler = replicationStrategy.getWriteResponseHandler(replicaPlan, () -> {
long delay = Math.max(0, System.currentTimeMillis() - baseComplete.get());
long delay = Math.max(0, currentTimeMillis() - baseComplete.get());
viewWriteMetrics.viewWriteLatency.update(delay, MILLISECONDS);
}, writeType, queryStartNanoTime);
BatchlogResponseHandler<IMutation> batchHandler = new ViewWriteMetricsWrapped(writeHandler, batchConsistencyLevel.blockFor(replicationStrategy), cleanup, queryStartNanoTime);
@ -1747,7 +1749,7 @@ public class StorageProxy implements StorageProxyMBean
if (group.queries.size() > 1)
throw new InvalidRequestException("SERIAL/LOCAL_SERIAL consistency may only be requested for one partition at a time");
long start = System.nanoTime();
long start = nanoTime();
SinglePartitionReadCommand command = group.queries.get(0);
TableMetadata metadata = command.metadata();
DecoratedKey key = command.partitionKey();
@ -1824,7 +1826,7 @@ public class StorageProxy implements StorageProxyMBean
}
finally
{
long latency = System.nanoTime() - start;
long latency = nanoTime() - start;
readMetrics.addNano(latency);
casReadMetrics.addNano(latency);
readMetricsMap.get(consistencyLevel).addNano(latency);
@ -1838,7 +1840,7 @@ public class StorageProxy implements StorageProxyMBean
private static PartitionIterator readRegular(SinglePartitionReadCommand.Group group, ConsistencyLevel consistencyLevel, long queryStartNanoTime)
throws UnavailableException, ReadFailureException, ReadTimeoutException
{
long start = System.nanoTime();
long start = nanoTime();
try
{
PartitionIterator result = fetchRows(group.queries, consistencyLevel, queryStartNanoTime);
@ -1876,7 +1878,7 @@ public class StorageProxy implements StorageProxyMBean
}
finally
{
long latency = System.nanoTime() - start;
long latency = nanoTime() - start;
readMetrics.addNano(latency);
readMetricsMap.get(consistencyLevel).addNano(latency);
// TODO avoid giving every command the same latency number. Can fix this in CASSADRA-5329
@ -2480,7 +2482,7 @@ public class StorageProxy implements StorageProxyMBean
logger.debug("Discarding hint for endpoint not part of ring: {}", target);
}
logger.trace("Adding hints for {}", validTargets);
HintsService.instance.write(hostIds, Hint.create(mutation, System.currentTimeMillis()));
HintsService.instance.write(hostIds, Hint.create(mutation, currentTimeMillis()));
validTargets.forEach(HintsService.instance.metrics::incrCreatedHints);
// Notify the handler only for CL == ANY
if (responseHandler != null && responseHandler.replicaPlan.consistencyLevel() == ConsistencyLevel.ANY)

8
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -134,6 +134,8 @@ import static org.apache.cassandra.index.SecondaryIndexManager.isIndexColumnFami
import static org.apache.cassandra.net.NoPayload.noPayload;
import static org.apache.cassandra.net.Verb.REPLICATION_DONE_REQ;
import static org.apache.cassandra.schema.MigrationManager.evolveSystemKeyspace;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* This abstraction contains the token/identifier of this node
@ -392,7 +394,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
setGossipTokens(tokens);
Gossiper.instance.forceNewerGeneration();
Gossiper.instance.start((int) (System.currentTimeMillis() / 1000));
Gossiper.instance.start((int) (currentTimeMillis() / 1000));
gossipActive = true;
}
}
@ -725,7 +727,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
initialized = true;
gossipActive = true;
Gossiper.instance.register(this);
Gossiper.instance.start((int) (System.currentTimeMillis() / 1000)); // needed for node-ring gathering.
Gossiper.instance.start((int) (currentTimeMillis() / 1000)); // needed for node-ring gathering.
Gossiper.instance.addLocalApplicationState(ApplicationState.NET_VERSION, valueFactory.networkVersion());
MessagingService.instance().listen();
}
@ -1703,7 +1705,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
if (existing != null)
{
long nanoDelay = schemaDelay * 1000000L;
if (Gossiper.instance.getEndpointStateForEndpoint(existing).getUpdateTimestamp() > (System.nanoTime() - nanoDelay))
if (Gossiper.instance.getEndpointStateForEndpoint(existing).getUpdateTimestamp() > (nanoTime() - nanoDelay))
throw new UnsupportedOperationException("Cannot replace a live node... ");
collisions.add(existing);
}

6
src/java/org/apache/cassandra/service/TruncateResponseHandler.java
View File

@ -34,6 +34,8 @@ import org.apache.cassandra.net.Message;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.config.DatabaseDescriptor.getTruncateRpcTimeout;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class TruncateResponseHandler implements RequestCallback<TruncateResponse>
{
@ -51,12 +53,12 @@ public class TruncateResponseHandler implements RequestCallback<TruncateResponse
assert 1 <= responseCount: "invalid response count " + responseCount;
this.responseCount = responseCount;
start = System.nanoTime();
start = nanoTime();
}
public void get() throws TimeoutException
{
long timeoutNanos = DatabaseDescriptor.getTruncateRpcTimeout(NANOSECONDS) - (System.nanoTime() - start);
long timeoutNanos = getTruncateRpcTimeout(NANOSECONDS) - (nanoTime() - start);
boolean completedInTime;
try
{

6
src/java/org/apache/cassandra/service/pager/AggregationQueryPager.java
View File

@ -29,6 +29,8 @@ import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.service.ClientState;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* {@code QueryPager} that takes care of fetching the pages for aggregation queries.
* <p>
@ -71,9 +73,9 @@ public final class AggregationQueryPager implements QueryPager
public PartitionIterator fetchPageInternal(int pageSize, ReadExecutionController executionController)
{
if (limits.isGroupByLimit())
return new GroupByPartitionIterator(pageSize, executionController, System.nanoTime());
return new GroupByPartitionIterator(pageSize, executionController, nanoTime());
return new AggregationPartitionIterator(pageSize, executionController, System.nanoTime());
return new AggregationPartitionIterator(pageSize, executionController, nanoTime());
}
@Override

4
src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java
View File

@ -30,6 +30,8 @@ import org.apache.cassandra.exceptions.RequestValidationException;
import org.apache.cassandra.exceptions.RequestExecutionException;
import org.apache.cassandra.service.ClientState;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Pager over a list of SinglePartitionReadQuery.
*
@ -158,7 +160,7 @@ public class MultiPartitionPager<T extends SinglePartitionReadQuery> implements
public PartitionIterator fetchPageInternal(int pageSize, ReadExecutionController executionController) throws RequestValidationException, RequestExecutionException
{
int toQuery = Math.min(remaining, pageSize);
return new PagersIterator(toQuery, null, null, executionController, System.nanoTime());
return new PagersIterator(toQuery, null, null, executionController, nanoTime());
}
private class PagersIterator extends AbstractIterator<RowIterator> implements PartitionIterator

3
src/java/org/apache/cassandra/service/paxos/AbstractPaxosCallback.java
View File

@ -29,6 +29,7 @@ import org.apache.cassandra.exceptions.WriteTimeoutException;
import org.apache.cassandra.net.RequestCallback;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public abstract class AbstractPaxosCallback<T> implements RequestCallback<T>
{
@ -54,7 +55,7 @@ public abstract class AbstractPaxosCallback<T> implements RequestCallback<T>
{
try
{
long timeout = DatabaseDescriptor.getWriteRpcTimeout(NANOSECONDS) - (System.nanoTime() - queryStartNanoTime);
long timeout = DatabaseDescriptor.getWriteRpcTimeout(NANOSECONDS) - (nanoTime() - queryStartNanoTime);
if (!latch.await(timeout, NANOSECONDS))
throw new WriteTimeoutException(WriteType.CAS, consistency, getResponseCount(), targets);
}

14
src/java/org/apache/cassandra/service/paxos/PaxosState.java
View File

@ -30,6 +30,8 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.UUIDGen;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class PaxosState
{
private static final Striped<Lock> LOCKS = Striped.lazyWeakLock(DatabaseDescriptor.getConcurrentWriters() * 1024);
@ -55,7 +57,7 @@ public class PaxosState
public static PrepareResponse prepare(Commit toPrepare)
{
long start = System.nanoTime();
long start = nanoTime();
try
{
Lock lock = LOCKS.get(toPrepare.update.partitionKey());
@ -89,14 +91,14 @@ public class PaxosState
}
finally
{
Keyspace.open(toPrepare.update.metadata().keyspace).getColumnFamilyStore(toPrepare.update.metadata().id).metric.casPrepare.addNano(System.nanoTime() - start);
Keyspace.open(toPrepare.update.metadata().keyspace).getColumnFamilyStore(toPrepare.update.metadata().id).metric.casPrepare.addNano(nanoTime() - start);
}
}
public static Boolean propose(Commit proposal)
{
long start = System.nanoTime();
long start = nanoTime();
try
{
Lock lock = LOCKS.get(proposal.update.partitionKey());
@ -124,13 +126,13 @@ public class PaxosState
}
finally
{
Keyspace.open(proposal.update.metadata().keyspace).getColumnFamilyStore(proposal.update.metadata().id).metric.casPropose.addNano(System.nanoTime() - start);
Keyspace.open(proposal.update.metadata().keyspace).getColumnFamilyStore(proposal.update.metadata().id).metric.casPropose.addNano(nanoTime() - start);
}
}
public static void commit(Commit proposal)
{
long start = System.nanoTime();
long start = nanoTime();
try
{
// There is no guarantee we will see commits in the right order, because messages
@ -155,7 +157,7 @@ public class PaxosState
}
finally
{
Keyspace.open(proposal.update.metadata().keyspace).getColumnFamilyStore(proposal.update.metadata().id).metric.casCommit.addNano(System.nanoTime() - start);
Keyspace.open(proposal.update.metadata().keyspace).getColumnFamilyStore(proposal.update.metadata().id).metric.casCommit.addNano(nanoTime() - start);
}
}
}

5
src/java/org/apache/cassandra/service/reads/DigestResolver.java
View File

@ -38,6 +38,7 @@ import org.apache.cassandra.service.reads.repair.NoopReadRepair;
import org.apache.cassandra.utils.ByteBufferUtil;
import static com.google.common.collect.Iterables.any;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class DigestResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<E>> extends ResponseResolver<E, P>
{
@ -102,7 +103,7 @@ public class DigestResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRea
public boolean responsesMatch()
{
long start = System.nanoTime();
long start = nanoTime();
// validate digests against each other; return false immediately on mismatch.
ByteBuffer digest = null;
@ -126,7 +127,7 @@ public class DigestResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRea
}
if (logger.isTraceEnabled())
logger.trace("responsesMatch: {} ms.", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
logger.trace("responsesMatch: {} ms.", TimeUnit.NANOSECONDS.toMillis(nanoTime() - start));
return true;
}

3
src/java/org/apache/cassandra/service/reads/ReadCallback.java
View File

@ -48,6 +48,7 @@ import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class ReadCallback<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<E>> implements RequestCallback<ReadResponse>
{
@ -91,7 +92,7 @@ public class ReadCallback<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
public boolean await(long timePastStart, TimeUnit unit)
{
long time = unit.toNanos(timePastStart) - (System.nanoTime() - queryStartNanoTime);
long time = unit.toNanos(timePastStart) - (nanoTime() - queryStartNanoTime);
try
{
return condition.await(time, TimeUnit.NANOSECONDS);

6
src/java/org/apache/cassandra/service/reads/range/RangeCommandIterator.java
View File

@ -52,6 +52,8 @@ import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.utils.CloseableIterator;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
class RangeCommandIterator extends AbstractIterator<RowIterator> implements PartitionIterator
{
private static final Logger logger = LoggerFactory.getLogger(RangeCommandIterator.class);
@ -90,7 +92,7 @@ class RangeCommandIterator extends AbstractIterator<RowIterator> implements Part
this.totalRangeCount = totalRangeCount;
this.queryStartNanoTime = queryStartNanoTime;
startTime = System.nanoTime();
startTime = nanoTime();
enforceStrictLiveness = command.metadata().enforceStrictLiveness();
}
@ -262,7 +264,7 @@ class RangeCommandIterator extends AbstractIterator<RowIterator> implements Part
}
finally
{
long latency = System.nanoTime() - startTime;
long latency = nanoTime() - startTime;
rangeMetrics.addNano(latency);
Keyspace.openAndGetStore(command.metadata()).metric.coordinatorScanLatency.update(latency, TimeUnit.NANOSECONDS);
}

5
src/java/org/apache/cassandra/service/reads/repair/BlockingPartitionRepair.java
View File

@ -53,6 +53,7 @@ import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.tracing.Tracing;
import static org.apache.cassandra.net.Verb.*;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class BlockingPartitionRepair
extends AbstractFuture<Object> implements RequestCallback<Object>
@ -146,7 +147,7 @@ public class BlockingPartitionRepair
public void sendInitialRepairs()
{
mutationsSentTime = System.nanoTime();
mutationsSentTime = nanoTime();
Replicas.assertFull(pendingRepairs.keySet());
for (Map.Entry<Replica, Mutation> entry: pendingRepairs.entrySet())
@ -177,7 +178,7 @@ public class BlockingPartitionRepair
public boolean awaitRepairsUntil(long timeoutAt, TimeUnit timeUnit)
{
long timeoutAtNanos = timeUnit.toNanos(timeoutAt);
long remaining = timeoutAtNanos - System.nanoTime();
long remaining = timeoutAtNanos - nanoTime();
try
{
return latch.await(remaining, TimeUnit.NANOSECONDS);

5
src/java/org/apache/cassandra/streaming/StreamSession.java
View File

@ -52,6 +52,7 @@ import org.apache.cassandra.utils.NoSpamLogger;
import static com.google.common.collect.Iterables.all;
import static org.apache.cassandra.net.MessagingService.current_version;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Handles the streaming a one or more streams to and from a specific remote node.
@ -790,14 +791,14 @@ public class StreamSession implements IEndpointStateChangeSubscriber
// send back file received message
messageSender.sendMessage(new ReceivedMessage(message.header.tableId, message.header.sequenceNumber));
StreamHook.instance.reportIncomingStream(message.header.tableId, message.stream, this, message.header.sequenceNumber);
long receivedStartNanos = System.nanoTime();
long receivedStartNanos = nanoTime();
try
{
receivers.get(message.header.tableId).received(message.stream);
}
finally
{
long latencyNanos = System.nanoTime() - receivedStartNanos;
long latencyNanos = nanoTime() - receivedStartNanos;
metrics.incomingProcessTime.update(latencyNanos, TimeUnit.NANOSECONDS);
long latencyMs = TimeUnit.NANOSECONDS.toMillis(latencyNanos);
int timeout = DatabaseDescriptor.getInternodeStreamingTcpUserTimeoutInMS();

6
src/java/org/apache/cassandra/streaming/async/NettyStreamingMessageSender.java
View File

@ -63,6 +63,8 @@ import org.apache.cassandra.streaming.messages.StreamMessage;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Responsible for sending {@link StreamMessage}s to a given peer. We manage an array of netty {@link Channel}s
* for sending {@link OutgoingStreamMessage} instances; all other {@link StreamMessage} types are sent via
@ -377,7 +379,7 @@ public class NettyStreamingMessageSender implements StreamingMessageSender
boolean acquirePermit(int logInterval)
{
long logIntervalNanos = TimeUnit.MINUTES.toNanos(logInterval);
long timeOfLastLogging = System.nanoTime();
long timeOfLastLogging = nanoTime();
while (true)
{
if (closed)
@ -388,7 +390,7 @@ public class NettyStreamingMessageSender implements StreamingMessageSender
return true;
// log a helpful message to operators in case they are wondering why a given session might not be making progress.
long now = System.nanoTime();
long now = nanoTime();
if (now - timeOfLastLogging > logIntervalNanos)
{
timeOfLastLogging = now;

6
src/java/org/apache/cassandra/streaming/management/StreamEventJMXNotifier.java
View File

@ -23,6 +23,8 @@ import javax.management.NotificationBroadcasterSupport;
import org.apache.cassandra.streaming.*;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
*/
public class StreamEventJMXNotifier extends NotificationBroadcasterSupport implements StreamEventHandler
@ -53,14 +55,14 @@ public class StreamEventJMXNotifier extends NotificationBroadcasterSupport imple
break;
case FILE_PROGRESS:
ProgressInfo progress = ((StreamEvent.ProgressEvent) event).progress;
long current = System.currentTimeMillis();
long current = currentTimeMillis();
if (current - progressLastSent >= PROGRESS_NOTIFICATION_INTERVAL || progress.isCompleted())
{
notif = new Notification(StreamEvent.ProgressEvent.class.getCanonicalName(),
StreamManagerMBean.OBJECT_NAME,
seq.getAndIncrement());
notif.setUserData(ProgressInfoCompositeData.toCompositeData(event.planId, progress));
progressLastSent = System.currentTimeMillis();
progressLastSent = currentTimeMillis();
}
else
{

4
src/java/org/apache/cassandra/tools/BootstrapMonitor.java
View File

@ -27,6 +27,8 @@ import org.apache.cassandra.utils.progress.ProgressEvent;
import org.apache.cassandra.utils.progress.ProgressEventType;
import org.apache.cassandra.utils.progress.jmx.JMXNotificationProgressListener;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public class BootstrapMonitor extends JMXNotificationProgressListener
{
private final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss,SSS");
@ -74,7 +76,7 @@ public class BootstrapMonitor extends JMXNotificationProgressListener
public void progress(String tag, ProgressEvent event)
{
ProgressEventType type = event.getType();
String message = String.format("[%s] %s", format.format(System.currentTimeMillis()), event.getMessage());
String message = String.format("[%s] %s", format.format(currentTimeMillis()), event.getMessage());
if (type == ProgressEventType.PROGRESS)
{
message = message + " (progress: " + (int)event.getProgressPercentage() + "%)";

8
src/java/org/apache/cassandra/tools/BulkLoader.java
View File

@ -43,6 +43,8 @@ import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.NativeSSTableLoaderClient;
import org.apache.cassandra.utils.OutputHandler;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class BulkLoader
{
public static void main(String args[]) throws BulkLoadException
@ -131,7 +133,7 @@ public class BulkLoader
public ProgressIndicator()
{
start = lastTime = System.nanoTime();
start = lastTime = nanoTime();
}
public void onSuccess(StreamState finalState)
@ -157,7 +159,7 @@ public class BulkLoader
progressInfo = ((StreamEvent.ProgressEvent) event).progress;
}
long time = System.nanoTime();
long time = nanoTime();
long deltaTime = time - lastTime;
StringBuilder sb = new StringBuilder();
@ -230,7 +232,7 @@ public class BulkLoader
private void printSummary(int connectionsPerHost)
{
long end = System.nanoTime();
long end = nanoTime();
long durationMS = ((end - start) / (1000000));
StringBuilder sb = new StringBuilder();

6
src/java/org/apache/cassandra/tools/JsonTransformer.java
View File

@ -58,6 +58,8 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public final class JsonTransformer
{
@ -282,7 +284,7 @@ public final class JsonTransformer
json.writeFieldName("expires_at");
json.writeString(dateString(TimeUnit.SECONDS, liveInfo.localExpirationTime()));
json.writeFieldName("expired");
json.writeBoolean(liveInfo.localExpirationTime() < (System.currentTimeMillis() / 1000));
json.writeBoolean(liveInfo.localExpirationTime() < (currentTimeMillis() / 1000));
}
json.writeEndObject();
objectIndenter.setCompact(false);
@ -497,7 +499,7 @@ public final class JsonTransformer
json.writeFieldName("expires_at");
json.writeString(dateString(TimeUnit.SECONDS, cell.localDeletionTime()));
json.writeFieldName("expired");
json.writeBoolean(!cell.isLive((int) (System.currentTimeMillis() / 1000)));
json.writeBoolean(!cell.isLive((int) (currentTimeMillis() / 1000)));
}
json.writeEndObject();
objectIndenter.setCompact(false);

4
src/java/org/apache/cassandra/tools/RepairRunner.java
View File

@ -35,6 +35,8 @@ import org.apache.cassandra.utils.progress.ProgressEvent;
import org.apache.cassandra.utils.progress.ProgressEventType;
import org.apache.cassandra.utils.progress.jmx.JMXNotificationProgressListener;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public class RepairRunner extends JMXNotificationProgressListener
{
private final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss,SSS");
@ -189,6 +191,6 @@ public class RepairRunner extends JMXNotificationProgressListener
private void printMessage(String message)
{
out.println(String.format("[%s] %s", this.format.format(System.currentTimeMillis()), message));
out.println(String.format("[%s] %s", this.format.format(currentTimeMillis()), message));
}
}

Some files were not shown because too many files have changed in this diff Show More