From 1743f1c69482f2773be772bf5d146fd8e1ba5533 Mon Sep 17 00:00:00 2001
From: leventov <leventov@ya.ru>
Date: Sat, 26 Dec 2015 17:27:30 +0300
Subject: [PATCH] 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

---
 .../chronicle/hash/ChronicleHashBuilder.java  |  2 +-
 .../DefaultEventualConsistencyStrategy.java   |  2 +-
 .../chronicle/map/ChronicleMapBuilder.java    |  2 +-
 ...anosecondPrecisionSystemTimeProvider.java} | 20 +++++++++----------
 .../stage/entry/ReplicatedMapEntryStages.java |  7 +------
 5 files changed, 14 insertions(+), 19 deletions(-)
 rename src/main/java/net/openhft/chronicle/map/{MicrosecondPrecisionSystemTimeProvider.java => NanosecondPrecisionSystemTimeProvider.java} (69%)

diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
index 5a1381679..74de5104c 100755
--- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
+++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
@@ -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
diff --git a/src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java b/src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java
index 2a3981de2..fc61b1249 100644
--- a/src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java
+++ b/src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java
@@ -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;
     }
     
diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
index a5ce0d020..53c01c37e 100755
--- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
@@ -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.
diff --git a/src/main/java/net/openhft/chronicle/map/MicrosecondPrecisionSystemTimeProvider.java b/src/main/java/net/openhft/chronicle/map/NanosecondPrecisionSystemTimeProvider.java
similarity index 69%
rename from src/main/java/net/openhft/chronicle/map/MicrosecondPrecisionSystemTimeProvider.java
rename to src/main/java/net/openhft/chronicle/map/NanosecondPrecisionSystemTimeProvider.java
index b3cf5ea43..0dab2e3f8 100644
--- a/src/main/java/net/openhft/chronicle/map/MicrosecondPrecisionSystemTimeProvider.java
+++ b/src/main/java/net/openhft/chronicle/map/NanosecondPrecisionSystemTimeProvider.java
@@ -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
diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java
index 8c2503f44..3eff2a9dc 100644
--- a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java
+++ b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java
@@ -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);
         }
     }