Write direct buffers in place through their NIO view above 512 bytes

benchmark/src/main/java/io/streamnative/lightproto/benchmark/LargeMessageBenchmark.java

@@ -0,0 +1,176 @@
+/**
+ * Copyright 2026 StreamNative
+ *
+ * Licensed 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 io.streamnative.lightproto.benchmark;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.streamnative.lightproto.tests.B;
+import io.streamnative.lightproto.tests.Repeated;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.common.api.proto.BaseCommand;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * Serialization of messages well above the small-message hot path, into pooled
+ * direct buffers (the Pulsar case): the topic-list shape at several sizes and a
+ * large bytes payload. Sizes straddle the candidate scratch/chunk thresholds.
+ */
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 3, time = 2)
+@Fork(value = 1)
+public class LargeMessageBenchmark {
+
+    private static BaseCommand topicList(int topics) {
+        List<String> list = new ArrayList<>(topics);
+        String base = "persistent://public/default/" + "t".repeat(520) + "-";
+        for (int i = 0; i < topics; i++) {
+            list.add(base + i);
+        }
+        BaseCommand cmd = new BaseCommand().setType(BaseCommand.Type.GET_TOPICS_OF_NAMESPACE_RESPONSE);
+        cmd.setGetTopicsOfNamespaceResponse().setRequestId(42).addAllTopics(list);
+        cmd.getSerializedSize();
+        return cmd;
+    }
+
+    private static ByteBuf directFor(int size) {
+        return PooledByteBufAllocator.DEFAULT.directBuffer(size);
+    }
+
+    /** Varint-dense shape (no bulk data): NIO's adversarial case. */
+    private static Repeated denseVarints(int count) {
+        Repeated r = new Repeated();
+        for (int i = 0; i < count; i++) {
+            r.addXInt64(i * 1000003L + 17);
+        }
+        r.getSerializedSize();
+        return r;
+    }
+
+    // Sub-4 KB points to locate the array-scratch / NIO crossover
+    private final BaseCommand topics600B = topicList(1);
+    private final BaseCommand topics1KB = topicList(2);
+    private final BaseCommand topics2KB = topicList(4);
+    private final BaseCommand topics3KB = topicList(6);
+    private final Repeated dense2KB = denseVarints(300);
+    private final Repeated dense8KB = denseVarints(1200);
+    private final ByteBuf buf600B = directFor(topics600B.getSerializedSize());
+    private final ByteBuf buf1KB = directFor(topics1KB.getSerializedSize());
+    private final ByteBuf buf2KB = directFor(topics2KB.getSerializedSize());
+    private final ByteBuf buf3KB = directFor(topics3KB.getSerializedSize());
+    private final ByteBuf bufDense2KB = directFor(dense2KB.getSerializedSize());
+    private final ByteBuf bufDense8KB = directFor(dense8KB.getSerializedSize());
+
+    private final BaseCommand topics6KB = topicList(11);      // ~6 KB: just above a 4 KB chunk
+    private final BaseCommand topics16KB = topicList(28);     // ~16 KB
+    private final BaseCommand topics100KB = topicList(180);   // ~100 KB
+    private final BaseCommand topics4MB = topicList(8192);    // ~4.6 MB: the Pulsar proxy test shape
+    private final B bytes2MB;
+
+    private final ByteBuf buf6KB = directFor(topics6KB.getSerializedSize());
+    private final ByteBuf buf16KB = directFor(topics16KB.getSerializedSize());
+    private final ByteBuf buf100KB = directFor(topics100KB.getSerializedSize());
+    private final ByteBuf buf4MB = directFor(topics4MB.getSerializedSize());
+    private final ByteBuf bufBytes2MB;
+
+    public LargeMessageBenchmark() {
+        byte[] payload = new byte[2 * 1024 * 1024];
+        new Random(7).nextBytes(payload);
+        bytes2MB = new B().setPayload(payload);
+        bytes2MB.getSerializedSize();
+        bufBytes2MB = directFor(bytes2MB.getSerializedSize());
+    }
+
+    @Benchmark
+    public void topicList6KB(Blackhole bh) {
+        buf6KB.clear();
+        bh.consume(topics6KB.writeTo(buf6KB));
+    }
+
+    @Benchmark
+    public void topicList16KB(Blackhole bh) {
+        buf16KB.clear();
+        bh.consume(topics16KB.writeTo(buf16KB));
+    }
+
+    @Benchmark
+    public void topicList100KB(Blackhole bh) {
+        buf100KB.clear();
+        bh.consume(topics100KB.writeTo(buf100KB));
+    }
+
+    @Benchmark
+    public void topicList4MB(Blackhole bh) {
+        buf4MB.clear();
+        bh.consume(topics4MB.writeTo(buf4MB));
+    }
+
+    @Benchmark
+    public void bytesPayload2MB(Blackhole bh) {
+        bufBytes2MB.clear();
+        bh.consume(bytes2MB.writeTo(bufBytes2MB));
+    }
+
+    @Benchmark
+    public void topicList600B(Blackhole bh) {
+        buf600B.clear();
+        bh.consume(topics600B.writeTo(buf600B));
+    }
+
+    @Benchmark
+    public void topicList1KB(Blackhole bh) {
+        buf1KB.clear();
+        bh.consume(topics1KB.writeTo(buf1KB));
+    }
+
+    @Benchmark
+    public void topicList2KB(Blackhole bh) {
+        buf2KB.clear();
+        bh.consume(topics2KB.writeTo(buf2KB));
+    }
+
+    @Benchmark
+    public void topicList3KB(Blackhole bh) {
+        buf3KB.clear();
+        bh.consume(topics3KB.writeTo(buf3KB));
+    }
+
+    @Benchmark
+    public void denseVarints2KB(Blackhole bh) {
+        bufDense2KB.clear();
+        bh.consume(dense2KB.writeTo(bufDense2KB));
+    }
+
+    @Benchmark
+    public void denseVarints8KB(Blackhole bh) {
+        bufDense8KB.clear();
+        bh.consume(dense8KB.writeTo(bufDense8KB));
+    }
+}

