From 5e8e7a5147f7cc6882642628ca3c726a951ffbb0 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Wed, 17 Jun 2026 11:23:02 -0400 Subject: [PATCH 1/5] HDDS-15592 use heap allocation in ChunkBuffer --- ...run-block-output-stream-write-benchmark.sh | 55 ++ .../storage/BenchmarkMockXceiverClient.java | 147 +++++ .../BlockOutputStreamWriteBenchmark.java | 528 ++++++++++++++++++ .../hadoop/ozone/common/ChunkBuffer.java | 20 +- .../ozone/common/IncrementalChunkBuffer.java | 9 +- 5 files changed, 757 insertions(+), 2 deletions(-) create mode 100755 hadoop-hdds/client/dev-support/run-block-output-stream-write-benchmark.sh create mode 100644 hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BenchmarkMockXceiverClient.java create mode 100644 hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStreamWriteBenchmark.java diff --git a/hadoop-hdds/client/dev-support/run-block-output-stream-write-benchmark.sh b/hadoop-hdds/client/dev-support/run-block-output-stream-write-benchmark.sh new file mode 100755 index 000000000000..37c950733b58 --- /dev/null +++ b/hadoop-hdds/client/dev-support/run-block-output-stream-write-benchmark.sh @@ -0,0 +1,55 @@ +#!/usr/bin/env bash +# 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. + +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" +SETTINGS="${HOME}/.m2/settings.xml" +MVN=(mvn -s "${SETTINGS}" -pl hadoop-hdds/client -q clean test-compile exec:java + -Dexec.mainClass=org.apache.hadoop.hdds.scm.storage.BlockOutputStreamWriteBenchmark + -Dexec.classpathScope=test) + +LABEL="${1:-workspace}" +shift || true + +# Quick usage reminder +cat <<'EOF' +Useful system properties (pass as -Dproperty=value after the label): + benchmark.scaling=true run scaling study (recommended) + benchmark.scaling.threads=1,2,7,14,28,42,56 override thread counts + benchmark.heapBuffer=true|false run only one allocation mode + benchmark.writeSize=4194304 single write size in bytes + benchmark.checksumType=NONE skip checksum overhead +EOF +echo + +echo "Running BlockOutputStreamWriteBenchmark (label=${LABEL}) ..." +cd "${ROOT}" + +# exec:java resolves dependency JARs from the local Maven repository, not from +# target/classes. Install interface-client (gRPC stubs) and common +# (ChunkBuffer / ALLOCATE_DIRECT) so the ratis-shaded JARs in ~/.m2 are current. +mvn -s "${SETTINGS}" -pl hadoop-hdds/interface-client,hadoop-hdds/common \ + -q install -DskipTests -DskipShade -DskipRecon -DskipDocs + +# exec:java runs the benchmark in Maven's own JVM, so MAVEN_OPTS applies here. +# The mock client serialises each request via request.toByteArray() without +# any Netty transport, so direct memory use is limited to the CodecBuffer +# working set (threads * poolCapacity * streamBufferSize). +export MAVEN_OPTS="${MAVEN_OPTS:-} -Xmx4g -XX:MaxDirectMemorySize=512m" +exec "${MVN[@]}" -Dbenchmark.label="${LABEL}" "$@" diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BenchmarkMockXceiverClient.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BenchmarkMockXceiverClient.java new file mode 100644 index 000000000000..aad6df1f318a --- /dev/null +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BenchmarkMockXceiverClient.java @@ -0,0 +1,147 @@ +/* + * 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.hdds.scm.storage; + +import java.io.IOException; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.GetCommittedBlockLengthResponseProto; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.PutBlockResponseProto; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType; +import org.apache.hadoop.hdds.scm.XceiverClientReply; +import org.apache.hadoop.hdds.scm.XceiverClientSpi; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.ratis.thirdparty.io.netty.buffer.ByteBuf; +import org.apache.ratis.thirdparty.io.netty.buffer.ByteBufOutputStream; +import org.apache.ratis.thirdparty.io.netty.buffer.PooledByteBufAllocator; + +/** + * Minimal xceiver client for {@link BlockOutputStreamWriteBenchmark}. + * + *

Replicates the production serialization path without real network I/O: + * the proto is serialized via {@code request.writeTo(OutputStream)} — which + * causes protobuf to use {@code OutputStreamEncoder}, the same encoder gRPC's + * {@code MessageFramer} uses — into a pooled direct {@code ByteBuf}, then the + * buffer is released immediately. + * + *

This exercises the full cross-domain copy chain: + *

