HDDS-15592 use heap allocation in ChunkBuffer

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}.
+ *
+ * <p>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.
+ *
+ * <p>This exercises the full cross-domain copy chain:
+ * <ul>
+ *   <li><b>direct chunk (pre-patch):</b> {@code NioByteString.writeTo} allocates
+ *       a temporary {@code byte[]}, copies from off-heap via {@code copyMemory}
+ *       (copy 1), then writes heap→direct ByteBuf (copy 2).</li>
+ *   <li><b>heap chunk (this patch):</b> {@code BoundedByteString.writeTo} writes
+ *       directly heap→direct ByteBuf (copy 1 only).</li>
+ * </ul>
+ */
+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<XceiverClientReply> 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<DatanodeDetails, ContainerCommandResponseProto> sendCommandOnAllNodes(
+      ContainerCommandRequestProto request) {
+    return null;
+  }
+}