diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBuffer.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBuffer.java index 600a7b56797a..6e06f502b2cf 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBuffer.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBuffer.java @@ -17,7 +17,9 @@ package org.apache.hadoop.ozone.common; +import com.google.common.base.Preconditions; import java.io.IOException; +import java.nio.BufferOverflowException; import java.nio.ByteBuffer; import java.nio.channels.GatheringByteChannel; import java.util.List; @@ -64,6 +66,34 @@ static ChunkBuffer wrap(List buffers) { return new ChunkBufferImplWithByteBufferList(buffers); } + default void checkArgument(byte[] b, int offset, int length) { + Objects.requireNonNull(b, "b == null"); + Preconditions.checkArgument(offset >= 0 && length >= 0, + "offset = %s, length = %s", offset, length); + Preconditions.checkArgument(length <= b.length - offset, + "length = %s out of range for array.length = %s, offset = %s", + length, b.length, offset); + if (length > remaining()) { + final BufferOverflowException boe = new BufferOverflowException(); + boe.initCause(new IllegalArgumentException( + "Failed to put since length = " + length + + " > this.remaining() = " + remaining())); + throw boe; + } + } + + default void checkArgument(ByteBuffer that) { + Objects.requireNonNull(that, "that == null"); + final int thatRemaining = that.remaining(); + if (thatRemaining > remaining()) { + final BufferOverflowException boe = new BufferOverflowException(); + boe.initCause(new IllegalArgumentException( + "Failed to put since that.remaining() = " + thatRemaining + + " > this.remaining() = " + remaining())); + throw boe; + } + } + /** Similar to {@link ByteBuffer#position()}. */ int position(); diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBufferImplWithByteBuffer.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBufferImplWithByteBuffer.java index d37b095621cf..5bc9300e9052 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBufferImplWithByteBuffer.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBufferImplWithByteBuffer.java @@ -118,6 +118,18 @@ public ChunkBuffer put(ByteBuffer b) { return this; } + @Override + public ChunkBuffer put(byte[] b, int offset, int length) { + checkArgument(b, offset, length); + buffer.put(b, offset, length); + return this; + } + + @Override + public ChunkBuffer put(byte[] b) { + return put(b, 0, b.length); + } + @Override public ChunkBuffer put(byte b) { buffer.put(b); diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBufferImplWithByteBufferList.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBufferImplWithByteBufferList.java index e67726e98fff..afbb4253ef73 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBufferImplWithByteBufferList.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBufferImplWithByteBufferList.java @@ -20,7 +20,6 @@ import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; import java.io.IOException; -import java.nio.BufferOverflowException; import java.nio.ByteBuffer; import java.nio.channels.GatheringByteChannel; import java.util.ArrayList; @@ -135,15 +134,8 @@ public ChunkBuffer clear() { @Override public ChunkBuffer put(ByteBuffer that) { - final int thisRemaining = remaining(); + checkArgument(that); int thatRemaining = that.remaining(); - if (thatRemaining > thisRemaining) { - final BufferOverflowException boe = new BufferOverflowException(); - boe.initCause(new IllegalArgumentException( - "Failed to put since that.remaining() = " + thatRemaining - + " > this.remaining() = " + thisRemaining)); - throw boe; - } while (thatRemaining > 0) { final ByteBuffer b = current(); @@ -157,6 +149,27 @@ public ChunkBuffer put(ByteBuffer that) { return this; } + @Override + public ChunkBuffer put(byte[] b, int offset, int length) { + checkArgument(b, offset, length); + + final int end = offset + length; + int off = offset; + while (off < end) { + final ByteBuffer cur = current(); + final int bytes = Math.min(cur.remaining(), end - off); + cur.put(b, off, bytes); + off += bytes; + advanceCurrent(); + } + return this; + } + + @Override + public ChunkBuffer put(byte[] b) { + return put(b, 0, b.length); + } + @Override public ChunkBuffer duplicate(int newPosition, int newLimit) { Preconditions.checkArgument(newPosition >= 0); diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/IncrementalChunkBuffer.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/IncrementalChunkBuffer.java index 1d90fbc3972a..8d20b42f1eca 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/IncrementalChunkBuffer.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/IncrementalChunkBuffer.java @@ -19,7 +19,6 @@ import com.google.common.base.Preconditions; import java.io.IOException; -import java.nio.BufferOverflowException; import java.nio.ByteBuffer; import java.nio.channels.GatheringByteChannel; import java.util.ArrayList; @@ -212,13 +211,7 @@ public ChunkBuffer clear() { @Override public ChunkBuffer put(ByteBuffer that) { - if (that.remaining() > this.remaining()) { - final BufferOverflowException boe = new BufferOverflowException(); - boe.initCause(new IllegalArgumentException( - "Failed to put since that.remaining() = " + that.remaining() - + " > this.remaining() = " + this.remaining())); - throw boe; - } + checkArgument(that); final int thatLimit = that.limit(); for (int p = position(); that.position() < thatLimit;) { @@ -231,6 +224,26 @@ public ChunkBuffer put(ByteBuffer that) { return this; } + @Override + public ChunkBuffer put(byte[] b, int offset, int length) { + checkArgument(b, offset, length); + + final int end = offset + length; + for (int p = position(); offset < end;) { + final ByteBuffer buf = getAndAllocateAtPosition(p); + final int min = Math.min(buf.remaining(), end - offset); + buf.put(b, offset, min); + offset += min; + p += min; + } + return this; + } + + @Override + public ChunkBuffer put(byte[] b) { + return put(b, 0, b.length); + } + @Override public ChunkBuffer duplicate(int newPosition, int newLimit) { Preconditions.checkArgument(newPosition >= 0); diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/ChunkBufferPutBenchmark.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/ChunkBufferPutBenchmark.java new file mode 100644 index 000000000000..3518691a2161 --- /dev/null +++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/ChunkBufferPutBenchmark.java @@ -0,0 +1,407 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.ozone.common; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.text.DecimalFormat; +import java.util.concurrent.ThreadLocalRandom; +import org.apache.hadoop.ozone.common.JfrByteBufferAllocations.AllocationStats; + +/** + * Microbenchmark for ChunkBuffer.put(byte[]) direct copy vs ByteBuffer.wrap path. + * + *

Focused on the scenarios where HDDS-15485 shows the clearest benefit: + *

+ * + *

Run from the repo root: + * 

+ *   mvn -pl hadoop-hdds/common -q test-compile exec:java \
+ *     -Dexec.mainClass=org.apache.hadoop.ozone.common.ChunkBufferPutBenchmark \
+ *     -Dexec.classpathScope=test \
+ *     -Dexec.args="--add-opens jdk.jfr/jdk.jfr=ALL-UNNAMED --add-opens jdk.jfr/jdk.jfr.consumer=ALL-UNNAMED"
+ *
+ * JFR ByteBuffer counts are sampled; put-op count reports exact wrap calls. + */ +public final class ChunkBufferPutBenchmark { + + private static final int WARMUP_SECONDS = 10; + private static final int BENCHMARK_SECONDS = 20; + private static final int ALLOCATION_BENCHMARK_SECONDS = 5; + private static final int THROUGHPUT_ROUNDS = 3; + private static final DecimalFormat MBPS = new DecimalFormat("#,##0.0"); + private static final DecimalFormat NS = new DecimalFormat("#,##0"); + private static final DecimalFormat PCT = new DecimalFormat("#,##0.0"); + private static final DecimalFormat RATIO = new DecimalFormat("0.00"); + private static final DecimalFormat COUNT = new DecimalFormat("#,##0"); + + /** ozone.client.stream.buffer.size default. */ + private static final int DEFAULT_CHUNK_SIZE = 4 * 1024 * 1024; + + /** Showcase: ozone.client.stream.buffer.increment for incremental buffer. */ + private static final int INCREMENTAL_BUFFER_INCREMENT = 64 * 1024; + + /** Hadoop io.file.buffer.size / FSDataOutputStream default. */ + private static final int HADOOP_FS_BUFFER_SIZE = 4 * 1024; + + private static final ThreadLocal SOURCE = ThreadLocal.withInitial(() -> { + byte[] bytes = new byte[DEFAULT_CHUNK_SIZE]; + ThreadLocalRandom.current().nextBytes(bytes); + return bytes; + }); + + private ChunkBufferPutBenchmark() { + } + + public static void main(String[] args) throws IOException { + System.out.println("ChunkBuffer.put(byte[]) microbenchmark (pre-allocated buffer, put-only)"); + System.out.println("JVM: " + System.getProperty("java.version") + + " on " + System.getProperty("os.arch")); + System.out.println(); + + final Scenario showcase = new Scenario( + "Incremental buffer showcase", + DEFAULT_CHUNK_SIZE, + INCREMENTAL_BUFFER_INCREMENT, + HADOOP_FS_BUFFER_SIZE); + + // Correctness self-check (verify=true) before timing; benchmark loops run + // with verify=false so the measurement is not polluted by read-back. + verifyCorrectness(showcase); + + System.out.println("=== Throughput showcase ==="); + runThroughputComparison(showcase); + System.out.println(); + + System.out.println("=== Allocation showcase ==="); + runAllocationComparison(showcase); + } + + private static void runThroughputComparison(Scenario scenario) { + printScenarioHeader(scenario); + warmupBothPaths(scenario); + final double[] improvements = new double[THROUGHPUT_ROUNDS]; + for (int round = 0; round < THROUGHPUT_ROUNDS; round++) { + final boolean directFirst = round % 2 == 0; + final Result direct; + final Result wrap; + if (directFirst) { + direct = benchmarkThroughput(scenario, ChunkBufferPutBenchmark::loopStreamFillDirect); + wrap = benchmarkThroughput(scenario, ChunkBufferPutBenchmark::loopStreamFillWrap); + } else { + wrap = benchmarkThroughput(scenario, ChunkBufferPutBenchmark::loopStreamFillWrap); + direct = benchmarkThroughput(scenario, ChunkBufferPutBenchmark::loopStreamFillDirect); + } + improvements[round] = wrap.nsPerOp / direct.nsPerOp - 1.0; + System.out.printf(" round %d:%n", round + 1); + printThroughputComparison(scenario.writeSize, direct, wrap, " "); + } + final double medianImprovement = median(improvements) * 100.0; + System.out.printf(" median improvement over %d rounds: %s%%%n", + THROUGHPUT_ROUNDS, PCT.format(medianImprovement)); + } + + private static void warmupBothPaths(Scenario scenario) { + final byte[] source = SOURCE.get(); + final int writesPerChunk = scenario.chunkSize / scenario.writeSize; + try (ChunkBuffer buffer = ChunkBuffer.allocate(scenario.chunkSize, + scenario.bufferIncrement)) { + for (int i = 0; i < 2; i++) { + loopStreamFillDirect(buffer, source, scenario.writeSize, writesPerChunk, + WARMUP_SECONDS, false); + loopStreamFillWrap(buffer, source, scenario.writeSize, writesPerChunk, + WARMUP_SECONDS, false); + } + } + } + + private static double median(double[] values) { + final double[] sorted = values.clone(); + java.util.Arrays.sort(sorted); + return sorted[sorted.length / 2]; + } + + private static void runAllocationComparison(Scenario scenario) throws IOException { + printScenarioHeader(scenario); + if (!JfrByteBufferAllocations.isAvailable()) { + System.out.println(" JFR unavailable; reporting wrap-call count from put ops only."); + } + final AllocationResult direct = benchmarkAllocations(scenario, + ChunkBufferPutBenchmark::loopStreamFillDirect); + final AllocationResult wrap = benchmarkAllocations(scenario, + ChunkBufferPutBenchmark::loopStreamFillWrap); + printAllocationComparison(direct, wrap); + } + + private static void printScenarioHeader(Scenario scenario) { + System.out.println("--- " + scenario.name + " ---"); + System.out.println("Config: " + scenario.config()); + System.out.println("Pattern: " + scenario.pattern()); + System.out.printf("Chunk=%dKB increment=%dKB write=%dKB%n", + scenario.chunkSize / 1024, + scenario.bufferIncrement / 1024, + scenario.writeSize / 1024); + } + + private static void printThroughputComparison(int writeSize, Result direct, Result wrap, + String prefix) { + final double speedup = wrap.nsPerOp / direct.nsPerOp; + final double pctFaster = (speedup - 1.0) * 100.0; + final double throughputSpeedup = direct.mbPerSec / wrap.mbPerSec; + final double throughputPct = (throughputSpeedup - 1.0) * 100.0; + System.out.printf("%sdirect put(byte[]): %s MB/s | %s ns/op | %.2fs | %s ops%n", + prefix, MBPS.format(direct.mbPerSec), NS.format(direct.nsPerOp), + direct.elapsedSeconds, COUNT.format(direct.ops)); + System.out.printf("%swrap put(ByteBuffer): %s MB/s | %s ns/op | %.2fs | %s ops%n", + prefix, MBPS.format(wrap.mbPerSec), NS.format(wrap.nsPerOp), + wrap.elapsedSeconds, COUNT.format(wrap.ops)); + System.out.printf("%simprovement: %s%% faster (%sx) per %dKB write; " + + "throughput %s%% (%sx)%n", + prefix, PCT.format(pctFaster), RATIO.format(speedup), writeSize / 1024, + PCT.format(throughputPct), RATIO.format(throughputSpeedup)); + } + + private static void printAllocationComparison(AllocationResult direct, + AllocationResult wrap) { + System.out.printf(" direct put(byte[]): %s put ops | %s ByteBuffer TLAB allocs | %s alloc bytes%n", + COUNT.format(direct.putOps), COUNT.format(direct.byteBufferAllocCount), + COUNT.format(direct.byteBufferAllocBytes)); + System.out.printf(" wrap put(ByteBuffer): %s put ops | %s ByteBuffer TLAB allocs | %s alloc bytes%n", + COUNT.format(wrap.putOps), COUNT.format(wrap.byteBufferAllocCount), + COUNT.format(wrap.byteBufferAllocBytes)); + System.out.printf(" ByteBuffer.wrap calls on wrap path (1 per put): %s%n", + COUNT.format(wrap.putOps)); + System.out.printf(" direct path avoids %s ByteBuffer.wrap calls per run%n", + COUNT.format(wrap.putOps)); + if (wrap.byteBufferAllocCount > 0 && direct.byteBufferAllocCount == 0) { + System.out.printf(" JFR confirms zero ByteBuffer TLAB allocations on direct path%n"); + } + if (wrap.byteBufferAllocCount > direct.byteBufferAllocCount) { + final long saved = wrap.byteBufferAllocCount - direct.byteBufferAllocCount; + System.out.printf(" JFR sampled ByteBuffer TLAB allocations avoided on direct path: %s%n", + COUNT.format(saved)); + System.out.println(" (JFR samples TLAB events; put-op count is the exact wrap-call metric)"); + } + } + + private static Result benchmarkThroughput(Scenario scenario, TimedLoop loop) { + final byte[] source = SOURCE.get(); + final int writesPerChunk = scenario.chunkSize / scenario.writeSize; + try (ChunkBuffer buffer = ChunkBuffer.allocate(scenario.chunkSize, + scenario.bufferIncrement)) { + loop.run(buffer, source, scenario.writeSize, writesPerChunk, WARMUP_SECONDS, false); + final LoopResult benchmark = loop.run(buffer, source, scenario.writeSize, + writesPerChunk, BENCHMARK_SECONDS, false); + return toResult(benchmark, scenario.writeSize); + } + } + + private static AllocationResult benchmarkAllocations(Scenario scenario, TimedLoop loop) + throws IOException { + final byte[] source = SOURCE.get(); + final int writesPerChunk = scenario.chunkSize / scenario.writeSize; + try (ChunkBuffer buffer = ChunkBuffer.allocate(scenario.chunkSize, + scenario.bufferIncrement)) { + loop.run(buffer, source, scenario.writeSize, writesPerChunk, WARMUP_SECONDS, false); + final LoopResult[] benchmark = new LoopResult[1]; + final AllocationStats stats = JfrByteBufferAllocations.measure( + () -> benchmark[0] = loop.run(buffer, source, scenario.writeSize, + writesPerChunk, ALLOCATION_BENCHMARK_SECONDS, false)); + final long putOps = benchmark[0].totalBytes / scenario.writeSize; + return new AllocationResult(putOps, stats.getByteBufferAllocCount(), + stats.getByteBufferAllocBytes()); + } + } + + private static Result toResult(LoopResult benchmark, int writeSize) { + final long elapsedNanos = benchmark.elapsedNanos; + final long benchmarkBytes = benchmark.totalBytes; + final double seconds = elapsedNanos / 1_000_000_000.0; + final double mbPerSec = benchmarkBytes / seconds / (1024.0 * 1024.0); + final long ops = benchmarkBytes / writeSize; + final double nsPerOp = (double) elapsedNanos / ops; + return new Result(mbPerSec, nsPerOp, seconds, ops); + } + + @FunctionalInterface + private interface TimedLoop { + LoopResult run(ChunkBuffer buffer, byte[] source, int writeSize, + int writesPerChunk, int seconds, boolean verify); + } + + private static LoopResult loopStreamFillDirect(ChunkBuffer buffer, byte[] source, + int writeSize, int writesPerChunk, int seconds, boolean verify) { + long totalBytes = 0; + final long start = System.nanoTime(); + final long deadline = start + seconds * 1_000_000_000L; + while (System.nanoTime() < deadline) { + buffer.clear(); + int off = 0; + for (int i = 0; i < writesPerChunk; i++) { + buffer.put(source, off, writeSize); + off += writeSize; + totalBytes += writeSize; + } + if (verify) { + verifyBuffer(buffer, source, off); + } + } + return new LoopResult(totalBytes, System.nanoTime() - start); + } + + private static LoopResult loopStreamFillWrap(ChunkBuffer buffer, byte[] source, + int writeSize, int writesPerChunk, int seconds, boolean verify) { + long totalBytes = 0; + final long start = System.nanoTime(); + final long deadline = start + seconds * 1_000_000_000L; + while (System.nanoTime() < deadline) { + buffer.clear(); + int off = 0; + for (int i = 0; i < writesPerChunk; i++) { + buffer.put(ByteBuffer.wrap(source, off, writeSize)); + off += writeSize; + totalBytes += writeSize; + } + if (verify) { + verifyBuffer(buffer, source, off); + } + } + return new LoopResult(totalBytes, System.nanoTime() - start); + } + + /** + * Reads back the first {@code bytesWritten} bytes and asserts they equal + * {@code source}. Used with verify=true to guarantee both put paths are + * byte-for-byte correct before (and independently of) timing them. + */ + private static void verifyBuffer(ChunkBuffer buffer, byte[] source, int bytesWritten) { + final ChunkBuffer slice = buffer.duplicate(0, bytesWritten); + int pos = 0; + for (final ByteBuffer backing : slice.asByteBufferList()) { + final ByteBuffer b = backing.duplicate(); + while (b.hasRemaining()) { + final byte actual = b.get(); + if (actual != source[pos]) { + throw new IllegalStateException("Verification failed at byte " + pos + + ": expected " + source[pos] + " but got " + actual); + } + pos++; + } + } + if (pos != bytesWritten) { + throw new IllegalStateException("Verification read " + pos + + " bytes, expected " + bytesWritten); + } + } + + private static void verifyCorrectness(Scenario scenario) { + final byte[] source = SOURCE.get(); + final int writesPerChunk = scenario.chunkSize / scenario.writeSize; + try (ChunkBuffer buffer = ChunkBuffer.allocate(scenario.chunkSize, + scenario.bufferIncrement)) { + loopStreamFillDirect(buffer, source, scenario.writeSize, writesPerChunk, 1, true); + loopStreamFillWrap(buffer, source, scenario.writeSize, writesPerChunk, 1, true); + } + System.out.println("Correctness self-check passed (verify=true): " + + "direct put(byte[]) and wrap put(ByteBuffer) produce identical bytes."); + System.out.println(); + } + + private static final class LoopResult { + private final long totalBytes; + private final long elapsedNanos; + + private LoopResult(long totalBytes, long elapsedNanos) { + this.totalBytes = totalBytes; + this.elapsedNanos = elapsedNanos; + } + } + + private static final class AllocationResult { + private final long putOps; + private final long byteBufferAllocCount; + private final long byteBufferAllocBytes; + + private AllocationResult(long putOps, long byteBufferAllocCount, + long byteBufferAllocBytes) { + this.putOps = putOps; + this.byteBufferAllocCount = byteBufferAllocCount; + this.byteBufferAllocBytes = byteBufferAllocBytes; + } + } + + private static final class Scenario { + private final String name; + private final int chunkSize; + private final int bufferIncrement; + private final int writeSize; + + private Scenario(String name, int chunkSize, int bufferIncrement, int writeSize) { + this.name = name; + this.chunkSize = chunkSize; + this.bufferIncrement = bufferIncrement; + this.writeSize = writeSize; + } + + /** {@code ChunkBuffer.allocate} picks the impl from capacity vs increment. */ + private String bufferKind() { + return bufferIncrement > 0 && bufferIncrement < chunkSize + ? "IncrementalChunkBuffer" : "ChunkBufferImplWithByteBuffer"; + } + + /** Config line derived from the numeric parameters (not hard coded). */ + private String config() { + return "ozone.client.stream.buffer.size=" + human(chunkSize) + + ", ozone.client.stream.buffer.increment=" + human(bufferIncrement) + + ", io.file.buffer.size=" + human(writeSize); + } + + /** Access pattern line derived from the numeric parameters (not hard coded). */ + private String pattern() { + return human(writeSize) + " stream fill into " + bufferKind() + + " (" + human(bufferIncrement) + " steps)"; + } + } + + private static String human(int bytes) { + if (bytes != 0 && bytes % (1024 * 1024) == 0) { + return bytes / (1024 * 1024) + "MB"; + } + if (bytes != 0 && bytes % 1024 == 0) { + return bytes / 1024 + "KB"; + } + return bytes + "B"; + } + + private static final class Result { + private final double mbPerSec; + private final double nsPerOp; + private final double elapsedSeconds; + private final long ops; + + private Result(double mbPerSec, double nsPerOp, double elapsedSeconds, long ops) { + this.mbPerSec = mbPerSec; + this.nsPerOp = nsPerOp; + this.elapsedSeconds = elapsedSeconds; + this.ops = ops; + } + } +} diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/JfrByteBufferAllocations.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/JfrByteBufferAllocations.java new file mode 100644 index 000000000000..9ba424f115ed --- /dev/null +++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/JfrByteBufferAllocations.java @@ -0,0 +1,133 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.ozone.common; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; + +/** + * Reads jdk.ObjectAllocationInNewTLAB ByteBuffer allocations via JFR reflection + * so test compilation does not require jdk.jfr on the compile classpath. + */ +final class JfrByteBufferAllocations { + + private static final boolean JFR_AVAILABLE = probeJfr(); + + private JfrByteBufferAllocations() { + } + + static boolean isAvailable() { + return JFR_AVAILABLE; + } + + static AllocationStats measure(Runnable workload) throws IOException { + if (!JFR_AVAILABLE) { + workload.run(); + return new AllocationStats(0, 0); + } + try { + final Class recordingClass = Class.forName("jdk.jfr.Recording"); + final Object recording = recordingClass.getConstructor().newInstance(); + final Object eventSettings = recordingClass.getMethod("enable", String.class) + .invoke(recording, "jdk.ObjectAllocationInNewTLAB"); + eventSettings.getClass().getMethod("withoutStackTrace").invoke(eventSettings); + recordingClass.getMethod("start").invoke(recording); + try { + workload.run(); + } finally { + recordingClass.getMethod("stop").invoke(recording); + } + return readByteBufferAllocations(recording); + } catch (ReflectiveOperationException e) { + throw new IOException("Failed to run JFR allocation measurement", e); + } + } + + private static AllocationStats readByteBufferAllocations(Object recording) + throws IOException, ReflectiveOperationException { + final Path dump = Files.createTempFile("chunkbuffer-put-bench", ".jfr"); + try { + recording.getClass().getMethod("dump", Path.class).invoke(recording, dump); + final Class recordingFileClass = Class.forName("jdk.jfr.consumer.RecordingFile"); + final Object recordingFile = recordingFileClass.getConstructor(Path.class).newInstance(dump); + final java.lang.reflect.Method hasMoreEvents = recordingFileClass.getMethod("hasMoreEvents"); + final java.lang.reflect.Method readEvent = recordingFileClass.getMethod("readEvent"); + final java.lang.reflect.Method close = recordingFileClass.getMethod("close"); + + long byteBufferAllocCount = 0; + long byteBufferAllocBytes = 0; + try { + while ((Boolean) hasMoreEvents.invoke(recordingFile)) { + final Object event = readEvent.invoke(recordingFile); + final Class eventClass = event.getClass(); + final Object eventType = eventClass.getMethod("getEventType").invoke(event); + final String eventName = (String) eventType.getClass().getMethod("getName") + .invoke(eventType); + if (!"jdk.ObjectAllocationInNewTLAB".equals(eventName)) { + continue; + } + final Object objectClass = eventClass.getMethod("getValue", String.class) + .invoke(event, "objectClass"); + if (objectClass == null) { + continue; + } + final String className = (String) objectClass.getClass().getMethod("getName") + .invoke(objectClass); + if (className.contains("ByteBuffer")) { + byteBufferAllocCount++; + byteBufferAllocBytes += (Long) eventClass.getMethod("getLong", String.class) + .invoke(event, "allocationSize"); + } + } + } finally { + close.invoke(recordingFile); + } + return new AllocationStats(byteBufferAllocCount, byteBufferAllocBytes); + } finally { + Files.deleteIfExists(dump); + } + } + + private static boolean probeJfr() { + try { + final Class frClass = Class.forName("jdk.jfr.FlightRecorder"); + return (Boolean) frClass.getMethod("isAvailable").invoke(null); + } catch (ReflectiveOperationException e) { + return false; + } + } + + static final class AllocationStats { + private final long byteBufferAllocCount; + private final long byteBufferAllocBytes; + + private AllocationStats(long byteBufferAllocCount, long byteBufferAllocBytes) { + this.byteBufferAllocCount = byteBufferAllocCount; + this.byteBufferAllocBytes = byteBufferAllocBytes; + } + + long getByteBufferAllocCount() { + return byteBufferAllocCount; + } + + long getByteBufferAllocBytes() { + return byteBufferAllocBytes; + } + } +} diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChunkBuffer.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChunkBuffer.java index d8520bebb8ad..3c4ba482268c 100644 --- a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChunkBuffer.java +++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChunkBuffer.java @@ -20,17 +20,23 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertThrows; +import com.google.common.collect.ImmutableList; import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.nio.BufferOverflowException; import java.nio.ByteBuffer; import java.nio.channels.Channels; import java.util.ArrayList; +import java.util.Arrays; import java.util.Iterator; import java.util.List; import java.util.NoSuchElementException; import java.util.concurrent.ThreadLocalRandom; +import java.util.function.IntFunction; +import java.util.stream.Stream; import org.apache.hadoop.hdds.utils.MockGatheringChannel; import org.apache.hadoop.hdds.utils.db.CodecBuffer; import org.apache.hadoop.hdds.utils.db.CodecTestUtil; @@ -38,6 +44,9 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; /** * Test {@link ChunkBuffer} implementations. @@ -57,6 +66,184 @@ public void after() throws Exception { CodecTestUtil.gc(); } + static Stream chunkBufferTypes() { + return Stream.of( + Arguments.of("ByteBuffer", (IntFunction) ChunkBuffer::allocate), + Arguments.of("Incremental", (IntFunction) cap -> + ChunkBuffer.allocate(cap, Math.max(1, cap / 3))), + Arguments.of("ByteBufferList", (IntFunction) cap -> { + final int first = cap / 2; + return ChunkBuffer.wrap(ImmutableList.of( + ByteBuffer.allocate(first), + ByteBuffer.allocate(cap - first))); + })); + } + + static Stream putTestCases() { + final List cases = new ArrayList<>(); + chunkBufferTypes().forEach(impl -> { + final String name = (String) impl.get()[0]; + @SuppressWarnings("unchecked") + final IntFunction factory = + (IntFunction) impl.get()[1]; + for (PutMethod method : PutMethod.values()) { + cases.add(Arguments.of(name, factory, method)); + } + }); + return cases.stream(); + } + + @ParameterizedTest(name = "put rejects null [{0} via {1}]") + @MethodSource("putTestCases") + void putRejectsNull(String ignored, IntFunction factory, + PutMethod method) { + try (ChunkBuffer buffer = factory.apply(16)) { + method.assertRejectsNull(buffer); + } + } + + @ParameterizedTest(name = "put(byte[]) rejects negative offset/length [{0}]") + @MethodSource("chunkBufferTypes") + void putByteArrayRejectsNegativeOffsetOrLength(String ignored, + IntFunction factory) { + final byte[] source = new byte[8]; + try (ChunkBuffer buffer = factory.apply(16)) { + assertThrows(IllegalArgumentException.class, () -> buffer.put(source, -1, 4)); + assertThrows(IllegalArgumentException.class, () -> buffer.put(source, 0, -1)); + } + } + + @ParameterizedTest(name = "put(byte[]) rejects out-of-range slice [{0}]") + @MethodSource("chunkBufferTypes") + void putByteArrayRejectsOutOfRangeSlice(String ignored, + IntFunction factory) { + final byte[] source = new byte[8]; + try (ChunkBuffer buffer = factory.apply(16)) { + assertThrows(IllegalArgumentException.class, () -> buffer.put(source, 5, 4)); + assertThrows(IllegalArgumentException.class, () -> buffer.put(source, 0, 9)); + } + } + + @ParameterizedTest(name = "put rejects buffer overflow [{0} via {1}]") + @MethodSource("putTestCases") + void putRejectsBufferOverflow(String type, IntFunction factory, + PutMethod method) { + final byte[] source = new byte[8]; + try (ChunkBuffer buffer = factory.apply(4)) { + final BufferOverflowException overflow = assertThrows( + BufferOverflowException.class, + () -> method.putSlice(buffer, source, 0, 5)); + if (method.expectsCheckArgumentOverflowCause(type)) { + assertInstanceOf(IllegalArgumentException.class, overflow.getCause()); + } + } + } + + @ParameterizedTest(name = "put rejects overflow after partial fill [{0} via {1}]") + @MethodSource("putTestCases") + void putRejectsOverflowAfterPartialFill(String type, IntFunction factory, + PutMethod method) throws IOException { + final byte[] first = {1, 2, 3, 4, 5}; + final byte[] second = {6, 7, 8, 9}; + try (ChunkBuffer buffer = factory.apply(8)) { + method.putSlice(buffer, first, 0, 5); + assertEquals(5, buffer.position()); + final BufferOverflowException overflow = assertThrows( + BufferOverflowException.class, () -> method.putSlice(buffer, second, 0, 4)); + if (method.expectsCheckArgumentOverflowCause(type)) { + assertInstanceOf(IllegalArgumentException.class, overflow.getCause()); + } + assertArrayEquals(first, readWrittenBytes(buffer)); + } + } + + @ParameterizedTest(name = "put writes array slice [{0} via {1}]") + @MethodSource("putTestCases") + void putWritesSlice(String ignored, IntFunction factory, PutMethod method) + throws IOException { + final byte[] source = {1, 2, 3, 4, 5, 6, 7, 8}; + try (ChunkBuffer buffer = factory.apply(4)) { + method.putSlice(buffer, source, 2, 4); + assertEquals(4, buffer.position()); + assertEquals(0, buffer.remaining()); + assertArrayEquals(new byte[] {3, 4, 5, 6}, readWrittenBytes(buffer)); + } + } + + @ParameterizedTest(name = "put writes randomized slice [{0} via {2}]") + @MethodSource("putTestCases") + void putWritesRandomizedSlice(String ignored, IntFunction factory, + PutMethod method) throws IOException { + // Randomize buffer size, data size, offset and length (per review feedback: + // exercise both put paths the way testImplWithByteBuffer randomizes inputs). + for (int iter = 0; iter < 20; iter++) { + final int capacity = nextInt(1 << 10) + 1; + final int length = nextInt(capacity + 1); + final byte[] source = new byte[length + nextInt(64)]; + ThreadLocalRandom.current().nextBytes(source); + final int offset = nextInt(source.length - length + 1); + try (ChunkBuffer buffer = factory.apply(capacity)) { + method.putSlice(buffer, source, offset, length); + assertEquals(length, buffer.position()); + assertArrayEquals(Arrays.copyOfRange(source, offset, offset + length), + readWrittenBytes(buffer)); + } + } + } + + @ParameterizedTest(name = "put zero-length payload is no-op [{0} via {1}]") + @MethodSource("putTestCases") + void putZeroLengthPayloadIsNoOp(String ignored, IntFunction factory, + PutMethod method) throws IOException { + final byte[] source = {1, 2, 3, 4}; + try (ChunkBuffer buffer = factory.apply(4)) { + method.putSlice(buffer, source, 0, 2); + method.putEmptySlice(buffer, source, 2, 2); + assertEquals(2, buffer.position()); + assertArrayEquals(new byte[] {1, 2}, readWrittenBytes(buffer)); + } + } + + @ParameterizedTest(name = "put fills entire buffer [{0} via {1}]") + @MethodSource("putTestCases") + void putFillsEntireBuffer(String ignored, IntFunction factory, + PutMethod method) throws IOException { + final byte[] source = new byte[32]; + ThreadLocalRandom.current().nextBytes(source); + try (ChunkBuffer buffer = factory.apply(source.length)) { + method.putAll(buffer, source); + assertEquals(source.length, buffer.position()); + assertArrayEquals(source, readWrittenBytes(buffer)); + } + } + + @ParameterizedTest(name = "put spans incremental chunks [{0} via {1}]") + @MethodSource("putTestCases") + void putSpansIncrementalChunks(String ignored, IntFunction ignoredFactory, + PutMethod method) throws IOException { + final byte[] source = new byte[10]; + ThreadLocalRandom.current().nextBytes(source); + try (ChunkBuffer buffer = ChunkBuffer.allocate(10, 3)) { + method.putSlice(buffer, source, 1, 7); + assertEquals(7, buffer.position()); + assertArrayEquals(Arrays.copyOfRange(source, 1, 8), readWrittenBytes(buffer)); + } + } + + @ParameterizedTest(name = "put spans list chunks [{0} via {1}]") + @MethodSource("putTestCases") + void putSpansListChunks(String ignored, IntFunction ignoredFactory, + PutMethod method) throws IOException { + final byte[] source = new byte[10]; + ThreadLocalRandom.current().nextBytes(source); + try (ChunkBuffer buffer = ChunkBuffer.wrap(ImmutableList.of( + ByteBuffer.allocate(4), ByteBuffer.allocate(6)))) { + method.putSlice(buffer, source, 1, 7); + assertEquals(7, buffer.position()); + assertArrayEquals(Arrays.copyOfRange(source, 1, 8), readWrittenBytes(buffer)); + } + } + @Test void testImplWithByteBuffer() throws IOException { runTestImplWithByteBuffer(1); @@ -226,4 +413,88 @@ private static void assertWrite(byte[] expected, ChunkBuffer impl) throws IOExce assertArrayEquals(expected, output.toByteArray()); assertFalse(impl.hasRemaining()); } + + private static byte[] readWrittenBytes(ChunkBuffer buffer) throws IOException { + final int written = buffer.position(); + final ChunkBuffer slice = buffer.duplicate(0, written); + final ByteArrayOutputStream output = new ByteArrayOutputStream(written); + slice.writeTo(new MockGatheringChannel(Channels.newChannel(output))); + return output.toByteArray(); + } + + private static ByteBuffer emptySlice(byte[] source, int offset, int length) { + final ByteBuffer slice = ByteBuffer.wrap(source, offset, length); + slice.position(slice.limit()); + return slice; + } + + /** + * Exercises {@link ChunkBuffer#put(byte[], int, int)} vs {@link ChunkBuffer#put(ByteBuffer)} + * with equivalent payloads. + */ + private enum PutMethod { + BYTE_ARRAY { + @Override + void putSlice(ChunkBuffer buffer, byte[] source, int offset, int length) { + buffer.put(source, offset, length); + } + + @Override + void putAll(ChunkBuffer buffer, byte[] source) { + buffer.put(source); + } + + @Override + void putEmptySlice(ChunkBuffer buffer, byte[] source, int offset, int length) { + buffer.put(source, offset, 0); + } + + @Override + void assertRejectsNull(ChunkBuffer buffer) { + assertThrows(NullPointerException.class, () -> buffer.put((byte[]) null)); + assertThrows(NullPointerException.class, () -> buffer.put(null, 0, 0)); + } + + @Override + boolean expectsCheckArgumentOverflowCause(String implType) { + return true; + } + }, + BYTE_BUFFER { + @Override + void putSlice(ChunkBuffer buffer, byte[] source, int offset, int length) { + buffer.put(ByteBuffer.wrap(source, offset, length)); + } + + @Override + void putAll(ChunkBuffer buffer, byte[] source) { + buffer.put(ByteBuffer.wrap(source)); + } + + @Override + void putEmptySlice(ChunkBuffer buffer, byte[] source, int offset, int length) { + buffer.put(emptySlice(source, offset, length)); + } + + @Override + void assertRejectsNull(ChunkBuffer buffer) { + assertThrows(NullPointerException.class, () -> buffer.put((ByteBuffer) null)); + } + + @Override + boolean expectsCheckArgumentOverflowCause(String implType) { + return !"ByteBuffer".equals(implType); + } + }; + + abstract void putSlice(ChunkBuffer buffer, byte[] source, int offset, int length); + + abstract void putAll(ChunkBuffer buffer, byte[] source); + + abstract void putEmptySlice(ChunkBuffer buffer, byte[] source, int offset, int length); + + abstract void assertRejectsNull(ChunkBuffer buffer); + + abstract boolean expectsCheckArgumentOverflowCause(String implType); + } }