HCOLL-392 Add entry checksums

src/main/java/net/openhft/chronicle/hash/ChecksumEntry.java

@@ -0,0 +1,43 @@
+/*
+ *      Copyright (C) 2015  higherfrequencytrading.com
+ *
+ *      This program is free software: you can redistribute it and/or modify
+ *      it under the terms of the GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License.
+ *
+ *      This program is distributed in the hope that it will be useful,
+ *      but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *      GNU Lesser General Public License for more details.
+ *
+ *      You should have received a copy of the GNU Lesser General Public License
+ *      along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package net.openhft.chronicle.hash;
+
+import net.openhft.chronicle.map.MapAbsentEntry;
+import net.openhft.chronicle.map.MapEntry;
+
+/**
+ * Abstracts entries of hash containers, created by {@link ChronicleHashBuilder}s
+ * with {@link ChronicleHashBuilder#checksumEntries(boolean)} configured to {@code true}.
+ */
+public interface ChecksumEntry {
+
+    /**
+     * Re-computes and stores checksum for the entry. This method <i>shouldn't</i> be called before
+     * or after ordinary operations like {@link MapAbsentEntry#doInsert(Data)}, {@link
+     * MapEntry#doReplaceValue(Data)}: it is performed automatically underneath. Call this method,
+     * only when value bytes was updated directly, for example though Chronicle-Value-generated
+     * interface.
+     */
+    void updateChecksum();
+
+    /**
+     * Computes checksum from the entry bytes and checks whether it is equal to the stored checksum.
+     *
+     * @return {@code true} if stored checksum equals to checksum computed from the entry bytes
+     */
+    boolean checkSum();
+}

src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java

@@ -566,6 +566,20 @@ public interface ChronicleHashBuilder<K, H extends ChronicleHash<K, ?, ?, ?>,
      */
     B aligned64BitMemoryOperationsAtomic(boolean aligned64BitMemoryOperationsAtomic);
 
+    /**
+     * Configures whether hash containers, created by this builder, should compute and store entry
+     * checksums. It could be used to detect data corruption during recovery after crashes.
+     *
+     * <p>By default, {@linkplain #createPersistedTo(File) persisted} hash containers, created by
+     * {@code ChronicleMapBuilder} <i>do</i> compute and store entry checksums, but hash containers,
+     * created in the process memory via {@link #create()} - don't.
+     *
+     * @param checksumEntries if entry checksums should be computed and stored
+     * @return this builder back
+     * @see ChecksumEntry
+     */
+    B checksumEntries(boolean checksumEntries);
+
     /**
      * Configures replication of the hash containers, created by this builder. See <a
      * href="https://github.com/OpenHFT/Chronicle-Map#tcp--udp-replication"> the section about

src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilderPrivateAPI.java

@@ -39,4 +39,6 @@ public interface ChronicleHashBuilderPrivateAPI<K> {
     long maxExtraTiers();
 
     boolean aligned64BitMemoryOperationsAtomic();
+
+    boolean checksumEntries();
 }

src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java

@@ -112,6 +112,8 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
     protected final long numberOfTiersInBulk;
     protected final int log2NumberOfTiersInBulk;
 
+    public final boolean checksumEntries;
+
     /////////////////////////////////////////////////
     // Bytes Store (essentially, the base address) and serialization-dependent offsets
     protected transient BytesStore ms;
@@ -210,6 +212,8 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
         log2NumberOfTiersInBulk = Maths.intLog2(numberOfTiersInBulk);
         tierBulkInnerOffsetToTiers = computeTierBulkInnerOffsetToTiers(numberOfTiersInBulk);
         tierBulkSizeInBytes = computeTierBulkBytesSize();
+
+        checksumEntries = privateAPI.checksumEntries();
     }
 
     private long segmentSize() {

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

@@ -0,0 +1,35 @@
+/*
+ *      Copyright (C) 2015  higherfrequencytrading.com
+ *
+ *      This program is free software: you can redistribute it and/or modify
+ *      it under the terms of the GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License.
+ *
+ *      This program is distributed in the hope that it will be useful,
+ *      but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *      GNU Lesser General Public License for more details.
+ *
+ *      You should have received a copy of the GNU Lesser General Public License
+ *      along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+public enum ChecksumHashing {
+    ;
+
+    /**
+     * A smart procedure copied from CityHash/FarmHash, see the full implementation in
+     * Zero-allocation-hashing or Chronicle-Algorithms
+     */
+    public static long hash8To16Bytes(long len, long first8Bytes, long last8Bytes) {
+        long K2 = 0x9ae16a3b2f90404fL;
+        long mul = K2 + (len << 1);
+        long a = first8Bytes + K2;
+        long c = ((last8Bytes >>> 37) | (last8Bytes << 27)) * mul + a;
+        long d = (((a >>> 25) | (a << 39)) + last8Bytes) * mul;
+        long a1 = (c ^ d) * mul ^ ((c ^ d) * mul >>> 47);
+        return ((d ^ a1) * mul ^ ((d ^ a1) * mul >>> 47)) * mul;
+    }
+}

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

