diff --git a/pom.xml b/pom.xml index 462c324b9..5586f7aa0 100755 --- a/pom.xml +++ b/pom.xml @@ -152,7 +152,7 @@ net.openhft stage-compiler - 1.8 + 1.9 provided diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ChecksumStrategy.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ChecksumStrategy.java index ad13d5b3f..8a005c7a6 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ChecksumStrategy.java +++ b/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; diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryChecksumStrategy.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryChecksumStrategy.java index ea4ac6ab7..8e4277ba0 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryChecksumStrategy.java +++ b/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; diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryStages.java index 48b7968cd..19badccd7 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryStages.java +++ b/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 implements HashEntry, 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 implements HashEntry, 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 implements HashEntry, 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() { diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/NoChecksumStrategy.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/NoChecksumStrategy.java index 5f2453d19..738118c20 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/NoChecksumStrategy.java +++ b/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 diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ReadLock.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ReadLock.java index 47b5efc94..7a0077682 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ReadLock.java +++ b/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); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java index c2869d223..5dc351874 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java +++ b/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); diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java index 6e48b0742..bca40d3b6 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java +++ b/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: diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java index 6cc36f5c3..a212b66fc 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java @@ -140,6 +140,8 @@ public abstract class HashSegmentIteration> steps--; hlp.initHashLookupPos(currentHashLookupPos); } + // force entry checksum update (delayedUpdateChecksum depends on keyOffset) + e.closeKeyOffset(); } } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java index 5cbc75c74..f6c89fdc2 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java @@ -92,6 +92,7 @@ public abstract class MapEntryStages extends HashEntryStages } 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) 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 7910abf41..15738be29 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 @@ -43,6 +43,7 @@ public abstract class ReplicatedMapEntryStages extends MapEntryStages 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); } } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java index 0ed6de1fb..a71fb4014 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java @@ -39,7 +39,6 @@ public abstract class MapSegmentIteration extends HashSegmentIteration< checkOnEachPublicOperation.checkOnEachPublicOperation(); try { entry.innerDefaultReplaceValue(newValue); - entry.checksumStrategy.computeAndStoreChecksum(); } finally { s.innerWriteLock.unlock(); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java index ff6e68535..0b4d4b2ac 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java @@ -59,7 +59,6 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI entry.innerDefaultReplaceValue(newValue); ru.updateChange(); e.updatedReplicationStateOnPresentEntry(); - e.checksumStrategy.computeAndStoreChecksum(); } finally { s.innerWriteLock.unlock(); } @@ -121,7 +120,6 @@ public abstract class ReplicatedMapSegmentIteration 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 extends MapSegmentI e.writeEntryPresent(); ru.updateChange(); e.updatedReplicationStateOnPresentEntry(); - e.checksumStrategy.computeAndStoreChecksum(); } finally { s.innerWriteLock.unlock(); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java index e92223b00..74142319f 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java @@ -79,7 +79,6 @@ public abstract class MapAbsent implements Absent { 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"); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java index 5629fd356..b2a1e1eb2 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java @@ -99,11 +99,6 @@ public abstract class MapQuery extends HashQuery @Override public void doReplaceValue(Data newValue) { - doReplaceValueWithoutChecksum(newValue); - e.checksumStrategy.computeAndStoreChecksum(); - } - - public void doReplaceValueWithoutChecksum(Data newValue) { putPrefix(); if (entryPresent()) { e.innerDefaultReplaceValue(newValue); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java index 3db29ebbc..f050d79f6 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java @@ -51,7 +51,6 @@ public abstract class ReplicatedMapAbsent extends MapAbsent { s.incrementModCount(); e.writeEntryPresent(); ru.updateChange(); - e.checksumStrategy.computeAndStoreChecksum(); } else { throw new IllegalStateException( "Entry is present in the map when doInsert() is called"); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java index b32101712..be1b37547 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java @@ -78,7 +78,6 @@ public abstract class ReplicatedMapQuery extends MapQuery 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 extends MapQuery @Override public void doReplaceValue(Data newValue) { - doReplaceValueWithoutChecksum(newValue); + super.doReplaceValue(newValue); ru.updateChange(); e.updatedReplicationStateOnPresentEntry(); - e.checksumStrategy.computeAndStoreChecksum(); } } diff --git a/src/test/java/net/openhft/chronicle/map/OldDeletedEntriesCleanupTest.java b/src/test/java/net/openhft/chronicle/map/OldDeletedEntriesCleanupTest.java index 8301c4bed..daa6a29d7 100644 --- a/src/test/java/net/openhft/chronicle/map/OldDeletedEntriesCleanupTest.java +++ b/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 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();