code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoBytesField.java

@@ -153,18 +153,17 @@ public void parseTextFormat(PrintWriter w) {
     }
 
     @Override
-    public void serialize(PrintWriter w) {
-        w.format("%s;\n", writeTagExpr(tagName()));
-        w.format("_i = LightProtoCodec.writeRawVarInt(_a, _i, _%sLen);\n", ccName);
+    public void serialize(PrintWriter w, WriteSink sink) {
+        w.format("%s;\n", writeTagExpr(tagName(), sink));
+        w.format("_i = LightProtoCodec.writeRawVarInt(%s, _i, _%sLen);\n", sink.var, ccName);
         w.format("if (_%sIdx == -1) {\n", ccName);
         // Use the absolute-indexed copy so we don't mutate the source buffer's
         // readerIndex; that allows the message to be re-serialized (e.g. on
         // gRPC retry) and lets two fields safely alias the same backing buffer.
-        w.format("    %s.getBytes(%s.readerIndex(), _a, _i, _%sLen);\n", ccName, ccName, ccName);
+        sink.copyBytes(w, ccName, ccName + ".readerIndex()", "_" + ccName + "Len");
         w.format("} else {\n");
-        w.format("    _parsedBuffer.getBytes(_%sIdx, _a, _i, _%sLen);\n", ccName, ccName);
+        sink.copyBytes(w, "_parsedBuffer", "_" + ccName + "Idx", "_" + ccName + "Len");
         w.format("}\n");
-        w.format("_i += _%sLen;\n", ccName);
     }
 
 

code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoField.java

@@ -173,7 +173,37 @@ public void fieldClear(PrintWriter w, String enclosingType) {
 
     abstract public void serializedSize(PrintWriter w);
 
-    abstract public void serialize(PrintWriter w);
+    /**
+     * Where generated serialization code writes. Both sinks are addressed by the int
+     * cursor {@code _i}, and every LightProtoCodec raw writer is overloaded for both,
+     * so a field emitter differs between them only in the sink variable and in how
+     * bulk data is copied out of a ByteBuf.
+     */
+    enum WriteSink {
+        /** {@code byte[] _a}: a heap buffer's backing array, or the scratch array. */
+        ARRAY("_a"),
+        /** {@code java.nio.ByteBuffer _nb}: a direct buffer's NIO view, written in place. */
+        NIO("_nb");
+
+        final String var;
+
+        WriteSink(String var) {
+            this.var = var;
+        }
+
+        /** Emits a copy of {@code len} bytes of ByteBuf {@code src} from {@code srcIdx} to the cursor, advancing it. */
+        void copyBytes(PrintWriter w, String src, String srcIdx, String len) {
+            if (this == ARRAY) {
+                w.format("%s.getBytes(%s, _a, _i, %s);\n", src, srcIdx, len);
+                w.format("_i += %s;\n", len);
+            } else {
+                w.format("_i = LightProtoCodec.copyRawBytes(%s, %s, _nb, _i, %s);\n", src, srcIdx, len);
+            }
+        }
+    }
+
+    /** Emit this field's serialization into the given sink; must produce identical bytes for both sinks. */
+    abstract public void serialize(PrintWriter w, WriteSink sink);
 
     abstract public void serializeJson(PrintWriter w);
 
@@ -232,11 +262,11 @@ public void parsePacked(PrintWriter w) {
 
     abstract protected String typeTag();
 
-    protected String writeTagExpr(String tag) {
+    protected String writeTagExpr(String tag, WriteSink sink) {
         if (field.getNumber() <= 15) {
-            return String.format("_i = LightProtoCodec.writeRawByte(_a, _i, %s)", tag);
+            return String.format("_i = LightProtoCodec.writeRawByte(%s, _i, %s)", sink.var, tag);
         } else {
-            return String.format("_i = LightProtoCodec.writeRawVarInt(_a, _i, %s)", tag);
+            return String.format("_i = LightProtoCodec.writeRawVarInt(%s, _i, %s)", sink.var, tag);
         }
     }
 

code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMapField.java

@@ -803,7 +803,7 @@ public void parseTextFormat(PrintWriter w) {
     }
 
     @Override
-    public void serialize(PrintWriter w) {
+    public void serialize(PrintWriter w, WriteSink sink) {
         w.format("for (int _entryIdx = 0; _entryIdx < _%sCount; _entryIdx++) {\n", ccName);
 
         // Compute entry size
@@ -818,16 +818,16 @@ public void serialize(PrintWriter w) {
         generateValueDataSize(w, "_entryIdx");
 
         // Write outer tag + entry size
-        w.format("    %s;\n", writeTagExpr(tagName()));
-        w.format("    _i = LightProtoCodec.writeRawVarInt(_a, _i, _entrySize);\n");
+        w.format("    %s;\n", writeTagExpr(tagName(), sink));
+        w.format("    _i = LightProtoCodec.writeRawVarInt(%s, _i, _entrySize);\n", sink.var);
 
         // Write key tag + key data
-        w.format("    _i = LightProtoCodec.writeRawByte(_a, _i, %s);\n", keyTagConstant());
-        generateSerializeKeyData(w, "_entryIdx");
+        w.format("    _i = LightProtoCodec.writeRawByte(%s, _i, %s);\n", sink.var, keyTagConstant());
+        generateSerializeKeyData(w, "_entryIdx", sink);
 
         // Write value tag + value data
-        w.format("    _i = LightProtoCodec.writeRawByte(_a, _i, %s);\n", valueTagConstant());
-        generateSerializeValueData(w, "_entryIdx");
+        w.format("    _i = LightProtoCodec.writeRawByte(%s, _i, %s);\n", sink.var, valueTagConstant());
+        generateSerializeValueData(w, "_entryIdx", sink);
 
         w.format("}\n");
     }
@@ -858,46 +858,43 @@ private void generateValueDataSize(PrintWriter w, String idxVar) {
         }
     }
 
-    private void generateSerializeKeyData(PrintWriter w, String idxVar) {
+    private void generateSerializeKeyData(PrintWriter w, String idxVar, WriteSink sink) {
         if (isStringKey()) {
             w.format("    LightProtoCodec.StringHolder _ksh = _%sKeys[%s];\n", ccName, idxVar);
-            w.format("    _i = LightProtoCodec.writeRawVarInt(_a, _i, _ksh.len);\n");
+            w.format("    _i = LightProtoCodec.writeRawVarInt(%s, _i, _ksh.len);\n", sink.var);
             w.format("    if (_ksh.idx == -1) {\n");
-            w.format("        _i = LightProtoCodec.writeRawString(_a, _i, _ksh.s, _ksh.len);\n");
+            w.format("        _i = LightProtoCodec.writeRawString(%s, _i, _ksh.s, _ksh.len);\n", sink.var);
             w.format("    } else {\n");
-            w.format("        _parsedBuffer.getBytes(_ksh.idx, _a, _i, _ksh.len);\n");
-            w.format("        _i += _ksh.len;\n");
+            sink.copyBytes(w, "_parsedBuffer", "_ksh.idx", "_ksh.len");
             w.format("    }\n");
         } else {
-            LightProtoNumberField.serializeNumber(w, keyField, String.format("_%sKeys[%s]", ccName, idxVar));
+            LightProtoNumberField.serializeNumber(w, keyField, String.format("_%sKeys[%s]", ccName, idxVar), sink);
         }
     }
 
-    private void generateSerializeValueData(PrintWriter w, String idxVar) {
+    private void generateSerializeValueData(PrintWriter w, String idxVar, WriteSink sink) {
         if (isStringValue()) {
             w.format("    LightProtoCodec.StringHolder _vsh = _%sValues[%s];\n", ccName, idxVar);
-            w.format("    _i = LightProtoCodec.writeRawVarInt(_a, _i, _vsh.len);\n");
+            w.format("    _i = LightProtoCodec.writeRawVarInt(%s, _i, _vsh.len);\n", sink.var);
             w.format("    if (_vsh.idx == -1) {\n");
-            w.format("        _i = LightProtoCodec.writeRawString(_a, _i, _vsh.s, _vsh.len);\n");
+            w.format("        _i = LightProtoCodec.writeRawString(%s, _i, _vsh.s, _vsh.len);\n", sink.var);
             w.format("    } else {\n");
-            w.format("        _parsedBuffer.getBytes(_vsh.idx, _a, _i, _vsh.len);\n");
-            w.format("        _i += _vsh.len;\n");
+            sink.copyBytes(w, "_parsedBuffer", "_vsh.idx", "_vsh.len");
             w.format("    }\n");
         } else if (isBytesValue()) {
             w.format("    LightProtoCodec.BytesHolder _vbh = _%sValues[%s];\n", ccName, idxVar);
-            w.format("    _i = LightProtoCodec.writeRawVarInt(_a, _i, _vbh.len);\n");
+            w.format("    _i = LightProtoCodec.writeRawVarInt(%s, _i, _vbh.len);\n", sink.var);
             w.format("    if (_vbh.idx == -1) {\n");
-            w.format("        _vbh.b.getBytes(_vbh.b.readerIndex(), _a, _i, _vbh.len);\n");
+            sink.copyBytes(w, "_vbh.b", "_vbh.b.readerIndex()", "_vbh.len");
             w.format("    } else {\n");
-            w.format("        _parsedBuffer.getBytes(_vbh.idx, _a, _i, _vbh.len);\n");
+            sink.copyBytes(w, "_parsedBuffer", "_vbh.idx", "_vbh.len");
             w.format("    }\n");
-            w.format("    _i += _vbh.len;\n");
         } else if (isMessageValue()) {
-            w.format("    _i = LightProtoCodec.writeRawVarInt(_a, _i, _%sValues[%s].getSerializedSize());\n",
-                    ccName, idxVar);
-            w.format("    _i = _%sValues[%s]._writeTo(_a, _i);\n", ccName, idxVar);
+            w.format("    _i = LightProtoCodec.writeRawVarInt(%s, _i, _%sValues[%s].getSerializedSize());\n",
+                    sink.var, ccName, idxVar);
+            w.format("    _i = _%sValues[%s]._writeTo(%s, _i);\n", ccName, idxVar, sink.var);
         } else {
-            LightProtoNumberField.serializeNumber(w, valueField, String.format("_%sValues[%s]", ccName, idxVar));
+            LightProtoNumberField.serializeNumber(w, valueField, String.format("_%sValues[%s]", ccName, idxVar), sink);
         }
     }