Replication updates with local identifier shouldn't use the same timestamp, because this allows two different updates which replace each other on different nodes, and don't converge. Extend default TimeProvider precision to nanoseconds, to exclude possibility of 'microsecond overflow' on frequent changes to the same entry

This commit is contained in:
leventov
2015-12-26 17:28:35 +03:00
parent 0ead52335b
commit 1743f1c694
5 changed files with 14 additions and 19 deletions
@@ -396,7 +396,7 @@ public interface ChronicleHashBuilder<K, H extends ChronicleHash<K, ?, ?, ?>,
* replication consensus protocol (conflicting data updates resolution).
*
* <p>Default time provider uses system time ({@link System#currentTimeMillis()}) in
* <i>microsecond</i> precision.
* <i>nanosecond</i> precision.
*
* @param timeProvider a new time provider for replication needs
* @return this builder back
@@ -49,7 +49,7 @@ public final class DefaultEventualConsistencyStrategy {
long originTimestamp = entry.originTimestamp();
boolean shouldAccept = remoteTimestamp > originTimestamp ||
(remoteTimestamp == originTimestamp &&
context.remoteIdentifier() <= entry.originIdentifier());
context.remoteIdentifier() < entry.originIdentifier());
return shouldAccept ? AcceptanceDecision.ACCEPT : AcceptanceDecision.DISCARD;
}
@@ -184,7 +184,7 @@ public final class ChronicleMapBuilder<K, V> implements
private boolean removeReturnsNull = false;
// replication
private TimeProvider timeProvider = MicrosecondPrecisionSystemTimeProvider.instance();
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.
@@ -21,31 +21,31 @@ 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.NANOSECONDS;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
final class MicrosecondPrecisionSystemTimeProvider extends TimeProvider {
final class NanosecondPrecisionSystemTimeProvider extends TimeProvider {
private static final long serialVersionUID = 0L;
private static final MicrosecondPrecisionSystemTimeProvider INSTANCE =
new MicrosecondPrecisionSystemTimeProvider();
private static final NanosecondPrecisionSystemTimeProvider INSTANCE =
new NanosecondPrecisionSystemTimeProvider();
public static MicrosecondPrecisionSystemTimeProvider instance() {
public static NanosecondPrecisionSystemTimeProvider instance() {
return INSTANCE;
}
private MicrosecondPrecisionSystemTimeProvider() {}
private NanosecondPrecisionSystemTimeProvider() {}
@Override
public long currentTime() {
return MILLISECONDS.toMicros(System.currentTimeMillis());
return MILLISECONDS.toNanos(System.currentTimeMillis());
}
@Override
public long systemTimeIntervalBetween(
long earlierTimeMicros, long laterTimeMicros, TimeUnit systemTimeIntervalUnit) {
long intervalMicros = laterTimeMicros - earlierTimeMicros;
return systemTimeIntervalUnit.convert(intervalMicros, MICROSECONDS);
long earlierTimeNanos, long laterTimeNanos, TimeUnit systemTimeIntervalUnit) {
long intervalNanos = laterTimeNanos - earlierTimeNanos;
return systemTimeIntervalUnit.convert(intervalNanos, NANOSECONDS);
}
@Override
@@ -128,12 +128,7 @@ public abstract class ReplicatedMapEntryStages<K, V> extends MapEntryStages<K, V
public void updatedReplicationStateOnPresentEntry() {
if (!ru.replicationUpdateInit()) {
s.innerWriteLock.lock();
long timestamp;
if (identifier() != mh.m().identifier()) {
timestamp = Math.max(timestamp() + 1, mh.m().timeProvider.currentTime());
} else {
timestamp = mh.m().timeProvider.currentTime();
}
long timestamp = Math.max(timestamp() + 1, mh.m().timeProvider.currentTime());
updateReplicationState(mh.m().identifier(), timestamp);
}
}