This is an automated email from the ASF dual-hosted git repository.
jojochuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 2e97c0fc07a HDDS-15485. Avoid ByteBuffer.wrap on
ChunkBuffer.put(byte[]) path (#11147)
2e97c0fc07a is described below
commit 2e97c0fc07a11f2576b2bb5334438a05e9c0273c
Author: Andrey Yarovoy <[email protected]>
AuthorDate: Fri Sep 18 19:28:50 2026 -0400
HDDS-15485. Avoid ByteBuffer.wrap on ChunkBuffer.put(byte[]) path (#11147)
Generated-by: Claude Code (Opus 4.8)
---
.../apache/hadoop/ozone/common/ChunkBuffer.java | 30 ++
.../common/ChunkBufferImplWithByteBuffer.java | 12 +
.../common/ChunkBufferImplWithByteBufferList.java | 31 +-
.../ozone/common/IncrementalChunkBuffer.java | 29 +-
.../ozone/common/ChunkBufferPutBenchmark.java | 407 +++++++++++++++++++++
.../ozone/common/JfrByteBufferAllocations.java | 133 +++++++
.../hadoop/ozone/common/TestChunkBuffer.java | 271 ++++++++++++++
7 files changed, 896 insertions(+), 17 deletions(-)
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 600a7b56797..6e06f502b2c 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<ByteBuffer> 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 d37b095621c..5bc9300e905 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 e67726e98ff..afbb4253ef7 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 1d90fbc3972..8d20b42f1ec 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 00000000000..3518691a216
--- /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.
+ *
+ * <p>Focused on the scenarios where HDDS-15485 shows the clearest benefit:
+ * <ul>
+ * <li>Throughput: 4KB stream fill with incremental buffer (64KB
increment)</li>
+ * <li>Allocations: same 4KB / 64KB-increment incremental buffer path (JFR +
wrap calls)</li>
+ * </ul>
+ *
+ * <p>Run from the repo root:
+ * <pre>
+ * 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"
+ * </pre>
+ * 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<byte[]> 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 00000000000..9ba424f115e
--- /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 d8520bebb8a..3c4ba482268 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<Arguments> chunkBufferTypes() {
+ return Stream.of(
+ Arguments.of("ByteBuffer", (IntFunction<ChunkBuffer>)
ChunkBuffer::allocate),
+ Arguments.of("Incremental", (IntFunction<ChunkBuffer>) cap ->
+ ChunkBuffer.allocate(cap, Math.max(1, cap / 3))),
+ Arguments.of("ByteBufferList", (IntFunction<ChunkBuffer>) cap -> {
+ final int first = cap / 2;
+ return ChunkBuffer.wrap(ImmutableList.of(
+ ByteBuffer.allocate(first),
+ ByteBuffer.allocate(cap - first)));
+ }));
+ }
+
+ static Stream<Arguments> putTestCases() {
+ final List<Arguments> cases = new ArrayList<>();
+ chunkBufferTypes().forEach(impl -> {
+ final String name = (String) impl.get()[0];
+ @SuppressWarnings("unchecked")
+ final IntFunction<ChunkBuffer> factory =
+ (IntFunction<ChunkBuffer>) 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<ChunkBuffer> 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<ChunkBuffer> 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<ChunkBuffer> 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<ChunkBuffer> 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<ChunkBuffer> 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<ChunkBuffer> 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<ChunkBuffer>
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<ChunkBuffer>
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<ChunkBuffer> 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<ChunkBuffer>
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<ChunkBuffer>
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);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