+ */ +final class BenchmarkMockXceiverClient extends XceiverClientSpi { + + private final Pipeline pipeline; + private final AtomicLong logIndex = new AtomicLong(); + + BenchmarkMockXceiverClient(Pipeline pipeline) { + this.pipeline = pipeline; + } + + @Override + public void connect() { + } + + @Override + public void close() { + } + + @Override + public Pipeline getPipeline() { + return pipeline; + } + + @Override + public XceiverClientReply sendCommandAsync(ContainerCommandRequestProto request) { + // Replicate: gRPC MessageFramer allocates a pooled direct ByteBuf via + // NettyWritableBufferAllocator, then serializes the proto into it through + // OutputStreamEncoder → WritableBufferOutputStream → ByteBuf.writeBytes(). + // For NioByteString (direct chunk data): copyMemory(direct→heap tmp) + write(heap→direct). + // For BoundedByteString (heap chunk data): write(heap→direct) only. + // The buffer is released immediately rather than being enqueued to a socket. + final int size = request.getSerializedSize(); + final ByteBuf frame = PooledByteBufAllocator.DEFAULT.directBuffer(size, size); + try { + final ByteBufOutputStream out = new ByteBufOutputStream(frame); + try { + request.writeTo(out); + } catch (IOException e) { + throw new RuntimeException(e); + } + } finally { + frame.release(); + } + + final ContainerCommandResponseProto.Builder builder = + ContainerCommandResponseProto.newBuilder() + .setResult(Result.SUCCESS) + .setCmdType(request.getCmdType()); + if (request.getCmdType() == Type.PutBlock) { + builder.setPutBlock(PutBlockResponseProto.newBuilder() + .setCommittedBlockLength( + GetCommittedBlockLengthResponseProto.newBuilder() + .setBlockID(request.getPutBlock().getBlockData().getBlockID()) + .setBlockLength(request.getPutBlock().getBlockData().getSize()) + .build()) + .build()); + } + final XceiverClientReply reply = + new XceiverClientReply(CompletableFuture.completedFuture(builder.build())); + reply.setLogIndex(logIndex.incrementAndGet()); + return reply; + } + + @Override + public ReplicationType getPipelineType() { + return pipeline.getType(); + } + + @Override + public CompletableFuture watchForCommit(long index) { + final ContainerCommandResponseProto response = + ContainerCommandResponseProto.newBuilder() + .setCmdType(Type.WriteChunk) + .setResult(Result.SUCCESS) + .build(); + final XceiverClientReply reply = + new XceiverClientReply(CompletableFuture.completedFuture(response)); + reply.setLogIndex(index); + return CompletableFuture.completedFuture(reply); + } + + @Override + public long getReplicatedMinCommitIndex() { + return logIndex.get(); + } + + @Override + public Map sendCommandOnAllNodes( + ContainerCommandRequestProto request) { + return null; + } +} diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStreamWriteBenchmark.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStreamWriteBenchmark.java new file mode 100644 index 000000000000..f05ee1743363 --- /dev/null +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStreamWriteBenchmark.java @@ -0,0 +1,528 @@ +/* + * 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.hdds.scm.storage; + +import static java.util.concurrent.Executors.newSingleThreadExecutor; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.text.DecimalFormat; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.hdds.client.BlockID; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType; +import org.apache.hadoop.hdds.scm.ByteStringConversion; +import org.apache.hadoop.hdds.scm.ContainerClientMetrics; +import org.apache.hadoop.hdds.scm.OzoneClientConfig; +import org.apache.hadoop.hdds.scm.StreamBufferArgs; +import org.apache.hadoop.hdds.scm.XceiverClientManager; +import org.apache.hadoop.hdds.scm.pipeline.MockPipeline; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.hadoop.ozone.common.ChunkBuffer; +import org.apache.ozone.test.GenericTestUtils; +import org.slf4j.event.Level; + +/** + * Client-side {@link RatisBlockOutputStream} write benchmark using mocked + * container RPCs ({@link BenchmarkMockXceiverClient}). + * + *

Mimics the write path exercised by {@code ozone freon dsfg} (DataStream + * File Generator) — each worker thread repeatedly fills a 4 MB block via + * sequential writes and closes the stream, driving the chunk serialisation and + * gRPC framing code at full speed without real network or disk I/O. + * + *

The benchmark compares two buffer-allocation strategies: + *

    + *
  • heap — {@code ByteBuffer.allocate()} backed chunks. + * {@code UnsafeByteOperations.unsafeWrap()} returns a + * {@code BoundedByteString} with a backing array, so gRPC serialises via + * a single {@code System.arraycopy} into the Netty wire buffer.
    • + *
  • direct — {@code ByteBuffer.allocateDirect()} backed chunks + * (the behaviour before this patch). {@code UnsafeByteOperations.unsafeWrap()} + * returns a {@code NioByteString}, forcing gRPC to allocate a temp byte[] + * and copy byte-by-byte.
    • + *
+ * + *

On-demand benchmark (not part of {@code mvn test}). Run from the repo root: + * 

+ *   mvn -pl hadoop-hdds/client -q test-compile exec:java \
+ *     -Dexec.mainClass=org.apache.hadoop.hdds.scm.storage.BlockOutputStreamWriteBenchmark \
+ *     -Dexec.classpathScope=test
+ *
+ * + *

Optional system properties: + *

    + *
  • {@code benchmark.label} – profile label (default {@code workspace})
    • + *
  • {@code benchmark.writeSize} – single write size in bytes (default: run + * 1 MB, 2 MB, 3 MB and 4 MB)
    • + *
  • {@code benchmark.heapBuffer} – {@code true} or {@code false} + * (default: run both)
    • + *
  • {@code benchmark.checksumType} – {@code NONE} or {@code CRC32} + * (default {@code CRC32})
    • + *
  • {@code benchmark.threads} – worker count (default {@code CPUs * 2})
    • + *
  • {@code benchmark.scaling=true} – run multiple thread counts for both + * allocation modes and print a scaling summary
    • + *
  • {@code benchmark.scaling.threads} – comma-separated thread counts for + * scaling (default {@code 1,2,7,14,28,42,56})
    • + *
  • {@code benchmark.scaling.writeSizes} – comma-separated write sizes in + * bytes (overrides default 1/2/3/4 MB list when set)
    • + *
+ * + *

Or use + * {@code hadoop-hdds/client/dev-support/run-block-output-stream-write-benchmark.sh}. + */ +public final class BlockOutputStreamWriteBenchmark { + + private static final int WARMUP_SECONDS = 10; + private static final int BENCHMARK_SECONDS = 20; + private static final int STREAM_BUFFER_SIZE = 4 * 1024 * 1024; + private static final int SOURCE_BUFFER_SIZE = STREAM_BUFFER_SIZE; + private static final int POOL_CAPACITY = 8; + + private static final int[] DEFAULT_WRITE_SIZES = { + 1024 * 1024, + 2 * 1024 * 1024, + 3 * 1024 * 1024, + 4 * 1024 * 1024 + }; + + private static final DecimalFormat MBPS = new DecimalFormat("#,##0.0"); + private static final DecimalFormat NS = new DecimalFormat("#,##0"); + private static final DecimalFormat COUNT = new DecimalFormat("#,##0"); + + private BlockOutputStreamWriteBenchmark() { + } + + public static void main(String[] args) throws Exception { + GenericTestUtils.setLogLevel(BufferPool.class, Level.INFO); + GenericTestUtils.setLogLevel(BlockOutputStream.class, Level.INFO); + + final String label = System.getProperty("benchmark.label", "workspace"); + final ChecksumType checksumType = ChecksumType.valueOf( + System.getProperty("benchmark.checksumType", ChecksumType.CRC32.name())); + final int cpus = Runtime.getRuntime().availableProcessors(); + + System.out.println("BlockOutputStream client write benchmark (mocked container RPCs)"); + System.out.println("Profile: " + label); + System.out.println("Comparing: heap ByteBuffer (this patch) vs direct ByteBuffer (pre-patch)"); + System.out.println("JVM: " + System.getProperty("java.version") + + " on " + System.getProperty("os.arch")); + System.out.printf("CPUs=%d sourceBuffer=%dMB streamBuffer=%dMB poolCapacity=%d checksum=%s%n", + cpus, SOURCE_BUFFER_SIZE / (1024 * 1024), + STREAM_BUFFER_SIZE / (1024 * 1024), POOL_CAPACITY, checksumType); + System.out.println(); + + final byte[] sourceBuffer = new byte[SOURCE_BUFFER_SIZE]; + ThreadLocalRandom.current().nextBytes(sourceBuffer); + + if (Boolean.parseBoolean(System.getProperty("benchmark.scaling", "false"))) { + runScalingStudy(sourceBuffer, checksumType, cpus); + } else { + final int threadCount = resolveThreadCount(cpus); + System.out.printf("threads=%d%n%n", threadCount); + final Pipeline pipeline = MockPipeline.createRatisPipeline(); + final BenchmarkMockXceiverClient client = new BenchmarkMockXceiverClient(pipeline); + try { + for (int writeSize : resolveWriteSizes()) { + for (boolean heapBuffer : resolveHeapBufferModes()) { + runProfile(sourceBuffer, writeSize, heapBuffer, checksumType, threadCount, client); + System.out.println(); + } + } + } finally { + client.close(); + } + } + } + + private static void runScalingStudy(byte[] sourceBuffer, ChecksumType checksumType, + int cpus) throws Exception { + final int[] threadCounts = resolveScalingThreadCounts(cpus); + System.out.printf("scaling thread counts: %s%n%n", formatThreadCounts(threadCounts)); + final List rows = new ArrayList<>(); + + // One shared gRPC client for the entire scaling study. Reusing the same + // event loop thread (and therefore the same PooledByteBufAllocator arena) + // across all phases lets the arena recycle pool chunks between phases. + // Creating a new client per phase assigns each new event loop to a + // different arena (round-robin), and 0%-usage chunks in old arenas cannot + // be reused by threads on new arenas, so direct memory grows unboundedly. + final Pipeline pipeline = MockPipeline.createRatisPipeline(); + final BenchmarkMockXceiverClient sharedClient = new BenchmarkMockXceiverClient(pipeline); + try { + for (int writeSize : resolveWriteSizes()) { + System.out.printf("===== writeSize=%dKB scaling study =====%n%n", writeSize / 1024); + for (int threadCount : threadCounts) { + for (boolean heapBuffer : new boolean[] {false, true}) { + final Result result = runProfile(sourceBuffer, writeSize, heapBuffer, + checksumType, threadCount, sharedClient); + rows.add(new ScalingRow(writeSize, threadCount, heapBuffer, result.mbPerSec)); + System.out.println(); + } + } + printScalingSummary(writeSize, threadCounts, rows); + System.out.println(); + } + } finally { + sharedClient.close(); + } + } + + private static void printScalingSummary(int writeSize, int[] threadCounts, + List allRows) { + final List rows = new ArrayList<>(); + for (ScalingRow row : allRows) { + if (row.writeSize == writeSize) { + rows.add(row); + } + } + + final double directBaseline = findMbPerSec(rows, 1, false); + final double heapBaseline = findMbPerSec(rows, 1, true); + + System.out.printf("--- scaling summary writeSize=%dKB ---%n", writeSize / 1024); + System.out.printf("%8s | %12s | %12s | %10s | %10s | %10s%n", + "threads", "direct MB/s", "heap MB/s", "heap/direct", "direct eff.", "heap eff."); + System.out.println("---------|--------------|--------------|------------|------------|------------"); + + for (int threadCount : threadCounts) { + final double direct = findMbPerSec(rows, threadCount, false); + final double heap = findMbPerSec(rows, threadCount, true); + final double heapVsDirect = direct == 0 ? 0 : heap / direct; + final double directEff = directBaseline == 0 ? 0 : direct / directBaseline / threadCount; + final double heapEff = heapBaseline == 0 ? 0 : heap / heapBaseline / threadCount; + System.out.printf("%8d | %12s | %12s | %10.2fx | %10.2f | %10.2f%n", + threadCount, MBPS.format(direct), MBPS.format(heap), heapVsDirect, + directEff, heapEff); + } + System.out.println(); + System.out.println("eff. = throughput vs 1-thread baseline / thread count (1.0 = linear scaling)"); + } + + private static double findMbPerSec(List rows, int threadCount, + boolean heapBuffer) { + for (ScalingRow row : rows) { + if (row.threadCount == threadCount && row.heapBuffer == heapBuffer) { + return row.mbPerSec; + } + } + return 0; + } + + private static final class ScalingRow { + private final int writeSize; + private final int threadCount; + private final boolean heapBuffer; + private final double mbPerSec; + + private ScalingRow(int writeSize, int threadCount, boolean heapBuffer, double mbPerSec) { + this.writeSize = writeSize; + this.threadCount = threadCount; + this.heapBuffer = heapBuffer; + this.mbPerSec = mbPerSec; + } + } + + private static int resolveThreadCount(int cpus) { + final String property = System.getProperty("benchmark.threads"); + if (property != null && !property.isEmpty()) { + return Integer.parseInt(property); + } + return cpus * 2; + } + + private static int[] resolveScalingThreadCounts(int cpus) { + final String property = System.getProperty("benchmark.scaling.threads"); + if (property == null || property.isEmpty()) { + return new int[] {1, 2, 7, 14, 28, 42, 56}; + } + return parseCommaSeparatedInts(property); + } + + private static String formatThreadCounts(int[] threadCounts) { + final StringBuilder builder = new StringBuilder(); + for (int i = 0; i < threadCounts.length; i++) { + if (i > 0) { + builder.append(", "); + } + builder.append(threadCounts[i]); + } + return builder.toString(); + } + + private static Result runProfile(byte[] sourceBuffer, int writeSize, + boolean heapBuffer, ChecksumType checksumType, int threadCount, + BenchmarkMockXceiverClient client) + throws Exception { + System.out.printf("--- writeSize=%dKB heapBuffer=%s threads=%d writes/stream=%d ---%n", + writeSize / 1024, heapBuffer, threadCount, SOURCE_BUFFER_SIZE / writeSize); + + final Result result = runProfileMeasured(sourceBuffer, writeSize, heapBuffer, + checksumType, threadCount, client); + + System.out.printf("elapsed: %.2fs%n", result.elapsedSeconds); + System.out.printf("bytes written: %s%n", COUNT.format(result.bytesWritten)); + System.out.printf("write ops (%dKB): %s%n", + writeSize / 1024, COUNT.format(result.writeOps)); + System.out.printf("blocks written: %s%n", COUNT.format(result.blocksWritten)); + System.out.printf("aggregate throughput: %s MB/s%n", MBPS.format(result.mbPerSec)); + System.out.printf("ns/write (aggregate): %s%n", NS.format(result.nsPerWrite)); + System.out.printf("writeChunks metric: %s%n", + COUNT.format(result.writeChunksDuringWrite)); + System.out.printf("flushes metric: %s%n", + COUNT.format(result.flushesDuringWrite)); + return result; + } + + private static Result runProfileMeasured(byte[] sourceBuffer, int writeSize, + boolean heapBuffer, ChecksumType checksumType, int threadCount, + BenchmarkMockXceiverClient client) + throws Exception { + runTimedWrite(sourceBuffer, writeSize, heapBuffer, checksumType, + threadCount, WARMUP_SECONDS, "warmup", client); + return runTimedWrite(sourceBuffer, writeSize, heapBuffer, checksumType, + threadCount, BENCHMARK_SECONDS, "benchmark", client); + } + + private static int[] resolveWriteSizes() { + final String scalingSizes = System.getProperty("benchmark.scaling.writeSizes"); + if (scalingSizes != null && !scalingSizes.isEmpty()) { + return parseCommaSeparatedInts(scalingSizes); + } + final String property = System.getProperty("benchmark.writeSize"); + if (property == null || property.isEmpty()) { + return DEFAULT_WRITE_SIZES; + } + return new int[] {Integer.parseInt(property)}; + } + + private static int[] parseCommaSeparatedInts(String property) { + final String[] parts = property.split(","); + final int[] values = new int[parts.length]; + for (int i = 0; i < parts.length; i++) { + values[i] = Integer.parseInt(parts[i].trim()); + } + return values; + } + + private static boolean[] resolveHeapBufferModes() { + final String property = System.getProperty("benchmark.heapBuffer"); + if (property == null || property.isEmpty()) { + return new boolean[] {false, true}; + } + return new boolean[] {Boolean.parseBoolean(property)}; + } + + private static Result runTimedWrite(byte[] sourceBuffer, int writeSize, + boolean heapBuffer, ChecksumType checksumType, int threadCount, + int seconds, String phase, BenchmarkMockXceiverClient client) throws Exception { + final long writeChunksBefore = + ContainerClientMetrics.acquire().getWriteChunksDuringWrite().value(); + final long flushesBefore = + ContainerClientMetrics.acquire().getFlushesDuringWrite().value(); + + final long start = System.nanoTime(); + final long deadline = start + seconds * 1_000_000_000L; + + final ExecutorService workers = Executors.newFixedThreadPool(threadCount); + try { + final List> futures = new ArrayList<>(threadCount); + for (int i = 0; i < threadCount; i++) { + futures.add(workers.submit(() -> + runWorker(sourceBuffer, writeSize, heapBuffer, checksumType, deadline, client))); + } + + long bytesWritten = 0; + long writeOps = 0; + long blocksWritten = 0; + for (Future future : futures) { + final WorkerResult workerResult = future.get(); + bytesWritten += workerResult.bytesWritten; + writeOps += workerResult.writeOps; + blocksWritten += workerResult.blocksWritten; + } + + final long elapsedNanos = System.nanoTime() - start; + final double elapsedSeconds = elapsedNanos / 1_000_000_000.0; + System.out.printf("%s complete (%.2fs)%n", phase, elapsedSeconds); + + return new Result( + bytesWritten, + writeOps, + blocksWritten, + elapsedSeconds, + bytesWritten / elapsedSeconds / (1024.0 * 1024.0), + writeOps == 0 ? 0 : (double) elapsedNanos / writeOps, + ContainerClientMetrics.acquire().getWriteChunksDuringWrite().value() + - writeChunksBefore, + ContainerClientMetrics.acquire().getFlushesDuringWrite().value() + - flushesBefore); + } finally { + workers.shutdownNow(); + try { + workers.awaitTermination(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + + private static WorkerResult runWorker(byte[] sourceBuffer, int writeSize, + boolean heapBuffer, ChecksumType checksumType, long deadline, + BenchmarkMockXceiverClient mockClient) + throws Exception { + long bytesWritten = 0; + long writeOps = 0; + long blocksWritten = 0; + + try (BenchmarkSession session = BenchmarkSession.open(heapBuffer, checksumType, mockClient)) { + while (System.nanoTime() < deadline) { + try (BlockOutputStream stream = session.newStream()) { + for (int offset = 0; offset + writeSize <= sourceBuffer.length + && System.nanoTime() < deadline; offset += writeSize) { + stream.write(sourceBuffer, offset, writeSize); + bytesWritten += writeSize; + writeOps++; + } + } + blocksWritten++; + } + } + return new WorkerResult(bytesWritten, writeOps, blocksWritten); + } + + private static final class WorkerResult { + private final long bytesWritten; + private final long writeOps; + private final long blocksWritten; + + private WorkerResult(long bytesWritten, long writeOps, long blocksWritten) { + this.bytesWritten = bytesWritten; + this.writeOps = writeOps; + this.blocksWritten = blocksWritten; + } + } + + private static final class BenchmarkSession implements AutoCloseable { + private final Pipeline pipeline; + private final ContainerClientMetrics metrics; + private final BufferPool bufferPool; + private final OzoneClientConfig config; + private final StreamBufferArgs streamBufferArgs; + private final XceiverClientManager xceiverClientManager; + private final ExecutorService responseExecutor; + + private BenchmarkSession(Pipeline pipeline, ContainerClientMetrics metrics, + BufferPool bufferPool, OzoneClientConfig config, StreamBufferArgs streamBufferArgs, + XceiverClientManager xceiverClientManager, ExecutorService responseExecutor) { + this.pipeline = pipeline; + this.metrics = metrics; + this.bufferPool = bufferPool; + this.config = config; + this.streamBufferArgs = streamBufferArgs; + this.xceiverClientManager = xceiverClientManager; + this.responseExecutor = responseExecutor; + } + + static BenchmarkSession open(boolean heapBuffer, + ChecksumType checksumType, BenchmarkMockXceiverClient mockClient) throws IOException { + final Pipeline pipeline = mockClient.getPipeline(); + final XceiverClientManager xcm = mock(XceiverClientManager.class); + when(xcm.acquireClient(any())).thenReturn(mockClient); + + final OzoneClientConfig config = new OzoneClientConfig(); + config.setStreamBufferSize(STREAM_BUFFER_SIZE); + config.setStreamBufferMaxSize(32 * 1024 * 1024); + config.setStreamBufferFlushDelay(false); + config.setStreamBufferFlushSize(16 * 1024 * 1024); + config.setChecksumType(checksumType); + config.setBytesPerChecksum(64 * 1024); + config.validate(); + + final StreamBufferArgs streamBufferArgs = + StreamBufferArgs.getDefaultStreamBufferArgs(pipeline.getReplicationConfig(), config); + + // Both modes use the same BufferPool; ChunkBuffer.ALLOCATE_DIRECT controls + // which allocator is called inside ChunkBuffer.allocate(), giving a true + // apples-to-apples comparison of heap vs direct buffer serialisation overhead. + ChunkBuffer.ALLOCATE_DIRECT.set(!heapBuffer); + + return new BenchmarkSession( + pipeline, + ContainerClientMetrics.acquire(), + new BufferPool(STREAM_BUFFER_SIZE, POOL_CAPACITY, + ByteStringConversion.createByteBufferConversion(true)), + config, + streamBufferArgs, + xcm, + newSingleThreadExecutor()); + } + + BlockOutputStream newStream() throws IOException { + return new RatisBlockOutputStream( + new BlockID(1L, 1L), + -1, + xceiverClientManager, + pipeline, + bufferPool, + config, + null, + metrics, + streamBufferArgs, + () -> responseExecutor); + } + + @Override + public void close() { + bufferPool.clearBufferPool(); + responseExecutor.shutdownNow(); + // mockClient is shared across all sessions in a phase; runTimedWrite owns it. + } + } + + private static final class Result { + private final long bytesWritten; + private final long writeOps; + private final long blocksWritten; + private final double elapsedSeconds; + private final double mbPerSec; + private final double nsPerWrite; + private final long writeChunksDuringWrite; + private final long flushesDuringWrite; + + private Result(long bytesWritten, long writeOps, long blocksWritten, double elapsedSeconds, + double mbPerSec, double nsPerWrite, long writeChunksDuringWrite, + long flushesDuringWrite) { + this.bytesWritten = bytesWritten; + this.writeOps = writeOps; + this.blocksWritten = blocksWritten; + this.elapsedSeconds = elapsedSeconds; + this.mbPerSec = mbPerSec; + this.nsPerWrite = nsPerWrite; + this.writeChunksDuringWrite = writeChunksDuringWrite; + this.flushesDuringWrite = flushesDuringWrite; + } + } +} 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..55ddcca9ddc4 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,8 +17,10 @@ package org.apache.hadoop.ozone.common; +import com.google.common.annotations.VisibleForTesting; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.concurrent.atomic.AtomicBoolean; import java.nio.channels.GatheringByteChannel; import java.util.List; import java.util.Objects; @@ -29,6 +31,16 @@ /** Buffer for a block chunk. */ public interface ChunkBuffer extends ChunkBufferToByteString, UncheckedAutoCloseable { + /** + * When true, {@link #allocate} uses direct ByteBuffers instead of heap ByteBuffers. + * Set only by {@link BlockOutputStreamWriteBenchmark} to compare allocation strategies + * within a single benchmark run; must never be modified in production code. + */ + @VisibleForTesting + // CHECKSTYLE:OFF VisibilityModifier + AtomicBoolean ALLOCATE_DIRECT = new AtomicBoolean(false); + // CHECKSTYLE:ON VisibilityModifier + /** Similar to {@link ByteBuffer#allocate(int)}. */ static ChunkBuffer allocate(int capacity) { return allocate(capacity, 0); @@ -45,7 +57,13 @@ static ChunkBuffer allocate(int capacity, int increment) { if (increment > 0 && increment < capacity) { return new IncrementalChunkBuffer(capacity, increment, false); } - CodecBuffer codecBuffer = CodecBuffer.allocateDirect(capacity); + // Heap buffer: UnsafeByteOperations.unsafeWrap() returns a BoundedByteString + // with a backing array, enabling a single System.arraycopy into the gRPC/Netty + // wire buffer instead of the slow byte-by-byte NioByteString path for direct buffers. + // ALLOCATE_DIRECT is flipped only by BlockOutputStreamWriteBenchmark. + CodecBuffer codecBuffer = ALLOCATE_DIRECT.get() + ? CodecBuffer.allocateDirect(capacity) + : CodecBuffer.allocateHeap(capacity); return new ChunkBufferImplWithByteBuffer(codecBuffer.asWritableByteBuffer(), codecBuffer); } 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..0aee09c80577 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 @@ -124,9 +124,16 @@ private ByteBuffer getAndAllocateAtIndex(int index) { } // allocate upto the given index + // Use heap buffer: UnsafeByteOperations.unsafeWrap on a heap ByteBuffer returns a + // BoundedByteString with a backing array, so gRPC serializes via a single + // System.arraycopy into the Netty wire buffer instead of the slow NioByteString + // direct-buffer path (temp-array allocation + byte-by-byte copy). + // ChunkBuffer.ALLOCATE_DIRECT is flipped only by BlockOutputStreamWriteBenchmark. ByteBuffer b = null; for (; i <= index; i++) { - final CodecBuffer c = CodecBuffer.allocateDirect(getBufferCapacityAtIndex(i)); + final CodecBuffer c = ChunkBuffer.ALLOCATE_DIRECT.get() + ? CodecBuffer.allocateDirect(getBufferCapacityAtIndex(i)) + : CodecBuffer.allocateHeap(getBufferCapacityAtIndex(i)); underlying.add(c); b = c.asWritableByteBuffer(); buffers.add(b); From 607652482e0e466533044e570ff0ab5c08749ade Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Thu, 18 Jun 2026 14:19:50 -0400 Subject: [PATCH 2/5] fixed checkstyle issues --- .../BlockOutputStreamWriteBenchmark.java | 21 ++++++++----------- .../hadoop/ozone/common/ChunkBuffer.java | 2 +- 2 files changed, 10 insertions(+), 13 deletions(-) diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStreamWriteBenchmark.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStreamWriteBenchmark.java index f05ee1743363..c33a68fc28ec 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStreamWriteBenchmark.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStreamWriteBenchmark.java @@ -176,8 +176,8 @@ private static void runScalingStudy(byte[] sourceBuffer, ChecksumType checksumTy System.out.printf("===== writeSize=%dKB scaling study =====%n%n", writeSize / 1024); for (int threadCount : threadCounts) { for (boolean heapBuffer : new boolean[] {false, true}) { - final Result result = runProfile(sourceBuffer, writeSize, heapBuffer, - checksumType, threadCount, sharedClient); + final Result result = runProfile(sourceBuffer, writeSize, heapBuffer, checksumType, + threadCount, sharedClient); rows.add(new ScalingRow(writeSize, threadCount, heapBuffer, result.mbPerSec)); System.out.println(); } @@ -301,9 +301,9 @@ private static Result runProfileMeasured(byte[] sourceBuffer, int writeSize, BenchmarkMockXceiverClient client) throws Exception { runTimedWrite(sourceBuffer, writeSize, heapBuffer, checksumType, - threadCount, WARMUP_SECONDS, "warmup", client); + threadCount, WARMUP_SECONDS, client); return runTimedWrite(sourceBuffer, writeSize, heapBuffer, checksumType, - threadCount, BENCHMARK_SECONDS, "benchmark", client); + threadCount, BENCHMARK_SECONDS, client); } private static int[] resolveWriteSizes() { @@ -337,7 +337,7 @@ private static boolean[] resolveHeapBufferModes() { private static Result runTimedWrite(byte[] sourceBuffer, int writeSize, boolean heapBuffer, ChecksumType checksumType, int threadCount, - int seconds, String phase, BenchmarkMockXceiverClient client) throws Exception { + int seconds, BenchmarkMockXceiverClient client) throws Exception { final long writeChunksBefore = ContainerClientMetrics.acquire().getWriteChunksDuringWrite().value(); final long flushesBefore = @@ -366,15 +366,13 @@ private static Result runTimedWrite(byte[] sourceBuffer, int writeSize, final long elapsedNanos = System.nanoTime() - start; final double elapsedSeconds = elapsedNanos / 1_000_000_000.0; - System.out.printf("%s complete (%.2fs)%n", phase, elapsedSeconds); + System.out.printf("complete (%.2fs)%n", elapsedSeconds); return new Result( bytesWritten, writeOps, blocksWritten, elapsedSeconds, - bytesWritten / elapsedSeconds / (1024.0 * 1024.0), - writeOps == 0 ? 0 : (double) elapsedNanos / writeOps, ContainerClientMetrics.acquire().getWriteChunksDuringWrite().value() - writeChunksBefore, ContainerClientMetrics.acquire().getFlushesDuringWrite().value() @@ -513,14 +511,13 @@ private static final class Result { private final long flushesDuringWrite; private Result(long bytesWritten, long writeOps, long blocksWritten, double elapsedSeconds, - double mbPerSec, double nsPerWrite, long writeChunksDuringWrite, - long flushesDuringWrite) { + long writeChunksDuringWrite, long flushesDuringWrite) { this.bytesWritten = bytesWritten; this.writeOps = writeOps; this.blocksWritten = blocksWritten; this.elapsedSeconds = elapsedSeconds; - this.mbPerSec = mbPerSec; - this.nsPerWrite = nsPerWrite; + this.mbPerSec = bytesWritten / elapsedSeconds / (1024.0 * 1024.0); + this.nsPerWrite = writeOps == 0 ? 0 : elapsedSeconds * 1e9 / writeOps; this.writeChunksDuringWrite = writeChunksDuringWrite; this.flushesDuringWrite = flushesDuringWrite; } 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 55ddcca9ddc4..45cd2c475b08 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 @@ -20,10 +20,10 @@ import com.google.common.annotations.VisibleForTesting; import java.io.IOException; import java.nio.ByteBuffer; -import java.util.concurrent.atomic.AtomicBoolean; import java.nio.channels.GatheringByteChannel; import java.util.List; import java.util.Objects; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.hadoop.hdds.utils.db.CodecBuffer; import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; import org.apache.ratis.util.UncheckedAutoCloseable; From 17f2f4fed0fa86faadca312b6e06fb2f625d2917 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Tue, 23 Jun 2026 22:23:46 -0400 Subject: [PATCH 3/5] Addressed review comments --- .../BlockOutputStreamWriteBenchmark.java | 120 ++++++------------ .../hadoop/ozone/common/ChunkBuffer.java | 17 +-- .../ozone/common/IncrementalChunkBuffer.java | 5 +- 3 files changed, 42 insertions(+), 100 deletions(-) diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStreamWriteBenchmark.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStreamWriteBenchmark.java index c33a68fc28ec..1fe13d40c43a 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStreamWriteBenchmark.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStreamWriteBenchmark.java @@ -40,7 +40,6 @@ import org.apache.hadoop.hdds.scm.XceiverClientManager; import org.apache.hadoop.hdds.scm.pipeline.MockPipeline; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; -import org.apache.hadoop.ozone.common.ChunkBuffer; import org.apache.ozone.test.GenericTestUtils; import org.slf4j.event.Level; @@ -53,17 +52,9 @@ * sequential writes and closes the stream, driving the chunk serialisation and * gRPC framing code at full speed without real network or disk I/O. * - *

The benchmark compares two buffer-allocation strategies: - *

    - *
  • heap — {@code ByteBuffer.allocate()} backed chunks. - * {@code UnsafeByteOperations.unsafeWrap()} returns a - * {@code BoundedByteString} with a backing array, so gRPC serialises via - * a single {@code System.arraycopy} into the Netty wire buffer.
    • - *
  • direct — {@code ByteBuffer.allocateDirect()} backed chunks - * (the behaviour before this patch). {@code UnsafeByteOperations.unsafeWrap()} - * returns a {@code NioByteString}, forcing gRPC to allocate a temp byte[] - * and copy byte-by-byte.
    • - *
+ *

Uses heap {@code ByteBuffer} backed chunks. {@code UnsafeByteOperations.unsafeWrap()} + * returns a {@code BoundedByteString} with a backing array, so gRPC serialises via + * a single {@code System.arraycopy} into the Netty wire buffer. * *

On-demand benchmark (not part of {@code mvn test}). Run from the repo root: * 

@@ -76,18 +67,16 @@
  * 
    
  *   
  • {@code benchmark.label} – profile label (default {@code workspace})
    • 
  *   
  • {@code benchmark.writeSize} – single write size in bytes (default: run
- *       1 MB, 2 MB, 3 MB and 4 MB)
    • 
- *   
  • {@code benchmark.heapBuffer} – {@code true} or {@code false}
- *       (default: run both)
    • 
+ *       256 KB, 512 KB, 1 MB, 2 MB and 4 MB)
  *   
  • {@code benchmark.checksumType} – {@code NONE} or {@code CRC32}
  *       (default {@code CRC32})
    • 
  *   
  • {@code benchmark.threads} – worker count (default {@code CPUs * 2})
    • 
- *   
  • {@code benchmark.scaling=true} – run multiple thread counts for both
- *       allocation modes and print a scaling summary
    • 
+ *   
  • {@code benchmark.scaling=true} – run multiple thread counts and print
+ *       a scaling summary
    • 
  *   
  • {@code benchmark.scaling.threads} – comma-separated thread counts for
- *       scaling (default {@code 1,2,7,14,28,42,56})
    • 
+ *       scaling (default {@code 1,2,4,7,14,28,42})
  *   
  • {@code benchmark.scaling.writeSizes} – comma-separated write sizes in
- *       bytes (overrides default 1/2/3/4 MB list when set)
    • 
+ *       bytes (overrides default list when set)
  * 

  *
  * 
Or use
@@ -102,9 +91,10 @@ public final class BlockOutputStreamWriteBenchmark {
   private static final int POOL_CAPACITY = 8;
 
   private static final int[] DEFAULT_WRITE_SIZES = {
+      256 * 1024,
+      512 * 1024,
       1024 * 1024,
       2 * 1024 * 1024,
-      3 * 1024 * 1024,
       4 * 1024 * 1024
   };
 
@@ -126,7 +116,6 @@ public static void main(String[] args) throws Exception {
 
     System.out.println("BlockOutputStream client write benchmark (mocked container RPCs)");
     System.out.println("Profile: " + label);
-    System.out.println("Comparing: heap ByteBuffer (this patch) vs direct ByteBuffer (pre-patch)");
     System.out.println("JVM: " + System.getProperty("java.version")
         + " on " + System.getProperty("os.arch"));
     System.out.printf("CPUs=%d sourceBuffer=%dMB streamBuffer=%dMB poolCapacity=%d checksum=%s%n",
@@ -146,10 +135,8 @@ public static void main(String[] args) throws Exception {
       final BenchmarkMockXceiverClient client = new BenchmarkMockXceiverClient(pipeline);
       try {
         for (int writeSize : resolveWriteSizes()) {
-          for (boolean heapBuffer : resolveHeapBufferModes()) {
-            runProfile(sourceBuffer, writeSize, heapBuffer, checksumType, threadCount, client);
-            System.out.println();
-          }
+          runProfile(sourceBuffer, writeSize, checksumType, threadCount, client);
+          System.out.println();
         }
       } finally {
         client.close();
@@ -175,12 +162,10 @@ private static void runScalingStudy(byte[] sourceBuffer, ChecksumType checksumTy
       for (int writeSize : resolveWriteSizes()) {
         System.out.printf("===== writeSize=%dKB scaling study =====%n%n", writeSize / 1024);
         for (int threadCount : threadCounts) {
-          for (boolean heapBuffer : new boolean[] {false, true}) {
-            final Result result = runProfile(sourceBuffer, writeSize, heapBuffer, checksumType,
-                threadCount, sharedClient);
-            rows.add(new ScalingRow(writeSize, threadCount, heapBuffer, result.mbPerSec));
-            System.out.println();
-          }
+          final Result result = runProfile(sourceBuffer, writeSize, checksumType,
+              threadCount, sharedClient);
+          rows.add(new ScalingRow(writeSize, threadCount, result.mbPerSec));
+          System.out.println();
         }
         printScalingSummary(writeSize, threadCounts, rows);
         System.out.println();
@@ -199,32 +184,24 @@ private static void printScalingSummary(int writeSize, int[] threadCounts,
       }
     }
 
-    final double directBaseline = findMbPerSec(rows, 1, false);
-    final double heapBaseline = findMbPerSec(rows, 1, true);
+    final double baseline = findMbPerSec(rows, 1);
 
     System.out.printf("--- scaling summary writeSize=%dKB ---%n", writeSize / 1024);
-    System.out.printf("%8s | %12s | %12s | %10s | %10s | %10s%n",
-        "threads", "direct MB/s", "heap MB/s", "heap/direct", "direct eff.", "heap eff.");
-    System.out.println("---------|--------------|--------------|------------|------------|------------");
+    System.out.printf("%8s | %12s | %10s%n", "threads", "MB/s", "efficiency");
+    System.out.println("---------|--------------|------------");
 
     for (int threadCount : threadCounts) {
-      final double direct = findMbPerSec(rows, threadCount, false);
-      final double heap = findMbPerSec(rows, threadCount, true);
-      final double heapVsDirect = direct == 0 ? 0 : heap / direct;
-      final double directEff = directBaseline == 0 ? 0 : direct / directBaseline / threadCount;
-      final double heapEff = heapBaseline == 0 ? 0 : heap / heapBaseline / threadCount;
-      System.out.printf("%8d | %12s | %12s | %10.2fx | %10.2f | %10.2f%n",
-          threadCount, MBPS.format(direct), MBPS.format(heap), heapVsDirect,
-          directEff, heapEff);
+      final double mbPerSec = findMbPerSec(rows, threadCount);
+      final double eff = baseline == 0 ? 0 : mbPerSec / baseline / threadCount;
+      System.out.printf("%8d | %12s | %10.2f%n", threadCount, MBPS.format(mbPerSec), eff);
     }
     System.out.println();
-    System.out.println("eff. = throughput vs 1-thread baseline / thread count (1.0 = linear scaling)");
+    System.out.println("efficiency = throughput vs 1-thread baseline / thread count (1.0 = linear scaling)");
   }
 
-  private static double findMbPerSec(List rows, int threadCount,
-      boolean heapBuffer) {
+  private static double findMbPerSec(List rows, int threadCount) {
     for (ScalingRow row : rows) {
-      if (row.threadCount == threadCount && row.heapBuffer == heapBuffer) {
+      if (row.threadCount == threadCount) {
         return row.mbPerSec;
       }
     }
@@ -234,13 +211,11 @@ private static double findMbPerSec(List rows, int threadCount,
   private static final class ScalingRow {
     private final int writeSize;
     private final int threadCount;
-    private final boolean heapBuffer;
     private final double mbPerSec;
 
-    private ScalingRow(int writeSize, int threadCount, boolean heapBuffer, double mbPerSec) {
+    private ScalingRow(int writeSize, int threadCount, double mbPerSec) {
       this.writeSize = writeSize;
       this.threadCount = threadCount;
-      this.heapBuffer = heapBuffer;
       this.mbPerSec = mbPerSec;
     }
   }
@@ -256,7 +231,7 @@ private static int resolveThreadCount(int cpus) {
   private static int[] resolveScalingThreadCounts(int cpus) {
     final String property = System.getProperty("benchmark.scaling.threads");
     if (property == null || property.isEmpty()) {
-      return new int[] {1, 2, 7, 14, 28, 42, 56};
+      return new int[] {1, 2, 4, 7, 14, 28, 42};
     }
     return parseCommaSeparatedInts(property);
   }
@@ -273,13 +248,13 @@ private static String formatThreadCounts(int[] threadCounts) {
   }
 
   private static Result runProfile(byte[] sourceBuffer, int writeSize,
-      boolean heapBuffer, ChecksumType checksumType, int threadCount,
+      ChecksumType checksumType, int threadCount,
       BenchmarkMockXceiverClient client)
       throws Exception {
-    System.out.printf("--- writeSize=%dKB heapBuffer=%s threads=%d writes/stream=%d ---%n",
-        writeSize / 1024, heapBuffer, threadCount, SOURCE_BUFFER_SIZE / writeSize);
+    System.out.printf("--- writeSize=%dKB threads=%d writes/stream=%d ---%n",
+        writeSize / 1024, threadCount, SOURCE_BUFFER_SIZE / writeSize);
 
-    final Result result = runProfileMeasured(sourceBuffer, writeSize, heapBuffer,
+    final Result result = runProfileMeasured(sourceBuffer, writeSize,
         checksumType, threadCount, client);
 
     System.out.printf("elapsed: %.2fs%n", result.elapsedSeconds);
@@ -297,13 +272,11 @@ private static Result runProfile(byte[] sourceBuffer, int writeSize,
   }
 
   private static Result runProfileMeasured(byte[] sourceBuffer, int writeSize,
-      boolean heapBuffer, ChecksumType checksumType, int threadCount,
+      ChecksumType checksumType, int threadCount,
       BenchmarkMockXceiverClient client)
       throws Exception {
-    runTimedWrite(sourceBuffer, writeSize, heapBuffer, checksumType,
-        threadCount, WARMUP_SECONDS, client);
-    return runTimedWrite(sourceBuffer, writeSize, heapBuffer, checksumType,
-        threadCount, BENCHMARK_SECONDS, client);
+    runTimedWrite(sourceBuffer, writeSize, checksumType, threadCount, WARMUP_SECONDS, client);
+    return runTimedWrite(sourceBuffer, writeSize, checksumType, threadCount, BENCHMARK_SECONDS, client);
   }
 
   private static int[] resolveWriteSizes() {
@@ -327,16 +300,8 @@ private static int[] parseCommaSeparatedInts(String property) {
     return values;
   }
 
-  private static boolean[] resolveHeapBufferModes() {
-    final String property = System.getProperty("benchmark.heapBuffer");
-    if (property == null || property.isEmpty()) {
-      return new boolean[] {false, true};
-    }
-    return new boolean[] {Boolean.parseBoolean(property)};
-  }
-
   private static Result runTimedWrite(byte[] sourceBuffer, int writeSize,
-      boolean heapBuffer, ChecksumType checksumType, int threadCount,
+      ChecksumType checksumType, int threadCount,
       int seconds, BenchmarkMockXceiverClient client) throws Exception {
     final long writeChunksBefore =
         ContainerClientMetrics.acquire().getWriteChunksDuringWrite().value();
@@ -351,7 +316,7 @@ private static Result runTimedWrite(byte[] sourceBuffer, int writeSize,
       final List> futures = new ArrayList<>(threadCount);
       for (int i = 0; i < threadCount; i++) {
         futures.add(workers.submit(() ->
-            runWorker(sourceBuffer, writeSize, heapBuffer, checksumType, deadline, client)));
+            runWorker(sourceBuffer, writeSize, checksumType, deadline, client)));
       }
 
       long bytesWritten = 0;
@@ -388,14 +353,14 @@ private static Result runTimedWrite(byte[] sourceBuffer, int writeSize,
   }
 
   private static WorkerResult runWorker(byte[] sourceBuffer, int writeSize,
-      boolean heapBuffer, ChecksumType checksumType, long deadline,
+      ChecksumType checksumType, long deadline,
       BenchmarkMockXceiverClient mockClient)
       throws Exception {
     long bytesWritten = 0;
     long writeOps = 0;
     long blocksWritten = 0;
 
-    try (BenchmarkSession session = BenchmarkSession.open(heapBuffer, checksumType, mockClient)) {
+    try (BenchmarkSession session = BenchmarkSession.open(checksumType, mockClient)) {
       while (System.nanoTime() < deadline) {
         try (BlockOutputStream stream = session.newStream()) {
           for (int offset = 0; offset + writeSize <= sourceBuffer.length
@@ -444,8 +409,8 @@ private BenchmarkSession(Pipeline pipeline, ContainerClientMetrics metrics,
       this.responseExecutor = responseExecutor;
     }
 
-    static BenchmarkSession open(boolean heapBuffer,
-        ChecksumType checksumType, BenchmarkMockXceiverClient mockClient) throws IOException {
+    static BenchmarkSession open(ChecksumType checksumType,
+        BenchmarkMockXceiverClient mockClient) throws IOException {
       final Pipeline pipeline = mockClient.getPipeline();
       final XceiverClientManager xcm = mock(XceiverClientManager.class);
       when(xcm.acquireClient(any())).thenReturn(mockClient);
@@ -462,11 +427,6 @@ static BenchmarkSession open(boolean heapBuffer,
       final StreamBufferArgs streamBufferArgs =
           StreamBufferArgs.getDefaultStreamBufferArgs(pipeline.getReplicationConfig(), config);
 
-      // Both modes use the same BufferPool; ChunkBuffer.ALLOCATE_DIRECT controls
-      // which allocator is called inside ChunkBuffer.allocate(), giving a true
-      // apples-to-apples comparison of heap vs direct buffer serialisation overhead.
-      ChunkBuffer.ALLOCATE_DIRECT.set(!heapBuffer);
-
       return new BenchmarkSession(
           pipeline,
           ContainerClientMetrics.acquire(),
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 45cd2c475b08..58691fac5d9c 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,13 +17,11 @@
 
 package org.apache.hadoop.ozone.common;
 
-import com.google.common.annotations.VisibleForTesting;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.GatheringByteChannel;
 import java.util.List;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.hadoop.hdds.utils.db.CodecBuffer;
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.ratis.util.UncheckedAutoCloseable;
@@ -31,16 +29,6 @@
 /** Buffer for a block chunk. */
 public interface ChunkBuffer extends ChunkBufferToByteString, UncheckedAutoCloseable {
 
-  /**
-   * When true, {@link #allocate} uses direct ByteBuffers instead of heap ByteBuffers.
-   * Set only by {@link BlockOutputStreamWriteBenchmark} to compare allocation strategies
-   * within a single benchmark run; must never be modified in production code.
-   */
-  @VisibleForTesting
-  // CHECKSTYLE:OFF VisibilityModifier
-  AtomicBoolean ALLOCATE_DIRECT = new AtomicBoolean(false);
-  // CHECKSTYLE:ON VisibilityModifier
-
   /** Similar to {@link ByteBuffer#allocate(int)}. */
   static ChunkBuffer allocate(int capacity) {
     return allocate(capacity, 0);
@@ -60,10 +48,7 @@ static ChunkBuffer allocate(int capacity, int increment) {
     // Heap buffer: UnsafeByteOperations.unsafeWrap() returns a BoundedByteString
     // with a backing array, enabling a single System.arraycopy into the gRPC/Netty
     // wire buffer instead of the slow byte-by-byte NioByteString path for direct buffers.
-    // ALLOCATE_DIRECT is flipped only by BlockOutputStreamWriteBenchmark.
-    CodecBuffer codecBuffer = ALLOCATE_DIRECT.get()
-        ? CodecBuffer.allocateDirect(capacity)
-        : CodecBuffer.allocateHeap(capacity);
+    CodecBuffer codecBuffer = CodecBuffer.allocateHeap(capacity);
     return new ChunkBufferImplWithByteBuffer(codecBuffer.asWritableByteBuffer(), codecBuffer);
   }
 
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 0aee09c80577..9e8efb1ad7e7 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
@@ -128,12 +128,9 @@ private ByteBuffer getAndAllocateAtIndex(int index) {
     // BoundedByteString with a backing array, so gRPC serializes via a single
     // System.arraycopy into the Netty wire buffer instead of the slow NioByteString
     // direct-buffer path (temp-array allocation + byte-by-byte copy).
-    // ChunkBuffer.ALLOCATE_DIRECT is flipped only by BlockOutputStreamWriteBenchmark.
     ByteBuffer b = null;
     for (; i <= index; i++) {
-      final CodecBuffer c = ChunkBuffer.ALLOCATE_DIRECT.get()
-          ? CodecBuffer.allocateDirect(getBufferCapacityAtIndex(i))
-          : CodecBuffer.allocateHeap(getBufferCapacityAtIndex(i));
+      final CodecBuffer c = CodecBuffer.allocateHeap(getBufferCapacityAtIndex(i));
       underlying.add(c);
       b = c.asWritableByteBuffer();
       buffers.add(b);

From 411de3b73ed34385c4d400585c6b388bccaf6877 Mon Sep 17 00:00:00 2001
From: Andrey Yarovoy 
Date: Tue, 23 Jun 2026 22:37:19 -0400
Subject: [PATCH 4/5] removed the shell script for running benchmark

---
 ...run-block-output-stream-write-benchmark.sh | 55 -------------------
 1 file changed, 55 deletions(-)
 delete mode 100755 hadoop-hdds/client/dev-support/run-block-output-stream-write-benchmark.sh

diff --git a/hadoop-hdds/client/dev-support/run-block-output-stream-write-benchmark.sh b/hadoop-hdds/client/dev-support/run-block-output-stream-write-benchmark.sh
deleted file mode 100755
index 37c950733b58..000000000000
--- a/hadoop-hdds/client/dev-support/run-block-output-stream-write-benchmark.sh
+++ /dev/null
@@ -1,55 +0,0 @@
-#!/usr/bin/env bash
-# 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.
-
-set -euo pipefail
-
-ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)"
-SETTINGS="${HOME}/.m2/settings.xml"
-MVN=(mvn -s "${SETTINGS}" -pl hadoop-hdds/client -q clean test-compile exec:java
-  -Dexec.mainClass=org.apache.hadoop.hdds.scm.storage.BlockOutputStreamWriteBenchmark
-  -Dexec.classpathScope=test)
-
-LABEL="${1:-workspace}"
-shift || true
-
-# Quick usage reminder
-cat <<'EOF'
-Useful system properties (pass as -Dproperty=value after the label):
-  benchmark.scaling=true                  run scaling study (recommended)
-  benchmark.scaling.threads=1,2,7,14,28,42,56   override thread counts
-  benchmark.heapBuffer=true|false         run only one allocation mode
-  benchmark.writeSize=4194304             single write size in bytes
-  benchmark.checksumType=NONE             skip checksum overhead
-EOF
-echo
-
-echo "Running BlockOutputStreamWriteBenchmark (label=${LABEL}) ..."
-cd "${ROOT}"
-
-# exec:java resolves dependency JARs from the local Maven repository, not from
-# target/classes.  Install interface-client (gRPC stubs) and common
-# (ChunkBuffer / ALLOCATE_DIRECT) so the ratis-shaded JARs in ~/.m2 are current.
-mvn -s "${SETTINGS}" -pl hadoop-hdds/interface-client,hadoop-hdds/common \
-  -q install -DskipTests -DskipShade -DskipRecon -DskipDocs
-
-# exec:java runs the benchmark in Maven's own JVM, so MAVEN_OPTS applies here.
-# The mock client serialises each request via request.toByteArray() without
-# any Netty transport, so direct memory use is limited to the CodecBuffer
-# working set (threads * poolCapacity * streamBufferSize).
-export MAVEN_OPTS="${MAVEN_OPTS:-} -Xmx4g -XX:MaxDirectMemorySize=512m"
-exec "${MVN[@]}" -Dbenchmark.label="${LABEL}" "$@"

From 23d52c2a5811cca1f8c201cab7bebd17ea4afcd4 Mon Sep 17 00:00:00 2001
From: Andrey Yarovoy 
Date: Fri, 10 Jul 2026 15:40:30 -0400
Subject: [PATCH 5/5] added an option to add backpressure in the benchmark

---
 .../storage/BenchmarkMockXceiverClient.java   | 28 +++++++++++++++++--
 1 file changed, 26 insertions(+), 2 deletions(-)

diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BenchmarkMockXceiverClient.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BenchmarkMockXceiverClient.java
index aad6df1f318a..4b44e0670a7e 100644
--- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BenchmarkMockXceiverClient.java
+++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/BenchmarkMockXceiverClient.java
@@ -20,6 +20,9 @@
 import java.io.IOException;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
@@ -56,6 +59,17 @@
  */
 final class BenchmarkMockXceiverClient extends XceiverClientSpi {
 
+  // Simulated datanode/network latency: holds each response ack for this many ms
+  // so in-flight chunk buffers accumulate in the BufferPool (backpressure), the
+  // same way a real datanode round-trip keeps buffers live until commit.
+  private static final long LATENCY_MS = Long.getLong("benchmark.dnLatencyMs", 0L);
+  private static final ScheduledExecutorService DELAY =
+      LATENCY_MS > 0 ? Executors.newScheduledThreadPool(4, r -> {
+        final Thread t = new Thread(r, "mock-dn-latency");
+        t.setDaemon(true);
+        return t;
+      }) : null;
+
   private final Pipeline pipeline;
   private final AtomicLong logIndex = new AtomicLong();
 
@@ -63,6 +77,16 @@ final class BenchmarkMockXceiverClient extends XceiverClientSpi {
     this.pipeline = pipeline;
   }
 
+  private  CompletableFuture completeMaybeDelayed(T value) {
+    final CompletableFuture future = new CompletableFuture<>();
+    if (DELAY != null) {
+      DELAY.schedule(() -> future.complete(value), LATENCY_MS, TimeUnit.MILLISECONDS);
+    } else {
+      future.complete(value);
+    }
+    return future;
+  }
+
   @Override
   public void connect() {
   }
@@ -111,7 +135,7 @@ public XceiverClientReply sendCommandAsync(ContainerCommandRequestProto request)
           .build());
     }
     final XceiverClientReply reply =
-        new XceiverClientReply(CompletableFuture.completedFuture(builder.build()));
+        new XceiverClientReply(completeMaybeDelayed(builder.build()));
     reply.setLogIndex(logIndex.incrementAndGet());
     return reply;
   }
@@ -131,7 +155,7 @@ public CompletableFuture watchForCommit(long index) {
     final XceiverClientReply reply =
         new XceiverClientReply(CompletableFuture.completedFuture(response));
     reply.setLogIndex(index);
-    return CompletableFuture.completedFuture(reply);
+    return completeMaybeDelayed(reply);
   }
 
   @Override