This is an automated email from the ASF dual-hosted git repository.
tabish121 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-protonj2.git
The following commit(s) were added to refs/heads/main by this push:
new ee168037 PROTON-2993 Read input buffers into heap buffers
ee168037 is described below
commit ee1680375cdfdd9c60a69520af972f23b12df28f
Author: Timothy Bish <[email protected]>
AuthorDate: Tue Oct 20 16:02:33 2026 -0400
PROTON-2993 Read input buffers into heap buffers
When decoding the binary payload read into a heap buffer to avoid any
dangling references to pooled buffers which can cause exhaustion of
the pool if references are long lived. Also improves performance of
the decode process by preventing use to cleaners for heap buffers that
do not need an explicit release when unreferenced.
---
.../primitives/AbstractBinaryTypeDecoder.java | 22 ++-----
.../org/apache/qpid/protonj2/types/Binary.java | 15 ++++-
.../qpid/protonj2/codec/benchmark/Benchmark.java | 1 +
.../codec/primitives/BinaryTypeCodecTest.java | 74 ++++++++++++++++++++++
4 files changed, 94 insertions(+), 18 deletions(-)
diff --git
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/primitives/AbstractBinaryTypeDecoder.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/primitives/AbstractBinaryTypeDecoder.java
index ab743d25..35454687 100644
---
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/primitives/AbstractBinaryTypeDecoder.java
+++
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/primitives/AbstractBinaryTypeDecoder.java
@@ -20,7 +20,7 @@ import java.io.IOException;
import java.io.InputStream;
import org.apache.qpid.protonj2.buffer.ProtonBuffer;
-import org.apache.qpid.protonj2.buffer.ProtonBufferAllocator;
+import org.apache.qpid.protonj2.buffer.impl.ProtonByteArrayBufferAllocator;
import org.apache.qpid.protonj2.codec.DecodeException;
import org.apache.qpid.protonj2.codec.DecoderState;
import org.apache.qpid.protonj2.codec.EncoderState;
@@ -36,7 +36,7 @@ public abstract class AbstractBinaryTypeDecoder extends
AbstractPrimitiveTypeDec
@Override
public Binary readValue(ProtonBuffer buffer, DecoderState state) throws
DecodeException {
- return new Binary(readValueAsBuffer(buffer, state));
+ return new Binary(readValueAsBuffer(buffer, state), false);
}
/**
@@ -53,19 +53,7 @@ public abstract class AbstractBinaryTypeDecoder extends
AbstractPrimitiveTypeDec
*/
@Override
public ProtonBuffer readValueAsBuffer(ProtonBuffer buffer, DecoderState
state) throws DecodeException {
- final int length = readSize(buffer, state);
-
- if (Integer.compareUnsigned(length, buffer.getReadableBytes()) > 0) {
- throw new DecodeException(
- String.format("Binary data size %d is specified to be greater
than the amount " +
- "of data available (%d)",
Integer.toUnsignedLong(length), buffer.getReadableBytes()));
- }
-
- final ProtonBuffer payload = buffer.copy(buffer.getReadOffset(),
length, true);
-
- buffer.advanceReadOffset(length);
-
- return payload;
+ return ProtonByteArrayBufferAllocator.wrapped(readValueAsArray(buffer,
state)).convertToReadOnly();
}
/**
@@ -112,7 +100,7 @@ public abstract class AbstractBinaryTypeDecoder extends
AbstractPrimitiveTypeDec
@Override
public Binary readValue(InputStream stream, StreamDecoderState state)
throws DecodeException {
- return new Binary(readValueAsBuffer(stream, state));
+ return new Binary(readValueAsBuffer(stream, state), false);
}
/**
@@ -129,7 +117,7 @@ public abstract class AbstractBinaryTypeDecoder extends
AbstractPrimitiveTypeDec
*/
@Override
public ProtonBuffer readValueAsBuffer(InputStream stream,
StreamDecoderState state) throws DecodeException {
- return
ProtonBufferAllocator.defaultAllocator().copy(readValueAsArray(stream,
state)).convertToReadOnly();
+ return ProtonByteArrayBufferAllocator.wrapped(readValueAsArray(stream,
state)).convertToReadOnly();
}
/**
diff --git a/protonj2/src/main/java/org/apache/qpid/protonj2/types/Binary.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/types/Binary.java
index d28d09ec..32ec2408 100644
--- a/protonj2/src/main/java/org/apache/qpid/protonj2/types/Binary.java
+++ b/protonj2/src/main/java/org/apache/qpid/protonj2/types/Binary.java
@@ -45,13 +45,26 @@ public final class Binary {
* @param buffer the {@link ProtonBuffer} to wrap or copy.
*/
public Binary(ProtonBuffer buffer) {
+ this(buffer, true);
+ }
+
+ /**
+ * Creates an {@link Binary} that wraps the given buffer or copies it
+ * if the given buffer is not read-only to preserves the immutable nature
+ * of this Binary instance. The caller can define if a cleaner is needed
+ * to release the buffer in cases where is not a heap backed instance.
+ *
+ * @param buffer the {@link ProtonBuffer} to wrap or copy.
+ * @param needsCleanup controls if the Binary will add the buffer to the
cleaner
+ */
+ public Binary(ProtonBuffer buffer, boolean needsCleanup) {
if (buffer != null && !buffer.isReadOnly()) {
this.buffer = buffer.copy(true);
} else {
this.buffer = buffer;
}
- if (buffer != null) {
+ if (buffer != null && needsCleanup) {
ProtonBufferUtils.registerCleanup(this, this.buffer);
}
}
diff --git
a/protonj2/src/test/java/org/apache/qpid/protonj2/codec/benchmark/Benchmark.java
b/protonj2/src/test/java/org/apache/qpid/protonj2/codec/benchmark/Benchmark.java
index 140b2186..c6407336 100644
---
a/protonj2/src/test/java/org/apache/qpid/protonj2/codec/benchmark/Benchmark.java
+++
b/protonj2/src/test/java/org/apache/qpid/protonj2/codec/benchmark/Benchmark.java
@@ -511,6 +511,7 @@ public class Benchmark implements Runnable {
private void benchmarkProperties() throws IOException {
Properties properties = new Properties();
+ properties.setUserId(new byte[] { 0, 10, 20, 30, 40, 50, 60, 70, 80,
90, 100 });
properties.setTo("queue:1-1024");
properties.setReplyTo("queue:1-11024-reply");
properties.setMessageId("ID:255f1297-5a71-4df1-8147-b2cdf850a56f:1");
diff --git
a/protonj2/src/test/java/org/apache/qpid/protonj2/codec/primitives/BinaryTypeCodecTest.java
b/protonj2/src/test/java/org/apache/qpid/protonj2/codec/primitives/BinaryTypeCodecTest.java
index d6ecec33..233b2a24 100644
---
a/protonj2/src/test/java/org/apache/qpid/protonj2/codec/primitives/BinaryTypeCodecTest.java
+++
b/protonj2/src/test/java/org/apache/qpid/protonj2/codec/primitives/BinaryTypeCodecTest.java
@@ -33,6 +33,7 @@ import java.util.Random;
import org.apache.qpid.protonj2.buffer.ProtonBuffer;
import org.apache.qpid.protonj2.buffer.ProtonBufferAllocator;
import org.apache.qpid.protonj2.buffer.ProtonBufferInputStream;
+import org.apache.qpid.protonj2.buffer.netty.Netty4ProtonBufferAllocator;
import org.apache.qpid.protonj2.codec.CodecTestSupport;
import org.apache.qpid.protonj2.codec.DecodeException;
import org.apache.qpid.protonj2.codec.EncodingCodes;
@@ -44,6 +45,10 @@ import org.apache.qpid.protonj2.types.Binary;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.util.ReferenceCountUtil;
+
/**
* Test the Binary codec for correctness
*/
@@ -627,4 +632,73 @@ public class BinaryTypeCodecTest extends CodecTestSupport {
StreamTypeDecoder<?> typeDecoder =
streamDecoder.readNextTypeDecoder(stream, streamDecoderState);
assertThrows(DecodeException.class, () ->
typeDecoder.skipValue(stream, streamDecoderState));
}
+
+ @Test
+ public void testDecodedBinaryDoesNotRetainPooledReadOnlyBuffer() throws
Exception {
+ doTestDecodedBinaryDoesNotRetainPooledReadOnlyBuffer(false, false);
+ }
+
+ @Test
+ public void testDecodedBinaryAsBufferDoesNotRetainPooledReadOnlyBuffer()
throws Exception {
+ doTestDecodedBinaryDoesNotRetainPooledReadOnlyBuffer(true, false);
+ }
+
+ @Test
+ public void testDecodedBinaryDoesNotRetainPooledReadOnlyBufferFS() throws
Exception {
+ doTestDecodedBinaryDoesNotRetainPooledReadOnlyBuffer(false, true);
+ }
+
+ @Test
+ public void testDecodedBinaryAsBufferDoesNotRetainPooledReadOnlyBufferFS()
throws Exception {
+ doTestDecodedBinaryDoesNotRetainPooledReadOnlyBuffer(true, true);
+ }
+
+ private void doTestDecodedBinaryDoesNotRetainPooledReadOnlyBuffer(boolean
asBuffer, boolean fromStream) throws Exception {
+ final byte[] input = new byte[512];
+
+ random.nextBytes(input);
+
+ final ProtonBuffer result;
+ final ByteBuf pooled;
+
+ try (ProtonBufferAllocator allocator = new
Netty4ProtonBufferAllocator(PooledByteBufAllocator.DEFAULT);
+ ProtonBuffer buffer = allocator.allocate(input.length + 10)) {
+
+ encoder.writeBinary(buffer, encoderState, input);
+ buffer.convertToReadOnly();
+
+ assertTrue(buffer.unwrap() instanceof ByteBuf);
+
+ pooled = (ByteBuf) buffer.unwrap();
+
+ assertEquals(1, ReferenceCountUtil.refCnt(pooled));
+
+ if (fromStream) {
+ try (InputStream stream = new ProtonBufferInputStream(buffer))
{
+ if (asBuffer) {
+ result = streamDecoder.readBinaryAsBuffer(stream,
streamDecoderState);
+ } else {
+ result = streamDecoder.readBinary(stream,
streamDecoderState).asProtonBuffer();
+ }
+ }
+ } else {
+ if (asBuffer) {
+ result = decoder.readBinaryAsBuffer(buffer, decoderState);
+ } else {
+ result = decoder.readBinary(buffer,
decoderState).asProtonBuffer();
+ }
+ }
+ }
+
+ assertEquals(0, ReferenceCountUtil.refCnt(pooled));
+ assertTrue(result.isReadOnly());
+ assertEquals(input.length, result.getReadableBytes());
+
+ final byte[] output = new byte[input.length];
+
+ result.readBytes(output, 0, output.length);
+
+ assertFalse(result.isReadable());
+ assertArrayEquals(input, output);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
