Make checksum computation delayed, on lock release

pom.xml

@@ -152,7 +152,7 @@
         <dependency>
             <groupId>net.openhft</groupId>
             <artifactId>stage-compiler</artifactId>
-            <version>1.8</version>
+            <version>1.9</version>
             <scope>provided</scope>
         </dependency>
 

src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ChecksumStrategy.java

@@ -16,9 +16,7 @@
 
 package net.openhft.chronicle.hash.impl.stage.entry;
 
-import net.openhft.chronicle.hash.ChecksumEntry;
-
-public interface ChecksumStrategy extends ChecksumEntry {
+public interface ChecksumStrategy {
 
     int CHECKSUM_STORED_BYTES = 4;
 

src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryChecksumStrategy.java

@@ -17,7 +17,6 @@
 package net.openhft.chronicle.hash.impl.stage.entry;
 
 import net.openhft.chronicle.algo.hashing.LongHashFunction;
-import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
 import net.openhft.sg.StageRef;
 import net.openhft.sg.Staged;
 
@@ -26,7 +25,6 @@ import static net.openhft.chronicle.hash.impl.stage.entry.ChecksumHashing.hash8T
 @Staged
 public class HashEntryChecksumStrategy implements ChecksumStrategy {
 
-    @StageRef CheckOnEachPublicOperation checkOnEachPublicOperation;
     @StageRef SegmentStages s;
     @StageRef HashEntryStages<?> e;
     @StageRef KeyHashCode h;
@@ -55,12 +53,6 @@ public class HashEntryChecksumStrategy implements ChecksumStrategy {
         return (int) ((checksum >>> 32) ^ checksum);
     }
 
-    @Override
-    public void updateChecksum() {
-        checkOnEachPublicOperation.checkOnEachPublicOperation();
-        computeAndStoreChecksum();
-    }
-
     @Override
     public boolean innerCheckSum() {
         int oldChecksum = s.segmentBS.readInt(e.entryEnd());
@@ -68,12 +60,6 @@ public class HashEntryChecksumStrategy implements ChecksumStrategy {
         return oldChecksum == checksum;
     }
 
-    @Override
-    public boolean checkSum() {
-        checkOnEachPublicOperation.checkOnEachPublicOperation();
-        return innerCheckSum();
-    }
-
     @Override
     public long extraEntryBytes() {
         return CHECKSUM_STORED_BYTES;

src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryStages.java

@@ -21,6 +21,7 @@ import net.openhft.chronicle.bytes.Bytes;
 import net.openhft.chronicle.hash.ChecksumEntry;
 import net.openhft.chronicle.hash.Data;
 import net.openhft.chronicle.hash.HashEntry;
+import net.openhft.chronicle.hash.impl.LocalLockState;
 import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder;
 import net.openhft.chronicle.hash.impl.stage.data.bytes.EntryKeyBytesData;
 import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
@@ -51,6 +52,8 @@ public abstract class HashEntryStages<K> implements HashEntry<K>, ChecksumEntry
 
     @Stage("EntryOffset") public long keySizeOffset = -1;
 
+    public abstract boolean entryOffsetInit();
+
     public void initEntryOffset() {
         keySizeOffset = s.entrySpaceOffset + pos * hh.h().chunkSize;
     }
@@ -67,6 +70,8 @@ public abstract class HashEntryStages<K> implements HashEntry<K>, ChecksumEntry
         this.keyOffset = keyOffset;
     }
 
+    public abstract void closeKeyOffset();
+
     public void readExistingEntry(long pos) {
         initPos(pos);
         Bytes segmentBytes = s.segmentBytesForRead();
@@ -111,14 +116,43 @@ public abstract class HashEntryStages<K> implements HashEntry<K>, ChecksumEntry
     public final ChecksumStrategy checksumStrategy = hh.h().checksumEntries ?
             hashEntryChecksumStrategy : NoChecksumStrategy.INSTANCE;
 
+    public boolean delayedUpdateChecksum = false;
+
+    public void initDelayedUpdateChecksum(boolean delayedUpdateChecksum) {
+        // makes delayedUpdateChecksum dependent on keySizeOffset and Locks stages, to trigger
+        // delayedUpdateChecksum close on these stages' close
+        assert entryOffsetInit() && keySizeOffset >= 0;
+        assert s.locksInit() && s.localLockState != LocalLockState.UNLOCKED;
+        assert delayedUpdateChecksum; // doesn't make sense to init to "uninit" false value
+        this.delayedUpdateChecksum = true;
+    }
+
+    abstract boolean delayedUpdateChecksumInit();
+
+    public void closeDelayedUpdateChecksum() {
+        if (hh.h().checksumEntries)
+            hashEntryChecksumStrategy.computeAndStoreChecksum();
+        delayedUpdateChecksum = false;
+    }
+
     @Override
     public void updateChecksum() {
-        checksumStrategy.updateChecksum();
+        checkOnEachPublicOperation.checkOnEachPublicOperation();
+        if (!hh.h().checksumEntries) {
+            throw new UnsupportedOperationException(
+                    "Checksum is not stored in this Chronicle Hash");
+        }
+        initDelayedUpdateChecksum(true);
     }
 
     @Override
     public boolean checkSum() {
-        return checksumStrategy.checkSum();
+        checkOnEachPublicOperation.checkOnEachPublicOperation();
+        if (!hh.h().checksumEntries) {
+            throw new UnsupportedOperationException(
+                    "Checksum is not stored in this Chronicle Hash");
+        }
+        return delayedUpdateChecksumInit() || checksumStrategy.innerCheckSum();
     }
 
     long entrySize() {

src/main/java/net/openhft/chronicle/hash/impl/stage/entry/NoChecksumStrategy.java

@@ -21,11 +21,6 @@ public enum NoChecksumStrategy implements ChecksumStrategy {
 
     @Override
     public void computeAndStoreChecksum() {
-        // no-op
-    }
-
-    @Override
-    public void updateChecksum() {
         throw new UnsupportedOperationException("Checksum is not stored in this Chronicle Hash");
     }
 
@@ -34,11 +29,6 @@ public enum NoChecksumStrategy implements ChecksumStrategy {
         return true;
     }
 
-    @Override
-    public boolean checkSum() {
-        throw new UnsupportedOperationException("Checksum is not stored in this Chronicle Hash");
-    }
-
     @Override
     public long extraEntryBytes() {
         return 0; // no extra bytes to store checksum

src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ReadLock.java

@@ -93,10 +93,11 @@ public class ReadLock implements InterProcessLock {
 
     @Override
     public void unlock() {
-        s.readUnlockAndDecrementCount();
-        s.setLocalLockState(UNLOCKED);
         // TODO what should close here?
         hlp.closeHashLookupPos();
         entry.closePos();
+        entry.closeKeyOffset();
+        s.readUnlockAndDecrementCount();
+        s.setLocalLockState(UNLOCKED);
     }
 }

src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java

@@ -483,6 +483,8 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface {
     @Stage("SegmentTier") public long tierIndex;
     @Stage("SegmentTier") public long tierBaseAddr;
 
+    public abstract boolean segmentTierInit();
+
     public void initSegmentTier() {
         tierIndex = segmentIndex + 1; // tiers are 1-counted
         tierBaseAddr = hh.h().segmentBaseAddr(segmentIndex);

src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java

@@ -32,6 +32,7 @@ public class UpdateLock implements InterProcessLock {
 
     @StageRef LogHolder logHolder;
     @StageRef SegmentStages s;
+    @StageRef HashEntryStages<?> entry;
     
     @Override
     public boolean isHeldByCurrentThread() {
@@ -163,6 +164,7 @@ public class UpdateLock implements InterProcessLock {
 
     @Override
     public void unlock() {
+        entry.closeDelayedUpdateChecksum();
         switch (s.localLockState) {
             case UNLOCKED:
             case READ_LOCKED:

src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java

@@ -140,6 +140,8 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>>
                             steps--;
                             hlp.initHashLookupPos(currentHashLookupPos);
                         }
+                        // force entry checksum update (delayedUpdateChecksum depends on keyOffset)
+                        e.closeKeyOffset();
                     }
                 }
             }

src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java

@@ -92,6 +92,7 @@ public abstract class MapEntryStages<K, V> extends HashEntryStages<K>
     }
 
     public void writeValue(Data<?> value) {
+        initDelayedUpdateChecksum(true);
         // In acquireContext(), replaceValue() is called for
         // 1) executing custom replaceValue() logic, if defined, from configured MapEntryOperations
         // 2) Update replication status (bits, timestamp)

src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java

@@ -43,6 +43,7 @@ public abstract class ReplicatedMapEntryStages<K, V> extends MapEntryStages<K, V
     }
 
     void updateReplicationState(long timestamp, byte identifier) {
+        initDelayedUpdateChecksum(true);
         Bytes segmentBytes = s.segmentBytesForWrite();
         segmentBytes.writePosition(replicationBytesOffset);
         segmentBytes.writeLong(timestamp);

src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java

@@ -95,32 +95,11 @@ public abstract class ReplicatedInput<K, V, R>
         mh.m().setLastModificationTime(riId, bootstrapTimestamp);
 
         q.initInputKey(replicatedInputKeyBytesValue);
-
-        boolean debugEnabled = lh.LOG.isDebugEnabled();
-
         s.innerUpdateLock.lock();
         if (isDeleted) {
-            if (debugEnabled) {
-                lh.LOG.debug("READING FROM SOURCE -  into local-id={}, remote={}, remove(key={})",
-                        mh.m().identifier(), riId, ks.inputKey);
-            }
             mh.m().remoteOperations.remove(this);
-            return;
-        }
-
-        String message = null;
-        if (debugEnabled) {
-            message = String.format(
-                    "READING FROM SOURCE -  into local-id=%d, remote-id=%d, put(key=%s,",
-                    mh.m().identifier(), riId, ks.inputKey);
-        }
-
-
-        mh.m().remoteOperations.put(this, replicatedInputValueBytesValue);
-
-
-        if (debugEnabled) {
-            lh.LOG.debug(message + "value=" + replicatedInputValueBytesValue + ")");
+        } else {
+            mh.m().remoteOperations.put(this, replicatedInputValueBytesValue);
         }
     }
 }

src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java

@@ -39,7 +39,6 @@ public abstract class MapSegmentIteration<K, V, R> extends HashSegmentIteration<
         checkOnEachPublicOperation.checkOnEachPublicOperation();
         try {
             entry.innerDefaultReplaceValue(newValue);
-            entry.checksumStrategy.computeAndStoreChecksum();
         } finally {
             s.innerWriteLock.unlock();
         }

src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java

@@ -59,7 +59,6 @@ public abstract class ReplicatedMapSegmentIteration<K, V, R> extends MapSegmentI
             entry.innerDefaultReplaceValue(newValue);
             ru.updateChange();
             e.updatedReplicationStateOnPresentEntry();
-            e.checksumStrategy.computeAndStoreChecksum();
         } finally {
             s.innerWriteLock.unlock();
         }
@@ -121,7 +120,6 @@ public abstract class ReplicatedMapSegmentIteration<K, V, R> extends MapSegmentI
             e.updatedReplicationStateOnPresentEntry();
             e.writeEntryDeleted();
             ru.updateChange();
-            e.checksumStrategy.computeAndStoreChecksum();
             s.deleted(s.deleted() + 1);
         } finally {
             s.innerWriteLock.unlock();
@@ -148,7 +146,6 @@ public abstract class ReplicatedMapSegmentIteration<K, V, R> extends MapSegmentI
                 e.writeEntryPresent();
                 ru.updateChange();
                 e.updatedReplicationStateOnPresentEntry();
-                e.checksumStrategy.computeAndStoreChecksum();
             } finally {
                 s.innerWriteLock.unlock();
             }

src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java

@@ -79,7 +79,6 @@ public abstract class MapAbsent<K, V> implements Absent<K, V> {
             s.incrementModCount();
             ks.setSearchState(PRESENT);
             q.initPresenceOfEntry(EntryPresence.PRESENT);
-            e.checksumStrategy.computeAndStoreChecksum();
         } else {
             throw new IllegalStateException(
                     "Entry is present in the map when doInsert() is called");

src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java

@@ -99,11 +99,6 @@ public abstract class MapQuery<K, V, R> extends HashQuery<K>
 
     @Override
     public void doReplaceValue(Data<V> newValue) {
-        doReplaceValueWithoutChecksum(newValue);
-        e.checksumStrategy.computeAndStoreChecksum();
-    }
-
-    public void doReplaceValueWithoutChecksum(Data<V> newValue) {
         putPrefix();
         if (entryPresent()) {
             e.innerDefaultReplaceValue(newValue);

src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java

@@ -51,7 +51,6 @@ public abstract class ReplicatedMapAbsent<K, V> extends MapAbsent<K, V> {
             s.incrementModCount();
             e.writeEntryPresent();
             ru.updateChange();
-            e.checksumStrategy.computeAndStoreChecksum();
         } else {
             throw new IllegalStateException(
                     "Entry is present in the map when doInsert() is called");

src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java

@@ -78,7 +78,6 @@ public abstract class ReplicatedMapQuery<K, V, R> extends MapQuery<K, V, R>
             e.updatedReplicationStateOnPresentEntry();
             e.writeEntryDeleted();
             ru.updateChange();
-            e.checksumStrategy.computeAndStoreChecksum();
             s.deleted(s.deleted() + 1);
         } else {
             throw new IllegalStateException("Entry is absent in the map when doRemove() is called");
@@ -95,9 +94,8 @@ public abstract class ReplicatedMapQuery<K, V, R> extends MapQuery<K, V, R>
 
     @Override
     public void doReplaceValue(Data<V> newValue) {
-        doReplaceValueWithoutChecksum(newValue);
+        super.doReplaceValue(newValue);
         ru.updateChange();
         e.updatedReplicationStateOnPresentEntry();
-        e.checksumStrategy.computeAndStoreChecksum();
     }
 }

src/test/java/net/openhft/chronicle/map/OldDeletedEntriesCleanupTest.java

@@ -123,7 +123,7 @@ public class OldDeletedEntriesCleanupTest {
     @Test
     public void testFloodReplicatedMapWithDeletedEntries() throws InterruptedException {
         try {
-            Random r = ThreadLocalRandom.current();
+            Random r = new Random(42);
             NavigableSet<Integer> put = new TreeSet<>();
             for (int i = 0; i < Builder.SIZE * 10; i++) {
                 if (r.nextBoolean() || put.isEmpty()) {
@@ -139,9 +139,11 @@ public class OldDeletedEntriesCleanupTest {
             System.out.println("\nwaiting till equal");
 
             waitTillEqual(15000);
-
-            if (!map1.equals(map2))
-                Assert.assertEquals(Maps.difference(map1, map2).toString(), map1, map2);
+            if (!map1.equals(map2)) {
+                Assert.assertEquals(
+                        Maps.difference(new TreeMap<>(map1), new TreeMap<>(map2)).toString(),
+                        new TreeMap<>(map1), new TreeMap<>(map2));
+            }
         } finally {
             map1.close();
             map2.close();