Fix incremental checksum computation for crc (#49)

pulsar-checksum/src/main/circe/include/int_types.h

@@ -15,7 +15,7 @@
  ******************************************************************************/
 #include <stddef.h> // size_t
 
-#ifdef _MSC_VER
+#if defined(_MSC_VER) && _MSC_VER < 1600 // stdint.h added in MSVC 2010
 
 typedef __int8 int8_t;
 typedef __int16 int16_t;
@@ -26,12 +26,18 @@ typedef unsigned __int16 uint16_t;
 typedef unsigned __int32 uint32_t;
 typedef unsigned __int64 uint64_t;
 
-# define SIZE_T_FORMAT "%Iu"
-
 #else
 
 # include <stdint.h>
 
+#endif
+
+#if defined(_MSC_VER) && _MSC_VER < 1900 // MSVC 2015
+
+# define SIZE_T_FORMAT "%Iu"
+
+#else
+
 # define SIZE_T_FORMAT "%zu"
 
 #endif

pulsar-checksum/src/main/java/com/scurrilous/circe/crc/ReflectedIntCrc.java

@@ -35,9 +35,13 @@ final class ReflectedIntCrc extends AbstractIntCrc {
         }
     }
 
+    @Override
+    protected int initial() {
+        return reflect(super.initial());
+    }
+
     @Override
     protected int resumeRaw(int crc, byte[] input, int index, int length) {
-        crc = reflect(crc);
         for (int i = 0; i < length; ++i)
             crc = table[(crc ^ input[index + i]) & 0xff] ^ (crc >>> 8);
         return crc;

pulsar-checksum/src/main/java/com/yahoo/pulsar/checksum/utils/Crc32cChecksum.java

@@ -46,7 +46,7 @@ public class Crc32cChecksum {
      */
     public static int computeChecksum(ByteBuf payload) {
         if (payload.hasMemoryAddress() && (CRC32C_HASH instanceof Sse42Crc32C)) {
-            return CRC32C_HASH.calculate(payload.memoryAddress(), payload.readableBytes());
+            return CRC32C_HASH.calculate(payload.memoryAddress() + payload.readerIndex(), payload.readableBytes());
         } else if (payload.hasArray()) {
             return CRC32C_HASH.calculate(payload.array(), payload.arrayOffset() + payload.readerIndex(),
                     payload.readableBytes());
@@ -54,5 +54,18 @@ public class Crc32cChecksum {
             return CRC32C_HASH.calculate(payload.nioBuffer());
         }
     }
+    
+    
+    public static int resumeChecksum(int previousChecksum, ByteBuf payload) {
+        if (payload.hasMemoryAddress() && (CRC32C_HASH instanceof Sse42Crc32C)) {
+            return CRC32C_HASH.resume(previousChecksum, payload.memoryAddress() + payload.readerIndex(),
+                    payload.readableBytes());
+        } else if (payload.hasArray()) {
+            return CRC32C_HASH.resume(previousChecksum, payload.array(), payload.arrayOffset() + payload.readerIndex(),
+                    payload.readableBytes());
+        } else {
+            return CRC32C_HASH.resume(previousChecksum, payload.nioBuffer());
+        }
+    }
 
 }

pulsar-checksum/src/test/java/com/scurrilous/circe/crc/CRCTest.java

@@ -31,6 +31,7 @@ import java.nio.charset.Charset;
 import org.testng.annotations.Test;
 
 import com.scurrilous.circe.HashProvider;
+import com.scurrilous.circe.IncrementalIntHash;
 import com.scurrilous.circe.params.CrcParameters;
 
 /**
@@ -160,4 +161,20 @@ public class CRCTest {
     public void testCRC64_XZ() {
         assertEquals(0x995dc9bbdf1939faL, PROVIDER.getIncrementalLong(CRC64_XZ).calculate(DIGITS));
     }
-}
+    
+    @Test
+    public void testCRC32CIncremental() {
+        // reflected
+        testIncremental(PROVIDER.getIncrementalInt(CRC32C));
+    }
+
+    private void testIncremental(IncrementalIntHash hash) {
+        final String data = "data";
+        final String combined = data + data;
+
+        final int dataChecksum = hash.calculate(data.getBytes(ASCII));
+        final int combinedChecksum = hash.calculate(combined.getBytes(ASCII));
+        final int incrementalChecksum = hash.resume(dataChecksum, data.getBytes(ASCII));
+        assertEquals(combinedChecksum, incrementalChecksum);
+    }
+}
\ No newline at end of file

pulsar-common/src/test/java/com/yahoo/pulsar/common/compression/Crc32cChecksumTest.java

@@ -96,4 +96,60 @@ public class Crc32cChecksumTest {
         payload.release();
         assertEquals(checksum, expectedChecksum);
     }
+    
+    @Test
+    public void testCrc32cIncremental() {
+        if (HARDWARE_CRC32C_HASH == null) {
+            return;
+        }
+
+        String data = "data-abcd-data-123-$%#";
+
+        for (int i = 0; i < 20; i++) {
+            String doubleData = data + data;
+
+            int doubleDataCrcHW = HARDWARE_CRC32C_HASH.calculate(doubleData.getBytes());
+            int data1CrcHW = HARDWARE_CRC32C_HASH.calculate(data.getBytes());
+            int data2CrcHW = HARDWARE_CRC32C_HASH.resume(data1CrcHW, data.getBytes());
+            assertEquals(doubleDataCrcHW, data2CrcHW);
+
+            int doubleDataCrcSW = SOFTWARE_CRC32C_HASH.calculate(doubleData.getBytes());
+            int data1CrcSW = SOFTWARE_CRC32C_HASH.calculate(data.getBytes());
+            int data2CrcSW = SOFTWARE_CRC32C_HASH.resume(data1CrcSW, data.getBytes());
+            assertEquals(doubleDataCrcSW, data2CrcSW);
+
+            assertEquals(doubleDataCrcHW, doubleDataCrcSW);
+
+            data += data;
+        }
+    }
+    
+    @Test
+    public void testCrc32cIncrementalUsingProvider() {
+
+        final byte[] data = "data".getBytes();
+        final byte[] doubleData = "datadata".getBytes();
+        ByteBuf payload = Unpooled.wrappedBuffer(data);
+        ByteBuf doublePayload = Unpooled.wrappedBuffer(doubleData);
+
+        int expectedChecksum = Crc32cChecksum.computeChecksum(doublePayload);
+        
+        // (1) heap-memory
+        int checksum = Crc32cChecksum.computeChecksum(payload);
+        int incrementalChecksum = Crc32cChecksum.resumeChecksum(checksum, payload);
+        assertEquals(expectedChecksum, incrementalChecksum);
+        payload.release();
+        doublePayload.release();
+        
+        // (2) direct-memory
+        payload = ByteBufAllocator.DEFAULT.directBuffer(data.length);
+        payload.writeBytes(data);
+        checksum = Crc32cChecksum.computeChecksum(payload);
+        incrementalChecksum = Crc32cChecksum.resumeChecksum(checksum, payload);
+        assertEquals(expectedChecksum, incrementalChecksum);
+        payload.release();
+        
+    
+    }
+    
 }