From 181f668734666487921e129e3a23fb37d4bb656a Mon Sep 17 00:00:00 2001 From: Martin Traverso Date: Tue, 10 Aug 2021 12:58:15 -0700 Subject: [PATCH] Properly target Java 8 Java 9+ added an overload of various methods in ByteBuffer. When compiling with Java 11+ and targeting Java 8 bytecode the resulting signatures are invalid for JDK 8, so accesses below result in NoSuchMethodError. Accessing the methods through the interface class works around the problem. Ideally, we'd use "javac --release 8", but that doesn't work because Unsafe is not available in the signature data for that profile. --- .github/workflows/main.yml | 9 +++- pom.xml | 17 +++++++ .../airlift/compress/lz4/Lz4Compressor.java | 50 +++++++++++-------- .../airlift/compress/lz4/Lz4Decompressor.java | 50 +++++++++++-------- .../airlift/compress/lzo/LzoCompressor.java | 50 +++++++++++-------- .../airlift/compress/lzo/LzoDecompressor.java | 50 +++++++++++-------- .../compress/snappy/SnappyCompressor.java | 50 +++++++++++-------- .../compress/snappy/SnappyDecompressor.java | 50 +++++++++++-------- .../airlift/compress/zstd/ZstdCompressor.java | 50 +++++++++++-------- .../compress/zstd/ZstdDecompressor.java | 50 +++++++++++-------- .../compress/AbstractTestCompression.java | 33 +++++++----- 11 files changed, 276 insertions(+), 183 deletions(-) diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 44edd706..dae365d4 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -10,9 +10,14 @@ jobs: java: ['8', '11', '15', '16'] steps: - uses: actions/checkout@v1 - - name: Setup JDK ${{ matrix.java }} + - name: Setup test JDK ${{ matrix.java }} uses: actions/setup-java@v1 with: java-version: ${{ matrix.java }} + - run: echo "test_java_home=$JAVA_HOME" >> $GITHUB_ENV + - name: Setup build JDK 16 + uses: actions/setup-java@v1 + with: + java-version: 16 - name: Maven Test - run: mvn install -B + run: mvn install -V -P ci -B -Dtest_java_home=${{ env.test_java_home }} diff --git a/pom.xml b/pom.xml index 88939945..d0e23a24 100644 --- a/pom.xml +++ b/pom.xml @@ -167,4 +167,21 @@ + + + + ci + + + + org.apache.maven.plugins + maven-surefire-plugin + + ${test_java_home}/bin/java + + + + + + diff --git a/src/main/java/io/airlift/compress/lz4/Lz4Compressor.java b/src/main/java/io/airlift/compress/lz4/Lz4Compressor.java index 6e87b66c..94b78560 100644 --- a/src/main/java/io/airlift/compress/lz4/Lz4Compressor.java +++ b/src/main/java/io/airlift/compress/lz4/Lz4Compressor.java @@ -15,6 +15,7 @@ import io.airlift.compress.Compressor; +import java.nio.Buffer; import java.nio.ByteBuffer; import static io.airlift.compress.lz4.Lz4RawCompressor.MAX_TABLE_SIZE; @@ -47,48 +48,55 @@ public int compress(byte[] input, int inputOffset, int inputLength, byte[] outpu @Override public void compress(ByteBuffer input, ByteBuffer output) { + // Java 9+ added an overload of various methods in ByteBuffer. When compiling with Java 11+ and targeting Java 8 bytecode + // the resulting signatures are invalid for JDK 8, so accesses below result in NoSuchMethodError. Accessing the + // methods through the interface class works around the problem + // Sidenote: we can't target "javac --release 8" because Unsafe is not available in the signature data for that profile + Buffer inputBuffer = input; + Buffer outputBuffer = output; + Object inputBase; long inputAddress; long inputLimit; - if (input.isDirect()) { + if (inputBuffer.isDirect()) { inputBase = null; - long address = getAddress(input); - inputAddress = address + input.position(); - inputLimit = address + input.limit(); + long address = getAddress(inputBuffer); + inputAddress = address + inputBuffer.position(); + inputLimit = address + inputBuffer.limit(); } - else if (input.hasArray()) { - inputBase = input.array(); - inputAddress = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.position(); - inputLimit = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.limit(); + else if (inputBuffer.hasArray()) { + inputBase = inputBuffer.array(); + inputAddress = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.position(); + inputLimit = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + input.getClass().getName()); + throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + inputBuffer.getClass().getName()); } Object outputBase; long outputAddress; long outputLimit; - if (output.isDirect()) { + if (outputBuffer.isDirect()) { outputBase = null; - long address = getAddress(output); - outputAddress = address + output.position(); - outputLimit = address + output.limit(); + long address = getAddress(outputBuffer); + outputAddress = address + outputBuffer.position(); + outputLimit = address + outputBuffer.limit(); } - else if (output.hasArray()) { - outputBase = output.array(); - outputAddress = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.position(); - outputLimit = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.limit(); + else if (outputBuffer.hasArray()) { + outputBase = outputBuffer.array(); + outputAddress = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.position(); + outputLimit = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + output.getClass().getName()); + throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + outputBuffer.getClass().getName()); } // HACK: Assure JVM does not collect Slice wrappers while compressing, since the // collection may trigger freeing of the underlying memory resulting in a segfault // There is no other known way to signal to the JVM that an object should not be // collected in a block, and technically, the JVM is allowed to eliminate these locks. - synchronized (input) { - synchronized (output) { + synchronized (inputBuffer) { + synchronized (outputBuffer) { int written = Lz4RawCompressor.compress( inputBase, inputAddress, @@ -97,7 +105,7 @@ else if (output.hasArray()) { outputAddress, outputLimit - outputAddress, table); - output.position(output.position() + written); + outputBuffer.position(outputBuffer.position() + written); } } } diff --git a/src/main/java/io/airlift/compress/lz4/Lz4Decompressor.java b/src/main/java/io/airlift/compress/lz4/Lz4Decompressor.java index 8e81592b..a5e47801 100644 --- a/src/main/java/io/airlift/compress/lz4/Lz4Decompressor.java +++ b/src/main/java/io/airlift/compress/lz4/Lz4Decompressor.java @@ -16,6 +16,7 @@ import io.airlift.compress.Decompressor; import io.airlift.compress.MalformedInputException; +import java.nio.Buffer; import java.nio.ByteBuffer; import static io.airlift.compress.lz4.UnsafeUtil.getAddress; @@ -40,50 +41,57 @@ public int decompress(byte[] input, int inputOffset, int inputLength, byte[] out public void decompress(ByteBuffer input, ByteBuffer output) throws MalformedInputException { + // Java 9+ added an overload of various methods in ByteBuffer. When compiling with Java 11+ and targeting Java 8 bytecode + // the resulting signatures are invalid for JDK 8, so accesses below result in NoSuchMethodError. Accessing the + // methods through the interface class works around the problem + // Sidenote: we can't target "javac --release 8" because Unsafe is not available in the signature data for that profile + Buffer inputBuffer = input; + Buffer outputBuffer = output; + Object inputBase; long inputAddress; long inputLimit; - if (input.isDirect()) { + if (inputBuffer.isDirect()) { inputBase = null; - long address = getAddress(input); - inputAddress = address + input.position(); - inputLimit = address + input.limit(); + long address = getAddress(inputBuffer); + inputAddress = address + inputBuffer.position(); + inputLimit = address + inputBuffer.limit(); } - else if (input.hasArray()) { - inputBase = input.array(); - inputAddress = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.position(); - inputLimit = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.limit(); + else if (inputBuffer.hasArray()) { + inputBase = inputBuffer.array(); + inputAddress = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.position(); + inputLimit = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + input.getClass().getName()); + throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + inputBuffer.getClass().getName()); } Object outputBase; long outputAddress; long outputLimit; - if (output.isDirect()) { + if (outputBuffer.isDirect()) { outputBase = null; - long address = getAddress(output); - outputAddress = address + output.position(); - outputLimit = address + output.limit(); + long address = getAddress(outputBuffer); + outputAddress = address + outputBuffer.position(); + outputLimit = address + outputBuffer.limit(); } - else if (output.hasArray()) { - outputBase = output.array(); - outputAddress = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.position(); - outputLimit = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.limit(); + else if (outputBuffer.hasArray()) { + outputBase = outputBuffer.array(); + outputAddress = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.position(); + outputLimit = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + output.getClass().getName()); + throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + outputBuffer.getClass().getName()); } // HACK: Assure JVM does not collect Slice wrappers while decompressing, since the // collection may trigger freeing of the underlying memory resulting in a segfault // There is no other known way to signal to the JVM that an object should not be // collected in a block, and technically, the JVM is allowed to eliminate these locks. - synchronized (input) { - synchronized (output) { + synchronized (inputBuffer) { + synchronized (outputBuffer) { int written = Lz4RawDecompressor.decompress(inputBase, inputAddress, inputLimit, outputBase, outputAddress, outputLimit); - output.position(output.position() + written); + outputBuffer.position(outputBuffer.position() + written); } } } diff --git a/src/main/java/io/airlift/compress/lzo/LzoCompressor.java b/src/main/java/io/airlift/compress/lzo/LzoCompressor.java index 6a279bd6..355f85ad 100644 --- a/src/main/java/io/airlift/compress/lzo/LzoCompressor.java +++ b/src/main/java/io/airlift/compress/lzo/LzoCompressor.java @@ -15,6 +15,7 @@ import io.airlift.compress.Compressor; +import java.nio.Buffer; import java.nio.ByteBuffer; import static io.airlift.compress.lzo.LzoRawCompressor.MAX_TABLE_SIZE; @@ -47,48 +48,55 @@ public int compress(byte[] input, int inputOffset, int inputLength, byte[] outpu @Override public void compress(ByteBuffer input, ByteBuffer output) { + // Java 9+ added an overload of various methods in ByteBuffer. When compiling with Java 11+ and targeting Java 8 bytecode + // the resulting signatures are invalid for JDK 8, so accesses below result in NoSuchMethodError. Accessing the + // methods through the interface class works around the problem + // Sidenote: we can't target "javac --release 8" because Unsafe is not available in the signature data for that profile + Buffer inputBuffer = input; + Buffer outputBuffer = output; + Object inputBase; long inputAddress; long inputLimit; - if (input.isDirect()) { + if (inputBuffer.isDirect()) { inputBase = null; - long address = getAddress(input); - inputAddress = address + input.position(); - inputLimit = address + input.limit(); + long address = getAddress(inputBuffer); + inputAddress = address + inputBuffer.position(); + inputLimit = address + inputBuffer.limit(); } - else if (input.hasArray()) { - inputBase = input.array(); - inputAddress = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.position(); - inputLimit = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.limit(); + else if (inputBuffer.hasArray()) { + inputBase = inputBuffer.array(); + inputAddress = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.position(); + inputLimit = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + input.getClass().getName()); + throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + inputBuffer.getClass().getName()); } Object outputBase; long outputAddress; long outputLimit; - if (output.isDirect()) { + if (outputBuffer.isDirect()) { outputBase = null; - long address = getAddress(output); - outputAddress = address + output.position(); - outputLimit = address + output.limit(); + long address = getAddress(outputBuffer); + outputAddress = address + outputBuffer.position(); + outputLimit = address + outputBuffer.limit(); } - else if (output.hasArray()) { - outputBase = output.array(); - outputAddress = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.position(); - outputLimit = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.limit(); + else if (outputBuffer.hasArray()) { + outputBase = outputBuffer.array(); + outputAddress = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.position(); + outputLimit = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + output.getClass().getName()); + throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + outputBuffer.getClass().getName()); } // HACK: Assure JVM does not collect Slice wrappers while compressing, since the // collection may trigger freeing of the underlying memory resulting in a segfault // There is no other known way to signal to the JVM that an object should not be // collected in a block, and technically, the JVM is allowed to eliminate these locks. - synchronized (input) { - synchronized (output) { + synchronized (inputBuffer) { + synchronized (outputBuffer) { int written = LzoRawCompressor.compress( inputBase, inputAddress, @@ -97,7 +105,7 @@ else if (output.hasArray()) { outputAddress, outputLimit - outputAddress, table); - output.position(output.position() + written); + outputBuffer.position(outputBuffer.position() + written); } } } diff --git a/src/main/java/io/airlift/compress/lzo/LzoDecompressor.java b/src/main/java/io/airlift/compress/lzo/LzoDecompressor.java index 86125d9e..123ae3ea 100644 --- a/src/main/java/io/airlift/compress/lzo/LzoDecompressor.java +++ b/src/main/java/io/airlift/compress/lzo/LzoDecompressor.java @@ -16,6 +16,7 @@ import io.airlift.compress.Decompressor; import io.airlift.compress.MalformedInputException; +import java.nio.Buffer; import java.nio.ByteBuffer; import static io.airlift.compress.lzo.UnsafeUtil.getAddress; @@ -40,50 +41,57 @@ public int decompress(byte[] input, int inputOffset, int inputLength, byte[] out public void decompress(ByteBuffer input, ByteBuffer output) throws MalformedInputException { + // Java 9+ added an overload of various methods in ByteBuffer. When compiling with Java 11+ and targeting Java 8 bytecode + // the resulting signatures are invalid for JDK 8, so accesses below result in NoSuchMethodError. Accessing the + // methods through the interface class works around the problem + // Sidenote: we can't target "javac --release 8" because Unsafe is not available in the signature data for that profile + Buffer inputBuffer = input; + Buffer outputBuffer = output; + Object inputBase; long inputAddress; long inputLimit; - if (input.isDirect()) { + if (inputBuffer.isDirect()) { inputBase = null; - long address = getAddress(input); - inputAddress = address + input.position(); - inputLimit = address + input.limit(); + long address = getAddress(inputBuffer); + inputAddress = address + inputBuffer.position(); + inputLimit = address + inputBuffer.limit(); } - else if (input.hasArray()) { - inputBase = input.array(); - inputAddress = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.position(); - inputLimit = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.limit(); + else if (inputBuffer.hasArray()) { + inputBase = inputBuffer.array(); + inputAddress = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.position(); + inputLimit = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + input.getClass().getName()); + throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + inputBuffer.getClass().getName()); } Object outputBase; long outputAddress; long outputLimit; - if (output.isDirect()) { + if (outputBuffer.isDirect()) { outputBase = null; - long address = getAddress(output); - outputAddress = address + output.position(); - outputLimit = address + output.limit(); + long address = getAddress(outputBuffer); + outputAddress = address + outputBuffer.position(); + outputLimit = address + outputBuffer.limit(); } - else if (output.hasArray()) { - outputBase = output.array(); - outputAddress = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.position(); - outputLimit = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.limit(); + else if (outputBuffer.hasArray()) { + outputBase = outputBuffer.array(); + outputAddress = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.position(); + outputLimit = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + output.getClass().getName()); + throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + outputBuffer.getClass().getName()); } // HACK: Assure JVM does not collect Slice wrappers while decompressing, since the // collection may trigger freeing of the underlying memory resulting in a segfault // There is no other known way to signal to the JVM that an object should not be // collected in a block, and technically, the JVM is allowed to eliminate these locks. - synchronized (input) { - synchronized (output) { + synchronized (inputBuffer) { + synchronized (outputBuffer) { int written = LzoRawDecompressor.decompress(inputBase, inputAddress, inputLimit, outputBase, outputAddress, outputLimit); - output.position(output.position() + written); + outputBuffer.position(outputBuffer.position() + written); } } } diff --git a/src/main/java/io/airlift/compress/snappy/SnappyCompressor.java b/src/main/java/io/airlift/compress/snappy/SnappyCompressor.java index 76e30ab7..a041c26f 100644 --- a/src/main/java/io/airlift/compress/snappy/SnappyCompressor.java +++ b/src/main/java/io/airlift/compress/snappy/SnappyCompressor.java @@ -15,6 +15,7 @@ import io.airlift.compress.Compressor; +import java.nio.Buffer; import java.nio.ByteBuffer; import static io.airlift.compress.snappy.UnsafeUtil.getAddress; @@ -45,48 +46,55 @@ public int compress(byte[] input, int inputOffset, int inputLength, byte[] outpu @Override public void compress(ByteBuffer input, ByteBuffer output) { + // Java 9+ added an overload of various methods in ByteBuffer. When compiling with Java 11+ and targeting Java 8 bytecode + // the resulting signatures are invalid for JDK 8, so accesses below result in NoSuchMethodError. Accessing the + // methods through the interface class works around the problem + // Sidenote: we can't target "javac --release 8" because Unsafe is not available in the signature data for that profile + Buffer inputBuffer = input; + Buffer outputBuffer = output; + Object inputBase; long inputAddress; long inputLimit; - if (input.isDirect()) { + if (inputBuffer.isDirect()) { inputBase = null; - long address = getAddress(input); - inputAddress = address + input.position(); - inputLimit = address + input.limit(); + long address = getAddress(inputBuffer); + inputAddress = address + inputBuffer.position(); + inputLimit = address + inputBuffer.limit(); } - else if (input.hasArray()) { - inputBase = input.array(); - inputAddress = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.position(); - inputLimit = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.limit(); + else if (inputBuffer.hasArray()) { + inputBase = inputBuffer.array(); + inputAddress = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.position(); + inputLimit = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + input.getClass().getName()); + throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + inputBuffer.getClass().getName()); } Object outputBase; long outputAddress; long outputLimit; - if (output.isDirect()) { + if (outputBuffer.isDirect()) { outputBase = null; - long address = getAddress(output); - outputAddress = address + output.position(); - outputLimit = address + output.limit(); + long address = getAddress(outputBuffer); + outputAddress = address + outputBuffer.position(); + outputLimit = address + outputBuffer.limit(); } - else if (output.hasArray()) { - outputBase = output.array(); - outputAddress = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.position(); - outputLimit = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.limit(); + else if (outputBuffer.hasArray()) { + outputBase = outputBuffer.array(); + outputAddress = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.position(); + outputLimit = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + output.getClass().getName()); + throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + outputBuffer.getClass().getName()); } // HACK: Assure JVM does not collect Slice wrappers while compressing, since the // collection may trigger freeing of the underlying memory resulting in a segfault // There is no other known way to signal to the JVM that an object should not be // collected in a block, and technically, the JVM is allowed to eliminate these locks. - synchronized (input) { - synchronized (output) { + synchronized (inputBuffer) { + synchronized (outputBuffer) { int written = SnappyRawCompressor.compress( inputBase, inputAddress, @@ -95,7 +103,7 @@ else if (output.hasArray()) { outputAddress, outputLimit, table); - output.position(output.position() + written); + outputBuffer.position(outputBuffer.position() + written); } } } diff --git a/src/main/java/io/airlift/compress/snappy/SnappyDecompressor.java b/src/main/java/io/airlift/compress/snappy/SnappyDecompressor.java index 98feb972..0a0b90e1 100644 --- a/src/main/java/io/airlift/compress/snappy/SnappyDecompressor.java +++ b/src/main/java/io/airlift/compress/snappy/SnappyDecompressor.java @@ -16,6 +16,7 @@ import io.airlift.compress.Decompressor; import io.airlift.compress.MalformedInputException; +import java.nio.Buffer; import java.nio.ByteBuffer; import static io.airlift.compress.snappy.UnsafeUtil.getAddress; @@ -48,50 +49,57 @@ public int decompress(byte[] input, int inputOffset, int inputLength, byte[] out public void decompress(ByteBuffer input, ByteBuffer output) throws MalformedInputException { + // Java 9+ added an overload of various methods in ByteBuffer. When compiling with Java 11+ and targeting Java 8 bytecode + // the resulting signatures are invalid for JDK 8, so accesses below result in NoSuchMethodError. Accessing the + // methods through the interface class works around the problem + // Sidenote: we can't target "javac --release 8" because Unsafe is not available in the signature data for that profile + Buffer inputBuffer = input; + Buffer outputBuffer = output; + Object inputBase; long inputAddress; long inputLimit; - if (input.isDirect()) { + if (inputBuffer.isDirect()) { inputBase = null; - long address = getAddress(input); - inputAddress = address + input.position(); - inputLimit = address + input.limit(); + long address = getAddress(inputBuffer); + inputAddress = address + inputBuffer.position(); + inputLimit = address + inputBuffer.limit(); } - else if (input.hasArray()) { - inputBase = input.array(); - inputAddress = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.position(); - inputLimit = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.limit(); + else if (inputBuffer.hasArray()) { + inputBase = inputBuffer.array(); + inputAddress = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.position(); + inputLimit = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + input.getClass().getName()); + throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + inputBuffer.getClass().getName()); } Object outputBase; long outputAddress; long outputLimit; - if (output.isDirect()) { + if (outputBuffer.isDirect()) { outputBase = null; - long address = getAddress(output); - outputAddress = address + output.position(); - outputLimit = address + output.limit(); + long address = getAddress(outputBuffer); + outputAddress = address + outputBuffer.position(); + outputLimit = address + outputBuffer.limit(); } - else if (output.hasArray()) { - outputBase = output.array(); - outputAddress = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.position(); - outputLimit = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.limit(); + else if (outputBuffer.hasArray()) { + outputBase = outputBuffer.array(); + outputAddress = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.position(); + outputLimit = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + output.getClass().getName()); + throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + outputBuffer.getClass().getName()); } // HACK: Assure JVM does not collect Slice wrappers while decompressing, since the // collection may trigger freeing of the underlying memory resulting in a segfault // There is no other known way to signal to the JVM that an object should not be // collected in a block, and technically, the JVM is allowed to eliminate these locks. - synchronized (input) { - synchronized (output) { + synchronized (inputBuffer) { + synchronized (outputBuffer) { int written = SnappyRawDecompressor.decompress(inputBase, inputAddress, inputLimit, outputBase, outputAddress, outputLimit); - output.position(output.position() + written); + outputBuffer.position(outputBuffer.position() + written); } } } diff --git a/src/main/java/io/airlift/compress/zstd/ZstdCompressor.java b/src/main/java/io/airlift/compress/zstd/ZstdCompressor.java index 28388a56..91f5bab8 100644 --- a/src/main/java/io/airlift/compress/zstd/ZstdCompressor.java +++ b/src/main/java/io/airlift/compress/zstd/ZstdCompressor.java @@ -15,6 +15,7 @@ import io.airlift.compress.Compressor; +import java.nio.Buffer; import java.nio.ByteBuffer; import static io.airlift.compress.zstd.Constants.MAX_BLOCK_SIZE; @@ -48,48 +49,55 @@ public int compress(byte[] input, int inputOffset, int inputLength, byte[] outpu @Override public void compress(ByteBuffer input, ByteBuffer output) { + // Java 9+ added an overload of various methods in ByteBuffer. When compiling with Java 11+ and targeting Java 8 bytecode + // the resulting signatures are invalid for JDK 8, so accesses below result in NoSuchMethodError. Accessing the + // methods through the interface class works around the problem + // Sidenote: we can't target "javac --release 8" because Unsafe is not available in the signature data for that profile + Buffer inputBuffer = input; + Buffer outputBuffer = output; + Object inputBase; long inputAddress; long inputLimit; - if (input.isDirect()) { + if (inputBuffer.isDirect()) { inputBase = null; - long address = getAddress(input); - inputAddress = address + input.position(); - inputLimit = address + input.limit(); + long address = getAddress(inputBuffer); + inputAddress = address + inputBuffer.position(); + inputLimit = address + inputBuffer.limit(); } - else if (input.hasArray()) { - inputBase = input.array(); - inputAddress = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.position(); - inputLimit = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.limit(); + else if (inputBuffer.hasArray()) { + inputBase = inputBuffer.array(); + inputAddress = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.position(); + inputLimit = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + input.getClass().getName()); + throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + inputBuffer.getClass().getName()); } Object outputBase; long outputAddress; long outputLimit; - if (output.isDirect()) { + if (outputBuffer.isDirect()) { outputBase = null; - long address = getAddress(output); - outputAddress = address + output.position(); - outputLimit = address + output.limit(); + long address = getAddress(outputBuffer); + outputAddress = address + outputBuffer.position(); + outputLimit = address + outputBuffer.limit(); } - else if (output.hasArray()) { - outputBase = output.array(); - outputAddress = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.position(); - outputLimit = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.limit(); + else if (outputBuffer.hasArray()) { + outputBase = outputBuffer.array(); + outputAddress = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.position(); + outputLimit = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + output.getClass().getName()); + throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + outputBuffer.getClass().getName()); } // HACK: Assure JVM does not collect Slice wrappers while compressing, since the // collection may trigger freeing of the underlying memory resulting in a segfault // There is no other known way to signal to the JVM that an object should not be // collected in a block, and technically, the JVM is allowed to eliminate these locks. - synchronized (input) { - synchronized (output) { + synchronized (inputBuffer) { + synchronized (outputBuffer) { int written = ZstdFrameCompressor.compress( inputBase, inputAddress, @@ -98,7 +106,7 @@ else if (output.hasArray()) { outputAddress, outputLimit, CompressionParameters.DEFAULT_COMPRESSION_LEVEL); - output.position(output.position() + written); + outputBuffer.position(outputBuffer.position() + written); } } } diff --git a/src/main/java/io/airlift/compress/zstd/ZstdDecompressor.java b/src/main/java/io/airlift/compress/zstd/ZstdDecompressor.java index 63655d9b..0f5c0db7 100644 --- a/src/main/java/io/airlift/compress/zstd/ZstdDecompressor.java +++ b/src/main/java/io/airlift/compress/zstd/ZstdDecompressor.java @@ -16,6 +16,7 @@ import io.airlift.compress.Decompressor; import io.airlift.compress.MalformedInputException; +import java.nio.Buffer; import java.nio.ByteBuffer; import static io.airlift.compress.zstd.UnsafeUtil.getAddress; @@ -42,50 +43,57 @@ public int decompress(byte[] input, int inputOffset, int inputLength, byte[] out public void decompress(ByteBuffer input, ByteBuffer output) throws MalformedInputException { + // Java 9+ added an overload of various methods in ByteBuffer. When compiling with Java 11+ and targeting Java 8 bytecode + // the resulting signatures are invalid for JDK 8, so accesses below result in NoSuchMethodError. Accessing the + // methods through the interface class works around the problem + // Sidenote: we can't target "javac --release 8" because Unsafe is not available in the signature data for that profile + Buffer inputBuffer = input; + Buffer outputBuffer = output; + Object inputBase; long inputAddress; long inputLimit; - if (input.isDirect()) { + if (inputBuffer.isDirect()) { inputBase = null; - long address = getAddress(input); - inputAddress = address + input.position(); - inputLimit = address + input.limit(); + long address = getAddress(inputBuffer); + inputAddress = address + inputBuffer.position(); + inputLimit = address + inputBuffer.limit(); } - else if (input.hasArray()) { - inputBase = input.array(); - inputAddress = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.position(); - inputLimit = ARRAY_BYTE_BASE_OFFSET + input.arrayOffset() + input.limit(); + else if (inputBuffer.hasArray()) { + inputBase = inputBuffer.array(); + inputAddress = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.position(); + inputLimit = ARRAY_BYTE_BASE_OFFSET + inputBuffer.arrayOffset() + inputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + input.getClass().getName()); + throw new IllegalArgumentException("Unsupported input ByteBuffer implementation " + inputBuffer.getClass().getName()); } Object outputBase; long outputAddress; long outputLimit; - if (output.isDirect()) { + if (outputBuffer.isDirect()) { outputBase = null; - long address = getAddress(output); - outputAddress = address + output.position(); - outputLimit = address + output.limit(); + long address = getAddress(outputBuffer); + outputAddress = address + outputBuffer.position(); + outputLimit = address + outputBuffer.limit(); } - else if (output.hasArray()) { - outputBase = output.array(); - outputAddress = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.position(); - outputLimit = ARRAY_BYTE_BASE_OFFSET + output.arrayOffset() + output.limit(); + else if (outputBuffer.hasArray()) { + outputBase = outputBuffer.array(); + outputAddress = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.position(); + outputLimit = ARRAY_BYTE_BASE_OFFSET + outputBuffer.arrayOffset() + outputBuffer.limit(); } else { - throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + output.getClass().getName()); + throw new IllegalArgumentException("Unsupported output ByteBuffer implementation " + outputBuffer.getClass().getName()); } // HACK: Assure JVM does not collect Slice wrappers while decompressing, since the // collection may trigger freeing of the underlying memory resulting in a segfault // There is no other known way to signal to the JVM that an object should not be // collected in a block, and technically, the JVM is allowed to eliminate these locks. - synchronized (input) { - synchronized (output) { + synchronized (inputBuffer) { + synchronized (outputBuffer) { int written = new ZstdFrameDecompressor().decompress(inputBase, inputAddress, inputLimit, outputBase, outputAddress, outputLimit); - output.position(output.position() + written); + outputBuffer.position(outputBuffer.position() + written); } } } diff --git a/src/test/java/io/airlift/compress/AbstractTestCompression.java b/src/test/java/io/airlift/compress/AbstractTestCompression.java index 2d66dae6..f1355a85 100644 --- a/src/test/java/io/airlift/compress/AbstractTestCompression.java +++ b/src/test/java/io/airlift/compress/AbstractTestCompression.java @@ -22,6 +22,7 @@ import javax.inject.Inject; import java.io.IOException; +import java.nio.Buffer; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; @@ -155,7 +156,7 @@ public void testDecompressByteBufferHeapToHeap(DataSet dataSet) ByteBuffer uncompressed = ByteBuffer.allocate(uncompressedOriginal.length); getDecompressor().decompress(compressed, uncompressed); - uncompressed.flip(); + ((Buffer) uncompressed).flip(); assertByteBufferEqual(ByteBuffer.wrap(uncompressedOriginal), uncompressed); } @@ -174,7 +175,7 @@ public void testDecompressByteBufferHeapToDirect(DataSet dataSet) ByteBuffer uncompressed = ByteBuffer.allocateDirect(uncompressedOriginal.length); getDecompressor().decompress(compressed, uncompressed); - uncompressed.flip(); + ((Buffer) uncompressed).flip(); assertByteBufferEqual(ByteBuffer.wrap(uncompressedOriginal), uncompressed); } @@ -193,7 +194,7 @@ public void testDecompressByteBufferDirectToHeap(DataSet dataSet) ByteBuffer uncompressed = ByteBuffer.allocate(uncompressedOriginal.length); getDecompressor().decompress(compressed, uncompressed); - uncompressed.flip(); + ((Buffer) uncompressed).flip(); assertByteBufferEqual(ByteBuffer.wrap(uncompressedOriginal), uncompressed); } @@ -212,7 +213,7 @@ public void testDecompressByteBufferDirectToDirect(DataSet dataSet) ByteBuffer uncompressed = ByteBuffer.allocateDirect(uncompressedOriginal.length); getDecompressor().decompress(compressed, uncompressed); - uncompressed.flip(); + ((Buffer) uncompressed).flip(); assertByteBufferEqual(ByteBuffer.wrap(uncompressedOriginal), uncompressed); } @@ -323,17 +324,17 @@ private void verifyCompressByteBuffer(Compressor compressor, ByteBuffer expected if (expected.remaining() > 1) { ByteBuffer duplicate = expected.duplicate(); duplicate.get(); // skip one byte - compressor.compress(duplicate, ByteBuffer.allocate(compressed.remaining())); + compressor.compress(duplicate, ByteBuffer.allocate(((Buffer) compressed).remaining())); } compressor.compress(expected.duplicate(), compressed); - compressed.flip(); + ((Buffer) compressed).flip(); - ByteBuffer uncompressed = ByteBuffer.allocate(expected.remaining()); + ByteBuffer uncompressed = ByteBuffer.allocate(((Buffer) expected).remaining()); // TODO: validate with "control" decompressor getDecompressor().decompress(compressed, uncompressed); - uncompressed.flip(); + ((Buffer) uncompressed).flip(); assertByteBufferEqual(expected.duplicate(), uncompressed); } @@ -409,21 +410,27 @@ public static void assertByteArraysEqual(byte[] left, int leftOffset, int leftLe private static void assertByteBufferEqual(ByteBuffer left, ByteBuffer right) { - int leftPosition = left.position(); - int rightPosition = right.position(); - for (int i = 0; i < Math.min(left.remaining(), right.remaining()); i++) { + Buffer leftBuffer = left; + Buffer rightBuffer = right; + + int leftPosition = leftBuffer.position(); + int rightPosition = rightBuffer.position(); + for (int i = 0; i < Math.min(leftBuffer.remaining(), rightBuffer.remaining()); i++) { if (left.get(leftPosition + i) != right.get(rightPosition + i)) { fail(String.format("Byte buffers differ at position %s: 0x%02X vs 0x%02X", i, left.get(leftPosition + i), right.get(rightPosition + i))); } } - assertEquals(left.remaining(), right.remaining(), String.format("Buffer lengths differ: %s vs %s", left.remaining(), left.remaining())); + assertEquals(leftBuffer.remaining(), rightBuffer.remaining(), String.format("Buffer lengths differ: %s vs %s", leftBuffer.remaining(), leftBuffer.remaining())); } private static ByteBuffer toDirectBuffer(byte[] data) { ByteBuffer direct = ByteBuffer.allocateDirect(data.length); - direct.put(data).flip(); + direct.put(data); + + ((Buffer) direct).flip(); + return direct; }