Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Make TimeProvider non-extensible and non-configurable; Fix a bug in

Browse Source 
ReplicatedQueryAlloc.CleanupAction
This commit is contained in:
leventov
2015-12-27 21:32:26 +03:00
parent 123fb177c6
commit 1221f11097
16 changed files with 116 additions and 244 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
-14
View File
@@ -21,7 +21,6 @@ import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.TimeProvider;
import net.openhft.chronicle.hash.serialization.*;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
@@ -391,19 +390,6 @@ public interface ChronicleHashBuilder<K, H extends ChronicleHash<K, ?, ?, ?>,
*/
B actualSegments(int actualSegments);
/**
* Configures a time provider, used by hash containers, created by this builder, for needs of
* replication consensus protocol (conflicting data updates resolution).
*
* <p>Default time provider uses system time ({@link System#currentTimeMillis()}) in
* <i>nanosecond</i> precision, also it returns non-decreasing time.
*
* @param timeProvider a new time provider for replication needs
* @return this builder back
* @see #replication(SingleChronicleHashReplication)
*/
B timeProvider(TimeProvider timeProvider);
/**
* Configures timeout after which entries, marked as removed in the Chronicle Hash, constructed
* by this builder, are allowed to be completely removed from the data structure. In replicated
src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java
+1 -1
View File
@@ -66,7 +66,7 @@ public final class DefaultEventualConsistencyStrategy {
return DISCARD;
// remoteTimestamp == originTimestamp && remoteIdentifier == originIdentifier below
// This could be, only if a node with the origin identifier was lost, a new Chronicle Hash
// instance was started up, but with timeProvider which for some reason is very late, so
// instance was started up, but with system time which for some reason is very late, so
// that it provides the same time, as the "old" node with this identifier, before it was
// lost. (This is almost a theoretical situation.) In this case, give advantage to fresh
// entry updates to the "new" node. Entries with the same id and timestamp, bootstrapped
src/main/java/net/openhft/chronicle/hash/replication/RemoteOperationContext.java
+4 -1
View File
@@ -27,7 +27,10 @@ import net.openhft.chronicle.hash.HashContext;
public interface RemoteOperationContext<K> extends HashContext<K> {
byte remoteIdentifier();
/**
* @see ReplicableEntry#originTimestamp()
*/
long remoteTimestamp();
byte currentNodeIdentifier();
src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java
+2 -10
View File
@@ -51,20 +51,12 @@ public interface ReplicableEntry {
byte originIdentifier();
/**
* The timestamp, associated with this entry. Originally, and by default, this timestamp means
* the time of the last.
* The timestamp of the last update to this entry.
*
* <p>When the entry is replicated, this timestamp comes to remote nodes as {@link
* RemoteOperationContext#remoteTimestamp()}.
*
* <p>On {@code ChronicleHash} local operations with entries, like {@link ChronicleMap#put(
* Object, Object)}, this timestamp is overwritten using {@linkplain
* ChronicleHashBuilder#timeProvider(TimeProvider) the configured time provider}. On remote
* operations, proxied through {@link MapRemoteOperations} or {@link SetRemoteOperations},
* you are free to overwrite this timestamp by any value, using {@link #updateOrigin(byte, long)
* } method.
*
* @return the timestamp, associated with this entry
* @return the timestamp of the last update to this entry
* @see TimeProvider
*/
long originTimestamp();
src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java
+42 -30
View File
@@ -19,52 +19,64 @@ package net.openhft.chronicle.hash.replication;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import java.io.Serializable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
/**
* 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.
* {@code TimeProvider} encapsulates time measurement for {@link ChronicleHash} replication needs.
* It is used to obtain timestamps of entry updates, and to determine, when deleted entries become
* eligible for complete purge from the Chronicle Hash data structure.
*
* <p>Possible reasons to implement a custom time provider:
* <ul>
* <li>To synchronize time between replicated nodes, when precision of the system time is
* not sufficient (it could diverge on remote servers).</li>
* <li>Optimization, if {@code System.currentTimeMillis()} is considered too slow</li>
* <li>If time machinery in replicated {@code ChronicleHash} system is completely hijacked,
* and used to store some {@code long} value with non-time meaning (see {@link
* ReplicableEntry#originTimestamp()}, {@link ReplicableEntry#updateOrigin(byte, long)},
* {@link RemoteOperationContext#remoteTimestamp()}).</li>
* </ul>
*
* <p>Subclasses should be immutable, because {@link ChronicleHashBuilder} doesn't make defensive
* copies.
*
* @see ChronicleHashBuilder#timeProvider(TimeProvider)
* @see ReplicableEntry#originTimestamp()
* @see RemoteOperationContext#remoteTimestamp()
* @see ChronicleHashBuilder#removedEntryCleanupTimeout(long, TimeUnit)
*/
public abstract class TimeProvider implements Serializable {
private static final long serialVersionUID = 0L;
public final class TimeProvider {
private TimeProvider() {}
private static final AtomicLong lastTimeHolder = new AtomicLong();
/**
* Returns the current time in this provider's context.
* Returns a non-decreasing number, assumed to be used as a "timestamp".
*
* <p>Approximate system time interval between two calls of this method is retrievable via
* {@link #systemTimeIntervalBetween(long, long, TimeUnit)}, applied to the returned values
* from those {@code currentTime()} calls.
*
* <p>Safe and scalable for concurrent use from multiple threads.
*
* @return the current timestamp
*/
public abstract long currentTime();
public static long currentTime() {
long now = MILLISECONDS.toNanos(System.currentTimeMillis());
while (true) {
long lastTime = lastTimeHolder.get();
if (now <= lastTime)
return lastTime;
if (lastTimeHolder.compareAndSet(lastTime, now))
return now;
}
}
/**
* 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.
* {@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
* than {@code earlierTime} was taken ("later" means there is a happens-before relationship
* between the two subject {@code currentTime()} calls)
* @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 systemTimeIntervalBetween(
long earlierTime, long laterTime, TimeUnit systemTimeIntervalUnit);
public static long systemTimeIntervalBetween(
long earlierTime, long laterTime, TimeUnit systemTimeIntervalUnit) {
long intervalNanos = laterTime - earlierTime;
return systemTimeIntervalUnit.convert(intervalNanos, NANOSECONDS);
}
}
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
-13
View File
@@ -183,8 +183,6 @@ public final class ChronicleMapBuilder<K, V> implements
private boolean putReturnsNull = false;
private boolean removeReturnsNull = false;
// replication
private TimeProvider timeProvider = NanosecondPrecisionSystemTimeProvider.instance();
/**
* Default timeout is 1 minute. Even loopback tests converge often in the course of seconds,
* let alone WAN replication over many nodes might take tens of seconds.
@@ -1134,7 +1132,6 @@ public final class ChronicleMapBuilder<K, V> implements
", entries=" + entries() +
", putReturnsNull=" + putReturnsNull() +
", removeReturnsNull=" + removeReturnsNull() +
", timeProvider=" + timeProvider() +
", keyBuilder=" + keyBuilder +
", valueBuilder=" + valueBuilder +
'}';
@@ -1151,12 +1148,6 @@ public final class ChronicleMapBuilder<K, V> implements
return toString().hashCode();
}
@Override
public ChronicleMapBuilder<K, V> timeProvider(TimeProvider timeProvider) {
this.timeProvider = timeProvider;
return this;
}
@Override
public ChronicleMapBuilder<K, V> removedEntryCleanupTimeout(
long removedEntryCleanupTimeout, TimeUnit unit) {
@@ -1175,10 +1166,6 @@ public final class ChronicleMapBuilder<K, V> implements
return this;
}
TimeProvider timeProvider() {
return timeProvider;
}
@Override
public ChronicleMapBuilder<K, V> keyReaderAndDataAccess(
SizedReader<K> keyReader, @NotNull DataAccess<K> keyDataAccess) {
src/main/java/net/openhft/chronicle/map/NanosecondPrecisionSystemTimeProvider.java
-79
View File
@@ -1,79 +0,0 @@
/*
* 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 <http://www.gnu.org/licenses/>.
*/
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.TimeProvider;
import java.io.ObjectStreamException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
final class NanosecondPrecisionSystemTimeProvider extends TimeProvider {
private static final long serialVersionUID = 0L;
private static final AtomicLong lastTimeHolder = new AtomicLong();
private static final NanosecondPrecisionSystemTimeProvider INSTANCE =
new NanosecondPrecisionSystemTimeProvider();
public static NanosecondPrecisionSystemTimeProvider instance() {
return INSTANCE;
}
private NanosecondPrecisionSystemTimeProvider() {}
@Override
public long currentTime() {
long now = MILLISECONDS.toNanos(System.currentTimeMillis());
while (true) {
long lastTime = lastTimeHolder.get();
if (now <= lastTime)
return lastTime;
if (lastTimeHolder.compareAndSet(lastTime, now))
return now;
}
}
@Override
public long systemTimeIntervalBetween(
long earlierTimeNanos, long laterTimeNanos, TimeUnit systemTimeIntervalUnit) {
long intervalNanos = laterTimeNanos - earlierTimeNanos;
return systemTimeIntervalUnit.convert(intervalNanos, NANOSECONDS);
}
@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;
}
}
src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java
+8 -7
View File
@@ -19,7 +19,6 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.core.io.Closeable;
import net.openhft.chronicle.hash.ReplicatedHashSegmentContext;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import net.openhft.chronicle.hash.replication.TimeProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,6 +28,9 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Predicate;
import static net.openhft.chronicle.hash.replication.TimeProvider.currentTime;
import static net.openhft.chronicle.hash.replication.TimeProvider.systemTimeIntervalBetween;
class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate<ReplicableEntry> {
private static final Logger LOG = LoggerFactory.getLogger(OldDeletedEntriesCleanup.class);
@@ -52,10 +54,9 @@ class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate<Replica
while (!shutdown) {
int segmentIndex = map.globalMutableState().getCurrentCleanupSegmentIndex();
int nextSegmentIndex;
TimeProvider timeProvider = map.timeProvider;
try (MapSegmentContext<?, ?, ?> context = map.segmentContext(segmentIndex)) {
if (segmentIndex == 0)
prevSegment0ScanStart = timeProvider.currentTime();
prevSegment0ScanStart = currentTime();
removedCompletely = 0;
if (((ReplicatedHashSegmentContext<?, ?>) context)
.forEachSegmentReplicableEntryWhile(this)) {
@@ -70,9 +71,9 @@ class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate<Replica
}
}
if (nextSegmentIndex == 0) {
long currentTime = timeProvider.currentTime();
long currentTime = currentTime();
TimeUnit cleanupTimeoutUnit = map.cleanupTimeoutUnit;
long mapScanTime = timeProvider.systemTimeIntervalBetween(
long mapScanTime = systemTimeIntervalBetween(
prevSegment0ScanStart, currentTime, cleanupTimeoutUnit);
LOG.debug("Old deleted entries scan time: {} {}", mapScanTime, cleanupTimeoutUnit);
long cleanupTimeout = map.cleanupTimeout;
@@ -93,8 +94,8 @@ class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate<Replica
if (shutdown)
return false;
if (e instanceof MapAbsentEntry) {
long deleteTimeout = map.timeProvider.systemTimeIntervalBetween(
e.originTimestamp(), map.timeProvider.currentTime(), map.cleanupTimeoutUnit);
long deleteTimeout = systemTimeIntervalBetween(
e.originTimestamp(), currentTime(), map.cleanupTimeoutUnit);
if (deleteTimeout > map.cleanupTimeout && !e.isChanged()) {
e.doRemoveCompletely();
removedCompletely++;
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+2 -7
View File
@@ -26,7 +26,6 @@ import net.openhft.chronicle.hash.impl.VanillaChronicleHash;
import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface;
import net.openhft.chronicle.hash.replication.AbstractReplication;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import net.openhft.chronicle.hash.replication.TimeProvider;
import net.openhft.chronicle.map.impl.CompiledReplicatedMapIterationContext;
import net.openhft.chronicle.map.impl.CompiledReplicatedMapQueryContext;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
@@ -48,6 +47,7 @@ import static net.openhft.chronicle.algo.MemoryUnit.*;
import static net.openhft.chronicle.algo.bitset.BitSetFrame.NOT_FOUND;
import static net.openhft.chronicle.algo.bytes.Access.checkedBytesStoreAccess;
import static net.openhft.chronicle.algo.bytes.Access.nativeAccess;
import static net.openhft.chronicle.hash.replication.TimeProvider.currentTime;
/**
* <h2>A Replicating Multi Master HashMap</h2> <p>Each remote hash map, mirrors its changes over to
@@ -97,7 +97,6 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
private static final long LAST_UPDATED_HEADER_SIZE = 128L * 8L;
private static final int SIZE_OF_BOOTSTRAP_TIME_STAMP = 8;
public transient TimeProvider timeProvider;
/**
* Default value is 0, that corresponds to "unset" identifier value (valid ids are positive)
*/
@@ -168,7 +167,6 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
void initTransientsFromBuilder(ChronicleMapBuilder<K, V> builder) {
super.initTransientsFromBuilder(builder);
this.remoteOperations = (MapRemoteOperations<K, V, R>) builder.remoteOperations;
this.timeProvider = builder.timeProvider();
cleanupRemovedEntries = builder.cleanupRemovedEntries;
cleanupTimeout = builder.cleanupTimeout;
cleanupTimeoutUnit = builder.cleanupTimeoutUnit;
@@ -490,7 +488,7 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
// if it is set as ZERO then the onPut() will set it to the current time, once the
// consumer has cycled through the bit set the timestamp will be set back to zero.
private AtomicLong bootStrapTimeStamp = new AtomicLong();
private long lastBootStrapTimeStamp = timeProvider.currentTime();
private long lastBootStrapTimeStamp = currentTime();
// records the current position of the cursor in the bitset
// TODO volatile?
@@ -551,9 +549,6 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
} else {
raiseExtraTierChange(tierIndex, bitIndex);
}
// assert timestamp > timeProvider.currentTime() - TimeUnit.SECONDS.toMillis(1) &&
// timestamp <= timeProvider.currentTime() : "timeStamp=" + timestamp + ", " +
// "currentTime=" + timeProvider.currentTime();
// todo improve this - use the timestamp from the entry its self
bootStrapTimeStamp.compareAndSet(0, timestamp);
if (modificationNotifier != null)
src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java
+3 -2
View File
@@ -25,6 +25,7 @@ import net.openhft.sg.Stage;
import net.openhft.sg.StageRef;
import net.openhft.sg.Staged;
import static net.openhft.chronicle.hash.replication.TimeProvider.currentTime;
import static net.openhft.chronicle.map.ReplicatedChronicleMap.ADDITIONAL_ENTRY_BYTES;
@Staged
@@ -128,7 +129,7 @@ public abstract class ReplicatedMapEntryStages<K, V> extends MapEntryStages<K, V
public void updatedReplicationStateOnPresentEntry() {
if (!ru.replicationUpdateInit()) {
s.innerWriteLock.lock();
long timestamp = Math.max(timestamp() + 1, mh.m().timeProvider.currentTime());
long timestamp = Math.max(timestamp() + 1, currentTime());
updateReplicationState(mh.m().identifier(), timestamp);
}
}
@@ -136,7 +137,7 @@ public abstract class ReplicatedMapEntryStages<K, V> extends MapEntryStages<K, V
public void updatedReplicationStateOnAbsentEntry() {
if (!ru.replicationUpdateInit()) {
s.innerWriteLock.lock();
updateReplicationState(mh.m().identifier(), mh.m().timeProvider.currentTime());
updateReplicationState(mh.m().identifier(), currentTime());
}
}
src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicatedQueryAlloc.java
+14 -9
View File
@@ -30,6 +30,9 @@ import net.openhft.sg.Staged;
import java.util.function.Consumer;
import static net.openhft.chronicle.hash.replication.TimeProvider.currentTime;
import static net.openhft.chronicle.hash.replication.TimeProvider.systemTimeIntervalBetween;
@Staged
public class ReplicatedQueryAlloc extends QueryAlloc {
@@ -64,15 +67,17 @@ public class ReplicatedQueryAlloc extends QueryAlloc {
@Override
public void accept(ReplicableEntry e) {
ReplicatedChronicleMap<?, ?, ?> map = mh.m();
if (e instanceof MapAbsentEntry && iterationContext.pos() != posToSkip) {
long deleteTimeout = map.timeProvider.currentTime() - e.originTimestamp();
map.timeProvider.systemTimeIntervalBetween(e.originTimestamp(),
map.timeProvider.currentTime(), map.cleanupTimeoutUnit);
if (deleteTimeout > map.cleanupTimeout && !e.isChanged()) {
e.doRemoveCompletely();
removedCompletely++;
}
}
if (!(e instanceof MapAbsentEntry) || iterationContext.pos() == posToSkip)
return;
long currentTime = currentTime();
if (e.originTimestamp() > currentTime)
return; // presumably unsynchronized clocks
long deleteTimeout = systemTimeIntervalBetween(
e.originTimestamp(), currentTime, map.cleanupTimeoutUnit);
if (deleteTimeout <= map.cleanupTimeout || e.isChanged())
return;
e.doRemoveCompletely();
removedCompletely++;
}
}
src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java
-7
View File
@@ -22,7 +22,6 @@ import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder;
import net.openhft.chronicle.hash.Data;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.TimeProvider;
import net.openhft.chronicle.hash.serialization.*;
import net.openhft.chronicle.map.*;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
@@ -199,12 +198,6 @@ public final class ChronicleSetBuilder<K>
return toString().hashCode();
}
@Override
public ChronicleSetBuilder<K> timeProvider(TimeProvider timeProvider) {
chronicleMapBuilder.timeProvider(timeProvider);
return this;
}
@Override
public ChronicleSetBuilder<K> removedEntryCleanupTimeout(
long removedEntryCleanupTimeout, TimeUnit unit) {