From 50963f5c5e9b7de957e3e9d0ccafd28cce97902d Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 1 Sep 2026 13:02:37 -0700 Subject: [PATCH 1/2] Add large-message serialize benchmark and non-array target identity sweep LargeMessageBenchmark serializes the Pulsar topic-list shape from 600 B to 4.6 MB, a varint-dense repeated-int64 message at 2 KB and 8 KB, and a 2 MB bytes payload into pooled direct buffers. NonArrayTargetIdentityTest checks that writeTo() to direct, offset and multi-component composite targets is byte-identical to the heap-array path for sizes swept byte by byte across every plausible internal boundary, for built and parsed messages and across repeated writes. --- .../benchmark/LargeMessageBenchmark.java | 176 +++++++++++++++ .../tests/NonArrayTargetIdentityTest.java | 204 ++++++++++++++++++ 2 files changed, 380 insertions(+) create mode 100644 benchmark/src/main/java/io/streamnative/lightproto/benchmark/LargeMessageBenchmark.java create mode 100644 tests/src/test/java/io/streamnative/lightproto/tests/NonArrayTargetIdentityTest.java diff --git a/benchmark/src/main/java/io/streamnative/lightproto/benchmark/LargeMessageBenchmark.java b/benchmark/src/main/java/io/streamnative/lightproto/benchmark/LargeMessageBenchmark.java new file mode 100644 index 0000000..53815b1 --- /dev/null +++ b/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 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)); + } +} diff --git a/tests/src/test/java/io/streamnative/lightproto/tests/NonArrayTargetIdentityTest.java b/tests/src/test/java/io/streamnative/lightproto/tests/NonArrayTargetIdentityTest.java new file mode 100644 index 0000000..aec9b5f --- /dev/null +++ b/tests/src/test/java/io/streamnative/lightproto/tests/NonArrayTargetIdentityTest.java @@ -0,0 +1,204 @@ +/** + * 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.tests; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.CompositeByteBuf; +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.buffer.Unpooled; +import java.util.ArrayList; +import java.util.List; +import java.util.Random; +import java.util.function.ToIntFunction; +import org.apache.pulsar.common.api.proto.BaseCommand; +import org.junit.jupiter.api.Test; + +/** + * Whatever strategy writeTo() uses for non-array targets (scratch, chunked + * scratch, NIO view, write-through), the bytes must equal the heap-array path + * (toByteArray()) for every message size — in particular around any internal + * chunk/threshold boundary — on pooled direct buffers, on multi-component + * composites, for freshly built and for parsed (lazy passthrough) messages, and + * across repeated writes of the same instance. + */ +public class NonArrayTargetIdentityTest { + + private static final int[] BOUNDARIES = { + 64, 512, 1024, 4096, 8192, 16384, 65536, 1024 * 1024 + }; + + private static byte[] drain(ByteBuf b) { + byte[] out = new byte[b.readableBytes()]; + b.readBytes(out); + return out; + } + + /** A composite whose capacity spans two direct components. */ + private static CompositeByteBuf splitComposite(int size) { + int first = Math.max(1, size / 3); + CompositeByteBuf c = Unpooled.compositeBuffer(); + c.addComponent(true, PooledByteBufAllocator.DEFAULT.directBuffer(first).writeZero(first)); + c.addComponent(true, PooledByteBufAllocator.DEFAULT.directBuffer(size - first + 1).writeZero(size - first + 1)); + c.writerIndex(0); + return c; + } + + private static void assertAllTargets(byte[] expected, int serializedSize, ToIntFunction msg) { + assertEquals(expected.length, serializedSize); + + ByteBuf direct = PooledByteBufAllocator.DEFAULT.directBuffer(expected.length); + try { + assertEquals(expected.length, msg.applyAsInt(direct)); + assertArrayEquals(expected, drain(direct), "direct"); + direct.clear(); + msg.applyAsInt(direct); + assertArrayEquals(expected, drain(direct), "direct, second write"); + } finally { + direct.release(); + } + + // A direct buffer with room to spare, written at a non-zero writerIndex + ByteBuf offset = PooledByteBufAllocator.DEFAULT.directBuffer(expected.length + 64); + try { + offset.writeZero(13); + msg.applyAsInt(offset); + offset.skipBytes(13); + assertArrayEquals(expected, drain(offset), "direct at offset"); + } finally { + offset.release(); + } + + CompositeByteBuf composite = splitComposite(expected.length); + try { + msg.applyAsInt(composite); + assertArrayEquals(expected, drain(composite), "composite"); + } finally { + composite.release(); + } + } + + private static void assertIdentity(S s) { + byte[] expected = s.toByteArray(); + assertAllTargets(expected, s.getSerializedSize(), s::writeTo); + S parsed = new S(); + parsed.parseFrom(expected); + assertAllTargets(expected, parsed.getSerializedSize(), parsed::writeTo); + } + + private static S strings(int count, int len, boolean nonAscii) { + S s = new S().setId("id"); + for (int i = 0; i < count; i++) { + String body = (nonAscii && i % 3 == 0 ? "λ∞≈" : "abc") + "x".repeat(len); + s.addName(body + i); + } + return s; + } + + @Test + public void testStringSizeSweepAroundBoundaries() { + // One long name: total size walks through every boundary byte by byte + for (int boundary : BOUNDARIES) { + for (int len = Math.max(0, boundary - 40); len <= boundary + 40; len++) { + S s = new S().setId("i"); + s.addName("y".repeat(len)); + assertIdentity(s); + } + } + } + + @Test + public void testManyStringsAcrossBoundaries() { + // Many ~500-byte elements: boundaries fall inside elements, tags and lengths + for (int count : new int[]{1, 7, 8, 9, 16, 17, 33, 130, 131, 132, 135, 2100, 8192}) { + assertIdentity(strings(count, 500, true)); + } + } + + @Test + public void testSmallStringsAcrossBoundaries() { + // Tiny elements: many tag/length pairs straddle the boundaries + for (int count : new int[]{300, 800, 1000, 1200, 1500, 3000, 6000}) { + assertIdentity(strings(count, 1, false)); + } + } + + @Test + public void testBytesPayloadSweep() { + Random rnd = new Random(1); + List sizes = new ArrayList<>(); + for (int boundary : BOUNDARIES) { + for (int d = -12; d <= 12; d += 3) { + sizes.add(Math.max(0, boundary + d)); + } + } + sizes.add(2 * 1024 * 1024 + 7); + for (int size : sizes) { + byte[] payload = new byte[size]; + rnd.nextBytes(payload); + B b = new B().setPayload(payload); + b.addExtraItem(new byte[]{1, 2, 3}); + b.addExtraItem(new byte[0]); + b.addExtraItem(payload.length > 100 ? java.util.Arrays.copyOf(payload, 100) : payload); + byte[] expected = b.toByteArray(); + assertAllTargets(expected, b.getSerializedSize(), b::writeTo); + B parsed = new B(); + parsed.parseFrom(expected); + assertAllTargets(expected, parsed.getSerializedSize(), parsed::writeTo); + } + } + + @Test + public void testNestedTreeAcrossBoundaries() { + for (int count : new int[]{1, 60, 120, 130, 140, 260, 2000, 2200, 9000}) { + M m = new M(); + m.setX().setA("a-value").setB("b-value"); + for (int i = 0; i < count; i++) { + M.KV kv = m.addItem(); + kv.setK("key-" + i); + kv.setV("value-" + "v".repeat(i % 17) + i); + if (i % 10 == 0) { + kv.setXx().setN(i); + } + } + byte[] expected = m.toByteArray(); + assertAllTargets(expected, m.getSerializedSize(), m::writeTo); + M parsed = new M(); + parsed.parseFrom(expected); + assertAllTargets(expected, parsed.getSerializedSize(), parsed::writeTo); + } + } + + @Test + public void testPulsarTopicListShape() { + for (int topics : new int[]{1, 7, 8, 9, 28, 180, 8192}) { + List 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); + byte[] expected = cmd.toByteArray(); + assertAllTargets(expected, cmd.getSerializedSize(), cmd::writeTo); + BaseCommand parsed = new BaseCommand(); + parsed.parseFrom(expected); + assertAllTargets(expected, parsed.getSerializedSize(), parsed::writeTo); + } + } +} From eb23039287d774e7d3cea6d9bb3730e3d402ca22 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 1 Sep 2026 13:08:20 -0700 Subject: [PATCH 2/2] Write direct buffers in place through their NIO view above 512 bytes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Since #12, writeTo() to a non-array buffer stages the whole message in a heap byte[] scratch and bulk-copies it. Messages above SCRATCH_RETAIN_MAX (1 MiB) never retain that scratch, so every write allocated a fresh full-size array — multi-MB G1-humongous allocations that OOMed Pulsar's proxy back-pressure test (apache/pulsar#26256, together with the clear() retention fixed in #19). Below the cap the copy itself was still paid. A single-region direct buffer exposes its memory as a java.nio.ByteBuffer through ByteBuf.internalNioBuffer(). Absolute puts on a DirectByteBuffer compile to a bounds check plus a jdk.internal.misc.Unsafe store — which, unlike sun.misc.Unsafe, carries no JDK 24+ deprecation check — so the message can be written in place: no scratch array and no bulk copy, at any size. writeTo() now dispatches heap buffers in place through the backing array (unchanged), single-region direct buffers larger than NIO_WRITE_MIN (512 bytes) through the NIO view, and everything else (small messages; composites and other buffers without a single NIO region) through the scratch path as before. Above the threshold no direct-buffer write touches the scratch, so it only grows past 512 bytes for composite targets. The threshold exists because the view's per-put cost is a fixed tax per message while the copy it saves grows with size. Interleaved JMH on pooled direct buffers (JDK 21/26): the view is 15-19% slower on the ~70-byte varint-dense MessageMetadata, at parity on BaseCommand, 20% faster at 600 bytes, and 35-40% faster from 6 KB to 100 KB; on the 2 MB / 4.6 MB cases it removes the per-write allocation (-70% / -55%) and matches the per-field ByteBuf-API write-through of #18, which it replaces. The field emitters are parameterized over the write sink (WriteSink.ARRAY / WriteSink.NIO): one emitter produces both _writeTo(byte[], int) and _writeTo(ByteBuffer, int), differing only in the sink variable and in how bulk data is copied out of a ByteBuf; every raw writer in LightProtoCodec is overloaded for both sinks. NonArrayTargetIdentityTest sweeps sizes byte by byte across every boundary (64 B .. 1 MiB) on direct, offset and multi-component composite targets, for built and parsed messages, repeated strings incl. non-ASCII, bytes payloads, nested trees and the Pulsar BaseCommand shape. NioWriteTest checks the routing flips exactly at NIO_WRITE_MIN, that composites keep the scratch path, and (via ThreadMXBean.getThreadAllocatedBytes) that 5 writes of a 4.6 MB topic list or a 5 MB payload allocate less than a quarter of one message. LargeMessageBenchmark covers 600 B .. 4.6 MB topic lists, varint-dense 2 KB / 8 KB messages and a 2 MB payload. --- .../generator/LightProtoBytesField.java | 11 +- .../lightproto/generator/LightProtoField.java | 38 +++- .../generator/LightProtoMapField.java | 49 +++-- .../generator/LightProtoMessage.java | 48 +++-- .../generator/LightProtoMessageField.java | 10 +- .../generator/LightProtoNumberField.java | 49 +++-- .../LightProtoRepeatedBytesField.java | 11 +- .../LightProtoRepeatedMessageField.java | 8 +- .../LightProtoRepeatedNumberField.java | 14 +- .../LightProtoRepeatedStringField.java | 11 +- .../generator/LightProtoStringField.java | 11 +- .../lightproto/generator/LightProtoCodec.java | 124 +++++++++++- .../lightproto/tests/NioWriteTest.java | 176 ++++++++++++++++++ 13 files changed, 443 insertions(+), 117 deletions(-) create mode 100644 tests/src/test/java/io/streamnative/lightproto/tests/NioWriteTest.java diff --git a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoBytesField.java b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoBytesField.java index 8200554..057ce10 100644 --- a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoBytesField.java +++ b/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); } diff --git a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoField.java b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoField.java index 2edf5c9..e5a1c1d 100644 --- a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoField.java +++ b/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); } } diff --git a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMapField.java b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMapField.java index 1db77c9..f5ffb38 100644 --- a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMapField.java +++ b/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); } } diff --git a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMessage.java b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMessage.java index 8d1bf13..846e12e 100644 --- a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMessage.java +++ b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMessage.java @@ -416,12 +416,23 @@ private void generateSerialize(PrintWriter w) { w.format(" int _writeIdx = _b.writerIndex();\n"); w.format(" _writeTo(_b.array(), _b.arrayOffset() + _writeIdx);\n"); w.format(" _b.writerIndex(_writeIdx + _serializedSize);\n"); + w.format(" } else if (_serializedSize > LightProtoCodec.NIO_WRITE_MIN && _b.nioBufferCount() == 1) {\n"); + // Single-region direct buffers are written in place through their NIO + // view: absolute puts on a DirectByteBuffer compile to a bounds check plus + // a jdk.internal.misc.Unsafe store (no JDK 24+ deprecation check), so no + // scratch array and no bulk copy are needed. Below NIO_WRITE_MIN the + // per-put cost outweighs the copy it saves, so small messages stay on the + // scratch path. + w.format(" _b.ensureWritable(_serializedSize);\n"); + w.format(" int _writeIdx = _b.writerIndex();\n"); + w.format(" java.nio.ByteBuffer _nb = _b.internalNioBuffer(_writeIdx, _serializedSize);\n"); + w.format(" _writeTo(_nb, _nb.position());\n"); + w.format(" _b.writerIndex(_writeIdx + _serializedSize);\n"); w.format(" } else {\n"); - // Direct, composite and other buffers: compose in a scratch array cached - // on this (typically pooled) instance and transfer with a single bulk - // write. Plain byte[] stores compile to raw memory accesses on every JDK, - // unlike sun.misc.Unsafe accesses which carry a per-call deprecation - // check since JDK 24. + // Small messages, and buffers without a single NIO region (composites): + // compose in a scratch array cached on this (typically pooled) instance + // and transfer with a single bulk write. Plain byte[] stores compile to + // raw memory accesses on every JDK. w.format(" byte[] _s = LightProtoCodec.scratchFor(this._scratch, _serializedSize);\n"); w.format(" if (_s.length <= LightProtoCodec.SCRATCH_RETAIN_MAX) {\n"); w.format(" this._scratch = _s;\n"); @@ -439,6 +450,24 @@ private void generateSerialize(PrintWriter w) { w.println(" * type into the same array."); w.println(" */"); w.format(" public int _writeTo(byte[] _a, int _i) {\n"); + emitWriteBody(w, LightProtoField.WriteSink.ARRAY); + w.format(" return _i;\n"); + w.format(" }\n"); + + w.println(" /**"); + w.println(" * Internal: serialize this message into a direct buffer's NIO view starting"); + w.println(" * at absolute index {@code _i}; returns the index after the last byte written."); + w.println(" * Public only so that generated messages in other packages can serialize"); + w.println(" * nested fields of this type into the same view."); + w.println(" */"); + w.format(" public int _writeTo(java.nio.ByteBuffer _nb, int _i) {\n"); + emitWriteBody(w, LightProtoField.WriteSink.NIO); + w.format(" return _i;\n"); + w.format(" }\n"); + } + + /** The field walk shared by both write sinks; only the sink variable and bulk copies differ. */ + private void emitWriteBody(PrintWriter w, LightProtoField.WriteSink sink) { if (hasRequiredFields()) { w.format(" checkRequiredFields();\n"); } @@ -447,22 +476,19 @@ private void generateSerialize(PrintWriter w) { // guard: set bits ascend, so the output order (and bytes) match the // declaration-order guard walk. Required fields always have their bit // set here — checkRequiredFields() has already thrown otherwise. - emitBitDrivenTraversal(w, f -> f.serialize(w)); + emitBitDrivenTraversal(w, f -> f.serialize(w, sink)); } else { for (LightProtoField f : fields) { String condition = f.serializeCondition(); if (condition != null) { w.format(" if (%s) {\n", condition); - f.serialize(w); + f.serialize(w, sink); w.format(" }\n"); } else { - f.serialize(w); + f.serialize(w, sink); } } } - - w.format(" return _i;\n"); - w.format(" }\n"); } private void generateGetSerializedSize(PrintWriter w) { diff --git a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMessageField.java b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMessageField.java index 3d241ef..16a5d4d 100644 --- a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMessageField.java +++ b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMessageField.java @@ -96,12 +96,12 @@ public void parseTextFormat(PrintWriter w) { } @Override - public void serialize(PrintWriter w) { - // Nested messages write into the same array: no per-child ensureWritable, + public void serialize(PrintWriter w, WriteSink sink) { + // Nested messages write into the same sink: no per-child ensureWritable, // buffer-address resolution or writerIndex round-trips. - w.format("%s;\n", writeTagExpr(tagName())); - w.format("_i = LightProtoCodec.writeRawVarInt(_a, _i, %s.getSerializedSize());\n", ccName); - w.format("_i = %s._writeTo(_a, _i);\n", ccName); + w.format("%s;\n", writeTagExpr(tagName(), sink)); + w.format("_i = LightProtoCodec.writeRawVarInt(%s, _i, %s.getSerializedSize());\n", sink.var, ccName); + w.format("_i = %s._writeTo(%s, _i);\n", ccName, sink.var); } @Override diff --git a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoNumberField.java b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoNumberField.java index f88a091..e82e80f 100644 --- a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoNumberField.java +++ b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoNumberField.java @@ -46,38 +46,35 @@ public LightProtoNumberField(ProtoFieldDescriptor field, int index) { super(field, index); } - static void serializeNumber(PrintWriter w, ProtoFieldDescriptor field, String name) { + static void serializeNumber(PrintWriter w, ProtoFieldDescriptor field, String name, WriteSink sink) { + String writer; + String value = name; if (field.isEnumField()) { - w.format(" _i = LightProtoCodec.writeRawVarInt(_a, _i, %s.getValue());\n", name); + writer = "writeRawVarInt"; + value = name + ".getValue()"; } else if (field.getProtoType().equals("bool")) { - w.format(" _i = LightProtoCodec.writeRawByte(_a, _i, %s ? 1 : 0);\n", name); - } else if (field.getProtoType().equals("int32")) { - w.format(" _i = LightProtoCodec.writeRawVarInt(_a, _i, %s);\n", name); - } else if (field.getProtoType().equals("uint32")) { - w.format(" _i = LightProtoCodec.writeRawVarInt(_a, _i, %s);\n", name); + writer = "writeRawByte"; + value = name + " ? 1 : 0"; + } else if (field.getProtoType().equals("int32") || field.getProtoType().equals("uint32")) { + writer = "writeRawVarInt"; } else if (field.getProtoType().equals("sint32")) { - w.format(" _i = LightProtoCodec.writeRawSignedVarInt(_a, _i, %s);\n", name); + writer = "writeRawSignedVarInt"; } else if (field.getProtoType().equals("sint64")) { - w.format(" _i = LightProtoCodec.writeRawSignedVarInt64(_a, _i, %s);\n", name); - } else if (field.getProtoType().equals("int64")) { - w.format(" _i = LightProtoCodec.writeRawVarInt64(_a, _i, %s);\n", name); - } else if (field.getProtoType().equals("uint64")) { - w.format(" _i = LightProtoCodec.writeRawVarInt64(_a, _i, %s);\n", name); - } else if (field.getProtoType().equals("fixed32")) { - w.format(" _i = LightProtoCodec.writeRawLittleEndian32(_a, _i, %s);\n", name); - } else if (field.getProtoType().equals("fixed64")) { - w.format(" _i = LightProtoCodec.writeRawLittleEndian64(_a, _i, %s);\n", name); - } else if (field.getProtoType().equals("sfixed32")) { - w.format(" _i = LightProtoCodec.writeRawLittleEndian32(_a, _i, %s);\n", name); - } else if (field.getProtoType().equals("sfixed64")) { - w.format(" _i = LightProtoCodec.writeRawLittleEndian64(_a, _i, %s);\n", name); + writer = "writeRawSignedVarInt64"; + } else if (field.getProtoType().equals("int64") || field.getProtoType().equals("uint64")) { + writer = "writeRawVarInt64"; + } else if (field.getProtoType().equals("fixed32") || field.getProtoType().equals("sfixed32")) { + writer = "writeRawLittleEndian32"; + } else if (field.getProtoType().equals("fixed64") || field.getProtoType().equals("sfixed64")) { + writer = "writeRawLittleEndian64"; } else if (field.getProtoType().equals("double")) { - w.format(" _i = LightProtoCodec.writeRawDouble(_a, _i, %s);\n", name); + writer = "writeRawDouble"; } else if (field.getProtoType().equals("float")) { - w.format(" _i = LightProtoCodec.writeRawFloat(_a, _i, %s);\n", name); + writer = "writeRawFloat"; } else { throw new IllegalArgumentException("Failed to write serializer for field: " + field.getProtoType()); } + w.format(" _i = LightProtoCodec.%s(%s, _i, %s);\n", writer, sink.var, value); } static String parseNumber(ProtoFieldDescriptor field) { @@ -188,9 +185,9 @@ public void parse(PrintWriter w) { } @Override - public void serialize(PrintWriter w) { - w.format("%s;\n", writeTagExpr(tagName())); - serializeNumber(w, field, ccName); + public void serialize(PrintWriter w, WriteSink sink) { + w.format("%s;\n", writeTagExpr(tagName(), sink)); + serializeNumber(w, field, ccName, sink); } @Override diff --git a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedBytesField.java b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedBytesField.java index 7ee3128..8c7f974 100644 --- a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedBytesField.java +++ b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedBytesField.java @@ -83,17 +83,16 @@ public void getter(PrintWriter w) { } @Override - public void serialize(PrintWriter w) { + public void serialize(PrintWriter w, WriteSink sink) { w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName); w.format(" LightProtoCodec.BytesHolder _bh = %s[i];\n", pluralName); - w.format(" %s;\n", writeTagExpr(tagName())); - w.format(" _i = LightProtoCodec.writeRawVarInt(_a, _i, _bh.len);\n"); + w.format(" %s;\n", writeTagExpr(tagName(), sink)); + w.format(" _i = LightProtoCodec.writeRawVarInt(%s, _i, _bh.len);\n", sink.var); w.format(" if (_bh.idx == -1) {\n"); - w.format(" _bh.b.getBytes(_bh.b.readerIndex(), _a, _i, _bh.len);\n"); + sink.copyBytes(w, "_bh.b", "_bh.b.readerIndex()", "_bh.len"); w.format(" } else {\n"); - w.format(" _parsedBuffer.getBytes(_bh.idx, _a, _i, _bh.len);\n"); + sink.copyBytes(w, "_parsedBuffer", "_bh.idx", "_bh.len"); w.format(" }\n"); - w.format(" _i += _bh.len;\n"); w.format("}\n"); } diff --git a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedMessageField.java b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedMessageField.java index 79e9e1d..df11de0 100644 --- a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedMessageField.java +++ b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedMessageField.java @@ -73,12 +73,12 @@ private String addForParseName() { } @Override - public void serialize(PrintWriter w) { + public void serialize(PrintWriter w, WriteSink sink) { w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName); w.format(" %s _item = %s[i];\n", field.getJavaType(), pluralName); - w.format(" %s;\n", writeTagExpr(tagName())); - w.format(" _i = LightProtoCodec.writeRawVarInt(_a, _i, _item.getSerializedSize());\n"); - w.format(" _i = _item._writeTo(_a, _i);\n"); + w.format(" %s;\n", writeTagExpr(tagName(), sink)); + w.format(" _i = LightProtoCodec.writeRawVarInt(%s, _i, _item.getSerializedSize());\n", sink.var); + w.format(" _i = _item._writeTo(%s, _i);\n", sink.var); w.format("}\n"); } diff --git a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedNumberField.java b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedNumberField.java index 7521086..0b3eb0f 100644 --- a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedNumberField.java +++ b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedNumberField.java @@ -81,29 +81,29 @@ public void getter(PrintWriter w) { } @Override - public void serialize(PrintWriter w) { + public void serialize(PrintWriter w, WriteSink sink) { int fixedSize = LightProtoNumberField.fixedDataSize(field); if (field.isPacked()) { - w.format(" %s;\n", writeTagExpr(tagName() + "_PACKED")); + w.format(" %s;\n", writeTagExpr(tagName() + "_PACKED", sink)); if (fixedSize >= 0) { - w.format(" _i = LightProtoCodec.writeRawVarInt(_a, _i, _%sCount * %d);\n", pluralName, fixedSize); + w.format(" _i = LightProtoCodec.writeRawVarInt(%s, _i, _%sCount * %d);\n", sink.var, pluralName, fixedSize); } else { w.format(" int _%sSize = 0;\n", pluralName); w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName); w.format(" %s _item = %s[i];\n", field.getJavaType(), pluralName); w.format(" _%sSize += %s;\n", pluralName, LightProtoNumberField.serializedSizeOfNumber(field, "_item")); w.format("}\n"); - w.format(" _i = LightProtoCodec.writeRawVarInt(_a, _i, _%sSize);\n", pluralName); + w.format(" _i = LightProtoCodec.writeRawVarInt(%s, _i, _%sSize);\n", sink.var, pluralName); } w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName); w.format(" %s _item = %s[i];\n", field.getJavaType(), pluralName); - LightProtoNumberField.serializeNumber(w, field, "_item"); + LightProtoNumberField.serializeNumber(w, field, "_item", sink); w.format("}\n"); } else { w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName); w.format(" %s _item = %s[i];\n", field.getJavaType(), pluralName); - w.format(" %s;\n", writeTagExpr(tagName())); - LightProtoNumberField.serializeNumber(w, field, "_item"); + w.format(" %s;\n", writeTagExpr(tagName(), sink)); + LightProtoNumberField.serializeNumber(w, field, "_item", sink); w.format("}\n"); } } diff --git a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedStringField.java b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedStringField.java index 4fab1da..75503df 100644 --- a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedStringField.java +++ b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedStringField.java @@ -81,16 +81,15 @@ public void getter(PrintWriter w) { } @Override - public void serialize(PrintWriter w) { + public void serialize(PrintWriter w, WriteSink sink) { w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName); w.format(" LightProtoCodec.StringHolder _sh = %s[i];\n", pluralName); - w.format(" %s;\n", writeTagExpr(tagName())); - w.format(" _i = LightProtoCodec.writeRawVarInt(_a, _i, _sh.len);\n"); + w.format(" %s;\n", writeTagExpr(tagName(), sink)); + w.format(" _i = LightProtoCodec.writeRawVarInt(%s, _i, _sh.len);\n", sink.var); w.format(" if (_sh.idx == -1) {\n"); - w.format(" _i = LightProtoCodec.writeRawString(_a, _i, _sh.s, _sh.len);\n"); + w.format(" _i = LightProtoCodec.writeRawString(%s, _i, _sh.s, _sh.len);\n", sink.var); w.format(" } else {\n"); - w.format(" _parsedBuffer.getBytes(_sh.idx, _a, _i, _sh.len);\n"); - w.format(" _i += _sh.len;\n"); + sink.copyBytes(w, "_parsedBuffer", "_sh.idx", "_sh.len"); w.format(" }\n"); w.format("}\n"); } diff --git a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoStringField.java b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoStringField.java index 2ef9795..5ad32fe 100644 --- a/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoStringField.java +++ b/code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoStringField.java @@ -101,14 +101,13 @@ public void serializedSize(PrintWriter w) { } @Override - public void serialize(PrintWriter w) { - w.format("%s;\n", writeTagExpr(tagName())); - w.format("_i = LightProtoCodec.writeRawVarInt(_a, _i, _%sBufferLen);\n", ccName); + public void serialize(PrintWriter w, WriteSink sink) { + w.format("%s;\n", writeTagExpr(tagName(), sink)); + w.format("_i = LightProtoCodec.writeRawVarInt(%s, _i, _%sBufferLen);\n", sink.var, ccName); w.format("if (_%sBufferIdx == -1) {\n", ccName); - w.format(" _i = LightProtoCodec.writeRawString(_a, _i, %s, _%sBufferLen);\n", ccName, ccName); + w.format(" _i = LightProtoCodec.writeRawString(%s, _i, %s, _%sBufferLen);\n", sink.var, ccName, ccName); w.format("} else {\n"); - w.format(" _parsedBuffer.getBytes(_%sBufferIdx, _a, _i, _%sBufferLen);\n", ccName, ccName); - w.format(" _i += _%sBufferLen;\n", ccName); + sink.copyBytes(w, "_parsedBuffer", "_" + ccName + "BufferIdx", "_" + ccName + "BufferLen"); w.format("}\n"); } diff --git a/code-generator/src/main/resources/io/streamnative/lightproto/generator/LightProtoCodec.java b/code-generator/src/main/resources/io/streamnative/lightproto/generator/LightProtoCodec.java index 00fb353..bb1a57b 100644 --- a/code-generator/src/main/resources/io/streamnative/lightproto/generator/LightProtoCodec.java +++ b/code-generator/src/main/resources/io/streamnative/lightproto/generator/LightProtoCodec.java @@ -390,16 +390,29 @@ static void writeString(ByteBuf b, String s, int bytesCount) { } } - // --- Array-based raw write methods for zero-overhead serialization --- - // Serialization composes into a plain byte[] with an int cursor: heap buffers - // are written in place through their backing array, other buffer types are - // composed in a reusable scratch array and transferred with a single bulk - // writeBytes(). Plain array stores compile to raw memory accesses on every JDK - // (no sun.misc.Unsafe in the hot loop — its memory-access methods carry a - // per-call deprecation check since JDK 24). - - // Scratch arrays larger than this are not retained on the message instance, - // so outlier messages don't pin large allocations. + // --- Raw write methods for zero-overhead serialization --- + // Serialization writes through an int cursor into one of two sinks, and every + // raw writer below is overloaded for both: + // - a plain byte[]: heap buffers in place through their backing array; small + // messages (and buffers without a single NIO region) composed in a reusable + // scratch array and transferred with one bulk writeBytes(); + // - a direct buffer's NIO view (ByteBuf.internalNioBuffer) for messages above + // NIO_WRITE_MIN, written in place: no scratch array and no bulk copy. + // Neither uses sun.misc.Unsafe in the hot loop (its memory-access methods carry + // a per-call deprecation check since JDK 24): array stores compile to raw + // memory accesses, and DirectByteBuffer puts to a bounds check plus a + // jdk.internal.misc.Unsafe store. + + // Messages strictly larger than this are written through the NIO view when the + // target is a single-region direct buffer. Below it the view's per-put cost + // outweighs the scratch copy it saves (~+15% on a 70-byte varint-dense message, + // versus -20% at 600 bytes and -35% from 6 KB up). + static final int NIO_WRITE_MIN = 512; + + // Scratch arrays larger than this are not retained on the message instance, so + // outlier messages don't pin large allocations. Direct-buffer messages above + // NIO_WRITE_MIN never touch the scratch, so it only grows past that for + // buffers without a single NIO region. static final int SCRATCH_RETAIN_MAX = 1024 * 1024; // clear() of a message larger than this (or of unknown size) releases the @@ -511,6 +524,97 @@ static int writeRawString(byte[] a, int i, String s, int bytesCount) { return i + bytesCount; } + // NIO-view overloads of the writers above. Absolute puts ignore the view's + // position; the view's byte order is honored explicitly for fixed-width values. + + static int writeRawByte(java.nio.ByteBuffer nb, int i, int value) { + nb.put(i, (byte) value); + return i + 1; + } + + static int writeRawVarInt(java.nio.ByteBuffer nb, int i, int n) { + if (n >= 0) { + while (true) { + if ((n & ~0x7F) == 0) { + nb.put(i++, (byte) n); + return i; + } + nb.put(i++, (byte) ((n & 0x7F) | 0x80)); + n >>>= 7; + } + } else { + return writeRawVarInt64(nb, i, n); + } + } + + static int writeRawVarInt64(java.nio.ByteBuffer nb, int i, long value) { + while (true) { + if ((value & ~0x7FL) == 0) { + nb.put(i++, (byte) value); + return i; + } + nb.put(i++, (byte) (((int) value & 0x7F) | 0x80)); + value >>>= 7; + } + } + + static int writeRawSignedVarInt(java.nio.ByteBuffer nb, int i, int n) { + return writeRawVarInt(nb, i, encodeZigZag32(n)); + } + + static int writeRawSignedVarInt64(java.nio.ByteBuffer nb, int i, long n) { + return writeRawVarInt64(nb, i, encodeZigZag64(n)); + } + + static int writeRawLittleEndian32(java.nio.ByteBuffer nb, int i, int value) { + nb.putInt(i, nb.order() == java.nio.ByteOrder.LITTLE_ENDIAN ? value : Integer.reverseBytes(value)); + return i + 4; + } + + static int writeRawLittleEndian64(java.nio.ByteBuffer nb, int i, long value) { + nb.putLong(i, nb.order() == java.nio.ByteOrder.LITTLE_ENDIAN ? value : Long.reverseBytes(value)); + return i + 8; + } + + static int writeRawFloat(java.nio.ByteBuffer nb, int i, float n) { + return writeRawLittleEndian32(nb, i, Float.floatToRawIntBits(n)); + } + + static int writeRawDouble(java.nio.ByteBuffer nb, int i, double n) { + return writeRawLittleEndian64(nb, i, Double.doubleToRawLongBits(n)); + } + + static int writeRawString(java.nio.ByteBuffer nb, int i, String s, int bytesCount) { + if (s.length() == bytesCount) { + // ASCII fast path: bulk-put the String's internal LATIN1 byte[] directly + if (HAS_UNSAFE && COMPACT_STRINGS) { + try { + Object _v = (Object) MH_GET_OBJECT.invokeExact((Object) s, STRING_VALUE_OFFSET); + nb.put(i, (byte[]) _v, 0, bytesCount); + } catch (Throwable t) { + throw new RuntimeException(t); + } + } else { + nb.put(i, s.getBytes(StandardCharsets.ISO_8859_1), 0, bytesCount); + } + } else { + nb.put(i, s.getBytes(StandardCharsets.UTF_8), 0, bytesCount); + } + return i + bytesCount; + } + + /** + * Copies len bytes of src starting at srcIdx into the view at absolute index i + * (through a temporary position/limit window, restored afterwards); returns i + len. + */ + static int copyRawBytes(ByteBuf src, int srcIdx, java.nio.ByteBuffer nb, int i, int len) { + int lim = nb.limit(); + nb.limit(i + len).position(i); + src.getBytes(srcIdx, nb); + nb.limit(lim); + return i + len; + } + static String readString(ByteBuf b, int index, int len) { if (HAS_UNSAFE && STRING_VALUE_OFFSET >= 0) { try { diff --git a/tests/src/test/java/io/streamnative/lightproto/tests/NioWriteTest.java b/tests/src/test/java/io/streamnative/lightproto/tests/NioWriteTest.java new file mode 100644 index 0000000..26a2ed0 --- /dev/null +++ b/tests/src/test/java/io/streamnative/lightproto/tests/NioWriteTest.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.tests; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.CompositeByteBuf; +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.buffer.Unpooled; +import java.lang.management.ManagementFactory; +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; +import java.util.Random; +import org.apache.pulsar.common.api.proto.BaseCommand; +import org.junit.jupiter.api.Test; + +/** + * Direct-buffer writes above {@code NIO_WRITE_MIN} go through the buffer's NIO + * view: no scratch array, no transient allocation, whatever the message size. + * At or below the threshold, and for buffers without a single NIO region, the + * scratch path is kept. + */ +public class NioWriteTest { + + private static byte[] scratchOf(Object msg) throws Exception { + Field f = msg.getClass().getDeclaredField("_scratch"); + f.setAccessible(true); + return (byte[]) f.get(msg); + } + + private static byte[] drain(ByteBuf b) { + byte[] out = new byte[b.readableBytes()]; + b.readBytes(out); + return out; + } + + private static S withNameOfSerializedSize(int size) { + // tag(1) + length varint + payload: search the payload length that lands on size + for (int len = Math.max(0, size - 6); len <= size; len++) { + S s = new S(); + s.addName("y".repeat(len)); + if (s.getSerializedSize() == size) { + return s; + } + } + throw new AssertionError("no single-name message of serialized size " + size); + } + + @Test + public void testThresholdRoutesToScratchOrView() throws Exception { + int t = LightProtoCodec.NIO_WRITE_MIN; + for (int size : new int[]{64, t - 1, t}) { + S s = withNameOfSerializedSize(size); + byte[] expected = s.toByteArray(); + ByteBuf direct = PooledByteBufAllocator.DEFAULT.directBuffer(size); + try { + s.writeTo(direct); + assertArrayEquals(expected, drain(direct)); + } finally { + direct.release(); + } + byte[] scratch = scratchOf(s); + assertNotNull(scratch, "size " + size + " should use the scratch path"); + assertTrue(scratch.length <= Math.max(64, t)); + } + for (int size : new int[]{t + 1, 4096, 100 * 1024}) { + S s = withNameOfSerializedSize(size); + byte[] expected = s.toByteArray(); + ByteBuf direct = PooledByteBufAllocator.DEFAULT.directBuffer(size); + try { + s.writeTo(direct); + assertArrayEquals(expected, drain(direct)); + } finally { + direct.release(); + } + assertNull(scratchOf(s), "size " + size + " should be written through the NIO view"); + } + } + + @Test + public void testCompositeKeepsScratchPath() throws Exception { + S s = withNameOfSerializedSize(3000); + byte[] expected = s.toByteArray(); + CompositeByteBuf composite = Unpooled.compositeBuffer(); + composite.addComponent(true, PooledByteBufAllocator.DEFAULT.directBuffer(1000).writeZero(1000)); + composite.addComponent(true, PooledByteBufAllocator.DEFAULT.directBuffer(2001).writeZero(2001)); + composite.writerIndex(0); + try { + s.writeTo(composite); + assertArrayEquals(expected, drain(composite)); + } finally { + composite.release(); + } + assertNotNull(scratchOf(s)); + } + + @Test + public void testAllocationFreeLargeWritesBytes() throws Exception { + byte[] payload = new byte[5 * 1024 * 1024]; + new Random(7).nextBytes(payload); + B b = new B().setPayload(payload); + assertAllocationFree(b::writeTo, b.getSerializedSize()); + assertNull(scratchOf(b)); + } + + @Test + public void testAllocationFreeLargeWritesTopicList() throws Exception { + // With -XX:-CompactStrings the ASCII fast path cannot bulk-put the String's + // internal byte[] and copies each string through a temporary array. + assumeTrue(!ManagementFactory.getRuntimeMXBean().getInputArguments().contains("-XX:-CompactStrings")); + + List topics = new ArrayList<>(); + String base = "persistent://public/default/" + "t".repeat(520) + "-"; + for (int i = 0; i < 8192; i++) { + topics.add(base + i); + } + BaseCommand cmd = new BaseCommand().setType(BaseCommand.Type.GET_TOPICS_OF_NAMESPACE_RESPONSE); + cmd.setGetTopicsOfNamespaceResponse().setRequestId(42).addAllTopics(topics); + assertAllocationFree(cmd::writeTo, cmd.getSerializedSize()); + assertNull(scratchOf(cmd)); + } + + private static void assertAllocationFree(java.util.function.ToIntFunction writeTo, int size) + throws Exception { + var mxBean = ManagementFactory.getThreadMXBean(); + assumeTrue(mxBean instanceof com.sun.management.ThreadMXBean); + com.sun.management.ThreadMXBean tb = (com.sun.management.ThreadMXBean) mxBean; + assumeTrue(tb.isThreadAllocatedMemorySupported()); + if (!tb.isThreadAllocatedMemoryEnabled()) { + tb.setThreadAllocatedMemoryEnabled(true); + } + assertTrue(size > 1024 * 1024); + + ByteBuf direct = PooledByteBufAllocator.DEFAULT.directBuffer(size + 64); + try { + for (int i = 0; i < 3; i++) { + direct.clear(); + writeTo.applyAsInt(direct); + } + long tid = Thread.currentThread().getId(); + long before = tb.getThreadAllocatedBytes(tid); + for (int i = 0; i < 5; i++) { + direct.clear(); + writeTo.applyAsInt(direct); + } + long allocated = tb.getThreadAllocatedBytes(tid) - before; + // Staging would allocate a fresh full-size array per write (5x the + // message size over this loop); the in-place view allocates none of it. + assertTrue(allocated < size / 4, + "expected allocation-free serialization but " + allocated + + " bytes were allocated for 5 writes of a " + size + "-byte message"); + } finally { + direct.release(); + } + } +}