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 d4c41f9..8200554 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 @@ -103,6 +103,16 @@ public void clear(PrintWriter w) { // parse() invalidates the stale buffer reference for fields present on the wire. } + @Override + public void clearRelease(PrintWriter w) { + w.format("%s = null;\n", ccName); + } + + @Override + public boolean needsRelease() { + return true; + } + @Override public void serializedSize(PrintWriter w) { w.format("_size += %s_SIZE;\n", tagName()); 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 5f0e843..2edf5c9 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 @@ -129,6 +129,26 @@ public void has(PrintWriter w) { abstract public void clear(PrintWriter w); + /** + * Emit this field's contribution to the generated {@code _clearAndRelease()}: + * like {@link #clear(PrintWriter)} but also releasing retained data references + * (cached Strings, ByteBuf refs), recursing into nested messages via their + * {@code _clearAndRelease()}. Fields that retain no references inherit this + * default, which emits the plain clear code. + */ + public void clearRelease(PrintWriter w) { + clear(w); + } + + /** + * Whether this field retains data references that {@link #clearRelease(PrintWriter)} + * must drop. Messages where no field does skip the clear() size gate entirely: + * their release path is behaviorally identical to the plain clear. + */ + public boolean needsRelease() { + return false; + } + public void fieldClear(PrintWriter w, String enclosingType) { w.format(" /** Clear the {@code %s} field. */\n", field.getName()); w.format(" public %s %s() {\n", enclosingType, Util.camelCase("clear", field.getName())); 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 95da500..1db77c9 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 @@ -971,6 +971,36 @@ public void clear(PrintWriter w) { w.format("_%sIndex = null;\n", ccName); } + @Override + public void clearRelease(PrintWriter w) { + if (isStringKey()) { + w.format("for (int i = 0; i < _%sCount; i++) {\n", ccName); + w.format(" _%sKeys[i].s = null;\n", ccName); + w.format("}\n"); + } + if (isStringValue()) { + w.format("for (int i = 0; i < _%sCount; i++) {\n", ccName); + w.format(" _%sValues[i].s = null;\n", ccName); + w.format("}\n"); + } else if (isBytesValue()) { + w.format("for (int i = 0; i < _%sCount; i++) {\n", ccName); + w.format(" _%sValues[i].b = null;\n", ccName); + w.format("}\n"); + } else if (isMessageValue()) { + // Forced recursion — see LightProtoMessageField#clearRelease. + w.format("for (int i = 0; i < _%sCount; i++) {\n", ccName); + w.format(" _%sValues[i]._clearAndRelease();\n", ccName); + w.format("}\n"); + } + w.format("_%sCount = 0;\n", ccName); + w.format("_%sIndex = null;\n", ccName); + } + + @Override + public boolean needsRelease() { + return true; + } + @Override public void materialize(PrintWriter w) { if (isStringKey()) { 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 1829821..8d1bf13 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 @@ -96,6 +96,7 @@ public void generate(PrintWriter w) { generateParseFrom(w); generateCheckRequiredFields(w); generateClear(w); + generateClearAndRelease(w); generateCopyFrom(w); if (generateJson) { @@ -285,6 +286,18 @@ private void emitBitDrivenTraversal(PrintWriter w, java.util.function.Consumer 0) {\n"); + w.format(" return _clearAndRelease();\n"); + w.format(" }\n"); + } boolean bitDriven = useBitDrivenClear(); for (LightProtoField f : fields) { if (bitDriven && f instanceof LightProtoMessageField && !f.isOneofMember()) { @@ -335,6 +348,33 @@ private void generateClear(PrintWriter w) { w.format(" }\n"); } + private void generateClearAndRelease(PrintWriter w) { + w.println(" /**"); + w.println(" * clear() variant for messages above CLEAR_RETAIN_MAX (or of unknown"); + w.println(" * size) that also releases the data references the O(1) clear() leaves"); + w.println(" * in place, so a reused (pooled or per-connection) instance doesn't pin"); + w.println(" * the last message's data. Recursion into nested messages is forced —"); + w.println(" * children don't re-check the size gate, or a large message spread over"); + w.println(" * many small children would release nothing. Public only so that"); + w.println(" * generated messages in other packages can release nested fields of"); + w.println(" * this type."); + w.println(" */"); + w.format(" public %s _clearAndRelease() {\n", message.getName()); + for (LightProtoField f : fields) { + f.clearRelease(w); + } + w.format(" _parsedBuffer = null;\n"); + w.format(" _cachedSize = -1;\n"); + for (int i = 0; i < bitFieldsCount(); i++) { + w.format(" _bitField%d = 0;\n", i); + } + for (ProtoOneofDescriptor oneof : oneofs) { + w.format(" _%sCase = 0;\n", Util.camelCase(oneof.getName())); + } + w.format(" return this;\n"); + w.format(" }\n"); + } + private void generateCopyFrom(PrintWriter w) { w.println(" /** Copy all fields from another message of the same type. */"); w.format("public %s copyFrom(%s _other) {\n", message.getName(), message.getName()); 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 218711b..3d241ef 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 @@ -111,6 +111,20 @@ public void clear(PrintWriter w) { w.format("}\n"); } + @Override + public void clearRelease(PrintWriter w) { + // Forced recursion: the child must not re-check its own size gate, or a + // large parent spread across many small children would release nothing. + w.format("if (%s()){\n", Util.camelCase("has", ccName)); + w.format(" %s._clearAndRelease();\n", ccName); + w.format("}\n"); + } + + @Override + public boolean needsRelease() { + return true; + } + @Override public void materialize(PrintWriter w) { w.format("if (%s()) {\n", Util.camelCase("has", ccName)); 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 9700531..7ee3128 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 @@ -211,6 +211,19 @@ public void clear(PrintWriter w) { w.format("_%sCount = 0;\n", pluralName); } + @Override + public void clearRelease(PrintWriter w) { + w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName); + w.format(" %s[i].b = null;\n", pluralName); + w.format("}\n"); + w.format("_%sCount = 0;\n", pluralName); + } + + @Override + public boolean needsRelease() { + return true; + } + @Override public void materialize(PrintWriter w) { w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName); 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 16a4049..79e9e1d 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 @@ -183,6 +183,20 @@ public void clear(PrintWriter w) { w.format("_%sCount = 0;\n", pluralName); } + @Override + public void clearRelease(PrintWriter w) { + // Forced recursion — see LightProtoMessageField#clearRelease. + w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName); + w.format(" %s[i]._clearAndRelease();\n", pluralName); + w.format("}\n"); + w.format("_%sCount = 0;\n", pluralName); + } + + @Override + public boolean needsRelease() { + return true; + } + @Override public void materialize(PrintWriter w) { w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName); 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 a0477b8..4fab1da 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 @@ -203,6 +203,19 @@ public void clear(PrintWriter w) { w.format("_%sCount = 0;\n", pluralName); } + @Override + public void clearRelease(PrintWriter w) { + w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName); + w.format(" %s[i].s = null;\n", pluralName); + w.format("}\n"); + w.format("_%sCount = 0;\n", pluralName); + } + + @Override + public boolean needsRelease() { + return true; + } + @Override public void materialize(PrintWriter w) { w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName); 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 ea37d75..2ef9795 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 @@ -83,6 +83,16 @@ public void clear(PrintWriter w) { // parse() invalidates the cached decoded String for fields present on the wire. } + @Override + public void clearRelease(PrintWriter w) { + w.format("%s = null;\n", ccName); + } + + @Override + public boolean needsRelease() { + return true; + } + @Override public void serializedSize(PrintWriter w) { w.format("_size += %s_SIZE;\n", tagName()); 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 77eb322..00fb353 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 @@ -402,6 +402,13 @@ static void writeString(ByteBuf b, String s, int bytesCount) { // so outlier messages don't pin large allocations. static final int SCRATCH_RETAIN_MAX = 1024 * 1024; + // clear() of a message larger than this (or of unknown size) releases the + // data references retained by the O(1) clear design, so a reused (pooled or + // per-connection) instance pins at most this much of the last message's + // data. The release walk costs O(element count), which is noise for any + // message this large; below the threshold the walk is skipped entirely. + static final int CLEAR_RETAIN_MAX = 64 * 1024; + /** Returns current if it can hold size bytes, otherwise a larger replacement. */ static byte[] scratchFor(byte[] current, int size) { if (current != null && current.length >= size) { diff --git a/tests/src/test/java/io/streamnative/lightproto/tests/ClearReleaseTest.java b/tests/src/test/java/io/streamnative/lightproto/tests/ClearReleaseTest.java new file mode 100644 index 0000000..0762c2d --- /dev/null +++ b/tests/src/test/java/io/streamnative/lightproto/tests/ClearReleaseTest.java @@ -0,0 +1,177 @@ +/** + * 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 java.lang.ref.WeakReference; + +import org.junit.jupiter.api.Test; + +/** + * clear() of a message above {@code CLEAR_RETAIN_MAX} (or of unknown size) must + * release the data references the O(1) clear leaves in place, so a reused + * (pooled or per-connection) instance doesn't pin the last message's data — + * the 0.8.0 regression that kept multi-MB topic lists alive on every Pulsar + * connection decoder. Below the threshold the O(1) clear is unchanged and may + * retain. + */ +public class ClearReleaseTest { + + /** Unique, non-interned marker string so a WeakReference observes liveness. */ + private static String marker() { + return new String("marker-" + System.nanoTime() + "-x".repeat(64)); + } + + private static void assertEventuallyCollected(WeakReference ref) throws InterruptedException { + for (int i = 0; i < 100 && ref.get() != null; i++) { + System.gc(); + Thread.sleep(10); + } + assertNull(ref.get(), "reference should have been released by clear()"); + } + + /** Adds ~size bytes of repeated names plus one weakly-tracked marker. */ + private static WeakReference fillNames(S s, int size) { + String m = marker(); + s.addName(m); + int chunk = 500; + for (int i = 0; i < size / chunk; i++) { + s.addName("n".repeat(chunk)); + } + return new WeakReference<>(m); + } + + @Test + public void testLargeBuiltMessageReleasesStringsOnClear() throws Exception { + S s = new S().setId("big"); + WeakReference ref = fillNames(s, 100 * 1024); + assertTrue(s.getSerializedSize() > LightProtoCodec.CLEAR_RETAIN_MAX); + + s.clear(); + assertEventuallyCollected(ref); + } + + @Test + public void testSmallMessageKeepsO1ClearRetention() { + S s = new S().setId("small"); + WeakReference ref = fillNames(s, 8 * 1024); + assertTrue(s.getSerializedSize() <= LightProtoCodec.CLEAR_RETAIN_MAX); + + s.clear(); + System.gc(); + // The O(1) clear path deliberately retains: the holder still references + // the string, so it must survive GC. + assertNotNull(ref.get()); + } + + @Test + public void testUnknownSizeReleasesConservatively() throws Exception { + // Never serialized nor cleanly parsed: _cachedSize is -1, so clear() + // cannot know the message was small and must take the release path. + S s = new S().setId("unsized"); + WeakReference ref = fillNames(s, 8 * 1024); + + s.clear(); + assertEventuallyCollected(ref); + } + + /** + * The Pulsar decoder shape: one reused message per connection, parseFrom() + * per command (which invokes clear() on the previous contents), getters + * materializing the strings. The next parse must release them. + */ + @Test + public void testParseReuseReleasesMaterializedStrings() throws Exception { + S big = new S().setId("big"); + fillNames(big, 100 * 1024); + byte[] bigBytes = big.toByteArray(); + byte[] smallBytes = new S().setId("small").toByteArray(); + + S reused = new S(); + reused.parseFrom(bigBytes); + WeakReference ref = new WeakReference<>(reused.getNameAt(0)); + + reused.parseFrom(smallBytes); + assertEquals("small", reused.getId()); + assertEventuallyCollected(ref); + } + + /** + * A large message spread over many small children: every child is far below + * the threshold, so the release must recurse unconditionally once the + * top-level gate triggers. + */ + @Test + public void testFanOutReleasesThroughForcedRecursion() throws Exception { + M m = new M(); + String v = marker(); + WeakReference ref = new WeakReference<>(v); + for (int i = 0; i < 2000; i++) { + M.KV kv = m.addItem(); + kv.setK("key-" + "k".repeat(40) + i); + kv.setV(i == 0 ? v : "val-" + "v".repeat(40) + i); + } + v = null; + assertTrue(m.getSerializedSize() > LightProtoCodec.CLEAR_RETAIN_MAX); + + m.clear(); + assertEventuallyCollected(ref); + } + + @Test + public void testLargeBytesPayloadReleasedOnClear() throws Exception { + byte[] payload = new byte[2 * 1024 * 1024]; + WeakReference ref = new WeakReference<>(payload); + B b = new B().setPayload(payload); + payload = null; + assertTrue(b.getSerializedSize() > LightProtoCodec.CLEAR_RETAIN_MAX); + + b.clear(); + assertEventuallyCollected(ref); + } + + /** Deep clear must leave the instance as reusable as the O(1) clear does. */ + @Test + public void testReuseAfterDeepClear() { + S s = new S().setId("big"); + fillNames(s, 100 * 1024); + assertTrue(s.getSerializedSize() > LightProtoCodec.CLEAR_RETAIN_MAX); + s.clear(); + + s.setId("after"); + s.addName("one"); + s.addName("two"); + byte[] reused = s.toByteArray(); + + S freshMsg = new S().setId("after"); + freshMsg.addName("one"); + freshMsg.addName("two"); + byte[] fresh = freshMsg.toByteArray(); + assertArrayEquals(fresh, reused); + + S parsed = new S(); + parsed.parseFrom(reused); + assertEquals("after", parsed.getId()); + assertEquals(2, parsed.getNamesCount()); + assertEquals("one", parsed.getNameAt(0)); + assertEquals("two", parsed.getNameAt(1)); + } +}