@@ -0,0 +1,30 @@
+/*
+ *      Copyright (C) 2015  higherfrequencytrading.com
+ *
+ *      This program is free software: you can redistribute it and/or modify
+ *      it under the terms of the GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License.
+ *
+ *      This program is distributed in the hope that it will be useful,
+ *      but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *      GNU Lesser General Public License for more details.
+ *
+ *      You should have received a copy of the GNU Lesser General Public License
+ *      along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+import net.openhft.chronicle.hash.ChecksumEntry;
+
+public interface ChecksumStrategy extends ChecksumEntry {
+
+    int CHECKSUM_STORED_BYTES = 4;
+
+    void computeAndStoreChecksum();
+
+    boolean innerCheckSum();
+
+    long extraEntryBytes();
+}

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

@@ -0,0 +1,42 @@
+/*
+ *      Copyright (C) 2015  higherfrequencytrading.com
+ *
+ *      This program is free software: you can redistribute it and/or modify
+ *      it under the terms of the GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License.
+ *
+ *      This program is distributed in the hope that it will be useful,
+ *      but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *      GNU Lesser General Public License for more details.
+ *
+ *      You should have received a copy of the GNU Lesser General Public License
+ *      along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+public interface Crc32 {
+
+    int crc32(long addr, long len);
+
+    enum Crc32Impl {
+        ;
+
+        static Crc32 crc32;
+        static {
+            try {
+                crc32 = NativeCrc32.INSTANCE;
+            } catch (Throwable e) {
+                // ignore
+            } finally {
+                if (crc32 == null)
+                    crc32 = FallbackJavaCrc32.INSTANCE;
+            }
+        }
+    }
+
+    static int compute(long addr, long len) {
+        return Crc32Impl.crc32.crc32(addr, len);
+    }
+}

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

@@ -0,0 +1,110 @@
+/*
+ *      Copyright 2014 The Netty Project
+ *      Copyright (C) 2015  higherfrequencytrading.com
+ *
+ *      This program is free software: you can redistribute it and/or modify
+ *      it under the terms of the GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License.
+ *
+ *      This program is distributed in the hope that it will be useful,
+ *      but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *      GNU Lesser General Public License for more details.
+ *
+ *      You should have received a copy of the GNU Lesser General Public License
+ *      along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+import static net.openhft.lang.io.NativeBytes.UNSAFE;
+
+/**
+ * Copied from
+ * https://github.com/netty/netty/blob/ed4a89082bb29b9e7d869c5d25d6b9ea8fc9d25b/
+ * codec/src/main/java/io/netty/handler/codec/compression/Crc32.java
+ */
+public enum FallbackJavaCrc32 implements Crc32 {
+    INSTANCE;
+
+    /**
+     * A static CRC lookup table.
+     */
+    private static final int[] crc32Table = {
+            0x00000000, 0x04c11db7, 0x09823b6e, 0x0d4326d9,
+            0x130476dc, 0x17c56b6b, 0x1a864db2, 0x1e475005,
+            0x2608edb8, 0x22c9f00f, 0x2f8ad6d6, 0x2b4bcb61,
+            0x350c9b64, 0x31cd86d3, 0x3c8ea00a, 0x384fbdbd,
+            0x4c11db70, 0x48d0c6c7, 0x4593e01e, 0x4152fda9,
+            0x5f15adac, 0x5bd4b01b, 0x569796c2, 0x52568b75,
+            0x6a1936c8, 0x6ed82b7f, 0x639b0da6, 0x675a1011,
+            0x791d4014, 0x7ddc5da3, 0x709f7b7a, 0x745e66cd,
+            0x9823b6e0, 0x9ce2ab57, 0x91a18d8e, 0x95609039,
+            0x8b27c03c, 0x8fe6dd8b, 0x82a5fb52, 0x8664e6e5,
+            0xbe2b5b58, 0xbaea46ef, 0xb7a96036, 0xb3687d81,
+            0xad2f2d84, 0xa9ee3033, 0xa4ad16ea, 0xa06c0b5d,
+            0xd4326d90, 0xd0f37027, 0xddb056fe, 0xd9714b49,
+            0xc7361b4c, 0xc3f706fb, 0xceb42022, 0xca753d95,
+            0xf23a8028, 0xf6fb9d9f, 0xfbb8bb46, 0xff79a6f1,
+            0xe13ef6f4, 0xe5ffeb43, 0xe8bccd9a, 0xec7dd02d,
+            0x34867077, 0x30476dc0, 0x3d044b19, 0x39c556ae,
+            0x278206ab, 0x23431b1c, 0x2e003dc5, 0x2ac12072,
+            0x128e9dcf, 0x164f8078, 0x1b0ca6a1, 0x1fcdbb16,
+            0x018aeb13, 0x054bf6a4, 0x0808d07d, 0x0cc9cdca,
+            0x7897ab07, 0x7c56b6b0, 0x71159069, 0x75d48dde,
+            0x6b93dddb, 0x6f52c06c, 0x6211e6b5, 0x66d0fb02,
+            0x5e9f46bf, 0x5a5e5b08, 0x571d7dd1, 0x53dc6066,
+            0x4d9b3063, 0x495a2dd4, 0x44190b0d, 0x40d816ba,
+            0xaca5c697, 0xa864db20, 0xa527fdf9, 0xa1e6e04e,
+            0xbfa1b04b, 0xbb60adfc, 0xb6238b25, 0xb2e29692,
+            0x8aad2b2f, 0x8e6c3698, 0x832f1041, 0x87ee0df6,
+            0x99a95df3, 0x9d684044, 0x902b669d, 0x94ea7b2a,
+            0xe0b41de7, 0xe4750050, 0xe9362689, 0xedf73b3e,
+            0xf3b06b3b, 0xf771768c, 0xfa325055, 0xfef34de2,
+            0xc6bcf05f, 0xc27dede8, 0xcf3ecb31, 0xcbffd686,
+            0xd5b88683, 0xd1799b34, 0xdc3abded, 0xd8fba05a,
+            0x690ce0ee, 0x6dcdfd59, 0x608edb80, 0x644fc637,
+            0x7a089632, 0x7ec98b85, 0x738aad5c, 0x774bb0eb,
+            0x4f040d56, 0x4bc510e1, 0x46863638, 0x42472b8f,
+            0x5c007b8a, 0x58c1663d, 0x558240e4, 0x51435d53,
+            0x251d3b9e, 0x21dc2629, 0x2c9f00f0, 0x285e1d47,
+            0x36194d42, 0x32d850f5, 0x3f9b762c, 0x3b5a6b9b,
+            0x0315d626, 0x07d4cb91, 0x0a97ed48, 0x0e56f0ff,
+            0x1011a0fa, 0x14d0bd4d, 0x19939b94, 0x1d528623,
+            0xf12f560e, 0xf5ee4bb9, 0xf8ad6d60, 0xfc6c70d7,
+            0xe22b20d2, 0xe6ea3d65, 0xeba91bbc, 0xef68060b,
+            0xd727bbb6, 0xd3e6a601, 0xdea580d8, 0xda649d6f,
+            0xc423cd6a, 0xc0e2d0dd, 0xcda1f604, 0xc960ebb3,
+            0xbd3e8d7e, 0xb9ff90c9, 0xb4bcb610, 0xb07daba7,
+            0xae3afba2, 0xaafbe615, 0xa7b8c0cc, 0xa379dd7b,
+            0x9b3660c6, 0x9ff77d71, 0x92b45ba8, 0x9675461f,
+            0x8832161a, 0x8cf30bad, 0x81b02d74, 0x857130c3,
+            0x5d8a9099, 0x594b8d2e, 0x5408abf7, 0x50c9b640,
+            0x4e8ee645, 0x4a4ffbf2, 0x470cdd2b, 0x43cdc09c,
+            0x7b827d21, 0x7f436096, 0x7200464f, 0x76c15bf8,
+            0x68860bfd, 0x6c47164a, 0x61043093, 0x65c52d24,
+            0x119b4be9, 0x155a565e, 0x18197087, 0x1cd86d30,
+            0x029f3d35, 0x065e2082, 0x0b1d065b, 0x0fdc1bec,
+            0x3793a651, 0x3352bbe6, 0x3e119d3f, 0x3ad08088,
+            0x2497d08d, 0x2056cd3a, 0x2d15ebe3, 0x29d4f654,
+            0xc5a92679, 0xc1683bce, 0xcc2b1d17, 0xc8ea00a0,
+            0xd6ad50a5, 0xd26c4d12, 0xdf2f6bcb, 0xdbee767c,
+            0xe3a1cbc1, 0xe760d676, 0xea23f0af, 0xeee2ed18,
+            0xf0a5bd1d, 0xf464a0aa, 0xf9278673, 0xfde69bc4,
+            0x89b8fd09, 0x8d79e0be, 0x803ac667, 0x84fbdbd0,
+            0x9abc8bd5, 0x9e7d9662, 0x933eb0bb, 0x97ffad0c,
+            0xafb010b1, 0xab710d06, 0xa6322bdf, 0xa2f33668,
+            0xbcb4666d, 0xb8757bda, 0xb5365d03, 0xb1f740b4
+    };
+
+
+    @Override
+    public int crc32(long addr, long len) {
+        int crc = 0xffffffff;
+        for (long limit = addr + len; addr < limit; addr++) {
+            int value = UNSAFE.getByte(addr) & 0xff;
+            crc = (crc << 8) ^ crc32Table[(((crc >> 24) ^ value) & 0xff)];
+        }
+        return ~crc;
+    }
+}

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

