You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2013/05/21 17:02:16 UTC
git commit: use nanotime consistently for node-local timeouts patch
by Mikhail Mazursky; reviewed by jbellis for CASSANDRA-5581
Updated Branches:
refs/heads/trunk 6f467d920 -> 1d2c12242
use nanotime consistently for node-local timeouts
patch by Mikhail Mazursky; reviewed by jbellis for CASSANDRA-5581
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/1d2c1224
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/1d2c1224
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/1d2c1224
Branch: refs/heads/trunk
Commit: 1d2c12242f20dfcc0b787129b4ad8eeef2768d6b
Parents: 6f467d9
Author: Jonathan Ellis <jb...@apache.org>
Authored: Tue May 21 10:02:12 2013 -0500
Committer: Jonathan Ellis <jb...@apache.org>
Committed: Tue May 21 10:02:12 2013 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/cache/AutoSavingCache.java | 12 +++---
.../org/apache/cassandra/db/ColumnFamilyStore.java | 21 +++++------
src/java/org/apache/cassandra/db/Memtable.java | 10 +++---
.../cassandra/db/compaction/CompactionManager.java | 4 +-
.../cassandra/db/compaction/CompactionTask.java | 5 ++-
.../org/apache/cassandra/gms/EndpointState.java | 7 +++-
src/java/org/apache/cassandra/gms/Gossiper.java | 13 ++++---
.../apache/cassandra/io/sstable/SSTableReader.java | 6 ++--
.../cassandra/locator/DynamicEndpointSnitch.java | 6 ++--
.../org/apache/cassandra/net/AsyncOneResponse.java | 10 +++---
.../org/apache/cassandra/net/MessagingService.java | 3 ++
.../cassandra/net/OutboundTcpConnection.java | 5 ++-
.../apache/cassandra/net/ResponseVerbHandler.java | 4 ++-
.../cassandra/service/AbstractReadExecutor.java | 3 +-
.../service/AbstractWriteResponseHandler.java | 8 ++--
.../org/apache/cassandra/service/ReadCallback.java | 14 ++++----
.../apache/cassandra/service/RowDataResolver.java | 5 ++-
.../cassandra/service/RowDigestResolver.java | 5 ++-
.../org/apache/cassandra/service/StorageProxy.java | 28 ++++++++-------
.../apache/cassandra/service/StorageService.java | 2 +-
.../cassandra/service/TruncateResponseHandler.java | 8 ++--
.../org/apache/cassandra/tools/BulkLoader.java | 10 +++---
.../org/apache/cassandra/tools/SSTableImport.java | 13 ++++---
.../org/apache/cassandra/utils/ExpiringMap.java | 11 ++++--
.../apache/cassandra/utils/SimpleCondition.java | 11 +++---
src/java/org/apache/cassandra/utils/Throttle.java | 6 ++--
.../db/compaction/LongCompactionsTest.java | 5 ++-
.../org/apache/cassandra/utils/LongBitSetTest.java | 25 +++++++------
.../DebuggableThreadPoolExecutorTest.java | 4 +-
test/unit/org/apache/cassandra/db/CleanupTest.java | 5 ++-
.../apache/cassandra/io/sstable/SSTableUtils.java | 1 -
.../org/apache/cassandra/stress/StressAction.java | 4 +-
33 files changed, 147 insertions(+), 128 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 48bec35..ba3dc20 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.0
+ * use nanotime consistently for node-local timeouts (CASSANDRA-5581)
* Avoid unnecessary second pass on name-based queries (CASSANDRA-5577)
* Experimental triggers (CASSANDRA-1311)
* JEMalloc support for off-heap allocation (CASSANDRA-3997)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/cache/AutoSavingCache.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cache/AutoSavingCache.java b/src/java/org/apache/cassandra/cache/AutoSavingCache.java
index 6a7e17d..bcb1b7f 100644
--- a/src/java/org/apache/cassandra/cache/AutoSavingCache.java
+++ b/src/java/org/apache/cassandra/cache/AutoSavingCache.java
@@ -102,7 +102,7 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
public int loadSaved(ColumnFamilyStore cfs)
{
int count = 0;
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
// old cache format that only saves keys
File path = getCachePath(cfs.table.getName(), cfs.name, null);
@@ -168,8 +168,8 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
}
}
if (logger.isDebugEnabled())
- logger.debug(String.format("completed reading (%d ms; %d keys) saved cache %s",
- System.currentTimeMillis() - start, count, path));
+ logger.debug("completed reading ({} ms; {} keys) saved cache {}",
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start), count, path);
return count;
}
@@ -228,7 +228,7 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
return;
}
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
HashMap<Pair<String, String>, SequentialWriter> writers = new HashMap<Pair<String, String>, SequentialWriter>();
@@ -272,10 +272,10 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
cacheFile.delete(); // ignore error if it didn't exist
if (!tmpFile.renameTo(cacheFile))
- logger.error("Unable to rename " + tmpFile + " to " + cacheFile);
+ logger.error("Unable to rename {} to {}", tmpFile, cacheFile);
}
- logger.info(String.format("Saved %s (%d items) in %d ms", cacheType, keys.size(), System.currentTimeMillis() - start));
+ logger.info("Saved {} ({} items) in {} ms", cacheType, keys.size(), TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
}
private SequentialWriter tempCacheFile(Pair<String, String> pathInfo)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 36c1db0..892e35f 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -536,15 +536,15 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
if (!isRowCacheEnabled())
return;
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
int cachedRowsRead = CacheService.instance.rowCache.loadSaved(this);
if (cachedRowsRead > 0)
- logger.info(String.format("completed loading (%d ms; %d keys) row cache for %s.%s",
- System.currentTimeMillis() - start,
+ logger.info("completed loading ({} ms; {} keys) row cache for {}.{}",
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start),
cachedRowsRead,
table.getName(),
- name));
+ name);
}
/**
@@ -1828,11 +1828,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
// sleep a little to make sure that our truncatedAt comes after any sstable
// that was part of the flushed we forced; otherwise on a tie, it won't get deleted.
- long starttime = System.currentTimeMillis();
- while ((System.currentTimeMillis() - starttime) < 1)
- {
- Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS);
- }
+ Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS);
}
else
{
@@ -1910,8 +1906,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
CompactionManager.instance.interruptCompactionFor(allMetadata, interruptValidation);
// wait for the interruption to be recognized
- long start = System.currentTimeMillis();
- while (System.currentTimeMillis() < start + 60000)
+ long start = System.nanoTime();
+ long delay = TimeUnit.MINUTES.toNanos(1);
+ while (System.nanoTime() - start < delay)
{
if (CompactionManager.instance.isCompacting(selfWithIndexes))
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
@@ -1924,7 +1921,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
if (!cfs.getDataTracker().getCompacting().isEmpty())
{
- logger.warn("Unable to cancel in-progress compactios for {}. Probably there is an unusually large row in progress somewhere. It is also possible that buggy code left some sstables compacting after it was done with them", metadata.cfName);
+ logger.warn("Unable to cancel in-progress compactions for {}. Probably there is an unusually large row in progress somewhere. It is also possible that buggy code left some sstables compacting after it was done with them", metadata.cfName);
}
}
logger.debug("Compactions successfully cancelled");
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/db/Memtable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java
index 88c8ffe..a19c4a7 100644
--- a/src/java/org/apache/cassandra/db/Memtable.java
+++ b/src/java/org/apache/cassandra/db/Memtable.java
@@ -112,7 +112,8 @@ public class Memtable
// actually only store DecoratedKey.
private final ConcurrentNavigableMap<RowPosition, AtomicSortedColumns> rows = new ConcurrentSkipListMap<RowPosition, AtomicSortedColumns>();
public final ColumnFamilyStore cfs;
- private final long creationTime;
+ private final long creationTime = System.currentTimeMillis();
+ private final long creationNano = System.nanoTime();
private final SlabAllocator allocator = new SlabAllocator();
// We really only need one column by allocator but one by memtable is not a big waste and avoids needing allocators to know about CFS
@@ -132,7 +133,6 @@ public class Memtable
public Memtable(ColumnFamilyStore cfs)
{
this.cfs = cfs;
- this.creationTime = System.currentTimeMillis();
this.initialComparator = cfs.metadata.comparator;
this.cfs.scheduleFlush();
@@ -201,7 +201,7 @@ public class Memtable
{
activelyMeasuring = Memtable.this;
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
// ConcurrentSkipListMap has cycles, so measureDeep will have to track a reference to EACH object it visits.
// So to reduce the memory overhead of doing a measurement, we break it up to row-at-a-time.
long deepSize = meter.measure(rows);
@@ -232,7 +232,7 @@ public class Memtable
cfs.liveRatio = (cfs.liveRatio + newRatio) / 2.0;
logger.info("{} liveRatio is {} (just-counted was {}). calculation took {}ms for {} columns",
- cfs, cfs.liveRatio, newRatio, System.currentTimeMillis() - start, objects);
+ cfs, cfs.liveRatio, newRatio, TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start), objects);
activelyMeasuring = null;
}
finally
@@ -333,7 +333,7 @@ public class Memtable
public boolean isExpired()
{
int period = cfs.metadata.getMemtableFlushPeriod();
- return period > 0 && (System.currentTimeMillis() >= creationTime + period);
+ return period > 0 && (System.nanoTime() - creationNano >= TimeUnit.MILLISECONDS.toNanos(period));
}
/**
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index e0d186f..1e44a07 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -475,7 +475,7 @@ public class CompactionManager implements CompactionManagerMBean
}
CompactionController controller = new CompactionController(cfs, Collections.singleton(sstable), getDefaultGcBefore(cfs));
- long startTime = System.currentTimeMillis();
+ long start = System.nanoTime();
long totalkeysWritten = 0;
@@ -576,7 +576,7 @@ public class CompactionManager implements CompactionManagerMBean
results.add(newSstable);
String format = "Cleaned up to %s. %,d to %,d (~%d%% of original) bytes for %,d keys. Time: %,dms.";
- long dTime = System.currentTimeMillis() - startTime;
+ long dTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
long startsize = sstable.onDiskLength();
long endsize = newSstable.onDiskLength();
double ratio = (double) endsize / (double) startsize;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
index e32956b..13fc11d 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.db.compaction;
import java.io.File;
import java.io.IOException;
import java.util.*;
+import java.util.concurrent.TimeUnit;
import com.google.common.base.Throwables;
import com.google.common.collect.Sets;
@@ -113,7 +114,7 @@ public class CompactionTask extends AbstractCompactionTask
// all the sstables (that existed when we started)
logger.info("Compacting {}", toCompact);
- long startTime = System.currentTimeMillis();
+ long start = System.nanoTime();
long totalkeysWritten = 0;
long estimatedTotalKeys = Math.max(cfs.metadata.getIndexInterval(), SSTableReader.getApproximateKeyCount(actuallyCompact, cfs.metadata));
@@ -249,7 +250,7 @@ public class CompactionTask extends AbstractCompactionTask
if (logger.isInfoEnabled())
{
// log a bunch of statistics about the result
- long dTime = System.currentTimeMillis() - startTime;
+ long dTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
long startsize = SSTable.getTotalBytes(toCompact);
long endsize = SSTable.getTotalBytes(sstables);
double ratio = (double)endsize / (double)startsize;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/gms/EndpointState.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/EndpointState.java b/src/java/org/apache/cassandra/gms/EndpointState.java
index c6575cd..a338345 100644
--- a/src/java/org/apache/cassandra/gms/EndpointState.java
+++ b/src/java/org/apache/cassandra/gms/EndpointState.java
@@ -50,7 +50,7 @@ public class EndpointState
EndpointState(HeartBeatState initialHbState)
{
hbState = initialHbState;
- updateTimestamp = System.currentTimeMillis();
+ updateTimestamp = System.nanoTime();
isAlive = true;
}
@@ -85,6 +85,9 @@ public class EndpointState
}
/* getters and setters */
+ /**
+ * @return System.nanoTime() when state was updated last time.
+ */
public long getUpdateTimestamp()
{
return updateTimestamp;
@@ -92,7 +95,7 @@ public class EndpointState
void updateTimestamp()
{
- updateTimestamp = System.currentTimeMillis();
+ updateTimestamp = System.nanoTime();
}
public boolean isAlive()
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/gms/Gossiper.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java
index d3c9343..c32cc41 100644
--- a/src/java/org/apache/cassandra/gms/Gossiper.java
+++ b/src/java/org/apache/cassandra/gms/Gossiper.java
@@ -223,7 +223,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
{
Long downtime = unreachableEndpoints.get(ep);
if (downtime != null)
- return System.currentTimeMillis() - downtime;
+ return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - downtime);
else
return 0L;
}
@@ -556,6 +556,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
private void doStatusCheck()
{
long now = System.currentTimeMillis();
+ long nowNano = System.nanoTime();
Set<InetAddress> eps = endpointStateMap.keySet();
for (InetAddress endpoint : eps)
@@ -567,11 +568,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
EndpointState epState = endpointStateMap.get(endpoint);
if (epState != null)
{
- long duration = now - epState.getUpdateTimestamp();
-
// check if this is a fat client. fat clients are removed automatically from
// gossip after FatClientTimeout. Do not remove dead states here.
- if (isFatClient(endpoint) && !justRemovedEndpoints.containsKey(endpoint) && (duration > FatClientTimeout))
+ if (isFatClient(endpoint)
+ && !justRemovedEndpoints.containsKey(endpoint)
+ && TimeUnit.NANOSECONDS.toMillis(nowNano - epState.getUpdateTimestamp()) > FatClientTimeout)
{
logger.info("FatClient " + endpoint + " has been silent for " + FatClientTimeout + "ms, removing from gossip");
removeEndpoint(endpoint); // will put it in justRemovedEndpoints to respect quarantine delay
@@ -783,7 +784,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
logger.trace("marking as down {}", addr);
localState.markDead();
liveEndpoints.remove(addr);
- unreachableEndpoints.put(addr, System.currentTimeMillis());
+ unreachableEndpoints.put(addr, System.nanoTime());
logger.info("InetAddress {} is now DOWN", addr);
for (IEndpointStateChangeSubscriber subscriber : subscribers)
subscriber.onDead(addr, localState);
@@ -1079,7 +1080,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
EndpointState epState = new EndpointState(new HeartBeatState(0));
epState.markDead();
endpointStateMap.put(ep, epState);
- unreachableEndpoints.put(ep, System.currentTimeMillis());
+ unreachableEndpoints.put(ep, System.nanoTime());
if (logger.isTraceEnabled())
logger.trace("Adding saved endpoint " + ep + " " + epState.getHeartBeatState().getGeneration());
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
index baba472..ecdddcc 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
@@ -166,7 +166,7 @@ public class SSTableReader extends SSTable
assert components.contains(Component.DATA) : "Data component is missing for sstable" + descriptor;
assert components.contains(Component.PRIMARY_INDEX) : "Primary index component is missing for sstable " + descriptor;
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
logger.info("Opening {} ({} bytes)", descriptor, new File(descriptor.filenameFor(COMPONENT_DATA)).length());
SSTableMetadata sstableMetadata = SSTableMetadata.serializer.deserialize(descriptor);
@@ -198,10 +198,10 @@ public class SSTableReader extends SSTable
if (validate)
sstable.validate();
- logger.debug("INDEX LOAD TIME for " + descriptor + ": " + (System.currentTimeMillis() - start) + " ms.");
+ logger.debug("INDEX LOAD TIME for {}: {} ms.", descriptor, TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
if (sstable.getKeyCache() != null)
- logger.debug(String.format("key cache contains %s/%s keys", sstable.getKeyCache().size(), sstable.getKeyCache().getCapacity()));
+ logger.debug("key cache contains {}/{} keys", sstable.getKeyCache().size(), sstable.getKeyCache().getCapacity());
return sstable;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java
index 978c2e2..ff8c70a 100644
--- a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java
+++ b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java
@@ -201,7 +201,7 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa
public void receiveTiming(InetAddress host, long latency) // this is cheap
{
- lastReceived.put(host, System.currentTimeMillis());
+ lastReceived.put(host, System.nanoTime());
ExponentiallyDecayingSample sample = samples.get(host);
if (sample == null)
@@ -237,8 +237,8 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa
double mean = entry.getValue().getSnapshot().getMedian();
if (mean > maxLatency)
maxLatency = mean;
- long timePenalty = lastReceived.containsKey(entry.getKey()) ? lastReceived.get(entry.getKey()) : System.currentTimeMillis();
- timePenalty = System.currentTimeMillis() - timePenalty;
+ long timePenalty = lastReceived.containsKey(entry.getKey()) ? lastReceived.get(entry.getKey()) : System.nanoTime();
+ timePenalty = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - timePenalty);
timePenalty = timePenalty > UPDATE_INTERVAL_IN_MS ? UPDATE_INTERVAL_IN_MS : timePenalty;
// a convenient place to remember this since we've already calculated it and need it later
penalties.put(entry.getKey(), timePenalty);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/net/AsyncOneResponse.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/AsyncOneResponse.java b/src/java/org/apache/cassandra/net/AsyncOneResponse.java
index b5370ff..bb3058e 100644
--- a/src/java/org/apache/cassandra/net/AsyncOneResponse.java
+++ b/src/java/org/apache/cassandra/net/AsyncOneResponse.java
@@ -34,12 +34,12 @@ public class AsyncOneResponse<T> implements IAsyncCallback<T>
private final AtomicBoolean done = new AtomicBoolean(false);
private final Lock lock = new ReentrantLock();
private final Condition condition;
- private final long startTime;
+ private final long start;
public AsyncOneResponse()
{
condition = lock.newCondition();
- startTime = System.currentTimeMillis();
+ start = System.nanoTime();
}
public T get(long timeout, TimeUnit tu) throws TimeoutException
@@ -52,9 +52,9 @@ public class AsyncOneResponse<T> implements IAsyncCallback<T>
{
if (!done.get())
{
- timeout = TimeUnit.MILLISECONDS.convert(timeout, tu);
- long overall_timeout = timeout - (System.currentTimeMillis() - startTime);
- bVal = overall_timeout > 0 && condition.await(overall_timeout, TimeUnit.MILLISECONDS);
+ timeout = tu.toNanos(timeout);
+ long overall_timeout = timeout - (System.nanoTime() - start);
+ bVal = overall_timeout > 0 && condition.await(overall_timeout, TimeUnit.NANOSECONDS);
}
}
catch (InterruptedException ex)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/net/MessagingService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java
index c0de7c9..5f2832d 100644
--- a/src/java/org/apache/cassandra/net/MessagingService.java
+++ b/src/java/org/apache/cassandra/net/MessagingService.java
@@ -744,6 +744,9 @@ public final class MessagingService implements MessagingServiceMBean
return callbacks.remove(messageId);
}
+ /**
+ * @return System.nanoTime() when callback was created.
+ */
public long getRegisteredCallbackAge(int messageId)
{
return callbacks.getAge(messageId);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/net/OutboundTcpConnection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/OutboundTcpConnection.java b/src/java/org/apache/cassandra/net/OutboundTcpConnection.java
index 62d5b82..52a415c 100644
--- a/src/java/org/apache/cassandra/net/OutboundTcpConnection.java
+++ b/src/java/org/apache/cassandra/net/OutboundTcpConnection.java
@@ -277,8 +277,9 @@ public class OutboundTcpConnection extends Thread
targetVersion = MessagingService.instance().getVersion(poolReference.endPoint());
- long start = System.currentTimeMillis();
- while (System.currentTimeMillis() < start + DatabaseDescriptor.getRpcTimeout())
+ long start = System.nanoTime();
+ long timeout = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getRpcTimeout());
+ while (System.nanoTime() - start < timeout)
{
try
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/net/ResponseVerbHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/ResponseVerbHandler.java b/src/java/org/apache/cassandra/net/ResponseVerbHandler.java
index 0245005..132e574 100644
--- a/src/java/org/apache/cassandra/net/ResponseVerbHandler.java
+++ b/src/java/org/apache/cassandra/net/ResponseVerbHandler.java
@@ -17,6 +17,8 @@
*/
package org.apache.cassandra.net;
+import java.util.concurrent.TimeUnit;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -28,7 +30,7 @@ public class ResponseVerbHandler implements IVerbHandler
public void doVerb(MessageIn message, int id)
{
- long latency = System.currentTimeMillis() - MessagingService.instance().getRegisteredCallbackAge(id);
+ long latency = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - MessagingService.instance().getRegisteredCallbackAge(id));
CallbackInfo callbackInfo = MessagingService.instance().removeRegisteredCallback(id);
if (callbackInfo == null)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/service/AbstractReadExecutor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/AbstractReadExecutor.java b/src/java/org/apache/cassandra/service/AbstractReadExecutor.java
index c49dabb..1c7c6f9 100644
--- a/src/java/org/apache/cassandra/service/AbstractReadExecutor.java
+++ b/src/java/org/apache/cassandra/service/AbstractReadExecutor.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.service;
import java.net.InetAddress;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.concurrent.StageManager;
@@ -170,7 +171,7 @@ public abstract class AbstractReadExecutor
if (cfs.sampleLatency > command.getTimeout())
return;
- if (!handler.await(cfs.sampleLatency))
+ if (!handler.await(cfs.sampleLatency, TimeUnit.MILLISECONDS))
{
InetAddress endpoint = unfiltered.get(handler.endpoints.size());
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java b/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java
index a97eec0..7be4c29 100644
--- a/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java
+++ b/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java
@@ -36,7 +36,7 @@ public abstract class AbstractWriteResponseHandler implements IAsyncCallback
{
private final SimpleCondition condition = new SimpleCondition();
protected final Table table;
- protected final long startTime;
+ protected final long start;
protected final Collection<InetAddress> naturalEndpoints;
protected final ConsistencyLevel consistencyLevel;
protected final Runnable callback;
@@ -56,7 +56,7 @@ public abstract class AbstractWriteResponseHandler implements IAsyncCallback
{
this.table = table;
this.pendingEndpoints = pendingEndpoints;
- this.startTime = System.currentTimeMillis();
+ this.start = System.nanoTime();
this.consistencyLevel = consistencyLevel;
this.naturalEndpoints = naturalEndpoints;
this.callback = callback;
@@ -65,12 +65,12 @@ public abstract class AbstractWriteResponseHandler implements IAsyncCallback
public void get() throws WriteTimeoutException
{
- long timeout = DatabaseDescriptor.getWriteRpcTimeout() - (System.currentTimeMillis() - startTime);
+ long timeout = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getWriteRpcTimeout()) - (System.nanoTime() - start);
boolean success;
try
{
- success = condition.await(timeout, TimeUnit.MILLISECONDS);
+ success = condition.await(timeout, TimeUnit.NANOSECONDS);
}
catch (InterruptedException ex)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/service/ReadCallback.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/ReadCallback.java b/src/java/org/apache/cassandra/service/ReadCallback.java
index 560e577..3e25aef 100644
--- a/src/java/org/apache/cassandra/service/ReadCallback.java
+++ b/src/java/org/apache/cassandra/service/ReadCallback.java
@@ -48,7 +48,7 @@ public class ReadCallback<TMessage, TResolved> implements IAsyncCallback<TMessag
public final IResponseResolver<TMessage, TResolved> resolver;
private final SimpleCondition condition = new SimpleCondition();
- final long startTime;
+ final long start;
private final int blockfor;
final List<InetAddress> endpoints;
private final IReadCommand command;
@@ -73,7 +73,7 @@ public class ReadCallback<TMessage, TResolved> implements IAsyncCallback<TMessag
this.blockfor = blockfor;
this.consistencyLevel = consistencyLevel;
this.resolver = resolver;
- this.startTime = System.currentTimeMillis();
+ this.start = System.nanoTime();
this.endpoints = endpoints;
}
@@ -82,12 +82,12 @@ public class ReadCallback<TMessage, TResolved> implements IAsyncCallback<TMessag
return new ReadCallback<TMessage, TResolved>(newResolver, consistencyLevel, blockfor, command, table, endpoints);
}
- public boolean await(long interimTimeout)
+ public boolean await(long timePastStart, TimeUnit unit)
{
- long timeout = interimTimeout - (System.currentTimeMillis() - startTime);
+ long time = unit.toNanos(timePastStart) - (System.nanoTime() - start);
try
{
- return condition.await(timeout, TimeUnit.MILLISECONDS);
+ return condition.await(time, TimeUnit.NANOSECONDS);
}
catch (InterruptedException ex)
{
@@ -97,14 +97,14 @@ public class ReadCallback<TMessage, TResolved> implements IAsyncCallback<TMessag
public TResolved get() throws ReadTimeoutException, DigestMismatchException
{
- long timeout = command.getTimeout() - (System.currentTimeMillis() - startTime);
- if (!await(timeout))
+ if (!await(command.getTimeout(), TimeUnit.MICROSECONDS))
{
ReadTimeoutException ex = new ReadTimeoutException(consistencyLevel, received.get(), blockfor, resolver.isDataPresent());
if (logger.isDebugEnabled())
logger.debug("Read timeout: {}", ex.toString());
throw ex;
}
+
return blockfor == 1 ? resolver.getData() : resolver.resolve();
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/service/RowDataResolver.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/RowDataResolver.java b/src/java/org/apache/cassandra/service/RowDataResolver.java
index 372056c..8533f4f 100644
--- a/src/java/org/apache/cassandra/service/RowDataResolver.java
+++ b/src/java/org/apache/cassandra/service/RowDataResolver.java
@@ -22,6 +22,7 @@ import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.Iterables;
@@ -56,7 +57,7 @@ public class RowDataResolver extends AbstractRowResolver
{
if (logger.isDebugEnabled())
logger.debug("resolving " + replies.size() + " responses");
- long startTime = System.currentTimeMillis();
+ long start = System.nanoTime();
ColumnFamily resolved;
if (replies.size() > 1)
@@ -93,7 +94,7 @@ public class RowDataResolver extends AbstractRowResolver
}
if (logger.isDebugEnabled())
- logger.debug("resolve: " + (System.currentTimeMillis() - startTime) + " ms.");
+ logger.debug("resolve: {} ms.", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
return new Row(key, resolved);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/service/RowDigestResolver.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/RowDigestResolver.java b/src/java/org/apache/cassandra/service/RowDigestResolver.java
index ad69d4f..7c4e394 100644
--- a/src/java/org/apache/cassandra/service/RowDigestResolver.java
+++ b/src/java/org/apache/cassandra/service/RowDigestResolver.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.service;
import java.nio.ByteBuffer;
+import java.util.concurrent.TimeUnit;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.ReadResponse;
@@ -60,7 +61,7 @@ public class RowDigestResolver extends AbstractRowResolver
if (logger.isDebugEnabled())
logger.debug("resolving " + replies.size() + " responses");
- long startTime = System.currentTimeMillis();
+ long start = System.nanoTime();
// validate digests against each other; throw immediately on mismatch.
// also extract the data reply, if any.
@@ -104,7 +105,7 @@ public class RowDigestResolver extends AbstractRowResolver
}
if (logger.isDebugEnabled())
- logger.debug("resolve: " + (System.currentTimeMillis() - startTime) + " ms.");
+ logger.debug("resolve: {} ms.", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
return new Row(key, data);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 2499caf..906a387 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -201,8 +201,9 @@ public class StorageProxy implements StorageProxyMBean
{
CFMetaData metadata = Schema.instance.getCFMetaData(table, cfName);
- long timedOut = System.currentTimeMillis() + DatabaseDescriptor.getCasContentionTimeout();
- while (System.currentTimeMillis() < timedOut)
+ long start = System.nanoTime();
+ long timeout = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getCasContentionTimeout());
+ while (System.nanoTime() - start < timeout)
{
// for simplicity, we'll do a single liveness check at the start of each attempt
Pair<List<InetAddress>, Integer> p = getPaxosParticipants(table, key);
@@ -1055,7 +1056,8 @@ public class StorageProxy implements StorageProxyMBean
ReadCommand command = commands.get(0);
CFMetaData metadata = Schema.instance.getCFMetaData(command.table, command.cfName);
- long timedOut = System.currentTimeMillis() + DatabaseDescriptor.getCasContentionTimeout();
+ long start = System.nanoTime();
+ long timeout = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getCasContentionTimeout());
while (true)
{
Pair<List<InetAddress>, Integer> p = getPaxosParticipants(command.table, command.key);
@@ -1065,7 +1067,7 @@ public class StorageProxy implements StorageProxyMBean
if (beginAndRepairPaxos(command.key, metadata, liveEndpoints, requiredParticipants) != null)
break;
- if (System.currentTimeMillis() >= timedOut)
+ if (System.nanoTime() - start >= timeout)
throw new WriteTimeoutException(WriteType.CAS, ConsistencyLevel.SERIAL, -1, -1);
}
@@ -1149,7 +1151,7 @@ public class StorageProxy implements StorageProxyMBean
rows.add(row);
}
if (logger.isDebugEnabled())
- logger.debug("Read: " + (System.currentTimeMillis() - exec.handler.startTime) + " ms.");
+ logger.debug("Read: {} ms.", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - exec.handler.start));
}
catch (DigestMismatchException ex)
{
@@ -1233,7 +1235,7 @@ public class StorageProxy implements StorageProxyMBean
{
private final ReadCommand command;
private final ReadCallback<ReadResponse, Row> handler;
- private final long start = System.currentTimeMillis();
+ private final long start = System.nanoTime();
LocalReadRunnable(ReadCommand command, ReadCallback<ReadResponse, Row> handler)
{
@@ -1249,7 +1251,7 @@ public class StorageProxy implements StorageProxyMBean
Table table = Table.open(command.table);
Row r = command.getRow(table);
ReadResponse result = ReadVerbHandler.getResponse(command, r);
- MessagingService.instance().addLatency(FBUtilities.getBroadcastAddress(), System.currentTimeMillis() - start);
+ MessagingService.instance().addLatency(FBUtilities.getBroadcastAddress(), TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
handler.response(result);
}
}
@@ -1258,7 +1260,7 @@ public class StorageProxy implements StorageProxyMBean
{
private final RangeSliceCommand command;
private final ReadCallback<RangeSliceReply, Iterable<Row>> handler;
- private final long start = System.currentTimeMillis();
+ private final long start = System.nanoTime();
LocalRangeSliceRunnable(RangeSliceCommand command, ReadCallback<RangeSliceReply, Iterable<Row>> handler)
{
@@ -1272,7 +1274,7 @@ public class StorageProxy implements StorageProxyMBean
logger.trace("LocalReadRunnable reading {}", command);
RangeSliceReply result = new RangeSliceReply(RangeSliceVerbHandler.executeLocally(command));
- MessagingService.instance().addLatency(FBUtilities.getBroadcastAddress(), System.currentTimeMillis() - start);
+ MessagingService.instance().addLatency(FBUtilities.getBroadcastAddress(), TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
handler.response(result);
}
}
@@ -1754,7 +1756,7 @@ public class StorageProxy implements StorageProxyMBean
*/
private static abstract class DroppableRunnable implements Runnable
{
- private final long constructionTime = System.currentTimeMillis();
+ private final long constructionTime = System.nanoTime();
private final MessagingService.Verb verb;
public DroppableRunnable(MessagingService.Verb verb)
@@ -1764,7 +1766,7 @@ public class StorageProxy implements StorageProxyMBean
public final void run()
{
- if (System.currentTimeMillis() > constructionTime + DatabaseDescriptor.getTimeout(verb))
+ if (TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - constructionTime) > DatabaseDescriptor.getTimeout(verb))
{
MessagingService.instance().incrementDroppedMessages(verb);
return;
@@ -1789,11 +1791,11 @@ public class StorageProxy implements StorageProxyMBean
*/
private static abstract class LocalMutationRunnable implements Runnable
{
- private final long constructionTime = System.currentTimeMillis();
+ private final long constructionTime = System.nanoTime();
public final void run()
{
- if (System.currentTimeMillis() > constructionTime + DatabaseDescriptor.getTimeout(MessagingService.Verb.MUTATION))
+ if (TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - constructionTime) > DatabaseDescriptor.getTimeout(MessagingService.Verb.MUTATION))
{
MessagingService.instance().incrementDroppedMessages(MessagingService.Verb.MUTATION);
HintRunnable runnable = new HintRunnable(FBUtilities.getBroadcastAddress())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/service/StorageService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 7b047da..a597d7f 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -641,7 +641,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
InetAddress existing = tokenMetadata.getEndpoint(token);
if (existing != null)
{
- if (Gossiper.instance.getEndpointStateForEndpoint(existing).getUpdateTimestamp() > (System.currentTimeMillis() - delay))
+ if (delay > TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - Gossiper.instance.getEndpointStateForEndpoint(existing).getUpdateTimestamp()))
throw new UnsupportedOperationException("Cannnot replace a token for a Live node... ");
current.add(existing);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/service/TruncateResponseHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/TruncateResponseHandler.java b/src/java/org/apache/cassandra/service/TruncateResponseHandler.java
index e17211c..3920b91 100644
--- a/src/java/org/apache/cassandra/service/TruncateResponseHandler.java
+++ b/src/java/org/apache/cassandra/service/TruncateResponseHandler.java
@@ -35,7 +35,7 @@ public class TruncateResponseHandler implements IAsyncCallback
protected final SimpleCondition condition = new SimpleCondition();
private final int responseCount;
protected final AtomicInteger responses = new AtomicInteger(0);
- private final long startTime;
+ private final long start;
public TruncateResponseHandler(int responseCount)
{
@@ -44,16 +44,16 @@ public class TruncateResponseHandler implements IAsyncCallback
assert 1 <= responseCount: "invalid response count " + responseCount;
this.responseCount = responseCount;
- startTime = System.currentTimeMillis();
+ start = System.nanoTime();
}
public void get() throws TimeoutException
{
- long timeout = DatabaseDescriptor.getTruncateRpcTimeout() - (System.currentTimeMillis() - startTime);
+ long timeout = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getTruncateRpcTimeout()) - (System.nanoTime() - start);
boolean success;
try
{
- success = condition.await(timeout, TimeUnit.MILLISECONDS); // TODO truncate needs a much longer timeout
+ success = condition.await(timeout, TimeUnit.NANOSECONDS); // TODO truncate needs a much longer timeout
}
catch (InterruptedException ex)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/tools/BulkLoader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tools/BulkLoader.java b/src/java/org/apache/cassandra/tools/BulkLoader.java
index 067cdb4..d2f3c4d 100644
--- a/src/java/org/apache/cassandra/tools/BulkLoader.java
+++ b/src/java/org/apache/cassandra/tools/BulkLoader.java
@@ -118,7 +118,7 @@ public class BulkLoader
static class ProgressIndicator
{
private final Map<InetAddress, Collection<PendingFile>> filesByHost;
- private long startTime;
+ private long start;
private long lastProgress;
private long lastTime;
@@ -129,7 +129,7 @@ public class BulkLoader
public void start()
{
- startTime = System.currentTimeMillis();
+ start = lastTime = System.nanoTime();
}
public boolean printProgress()
@@ -160,15 +160,15 @@ public class BulkLoader
sb.append(" ").append(completed).append("/").append(pendings.size());
sb.append(" (").append(size == 0 ? 100L : progress * 100L / size).append(")] ");
}
- long time = System.currentTimeMillis();
- long deltaTime = time - lastTime;
+ long time = System.nanoTime();
+ long deltaTime = TimeUnit.NANOSECONDS.toMillis(time - lastTime);
lastTime = time;
long deltaProgress = totalProgress - lastProgress;
lastProgress = totalProgress;
sb.append("[total: ").append(totalSize == 0 ? 100L : totalProgress * 100L / totalSize).append(" - ");
sb.append(mbPerSec(deltaProgress, deltaTime)).append("MB/s");
- sb.append(" (avg: ").append(mbPerSec(totalProgress, time - startTime)).append("MB/s)]");
+ sb.append(" (avg: ").append(mbPerSec(totalProgress, TimeUnit.NANOSECONDS.toMillis(time - start))).append("MB/s)]");
System.out.print(sb.toString());
return done;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/tools/SSTableImport.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tools/SSTableImport.java b/src/java/org/apache/cassandra/tools/SSTableImport.java
index f25a60b..5fddd65 100644
--- a/src/java/org/apache/cassandra/tools/SSTableImport.java
+++ b/src/java/org/apache/cassandra/tools/SSTableImport.java
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
@@ -327,7 +328,7 @@ public class SSTableImport
private int importUnsorted(String jsonFile, ColumnFamily columnFamily, String ssTablePath, IPartitioner<?> partitioner) throws IOException
{
int importedKeys = 0;
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
JsonParser parser = getParser(jsonFile);
@@ -369,9 +370,9 @@ public class SSTableImport
importedKeys++;
- long current = System.currentTimeMillis();
+ long current = System.nanoTime();
- if (current - start >= 5000) // 5 secs.
+ if (TimeUnit.NANOSECONDS.toSeconds(current - start) >= 5) // 5 secs.
{
System.out.printf("Currently imported %d keys.%n", importedKeys);
start = current;
@@ -390,7 +391,7 @@ public class SSTableImport
IPartitioner<?> partitioner) throws IOException
{
int importedKeys = 0; // already imported keys count
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
JsonParser parser = getParser(jsonFile);
@@ -449,9 +450,9 @@ public class SSTableImport
importedKeys++;
lineNumber++;
- long current = System.currentTimeMillis();
+ long current = System.nanoTime();
- if (current - start >= 5000) // 5 secs.
+ if (TimeUnit.NANOSECONDS.toSeconds(current - start) >= 5) // 5 secs.
{
System.out.printf("Currently imported %d keys.%n", importedKeys);
start = current;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/utils/ExpiringMap.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/ExpiringMap.java b/src/java/org/apache/cassandra/utils/ExpiringMap.java
index 7ec57ca..7eec40e 100644
--- a/src/java/org/apache/cassandra/utils/ExpiringMap.java
+++ b/src/java/org/apache/cassandra/utils/ExpiringMap.java
@@ -48,12 +48,12 @@ public class ExpiringMap<K, V>
assert value != null;
this.value = value;
this.timeout = timeout;
- this.createdAt = System.currentTimeMillis();
+ this.createdAt = System.nanoTime();
}
- private boolean isReadyToDieAt(long time)
+ private boolean isReadyToDieAt(long atNano)
{
- return ((time - createdAt) > timeout);
+ return atNano - createdAt > TimeUnit.MILLISECONDS.toNanos(timeout);
}
}
@@ -85,7 +85,7 @@ public class ExpiringMap<K, V>
{
public void run()
{
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
int n = 0;
for (Map.Entry<K, CacheableObject<V>> entry : cache.entrySet())
{
@@ -153,6 +153,9 @@ public class ExpiringMap<K, V>
return co == null ? null : co.value;
}
+ /**
+ * @return System.nanoTime() when key was put into the map.
+ */
public long getAge(K key)
{
CacheableObject<V> co = cache.get(key);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/utils/SimpleCondition.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/SimpleCondition.java b/src/java/org/apache/cassandra/utils/SimpleCondition.java
index be7e654..6086f3b 100644
--- a/src/java/org/apache/cassandra/utils/SimpleCondition.java
+++ b/src/java/org/apache/cassandra/utils/SimpleCondition.java
@@ -41,13 +41,12 @@ public class SimpleCondition implements Condition
public synchronized boolean await(long time, TimeUnit unit) throws InterruptedException
{
- // micro/nanoseconds not supported
- assert unit == TimeUnit.DAYS || unit == TimeUnit.HOURS || unit == TimeUnit.MINUTES || unit == TimeUnit.SECONDS || unit == TimeUnit.MILLISECONDS;
-
- long end = System.currentTimeMillis() + unit.convert(time, TimeUnit.MILLISECONDS);
- while (!set && end > System.currentTimeMillis())
+ long start = System.nanoTime();
+ long timeout = unit.toNanos(time);
+ long elapsed;
+ while (!set && (elapsed = System.nanoTime() - start) < timeout)
{
- TimeUnit.MILLISECONDS.timedWait(this, end - System.currentTimeMillis());
+ TimeUnit.NANOSECONDS.timedWait(this, timeout - elapsed);
}
return set;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/src/java/org/apache/cassandra/utils/Throttle.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/Throttle.java b/src/java/org/apache/cassandra/utils/Throttle.java
index fc19a6f..d33fab2 100644
--- a/src/java/org/apache/cassandra/utils/Throttle.java
+++ b/src/java/org/apache/cassandra/utils/Throttle.java
@@ -72,12 +72,12 @@ public class Throttle
logger.debug("{} target throughput now {} bytes/ms.", this, newTargetBytesPerMS);
targetBytesPerMS = newTargetBytesPerMS;
bytesAtLastDelay += bytesDelta;
- timeAtLastDelay = System.currentTimeMillis();
+ timeAtLastDelay = System.nanoTime();
return;
}
// time passed since last delay
- long msSinceLast = System.currentTimeMillis() - timeAtLastDelay;
+ long msSinceLast = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - timeAtLastDelay);
// the excess bytes in this period
long excessBytes = bytesDelta - msSinceLast * targetBytesPerMS;
@@ -91,7 +91,7 @@ public class Throttle
Uninterruptibles.sleepUninterruptibly(timeToDelay, TimeUnit.MILLISECONDS);
}
bytesAtLastDelay += bytesDelta;
- timeAtLastDelay = System.currentTimeMillis();
+ timeAtLastDelay = System.nanoTime();
}
@Override
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/test/long/org/apache/cassandra/db/compaction/LongCompactionsTest.java
----------------------------------------------------------------------
diff --git a/test/long/org/apache/cassandra/db/compaction/LongCompactionsTest.java b/test/long/org/apache/cassandra/db/compaction/LongCompactionsTest.java
index 36f1e71..a1323cb 100644
--- a/test/long/org/apache/cassandra/db/compaction/LongCompactionsTest.java
+++ b/test/long/org/apache/cassandra/db/compaction/LongCompactionsTest.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import org.apache.cassandra.config.Schema;
import org.junit.Test;
@@ -96,7 +97,7 @@ public class LongCompactionsTest extends SchemaLoader
// give garbage collection a bit of time to catch up
Thread.sleep(1000);
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
final int gcBefore = (int) (System.currentTimeMillis() / 1000) - Schema.instance.getCFMetaData(TABLE1, "Standard1").getGcGraceSeconds();
new CompactionTask(store, sstables, gcBefore).execute(null);
System.out.println(String.format("%s: sstables=%d rowsper=%d colsper=%d: %d ms",
@@ -104,7 +105,7 @@ public class LongCompactionsTest extends SchemaLoader
sstableCount,
rowsPerSSTable,
colsPerRow,
- System.currentTimeMillis() - start));
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)));
}
@Test
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/test/long/org/apache/cassandra/utils/LongBitSetTest.java
----------------------------------------------------------------------
diff --git a/test/long/org/apache/cassandra/utils/LongBitSetTest.java b/test/long/org/apache/cassandra/utils/LongBitSetTest.java
index 7941fae..5c5f4e3 100644
--- a/test/long/org/apache/cassandra/utils/LongBitSetTest.java
+++ b/test/long/org/apache/cassandra/utils/LongBitSetTest.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.utils;
import java.util.Random;
+import java.util.concurrent.TimeUnit;
import junit.framework.Assert;
@@ -80,36 +81,36 @@ public class LongBitSetTest
StringBuffer buffer = new StringBuffer();
// start off fresh.
System.gc();
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
for (long i = 0; i < size_to_test; i++)
obs.set(i);
- buffer.append("||").append(System.currentTimeMillis() - start);
+ buffer.append("||").append(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
- start = System.currentTimeMillis();
+ start = System.nanoTime();
for (long i = 0; i < size_to_test; i++)
obs.get(i);
- buffer.append("|").append(System.currentTimeMillis() - start);
+ buffer.append("|").append(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
- start = System.currentTimeMillis();
+ start = System.nanoTime();
for (long i = 0; i < size_to_test; i++)
obs.clear(i);
- buffer.append("|").append(System.currentTimeMillis() - start);
+ buffer.append("|").append(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
System.gc();
- start = System.currentTimeMillis();
+ start = System.nanoTime();
for (long i = 0; i < size_to_test; i++)
offbs.set(i);
- buffer.append("|").append(System.currentTimeMillis() - start);
+ buffer.append("|").append(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
- start = System.currentTimeMillis();
+ start = System.nanoTime();
for (long i = 0; i < size_to_test; i++)
offbs.get(i);
+ buffer.append("|").append(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
- buffer.append("|").append(System.currentTimeMillis() - start);
- start = System.currentTimeMillis();
+ start = System.nanoTime();
for (long i = 0; i < size_to_test; i++)
offbs.clear(i);
- buffer.append("|").append(System.currentTimeMillis() - start).append("|");
+ buffer.append("|").append(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)).append("|");
logger.info(buffer.toString());
// System.out.println(buffer.toString());
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/test/unit/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutorTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutorTest.java b/test/unit/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutorTest.java
index 4b0e9c4..aba84ea 100644
--- a/test/unit/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutorTest.java
+++ b/test/unit/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutorTest.java
@@ -46,7 +46,7 @@ public class DebuggableThreadPoolExecutorTest
Thread.sleep(50);
}
};
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
for (int i = 0; i < 10; i++)
{
executor.execute(runnable);
@@ -54,7 +54,7 @@ public class DebuggableThreadPoolExecutorTest
assert q.size() > 0 : q.size();
while (executor.getCompletedTaskCount() < 10)
continue;
- long delta = System.currentTimeMillis() - start;
+ long delta = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
assert delta >= 9 * 50 : delta;
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/test/unit/org/apache/cassandra/db/CleanupTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/CleanupTest.java b/test/unit/org/apache/cassandra/db/CleanupTest.java
index b4911d7..513db98 100644
--- a/test/unit/org/apache/cassandra/db/CleanupTest.java
+++ b/test/unit/org/apache/cassandra/db/CleanupTest.java
@@ -26,6 +26,7 @@ import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
@@ -104,8 +105,8 @@ public class CleanupTest extends SchemaLoader
assertEquals(LOOPS, rows.size());
SecondaryIndex index = cfs.indexManager.getIndexForColumn(COLUMN);
- long start = System.currentTimeMillis();
- while (!index.isIndexBuilt(COLUMN) && System.currentTimeMillis() < start + 10000)
+ long start = System.nanoTime();
+ while (!index.isIndexBuilt(COLUMN) && System.nanoTime() - start < TimeUnit.SECONDS.toNanos(10))
Thread.sleep(10);
// verify we get it back w/ index query too
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/test/unit/org/apache/cassandra/io/sstable/SSTableUtils.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableUtils.java b/test/unit/org/apache/cassandra/io/sstable/SSTableUtils.java
index c5c41c3..01b174b 100644
--- a/test/unit/org/apache/cassandra/io/sstable/SSTableUtils.java
+++ b/test/unit/org/apache/cassandra/io/sstable/SSTableUtils.java
@@ -199,7 +199,6 @@ public class SSTableUtils
{
File datafile = (dest == null) ? tempSSTableFile(ksname, cfname, generation) : new File(dest.filenameFor(Component.DATA));
SSTableWriter writer = new SSTableWriter(datafile.getAbsolutePath(), expectedSize);
- long start = System.currentTimeMillis();
while (appender.append(writer)) { /* pass */ }
SSTableReader reader = writer.closeAndOpenReader();
// mark all components for removal
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1d2c1224/tools/stress/src/org/apache/cassandra/stress/StressAction.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/StressAction.java b/tools/stress/src/org/apache/cassandra/stress/StressAction.java
index 7de96a5..13ac611 100644
--- a/tools/stress/src/org/apache/cassandra/stress/StressAction.java
+++ b/tools/stress/src/org/apache/cassandra/stress/StressAction.java
@@ -91,7 +91,7 @@ public class StressAction extends Thread
int interval = client.getProgressInterval();
int epochIntervals = client.getProgressInterval() * 10;
- long testStartTime = System.currentTimeMillis();
+ long testStartTime = System.nanoTime();
StressStatistics stats = new StressStatistics(client, output);
@@ -132,7 +132,7 @@ public class StressAction extends Thread
int opDelta = total - oldTotal;
int keyDelta = keyCount - oldKeyCount;
- long currentTimeInSeconds = (System.currentTimeMillis() - testStartTime) / 1000;
+ long currentTimeInSeconds = TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - testStartTime);
output.println(String.format("%d,%d,%d,%.1f,%.1f,%.1f,%d",
total,