diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java index de9703502..3f488f4b2 100755 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java @@ -391,7 +391,8 @@ public interface ChronicleHashBuilder, * Configures a time provider, used by hash containers, created by this builder, for needs of * replication consensus protocol (conflicting data updates resolution). * - *

Default time provider is {@link TimeProvider#SYSTEM}. + *

Default time provider uses system time ({@link System#currentTimeMillis()}) in + * microsecond precision. * * @param timeProvider a new time provider for replication needs * @return this builder back diff --git a/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java b/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java index 0e23d014b..0a4d49203 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java @@ -16,84 +16,55 @@ package net.openhft.chronicle.hash.replication; +import net.openhft.chronicle.hash.ChronicleHash; import net.openhft.chronicle.hash.ChronicleHashBuilder; -import net.openhft.chronicle.map.ChronicleMap; -import net.openhft.chronicle.set.ChronicleSet; -import java.io.ObjectStreamException; import java.io.Serializable; import java.util.concurrent.TimeUnit; /** - * TimeProvider was aims to possibly later provide an optimization to - * {@link System#currentTime()} on every call to replicated {@link ChronicleMap} - * or {@link ChronicleSet}. + * TimeProvider abstracts time measurement for {@link ChronicleHash} replication needs. {@code + * TimeProvider} is specified to be used in a replicated {@code ChronicleHash} via {@link + * ChronicleHashBuilder#timeProvider(TimeProvider)} method. The simplest (including default) + * providers rely on system time ({@link System#currentTimeMillis()}), but more complex ones could + * use some atomic counters, or different sources of time. + * + *

Possible reasons to implement a custom time provider: + *

* *

Subclasses should be immutable, because {@link ChronicleHashBuilder} doesn't make defensive * copies. * - * @author Rob Austin. * @see ChronicleHashBuilder#timeProvider(TimeProvider) */ public abstract class TimeProvider implements Serializable { private static final long serialVersionUID = 0L; /** - * Delegates {@link #currentTime()} to {@link System#currentTime()}. + * Returns the current time in this provider's context. */ - public static final TimeProvider SYSTEM = new System(); - public abstract long currentTime(); /** - * {@code timeProvider.scale(System.currentTimeMillis(), TimeUnit.MILLISECONDS)} should - * result to {@link #currentTime()}. + * Returns system time interval (i. e. wall time interval) between two time values, taken using + * this {@code TimeProvider}'s {@link #currentTime()} method, with the highest possible + * precision, in the given time units. + * + * @param earlierTime {@link #currentTime()} result, taken at some moment in the past (earlier) + * @param laterTime {@link #currentTime()} result, taken at some moment in the past, but later + * than {@code earlierTime} was taken + * @param systemTimeIntervalUnit the time units to return system time interval in + * @return wall time interval between the specified moments in the given time unit */ - public abstract long scale(long time, TimeUnit unit); - - /** - * {@code timeProvider.unscale(timeProvider.currentTime(), TimeUnit.NANOSECONDS)} should result - * to something close to {@code System.nanoTime()}. - */ - public abstract long unscale(long time, TimeUnit toUnit); - - private static class System extends TimeProvider { - private static final long serialVersionUID = 1L; - private static final long SCALE = 1000; - - @Override - public long currentTime() { - return java.lang.System.currentTimeMillis() * SCALE; - } - - @Override - public long scale(long time, TimeUnit unit) { - return TimeUnit.MILLISECONDS.convert(time, unit) * SCALE; - } - - @Override - public long unscale(long time, TimeUnit toUnit) { - return toUnit.convert(time, TimeUnit.MILLISECONDS) / SCALE; - } - - @Override - public boolean equals(Object o) { - return o != null && o.getClass() == getClass(); - } - - @Override - public int hashCode() { - return getClass().hashCode(); - } - - @Override - public String toString() { - return getClass().getSimpleName(); - } - - private Object readResolve() throws ObjectStreamException { - return SYSTEM; - } - } + public abstract long systemTimeIntervalBetween( + long earlierTime, long laterTime, TimeUnit systemTimeIntervalUnit); } diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 4152cca5c..7fcbc0250 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -180,7 +180,7 @@ public final class ChronicleMapBuilder implements private boolean removeReturnsNull = false; // replication - private TimeProvider timeProvider = TimeProvider.SYSTEM; + private TimeProvider timeProvider = MicrosecondPrecisionSystemTimeProvider.instance(); long cleanupTimeout = 1; private boolean cleanupRemovedEntries = true; TimeUnit cleanupTimeoutUnit = TimeUnit.SECONDS; @@ -1724,8 +1724,7 @@ public final class ChronicleMapBuilder implements map.globalStateLock(); try { if (!map.globalMutableState.isCurrentlyCleanupIterated()) { - OldDeletedEntriesCleanup cleanup = new OldDeletedEntriesCleanup(map, - timeProvider.scale(cleanupTimeout, cleanupTimeoutUnit)); + OldDeletedEntriesCleanup cleanup = new OldDeletedEntriesCleanup(map); NamedThreadFactory threadFactory = new NamedThreadFactory("cleanup thread for map persisted at " + map.file()); ExecutorService executor = Executors.newSingleThreadExecutor(threadFactory); diff --git a/src/main/java/net/openhft/chronicle/map/MicrosecondPrecisionSystemTimeProvider.java b/src/main/java/net/openhft/chronicle/map/MicrosecondPrecisionSystemTimeProvider.java new file mode 100644 index 000000000..b3cf5ea43 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/MicrosecondPrecisionSystemTimeProvider.java @@ -0,0 +1,69 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program. If not, see . + */ + +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.replication.TimeProvider; + +import java.io.ObjectStreamException; +import java.util.concurrent.TimeUnit; + +import static java.util.concurrent.TimeUnit.MICROSECONDS; +import static java.util.concurrent.TimeUnit.MILLISECONDS; + +final class MicrosecondPrecisionSystemTimeProvider extends TimeProvider { + private static final long serialVersionUID = 0L; + + private static final MicrosecondPrecisionSystemTimeProvider INSTANCE = + new MicrosecondPrecisionSystemTimeProvider(); + + public static MicrosecondPrecisionSystemTimeProvider instance() { + return INSTANCE; + } + + private MicrosecondPrecisionSystemTimeProvider() {} + + @Override + public long currentTime() { + return MILLISECONDS.toMicros(System.currentTimeMillis()); + } + + @Override + public long systemTimeIntervalBetween( + long earlierTimeMicros, long laterTimeMicros, TimeUnit systemTimeIntervalUnit) { + long intervalMicros = laterTimeMicros - earlierTimeMicros; + return systemTimeIntervalUnit.convert(intervalMicros, MICROSECONDS); + } + + @Override + public boolean equals(Object o) { + return o != null && o.getClass() == getClass(); + } + + @Override + public int hashCode() { + return getClass().hashCode(); + } + + @Override + public String toString() { + return getClass().getSimpleName(); + } + + private Object readResolve() throws ObjectStreamException { + return INSTANCE; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java b/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java index fbdf70d63..c363917fd 100644 --- a/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java +++ b/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java @@ -32,16 +32,14 @@ public class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate< private static final Logger LOG = LoggerFactory.getLogger(OldDeletedEntriesCleanup.class); private final ReplicatedChronicleMap map; - private final long timeout; private final int[] segmentsPermutation; private final int[] inverseSegmentsPermutation; private volatile boolean shutdown; private long prevSegment0ScanStart = -1; private long removedCompletely; - public OldDeletedEntriesCleanup(ReplicatedChronicleMap map, long timeout) { + public OldDeletedEntriesCleanup(ReplicatedChronicleMap map) { this.map = map; - this.timeout = timeout; segmentsPermutation = randomPermutation(map.segments()); inverseSegmentsPermutation = inversePermutation(segmentsPermutation); } @@ -53,11 +51,13 @@ public class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate< TimeProvider timeProvider = map.timeProvider; if (segmentIndex == 0 && prevSegment0ScanStart >= 0) { long currentTime = timeProvider.currentTime(); - long mapScanTime = currentTime - prevSegment0ScanStart; - LOG.debug("Old deleted entries scan time: {}", mapScanTime); - if (mapScanTime < timeout) { - long timeToSleep = - timeProvider.unscale(timeout - mapScanTime, TimeUnit.MILLISECONDS); + TimeUnit cleanupTimeoutUnit = map.cleanupTimeoutUnit; + long mapScanTime = timeProvider.systemTimeIntervalBetween( + prevSegment0ScanStart, currentTime, cleanupTimeoutUnit); + LOG.debug("Old deleted entries scan time: {} {}", mapScanTime, cleanupTimeoutUnit); + long cleanupTimeout = map.cleanupTimeout; + if (mapScanTime < cleanupTimeout) { + long timeToSleep = cleanupTimeoutUnit.toMillis(cleanupTimeout - mapScanTime); if (timeToSleep > 0) { try { Thread.sleep(timeToSleep); @@ -66,8 +66,7 @@ public class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate< } } else { - timeToSleep = timeProvider.unscale(timeout - mapScanTime, - TimeUnit.NANOSECONDS); + timeToSleep = cleanupTimeoutUnit.toNanos(cleanupTimeout - mapScanTime); try { Thread.sleep(0, (int) timeToSleep); } catch (InterruptedException e) { @@ -96,8 +95,9 @@ public class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate< if (shutdown) return false; if (e instanceof MapAbsentEntry) { - long deleteTimeout = map.timeProvider.currentTime() - e.originTimestamp(); - if (deleteTimeout > timeout && !e.isChanged()) { + long deleteTimeout = map.timeProvider.systemTimeIntervalBetween( + e.originTimestamp(), map.timeProvider.currentTime(), map.cleanupTimeoutUnit); + if (deleteTimeout > map.cleanupTimeout && !e.isChanged()) { e.doRemoveCompletely(); removedCompletely++; } diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 44a0d5f0a..a12ab6018 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -42,6 +42,7 @@ import java.io.IOException; import java.io.ObjectInputStream; import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReferenceArray; @@ -113,6 +114,7 @@ public class ReplicatedChronicleMap remoteOperations; transient CompiledReplicatedMapQueryContext remoteOpContext; @@ -133,7 +135,8 @@ public class ReplicatedChronicleMap extends MapEntryStages moments = new ArrayList<>(); Random rnd = new Random(4); + { + moments.add(System.currentTimeMillis()); + } + @Override public long currentTime() { - if (rnd.nextBoolean()) + if (rnd.nextBoolean()) { + moments.add(System.currentTimeMillis()); return t++; - else + } else { return t; + } } @Override - public long scale(long time, TimeUnit unit) { - return TimeUnit.MILLISECONDS.convert(time, unit); - } - - @Override - public long unscale(long time, TimeUnit toUnit) { - return toUnit.convert(time, TimeUnit.MILLISECONDS); + public long systemTimeIntervalBetween( + long earlierTime, long laterTime, TimeUnit systemTimeIntervalUnit) { + long earlierMillis = moments.get((int) earlierTime); + long laterMillis = moments.get((int) laterTime); + long intervalMillis = laterMillis - earlierMillis; + return systemTimeIntervalUnit.convert(intervalMillis, TimeUnit.MILLISECONDS); } };