@@ -17,14 +17,12 @@
 package net.openhft.chronicle.hash.impl.stage.entry;
 
 import net.openhft.chronicle.algo.bytes.Access;
-import net.openhft.chronicle.bytes.BytesStore;
-import net.openhft.chronicle.bytes.NativeBytesStore;
+import net.openhft.chronicle.hash.ChecksumEntry;
 import net.openhft.chronicle.hash.Data;
 import net.openhft.chronicle.hash.HashEntry;
 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;
-import net.openhft.lang.io.Bytes;
 import net.openhft.sg.Stage;
 import net.openhft.sg.StageRef;
 import net.openhft.sg.Staged;
@@ -35,7 +33,7 @@ import static net.openhft.chronicle.algo.bytes.Access.nativeAccess;
 
 
 @Staged
-public abstract class HashEntryStages<K> implements HashEntry<K> {
+public abstract class HashEntryStages<K> implements HashEntry<K>, ChecksumEntry {
 
     @StageRef public VanillaChronicleHashHolder<?, ?, ?> hh;
     @StageRef public SegmentStages s;
@@ -107,8 +105,22 @@ public abstract class HashEntryStages<K> implements HashEntry<K> {
         return keyEnd();
     }
 
+    @StageRef HashKeyCrc32PayloadChecksumStrategy hashKeyCrc32PayloadChecksumStrategy;
+    public final ChecksumStrategy checksumStrategy = hh.h().checksumEntries ?
+            hashKeyCrc32PayloadChecksumStrategy : NoChecksumStrategy.INSTANCE;
+
+    @Override
+    public void updateChecksum() {
+        checksumStrategy.updateChecksum();
+    }
+
+    @Override
+    public boolean checkSum() {
+        return checksumStrategy.checkSum();
+    }
+
     long entrySize() {
-        return entryEnd() - keySizeOffset;
+        return checksumStrategy.extraEntryBytes() + entryEnd() - keySizeOffset;
     }
 
     @StageRef EntryKeyBytesData<K> entryKey;

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

@@ -0,0 +1,80 @@
+/*
+ *      Copyright (C) 2015  higherfrequencytrading.com
+ *
+ *      This program is free software: you can redistribute it and/or modify
+ *      it under the terms of the GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License.
+ *
+ *      This program is distributed in the hope that it will be useful,
+ *      but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *      GNU Lesser General Public License for more details.
+ *
+ *      You should have received a copy of the GNU Lesser General Public License
+ *      along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+import static net.openhft.chronicle.hash.impl.stage.entry.ChecksumHashing.hash8To16Bytes;
+
+@Staged
+public class HashKeyCrc32PayloadChecksumStrategy implements ChecksumStrategy {
+
+    @StageRef CheckOnEachPublicOperation checkOnEachPublicOperation;
+    @StageRef SegmentStages s;
+    @StageRef HashEntryStages<?> e;
+    @StageRef KeyHashCode h;
+
+    @Override
+    public void computeAndStoreChecksum() {
+        int crc = computeChecksum();
+        s.segmentBS.writeInt(e.entryEnd(), crc);
+    }
+
+    private int computeChecksum() {
+        long keyHashCode = h.keyHashCode();
+
+        long keyEnd = e.keyEnd();
+        long len = e.entryEnd() - keyEnd;
+
+        long checksum;
+        if (len > 0) {
+            long addr = s.segmentBaseAddr + keyEnd;
+            int payloadCrc = Crc32.compute(addr, len);
+            checksum = hash8To16Bytes(e.keySize, keyHashCode, payloadCrc);
+        } else {
+            // non replicated ChronicleSet has no payload
+            checksum = hash8To16Bytes(e.keySize, keyHashCode, keyHashCode);
+        }
+        return (int) ((checksum >>> 32) ^ checksum);
+    }
+
+    @Override
+    public void updateChecksum() {
+        checkOnEachPublicOperation.checkOnEachPublicOperation();
+        computeAndStoreChecksum();
+    }
+
+    @Override
+    public boolean innerCheckSum() {
+        int oldChecksum = s.segmentBS.readInt(e.entryEnd());
+        int crc = computeChecksum();
+        return oldChecksum == crc;
+    }
+
+    @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/NativeCrc32.java

@@ -0,0 +1,51 @@
+/*
+ *      Copyright (C) 2015  higherfrequencytrading.com
+ *
+ *      This program is free software: you can redistribute it and/or modify
+ *      it under the terms of the GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License.
+ *
+ *      This program is distributed in the hope that it will be useful,
+ *      but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *      GNU Lesser General Public License for more details.
+ *
+ *      You should have received a copy of the GNU Lesser General Public License
+ *      along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+import java.util.zip.CRC32;
+
+public enum NativeCrc32 implements Crc32 {
+    INSTANCE;
+
+    static final MethodHandle updateByteBuffer;
+    static {
+        try {
+            MethodHandles.Lookup lookup = MethodHandles.lookup();
+            updateByteBuffer = lookup.findStatic(CRC32.class, "updateByteBuffer",
+                    MethodType.methodType(int.class, long.class, int.class, int.class));
+        } catch (NoSuchMethodException e) {
+            throw new RuntimeException(e);
+        } catch (IllegalAccessException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public int crc32(long addr, long len) {
+        if (len > Integer.MAX_VALUE)
+            throw new UnsupportedOperationException("Native Crc32 checksum doesn't support " +
+                    "entries bigger than 2^31 - 1 in size, " + len + " given");
+        try {
+            return (int) updateByteBuffer.invoke(addr, 0, (int) len);
+        } catch (Throwable throwable) {
+            throw new RuntimeException(throwable);
+        }
+    }
+}

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

@@ -0,0 +1,46 @@
+/*
+ *      Copyright (C) 2015  higherfrequencytrading.com
+ *
+ *      This program is free software: you can redistribute it and/or modify
+ *      it under the terms of the GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License.
+ *
+ *      This program is distributed in the hope that it will be useful,
+ *      but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *      GNU Lesser General Public License for more details.
+ *
+ *      You should have received a copy of the GNU Lesser General Public License
+ *      along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+public enum NoChecksumStrategy implements ChecksumStrategy {
+    INSTANCE;
+
+    @Override
+    public void computeAndStoreChecksum() {
+        // no-op
+    }
+
+    @Override
+    public void updateChecksum() {
+        throw new UnsupportedOperationException("Checksum is not stored in this Chronicle Hash");
+    }
+
+    @Override
+    public boolean innerCheckSum() {
+        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/iter/IterationKeyHashCode.java

@@ -0,0 +1,44 @@
+/*
+ *      Copyright (C) 2015  higherfrequencytrading.com
+ *
+ *      This program is free software: you can redistribute it and/or modify
+ *      it under the terms of the GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License.
+ *
+ *      This program is distributed in the hope that it will be useful,
+ *      but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *      GNU Lesser General Public License for more details.
+ *
+ *      You should have received a copy of the GNU Lesser General Public License
+ *      along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.iter;
+
+import net.openhft.chronicle.algo.hashing.LongHashFunction;
+import net.openhft.chronicle.hash.impl.stage.entry.HashEntryStages;
+import net.openhft.chronicle.hash.impl.stage.entry.KeyHashCode;
+import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+@Staged
+public class IterationKeyHashCode implements KeyHashCode {
+
+    @StageRef SegmentStages s;
+    @StageRef HashEntryStages<?> e;
+
+    long keyHash = 0;
+
+    void initKeyHash() {
+        long addr = s.segmentBaseAddr + e.keyOffset;
+        long len = e.keySize;
+        keyHash = LongHashFunction.city_1_1().hashMemory(addr, len);
+    }
+
+    @Override
+    public long keyHashCode() {
+        return keyHash;
+    }
+}

src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java

@@ -20,6 +20,7 @@ import net.openhft.chronicle.core.OS;
 import net.openhft.chronicle.hash.ChronicleHashBuilder;
 import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI;
 import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder;
+import net.openhft.chronicle.hash.impl.stage.entry.ChecksumStrategy;
 import net.openhft.chronicle.hash.impl.util.math.PoissonDistribution;
 import net.openhft.chronicle.hash.replication.*;
 import net.openhft.chronicle.hash.serialization.*;
@@ -162,7 +163,10 @@ public final class ChronicleMapBuilder<K, V> implements
     private double maxBloatFactor = 1.0;
     private boolean allowSegmentTiering = true;
     private double nonTieredSegmentsPercentile = 0.99999;
-    private boolean aligned64BitMemoryOperationsAtomic = OS.is64Bit() ? true : false;
+    private boolean aligned64BitMemoryOperationsAtomic = OS.is64Bit();
+
+    enum ChecksumEntries {YES, NO, IF_PERSISTED}
+    private ChecksumEntries checksumEntries = ChecksumEntries.IF_PERSISTED;
 
     private boolean putReturnsNull = false;
     private boolean removeReturnsNull = false;
@@ -534,6 +538,8 @@ public final class ChronicleMapBuilder<K, V> implements
         size += keySize;
         if (replicated)
             size += ReplicatedChronicleMap.ADDITIONAL_ENTRY_BYTES;
+        if (checksumEntries())
+            size += ChecksumStrategy.CHECKSUM_STORED_BYTES;
         double valueSize = averageValueSize();
         size += averageSizeEncodingSize(valueBuilder, valueSize);
         Alignment alignment = valueAlignment();
@@ -1240,6 +1246,21 @@ public final class ChronicleMapBuilder<K, V> implements
         return this;
     }
 
+    @Override
+    public ChronicleMapBuilder<K, V> checksumEntries(boolean checksumEntries) {
+        this.checksumEntries = checksumEntries ? ChecksumEntries.YES : ChecksumEntries.NO;
+        return this;
+    }
+
+    boolean checksumEntries() {
+        switch (checksumEntries) {
+            case NO: return false;
+            case YES: return true;
+            case IF_PERSISTED: return persisted;
+            default: throw new AssertionError();
+        }
+    }
+
     boolean aligned64BitMemoryOperationsAtomic() {
         return aligned64BitMemoryOperationsAtomic;
     }

src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderPrivateAPI.java

@@ -76,4 +76,9 @@ class ChronicleMapBuilderPrivateAPI<K> implements ChronicleHashBuilderPrivateAPI
     public boolean aligned64BitMemoryOperationsAtomic() {
         return b.aligned64BitMemoryOperationsAtomic();
     }
+
+    @Override
+    public boolean checksumEntries() {
+        return b.checksumEntries();
+    }
 }

src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java

@@ -21,6 +21,7 @@ import net.openhft.chronicle.hash.impl.stage.entry.*;
 import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop;
 import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder;
 import net.openhft.chronicle.hash.impl.stage.hash.ThreadLocalCopiesHolder;
+import net.openhft.chronicle.hash.impl.stage.iter.IterationKeyHashCode;
 import net.openhft.chronicle.hash.impl.stage.iter.IterationSegmentStages;
 import net.openhft.chronicle.hash.impl.stage.query.KeySearch;
 import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData;
@@ -46,6 +47,7 @@ import net.openhft.sg.Staged;
 
         KeyBytesInterop.class,
         IterationSegmentStages.class,
+        IterationKeyHashCode.class,
         HashLookupPos.class,
         IterationCheckOnEachPublicOperation.class,
         AllocatedChunks.class,
@@ -65,6 +67,8 @@ import net.openhft.sg.Staged;
         EntryValueBytesData.class,
 
         WrappedValueInstanceData.class,
+
+        HashKeyCrc32PayloadChecksumStrategy.class,
 })
 public class MapIterationContext {
 }

src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java

@@ -19,10 +19,7 @@ package net.openhft.chronicle.map.impl;
 import net.openhft.chronicle.hash.impl.stage.data.bytes.EntryKeyBytesData;
 import net.openhft.chronicle.hash.impl.stage.data.bytes.InputKeyBytesData;
 import net.openhft.chronicle.hash.impl.stage.data.instance.InputKeyInstanceData;
-import net.openhft.chronicle.hash.impl.stage.entry.InputKeyHashCode;
-import net.openhft.chronicle.hash.impl.stage.entry.ReadLock;
-import net.openhft.chronicle.hash.impl.stage.entry.UpdateLock;
-import net.openhft.chronicle.hash.impl.stage.entry.WriteLock;
+import net.openhft.chronicle.hash.impl.stage.entry.*;
 import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop;
 import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder;
 import net.openhft.chronicle.hash.impl.stage.hash.ThreadLocalCopiesHolder;
@@ -90,6 +87,8 @@ nested = {
         AcquireHandle.class,
         DefaultReturnValue.class,
         UsingReturnValue.class,
+
+        HashKeyCrc32PayloadChecksumStrategy.class,
 })
 public class MapQueryContext {
 }

src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java

@@ -21,6 +21,7 @@ import net.openhft.chronicle.hash.impl.stage.entry.*;
 import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop;
 import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder;
 import net.openhft.chronicle.hash.impl.stage.hash.ThreadLocalCopiesHolder;
+import net.openhft.chronicle.hash.impl.stage.iter.IterationKeyHashCode;
 import net.openhft.chronicle.hash.impl.stage.iter.IterationSegmentStages;
 import net.openhft.chronicle.hash.impl.stage.query.KeySearch;
 import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData;
@@ -47,6 +48,7 @@ import net.openhft.sg.Staged;
 
         KeyBytesInterop.class,
         IterationSegmentStages.class,
+        IterationKeyHashCode.class,
         HashLookupPos.class,
         IterationCheckOnEachPublicOperation.class,
         AllocatedChunks.class,
@@ -73,7 +75,9 @@ import net.openhft.sg.Staged;
         DummyValueZeroData.class,
 
         ReplicatedMapAbsentDelegatingForIteration.class,
-        ReplicatedMapEntryDelegating.class
+        ReplicatedMapEntryDelegating.class,
+
+        HashKeyCrc32PayloadChecksumStrategy.class,
 })
 public class ReplicatedMapIterationContext {
 }

src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java

@@ -19,10 +19,7 @@ package net.openhft.chronicle.map.impl;
 import net.openhft.chronicle.hash.impl.stage.data.bytes.EntryKeyBytesData;
 import net.openhft.chronicle.hash.impl.stage.data.bytes.InputKeyBytesData;
 import net.openhft.chronicle.hash.impl.stage.data.instance.InputKeyInstanceData;
-import net.openhft.chronicle.hash.impl.stage.entry.InputKeyHashCode;
-import net.openhft.chronicle.hash.impl.stage.entry.ReadLock;
-import net.openhft.chronicle.hash.impl.stage.entry.UpdateLock;
-import net.openhft.chronicle.hash.impl.stage.entry.WriteLock;
+import net.openhft.chronicle.hash.impl.stage.entry.*;
 import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop;
 import net.openhft.chronicle.hash.impl.stage.hash.LogHolder;
 import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder;
@@ -109,6 +106,8 @@ nested = {
         ReplicatedInputValueBytesData.class,
 
         DummyValueZeroData.class,
+
+        HashKeyCrc32PayloadChecksumStrategy.class,
 })
 public class ReplicatedMapQueryContext {
 }

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

@@ -39,6 +39,7 @@ 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

@@ -57,6 +57,7 @@ public abstract class ReplicatedMapSegmentIteration<K, V, R> extends MapSegmentI
             entry.innerDefaultReplaceValue(newValue);
             ru.updateChange();
             e.updatedReplicationStateOnPresentEntry();
+            e.checksumStrategy.computeAndStoreChecksum();
         } finally {
             s.innerWriteLock.unlock();
         }
@@ -118,6 +119,7 @@ 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();
@@ -144,6 +146,7 @@ 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

@@ -78,6 +78,7 @@ public abstract class MapAbsent<K, V> implements MapAbsentEntry<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

@@ -87,6 +87,11 @@ 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

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

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

@@ -78,6 +78,7 @@ 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");
@@ -94,8 +95,9 @@ public abstract class ReplicatedMapQuery<K, V, R> extends MapQuery<K, V, R>
 
     @Override
     public void doReplaceValue(Data<V> newValue) {
-        super.doReplaceValue(newValue);
+        doReplaceValueWithoutChecksum(newValue);
         ru.updateChange();
         e.updatedReplicationStateOnPresentEntry();
+        e.checksumStrategy.computeAndStoreChecksum();
     }
 }

src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java

@@ -293,6 +293,12 @@ public final class ChronicleSetBuilder<K>
         return this;
     }
 
+    @Override
+    public ChronicleSetBuilder<K> checksumEntries(boolean checksumEntries) {
+        chronicleMapBuilder.checksumEntries(checksumEntries);
+        return this;
+    }
+
     @Override
     public ChronicleSetBuilder<K> replication(SingleChronicleHashReplication replication) {
         chronicleMapBuilder.replication(replication);