diff --git a/circe-checksum/src/main/java/com/scurrilous/circe/checksum/Java9IntHash.java b/circe-checksum/src/main/java/com/scurrilous/circe/checksum/Java9IntHash.java index f2d7276ec0c..92db94c44f0 100644 --- a/circe-checksum/src/main/java/com/scurrilous/circe/checksum/Java9IntHash.java +++ b/circe-checksum/src/main/java/com/scurrilous/circe/checksum/Java9IntHash.java @@ -20,15 +20,22 @@ import io.netty.buffer.ByteBuf; import io.netty.util.concurrent.FastThreadLocal; -import java.lang.reflect.InvocationTargetException; +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; import java.lang.reflect.Method; import lombok.CustomLog; @CustomLog public class Java9IntHash implements IntHash { static final boolean HAS_JAVA9_CRC32C; - private static final Method UPDATE_BYTES; - private static final Method UPDATE_DIRECT_BYTEBUFFER; + + // Method handles rather than java.lang.reflect.Method: Method.invoke takes its arguments as an + // Object[], so every call boxes the checksum, the address and the offsets and allocates the + // array. Since this runs once per checksummed buffer, that showed up as ~9% of all allocation + // in a broker under a write-heavy workload. invokeExact on a static final handle passes the + // primitives straight through and lets the JIT inline the target, allocating nothing. + private static final MethodHandle UPDATE_BYTES; + private static final MethodHandle UPDATE_DIRECT_BYTEBUFFER; private static final String CRC32C_CLASS_NAME = "java.util.zip.CRC32C"; @@ -41,16 +48,25 @@ protected byte[] initialValue() { static { boolean hasJava9CRC32C = false; - Method updateBytes = null; - Method updateDirectByteBuffer = null; + MethodHandle updateBytes = null; + MethodHandle updateDirectByteBuffer = null; try { Class c = Class.forName(CRC32C_CLASS_NAME); - updateBytes = c.getDeclaredMethod("updateBytes", int.class, byte[].class, int.class, int.class); - updateBytes.setAccessible(true); - updateDirectByteBuffer = + MethodHandles.Lookup lookup = MethodHandles.lookup(); + + // The methods are private to java.util.zip, so they are made accessible first and then + // unreflected: Lookup.unreflect skips its own access check for a method whose accessible + // flag is already set, which is what lets this reach them without a Java 9+ lookup API. + Method updateBytesMethod = + c.getDeclaredMethod("updateBytes", int.class, byte[].class, int.class, int.class); + updateBytesMethod.setAccessible(true); + updateBytes = lookup.unreflect(updateBytesMethod); + + Method updateDirectByteBufferMethod = c.getDeclaredMethod("updateDirectByteBuffer", int.class, long.class, int.class, int.class); - updateDirectByteBuffer.setAccessible(true); + updateDirectByteBufferMethod.setAccessible(true); + updateDirectByteBuffer = lookup.unreflect(updateDirectByteBufferMethod); hasJava9CRC32C = true; } catch (Exception e) { @@ -76,9 +92,11 @@ public int calculate(ByteBuf buffer, int offset, int len) { private int updateDirectByteBuffer(int current, long address, int offset, int length) { try { - return (int) UPDATE_DIRECT_BYTEBUFFER.invoke(null, current, address, offset, offset + length); - } catch (IllegalAccessException | InvocationTargetException e) { - throw new RuntimeException(e); + // The argument and return types have to match the handle's type exactly for + // invokeExact: (int, long, int, int)int. + return (int) UPDATE_DIRECT_BYTEBUFFER.invokeExact(current, address, offset, offset + length); + } catch (Throwable t) { + throw asUnchecked(t); } } @@ -97,10 +115,28 @@ public boolean acceptsMemoryAddressBuffer() { private static int updateBytes(int current, byte[] array, int offset, int length) { try { - return (int) UPDATE_BYTES.invoke(null, current, array, offset, offset + length); - } catch (IllegalAccessException | InvocationTargetException e) { - throw new RuntimeException(e); + // The argument and return types have to match the handle's type exactly for + // invokeExact: (int, byte[], int, int)int. + return (int) UPDATE_BYTES.invokeExact(current, array, offset, offset + length); + } catch (Throwable t) { + throw asUnchecked(t); + } + } + + /** + * Adapts a failure from {@link MethodHandle#invokeExact}, which is declared to throw + * {@link Throwable}, to something this method can throw. Unlike {@code Method.invoke}, an + * exception raised by the target is not wrapped, so it is passed through unchanged when it + * already is unchecked. + */ + private static RuntimeException asUnchecked(Throwable t) { + if (t instanceof Error) { + throw (Error) t; + } + if (t instanceof RuntimeException) { + return (RuntimeException) t; } + return new RuntimeException(t); } @Override diff --git a/circe-checksum/src/test/java/com/scurrilous/circe/checksum/Java9IntHashTest.java b/circe-checksum/src/test/java/com/scurrilous/circe/checksum/Java9IntHashTest.java index 3fb57b4a1aa..50754d02398 100644 --- a/circe-checksum/src/test/java/com/scurrilous/circe/checksum/Java9IntHashTest.java +++ b/circe-checksum/src/test/java/com/scurrilous/circe/checksum/Java9IntHashTest.java @@ -25,6 +25,7 @@ import java.util.Random; import lombok.extern.slf4j.Slf4j; import org.junit.Assert; +import org.junit.Assume; import org.junit.Test; @Slf4j @@ -102,6 +103,51 @@ public void calculateCheckSumUsingNoArrayNoMemoryAddrByteBuf() { b2.release(); } + /** + * The three ways {@link Java9IntHash#resume(int, ByteBuf, int, int)} can reach the data, checked + * against {@link Java8IntHash} as an independent implementation of the same algorithm. + * + *

The direct path is the one worth pinning: it is the only caller of the JDK's + * {@code updateDirectByteBuffer}, whose {@code (int, long, int, int)int} shape has to be matched + * exactly by the method handle invocation. Getting that wrong is not a compile error — it fails + * at runtime with a {@code WrongMethodTypeException} — and none of the other cases would catch + * it, since they take the {@code updateBytes} path instead. + */ + @Test + public void matchesJava8ImplementationOnEveryBufferKind() { + Assume.assumeTrue("java.util.zip.CRC32C is not reachable, so Java9IntHash is not in use", + Java9IntHash.HAS_JAVA9_CRC32C); + + byte[] data = new byte[8192]; + new Random(42).nextBytes(data); + + ByteBuf direct = ByteBufAllocator.DEFAULT.directBuffer(data.length); + ByteBuf heap = ByteBufAllocator.DEFAULT.heapBuffer(data.length); + try { + direct.writeBytes(data); + heap.writeBytes(data); + Assert.assertTrue("expected a buffer with a memory address to cover the direct path", + direct.hasMemoryAddress()); + + Java9IntHash java9 = new Java9IntHash(); + Java8IntHash java8 = new Java8IntHash(); + + Assert.assertEquals(java8.calculate(direct), java9.calculate(direct)); + Assert.assertEquals(java8.calculate(heap), java9.calculate(heap)); + Assert.assertEquals(java8.calculate(new NoArrayNoMemoryAddrByteBuff(heap)), + java9.calculate(new NoArrayNoMemoryAddrByteBuff(heap))); + + // Resuming has to agree too: it is the incremental form the ledger write path uses. + int half = data.length / 2; + int java9Resumed = java9.resume(java9.calculate(direct.slice(0, half)), + direct.slice(half, data.length - half)); + Assert.assertEquals(java8.calculate(direct), java9Resumed); + } finally { + direct.release(); + heap.release(); + } + } + public static class NoArrayNoMemoryAddrByteBuff extends DuplicatedByteBuf { public NoArrayNoMemoryAddrByteBuff(ByteBuf buffer) {