From 34e942e0ff1ca76b03a7e634f462aaea40c19104 Mon Sep 17 00:00:00 2001
From: leventov <leventov@ya.ru>
Date: Wed, 6 Jan 2016 03:57:30 +0300
Subject: [PATCH] HCOLL-417 recognize modification iterators appearance
 cross-process

---
 .../chronicle/map/ReplicatedChronicleMap.java | 29 +++++++++++++++++--
 .../map/ReplicatedGlobalMutableState.java     |  4 +++
 ...ateProxyMapUpdatedViaSharedMemoryTest.java |  1 -
 3 files changed, 30 insertions(+), 4 deletions(-)

diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
index 9900527cf..e0695d17f 100644
--- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
@@ -325,12 +325,24 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
             if (modificationIterator != null)
                 return modificationIterator;
 
+            ReplicatedGlobalMutableState globalMutableState = globalMutableState();
             boolean modificationIteratorInit =
-                    globalMutableState().getModificationIteratorInitAt(remoteIdentifier);
+                    globalMutableState.getModificationIteratorInitAt(remoteIdentifier);
             final ModificationIterator modIter =
                     new ModificationIterator(remoteIdentifier, modificationIteratorInit);
-            if (!modificationIteratorInit)
-                globalMutableState().setModificationIteratorInitAt(remoteIdentifier, true);
+            if (!modificationIteratorInit) {
+                globalMutableState.setModificationIteratorInitAt(remoteIdentifier, true);
+                // This doesn't need to be volatile update, because modification iterators count
+                // is checked (also non-volatile) in the beginning of raiseChange()/dropChange()
+                // methods, the risk is that a new modification iterator is added, dirtyEntries()
+                // already completed, and then we call raiseChange() and miss the new iterator.
+                // raiseChange() is called, when the segment lock is held on update level, as well
+                // as dirtyEntries() (that is segment iteration, which is always performed on
+                // update-level lock), so the two actions (dirtyEntries() and raiseChange())
+                // are serialized between each other, => change to ModificationIteratorsCount is
+                // visible.
+                globalMutableState.addModificationIteratorsCount(1);
+            }
             //noinspection unchecked
             assignedModificationIterators = Stream
                     .concat(Arrays.stream(assignedModificationIterators), Stream.of(modIter))
@@ -352,6 +364,13 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
         return assignedModificationIterators;
     }
 
+    private void updateModificationIteratorsArray() {
+        if (globalMutableState().getModificationIteratorsCount() !=
+                assignedModificationIterators.length) {
+            acquireAllModificationIterators();
+        }
+    }
+
     public void raiseChange(long tierIndex, long pos) {
         // -1 is invalid remoteIdentifier => raise change for all
         raiseChangeForAllExcept(tierIndex, pos, (byte) -1);
@@ -362,6 +381,7 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
     }
 
     public void raiseChangeForAllExcept(long tierIndex, long pos, byte remoteIdentifier) {
+        updateModificationIteratorsArray();
         if (tierIndex <= actualSegments) {
             long segmentIndex = tierIndex - 1;
             long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize;
@@ -382,6 +402,7 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
     }
 
     public void dropChange(long tierIndex, long pos) {
+        updateModificationIteratorsArray();
         if (tierIndex <= actualSegments) {
             long segmentIndex = tierIndex - 1;
             long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize;
@@ -404,6 +425,7 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
     }
 
     public void moveChange(long oldTierIndex, long oldPos, long newTierIndex, long newPos) {
+        updateModificationIteratorsArray();
         if (oldTierIndex <= actualSegments) {
             long oldSegmentIndex = oldTierIndex - 1;
             long oldOffsetToTierBitSet = oldSegmentIndex * tierModIterBitSetOuterSize;
@@ -457,6 +479,7 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
     }
 
     public boolean isChanged(long tierIndex, long pos) {
+        updateModificationIteratorsArray();
         if (tierIndex <= actualSegments) {
             long segmentIndex = tierIndex - 1;
             long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize;
diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedGlobalMutableState.java b/src/main/java/net/openhft/chronicle/map/ReplicatedGlobalMutableState.java
index 5a3736f4f..dd9e2de51 100644
--- a/src/main/java/net/openhft/chronicle/map/ReplicatedGlobalMutableState.java
+++ b/src/main/java/net/openhft/chronicle/map/ReplicatedGlobalMutableState.java
@@ -30,4 +30,8 @@ interface ReplicatedGlobalMutableState extends VanillaGlobalMutableState {
     @Array(length = 128)
     boolean getModificationIteratorInitAt(int index);
     void setModificationIteratorInitAt(int index, boolean init);
+
+    @Group(7)
+    int getModificationIteratorsCount();
+    int addModificationIteratorsCount(int addition);
 }
diff --git a/src/test/java/net/openhft/chronicle/map/ReplicateProxyMapUpdatedViaSharedMemoryTest.java b/src/test/java/net/openhft/chronicle/map/ReplicateProxyMapUpdatedViaSharedMemoryTest.java
index 620d00f08..d745de10b 100644
--- a/src/test/java/net/openhft/chronicle/map/ReplicateProxyMapUpdatedViaSharedMemoryTest.java
+++ b/src/test/java/net/openhft/chronicle/map/ReplicateProxyMapUpdatedViaSharedMemoryTest.java
@@ -50,7 +50,6 @@ public class ReplicateProxyMapUpdatedViaSharedMemoryTest {
             TcpTransportAndNetworkConfig serverConfig = TcpTransportAndNetworkConfig.of(7077)
                     .heartBeatInterval(500, TimeUnit.MILLISECONDS);
             replicationMap = ChronicleMapBuilder.of(String.class, String.class)
-                    .entries(1)
                     .replication((byte) 1, serverConfig).createPersistedTo(tempFile);
         }