This is an automated email from the ASF dual-hosted git repository.
tabish121 pushed a commit to branch 1.x
in repository https://gitbox.apache.org/repos/asf/qpid-protonj2.git
The following commit(s) were added to refs/heads/1.x by this push:
new 7f8ebc11 PROTON-3002 Add additional thread safety fixes to codec
implementations
7f8ebc11 is described below
commit 7f8ebc11872bdb2b7368ce17245422c751822b07
Author: Timothy Bish <[email protected]>
AuthorDate: Wed Nov 4 18:40:19 2026 -0500
PROTON-3002 Add additional thread safety fixes to codec implementations
Adds a few more missed thread safety fixes for the encoder and both
decoder implementations as well as some additional checks that ensure
descriptive errors are thrown in more cases.
(cherry picked from commit e035e68806296465da009f8083c82cd69fb506fb)
---
.../protonj2/codec/decoders/ProtonDecoder.java | 40 ++++++--
.../codec/decoders/ProtonStreamDecoder.java | 29 ++++--
.../protonj2/codec/encoders/ProtonEncoder.java | 5 +-
.../codec/UnknownDescribedTypeCodecTest.java | 66 +++++++++++++
.../protonj2/codec/decoders/ProtonDecoderTest.java | 10 ++
.../codec/decoders/ProtonStreamDecoderTest.java | 10 ++
.../codec/decoders/TestDescribedTypeDecoder.java | 102 +++++++++++++++++++++
7 files changed, 244 insertions(+), 18 deletions(-)
diff --git
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonDecoder.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonDecoder.java
index f5b0a475..c164fc31 100644
---
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonDecoder.java
+++
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonDecoder.java
@@ -21,8 +21,10 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import org.apache.qpid.protonj2.buffer.ProtonBuffer;
@@ -174,6 +176,7 @@ public class ProtonDecoder implements Decoder {
// Registry of decoders for described types which are not registered which
can be updated with a
// limited number of cached decoders to speed up processing
private final Map<Object, UnknownDescribedTypeDecoder>
unknownDescribedTypeDecoders = new ConcurrentHashMap<>();
+ private final AtomicInteger unknownDescribedTypeDecodersSlots = new
AtomicInteger(UNKNOWN_DESCRIBED_TYPES_CACHE_LIMIT);
// Quick access to decoders that handle AMQP types like Transfer,
Properties etc.
private final DescribedTypeDecoder<?>[] amqpTypeDecoders = new
DescribedTypeDecoder[256];
@@ -202,7 +205,7 @@ public class ProtonDecoder implements Decoder {
private final AbstractSymbolTypeDecoder localSymbol8Decoder;
private final AbstractSymbolTypeDecoder localSymbol32Decoder;
- private ProtonDecoderState singleThreadedState;
+ private volatile ProtonDecoderState singleThreadedState;
public ProtonDecoder() {
this(DecoderMode.DEFAULT, null);
@@ -241,6 +244,7 @@ public class ProtonDecoder implements Decoder {
@Override
public ProtonDecoderState getCachedDecoderState() {
ProtonDecoderState state = singleThreadedState;
+
if (state == null) {
singleThreadedState = state = newDecoderState();
}
@@ -337,8 +341,14 @@ public class ProtonDecoder implements Decoder {
state.decreaseDepth();
}
} else {
- throw new DecodeException(String.format(
- "Cannot decode a type that is using a reserved
type descriptor: %s", peekNextTypeDecoder(buffer, state).getTypeClass()));
+ final TypeDecoder<?> typeDecoder =
peekNextTypeDecoder(buffer, state);
+
+ if (typeDecoder == null) {
+ throw ex;
+ } else {
+ throw new DecodeException(String.format(
+ "Cannot decode a type that is using a reserved
type descriptor: %s", typeDecoder.getTypeClass()));
+ }
}
} else {
throw new DecodeException("Cannot decode reserved
descriptor type in SASL mode.");
@@ -346,6 +356,10 @@ public class ProtonDecoder implements Decoder {
}
}
+ if (descriptor == null) {
+ throw new DecodeException("Expected Descriptor type but read null
encoding");
+ }
+
TypeDecoder<?> typeDecoder = describedTypeDecoders.get(descriptor);
if (typeDecoder == null) {
typeDecoder = unknownDescribedTypeDecoders.get(descriptor);
@@ -369,15 +383,20 @@ public class ProtonDecoder implements Decoder {
@Override
public <V> ProtonDecoder
registerDescribedTypeDecoder(DescribedTypeDecoder<V> decoder) {
- DescribedTypeDecoder<?> describedTypeDecoder = decoder;
+ Objects.requireNonNull(decoder, "Stream described type decoder cannot
be null");
+
+ if (decoder.getDescriptorCode() == null ||
decoder.getDescriptorSymbol() == null) {
+ throw new IllegalArgumentException(
+ "DescribedTypeDecoder instance must return both a Symbol and
an UnsignedLong descriptor value");
+ }
// Cache AMQP type decoders in the quick lookup array.
if (decoder.getDescriptorCode().compareTo(amqpTypeDecoders.length) <
0) {
amqpTypeDecoders[decoder.getDescriptorCode().intValue()] = decoder;
}
- describedTypeDecoders.put(describedTypeDecoder.getDescriptorCode(),
describedTypeDecoder);
- describedTypeDecoders.put(describedTypeDecoder.getDescriptorSymbol(),
describedTypeDecoder);
+ describedTypeDecoders.put(decoder.getDescriptorCode(), decoder);
+ describedTypeDecoders.put(decoder.getDescriptorSymbol(), decoder);
decoder.decoderRegistered(this);
@@ -1033,8 +1052,13 @@ public class ProtonDecoder implements Decoder {
}
};
- if (canCache && unknownDescribedTypeDecoders.size() <
UNKNOWN_DESCRIBED_TYPES_CACHE_LIMIT) {
- unknownDescribedTypeDecoders.putIfAbsent(descriptor, typeDecoder);
+ if (canCache &&
unknownDescribedTypeDecodersSlots.getAndUpdate(available -> available > 0 ?
available - 1 : available) > 0) {
+ if (unknownDescribedTypeDecodersSlots.getAndDecrement() > 0) {
+ final UnknownDescribedTypeDecoder previous =
unknownDescribedTypeDecoders.putIfAbsent(descriptor, typeDecoder);
+ if (previous != null) {
+ unknownDescribedTypeDecodersSlots.incrementAndGet();
+ }
+ }
}
return typeDecoder;
diff --git
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonStreamDecoder.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonStreamDecoder.java
index 7dcc05fe..7fbcc283 100644
---
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonStreamDecoder.java
+++
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonStreamDecoder.java
@@ -21,11 +21,12 @@ import java.io.InputStream;
import java.lang.reflect.Array;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.qpid.protonj2.buffer.ProtonBuffer;
import org.apache.qpid.protonj2.codec.DecodeException;
@@ -172,16 +173,17 @@ public final class ProtonStreamDecoder implements
StreamDecoder {
// Registry of decoders for described types which can be updated with user
defined
// decoders as well as the default decoders.
- private final Map<Object, StreamDescribedTypeDecoder<?>>
describedTypeDecoders = new HashMap<>();
+ private final Map<Object, StreamDescribedTypeDecoder<?>>
describedTypeDecoders = new ConcurrentHashMap<>();
// Registry of decoders for described types which are not registered which
can be updated with a
// limited number of cached decoders to speed up processing
private final Map<Object, UnknownDescribedTypeDecoder>
unknownDescribedTypeDecoders = new ConcurrentHashMap<>();
+ private final AtomicInteger unknownDescribedTypeDecodersSlots = new
AtomicInteger(UNKNOWN_DESCRIBED_TYPES_CACHE_LIMIT);
// Quick access to decoders that handle AMQP types like Transfer,
Properties etc.
private final StreamDescribedTypeDecoder<?>[] amqpTypeDecoders = new
StreamDescribedTypeDecoder[256];
- private ProtonStreamDecoderState singleThreadedState;
+ private volatile ProtonStreamDecoderState singleThreadedState;
// Internal Decoders used to prevent user to access Proton specific
decoding methods
private static final Symbol8TypeDecoder symbol8Decoder;
@@ -230,6 +232,7 @@ public final class ProtonStreamDecoder implements
StreamDecoder {
@Override
public ProtonStreamDecoderState getCachedDecoderState() {
ProtonStreamDecoderState state = singleThreadedState;
+
if (state == null) {
singleThreadedState = state = newDecoderState();
}
@@ -362,15 +365,20 @@ public final class ProtonStreamDecoder implements
StreamDecoder {
@Override
public <V> ProtonStreamDecoder
registerDescribedTypeDecoder(StreamDescribedTypeDecoder<V> decoder) {
- StreamDescribedTypeDecoder<?> describedTypeDecoder = decoder;
+ Objects.requireNonNull(decoder, "Stream described type decoder cannot
be null");
+
+ if (decoder.getDescriptorCode() == null ||
decoder.getDescriptorSymbol() == null) {
+ throw new IllegalArgumentException(
+ "DescribedTypeDecoder instance must return both a Symbol and
an UnsignedLong descriptor value");
+ }
// Cache AMQP type decoders in the quick lookup array.
if (decoder.getDescriptorCode().compareTo(amqpTypeDecoders.length) <
0) {
amqpTypeDecoders[decoder.getDescriptorCode().intValue()] = decoder;
}
- describedTypeDecoders.put(describedTypeDecoder.getDescriptorCode(),
describedTypeDecoder);
- describedTypeDecoders.put(describedTypeDecoder.getDescriptorSymbol(),
describedTypeDecoder);
+ describedTypeDecoders.put(decoder.getDescriptorCode(), decoder);
+ describedTypeDecoders.put(decoder.getDescriptorSymbol(), decoder);
return this;
}
@@ -1016,8 +1024,13 @@ public final class ProtonStreamDecoder implements
StreamDecoder {
}
};
- if (canCache && unknownDescribedTypeDecoders.size() <
UNKNOWN_DESCRIBED_TYPES_CACHE_LIMIT) {
- unknownDescribedTypeDecoders.putIfAbsent(descriptor,
streamTypeDecoder);
+ if (canCache && unknownDescribedTypeDecodersSlots.get() > 0) {
+ if (canCache &&
unknownDescribedTypeDecodersSlots.getAndUpdate(available -> available > 0 ?
available - 1 : available) > 0) {
+ final UnknownDescribedTypeDecoder previous =
unknownDescribedTypeDecoders.putIfAbsent(descriptor, streamTypeDecoder);
+ if (previous != null) {
+ unknownDescribedTypeDecodersSlots.incrementAndGet();
+ }
+ }
}
return streamTypeDecoder;
diff --git
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/encoders/ProtonEncoder.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/encoders/ProtonEncoder.java
index 8fecf306..9f8ff698 100644
---
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/encoders/ProtonEncoder.java
+++
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/encoders/ProtonEncoder.java
@@ -106,7 +106,7 @@ public final class ProtonEncoder implements Encoder {
private static final UnsignedLongTypeEncoder ulongEncoder =
UnsignedLongTypeEncoder.INSTANCE;
private static final DeliveryTagEncoder deliveryTagEncoder =
DeliveryTagEncoder.INSTANCE;
- private ProtonEncoderState singleThreadedState;
+ private volatile ProtonEncoderState singleThreadedState;
private final Map<Class<?>, TypeEncoder<?>> typeEncoders = new
ConcurrentHashMap<>();
{
@@ -160,11 +160,12 @@ public final class ProtonEncoder implements Encoder {
@Override
public ProtonEncoderState getCachedEncoderState() {
ProtonEncoderState state = singleThreadedState;
+
if (state == null) {
singleThreadedState = state = newEncoderState();
}
- return singleThreadedState.reset();
+ return state.reset();
}
@Override
diff --git
a/protonj2/src/test/java/org/apache/qpid/protonj2/codec/UnknownDescribedTypeCodecTest.java
b/protonj2/src/test/java/org/apache/qpid/protonj2/codec/UnknownDescribedTypeCodecTest.java
index 9717c163..5b892236 100644
---
a/protonj2/src/test/java/org/apache/qpid/protonj2/codec/UnknownDescribedTypeCodecTest.java
+++
b/protonj2/src/test/java/org/apache/qpid/protonj2/codec/UnknownDescribedTypeCodecTest.java
@@ -650,4 +650,70 @@ public class UnknownDescribedTypeCodecTest extends
CodecTestSupport {
assertThrows(DecodeException.class, () ->
decoder.readObject(buffer.copy(true), decoderState));
}
}
+
+ @Test
+ public void testDecodeExceptionThrownForInvalidDescribedTypeDescriptor()
throws IOException {
+ doTestDecodeExceptionThrownForInvliadDescribefTypeDescriptor(false);
+ }
+
+ @Test
+ public void testDecodeExceptionThrownForInvalidDescribedTypeDescriptorFS()
throws IOException {
+ doTestDecodeExceptionThrownForInvliadDescribefTypeDescriptor(true);
+ }
+
+ private void
doTestDecodeExceptionThrownForInvliadDescribefTypeDescriptor(boolean
fromStream) throws IOException {
+ ProtonBuffer buffer =
ProtonBufferAllocator.defaultAllocator().allocate();
+
+ buffer.writeByte((byte) 0); // Described Type Indicator
+ buffer.writeByte(EncodingCodes.SMALLINT);
+ buffer.writeByte(NoLocalType.DESCRIPTOR_CODE.byteValue());
+ buffer.writeByte(EncodingCodes.LIST0);
+
+ if (fromStream) {
+ InputStream stream = new
ProtonBufferInputStream(buffer.copy(true));
+ assertThrows(DecodeException.class, () ->
streamDecoder.readObject(stream, streamDecoderState));
+ } else {
+ assertThrows(DecodeException.class, () ->
decoder.readObject(buffer.copy(true), decoderState));
+ }
+
+ if (fromStream) {
+ InputStream stream = new
ProtonBufferInputStream(buffer.copy(true));
+ assertThrows(DecodeException.class, () ->
streamDecoder.readNextTypeDecoder(stream, streamDecoderState));
+ } else {
+ assertThrows(DecodeException.class, () ->
decoder.readNextTypeDecoder(buffer.copy(true), decoderState));
+ }
+ }
+
+ @Test
+ public void
testDecodeExceptionThrownForInvalidNullDescribedTypeDescriptor() throws
IOException {
+
doTestDecodeExceptionThrownForInvalidNullDescribedTypeDescriptor(false);
+ }
+
+ @Test
+ public void
testDecodeExceptionThrownForInvalidNulDescribedTypeDescriptorFS() throws
IOException {
+ doTestDecodeExceptionThrownForInvalidNullDescribedTypeDescriptor(true);
+ }
+
+ private void
doTestDecodeExceptionThrownForInvalidNullDescribedTypeDescriptor(boolean
fromStream) throws IOException {
+ ProtonBuffer buffer =
ProtonBufferAllocator.defaultAllocator().allocate();
+
+ buffer.writeByte((byte) 0); // Described Type Indicator
+ buffer.writeByte(EncodingCodes.NULL);
+ buffer.writeByte(NoLocalType.DESCRIPTOR_CODE.byteValue());
+ buffer.writeByte(EncodingCodes.LIST0);
+
+ if (fromStream) {
+ InputStream stream = new
ProtonBufferInputStream(buffer.copy(true));
+ assertThrows(DecodeException.class, () ->
streamDecoder.readObject(stream, streamDecoderState));
+ } else {
+ assertThrows(DecodeException.class, () ->
decoder.readObject(buffer.copy(true), decoderState));
+ }
+
+ if (fromStream) {
+ InputStream stream = new
ProtonBufferInputStream(buffer.copy(true));
+ assertThrows(DecodeException.class, () ->
streamDecoder.readNextTypeDecoder(stream, streamDecoderState));
+ } else {
+ assertThrows(DecodeException.class, () ->
decoder.readNextTypeDecoder(buffer.copy(true), decoderState));
+ }
+ }
}
diff --git
a/protonj2/src/test/java/org/apache/qpid/protonj2/codec/decoders/ProtonDecoderTest.java
b/protonj2/src/test/java/org/apache/qpid/protonj2/codec/decoders/ProtonDecoderTest.java
index a965ea7e..83460328 100644
---
a/protonj2/src/test/java/org/apache/qpid/protonj2/codec/decoders/ProtonDecoderTest.java
+++
b/protonj2/src/test/java/org/apache/qpid/protonj2/codec/decoders/ProtonDecoderTest.java
@@ -52,6 +52,16 @@ public class ProtonDecoderTest extends CodecTestSupport {
assertSame(first, decoder.getCachedDecoderState());
}
+ @Test
+ public void
testDescriptedTypeDecoderRegistrationChecksForNullDescriptors() {
+ assertThrows(IllegalArgumentException.class,
+ () -> decoder.registerDescribedTypeDecoder(new
TestDescribedTypeDecoder(null, null)));
+ assertThrows(IllegalArgumentException.class,
+ () -> decoder.registerDescribedTypeDecoder(new
TestDescribedTypeDecoder(null, new UnsignedLong(0))));
+ assertThrows(IllegalArgumentException.class,
+ () -> decoder.registerDescribedTypeDecoder(new
TestDescribedTypeDecoder(Symbol.getSymbol("test"), null)));
+ }
+
@Test
public void testReadNullFromReadObjectForNullEncoding() throws IOException
{
ProtonBuffer buffer =
ProtonBufferAllocator.defaultAllocator().allocate();
diff --git
a/protonj2/src/test/java/org/apache/qpid/protonj2/codec/decoders/ProtonStreamDecoderTest.java
b/protonj2/src/test/java/org/apache/qpid/protonj2/codec/decoders/ProtonStreamDecoderTest.java
index c5539aa1..2154d568 100644
---
a/protonj2/src/test/java/org/apache/qpid/protonj2/codec/decoders/ProtonStreamDecoderTest.java
+++
b/protonj2/src/test/java/org/apache/qpid/protonj2/codec/decoders/ProtonStreamDecoderTest.java
@@ -55,6 +55,16 @@ public class ProtonStreamDecoderTest extends
CodecTestSupport {
assertSame(first, streamDecoder.getCachedDecoderState());
}
+ @Test
+ public void
testDescriptedTypeDecoderRegistrationChecksForNullDescriptors() {
+ assertThrows(IllegalArgumentException.class,
+ () -> streamDecoder.registerDescribedTypeDecoder(new
TestDescribedTypeDecoder(null, null)));
+ assertThrows(IllegalArgumentException.class,
+ () -> streamDecoder.registerDescribedTypeDecoder(new
TestDescribedTypeDecoder(null, new UnsignedLong(0))));
+ assertThrows(IllegalArgumentException.class,
+ () -> streamDecoder.registerDescribedTypeDecoder(new
TestDescribedTypeDecoder(Symbol.getSymbol("test"), null)));
+ }
+
@Test
public void testReadNullFromReadObjectForNullEncoding() throws IOException
{
ProtonBuffer buffer =
ProtonBufferAllocator.defaultAllocator().allocate();
diff --git
a/protonj2/src/test/java/org/apache/qpid/protonj2/codec/decoders/TestDescribedTypeDecoder.java
b/protonj2/src/test/java/org/apache/qpid/protonj2/codec/decoders/TestDescribedTypeDecoder.java
new file mode 100644
index 00000000..b812896b
--- /dev/null
+++
b/protonj2/src/test/java/org/apache/qpid/protonj2/codec/decoders/TestDescribedTypeDecoder.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.qpid.protonj2.codec.decoders;
+
+import java.io.InputStream;
+
+import org.apache.qpid.protonj2.buffer.ProtonBuffer;
+import org.apache.qpid.protonj2.codec.DecodeException;
+import org.apache.qpid.protonj2.codec.DecoderState;
+import org.apache.qpid.protonj2.codec.StreamDecoderState;
+import org.apache.qpid.protonj2.types.Symbol;
+import org.apache.qpid.protonj2.types.UnsignedLong;
+
+public final class TestDescribedTypeDecoder extends
AbstractDescribedTypeDecoder<String> {
+
+ private final Symbol symbolDescriptor;
+ private final UnsignedLong ulongDescriptor;
+
+ public TestDescribedTypeDecoder(Symbol symbol, UnsignedLong ulong) {
+ this.symbolDescriptor = symbol;
+ this.ulongDescriptor = ulong;
+ }
+
+ @Override
+ public String readValue(InputStream stream, StreamDecoderState state)
throws DecodeException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String readValue(ProtonBuffer buffer, DecoderState state) throws
DecodeException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void skipValue(InputStream stream, StreamDecoderState state) throws
DecodeException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void skipValue(ProtonBuffer buffer, DecoderState state) throws
DecodeException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String[] readArrayElements(InputStream stream, StreamDecoderState
state, int count) throws DecodeException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String[] readArrayElements(ProtonBuffer buffer, DecoderState state,
int count) throws DecodeException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int readSize(ProtonBuffer buffer, DecoderState state) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isPrimitive() {
+ return false;
+ }
+
+ @Override
+ public boolean isNull() {
+ return false;
+ }
+
+ @Override
+ public boolean isArrayType() {
+ return false;
+ }
+
+ @Override
+ public Class<String> getTypeClass() {
+ return String.class;
+ }
+
+ @Override
+ public Symbol getDescriptorSymbol() {
+ return symbolDescriptor;
+ }
+
+ @Override
+ public UnsignedLong getDescriptorCode() {
+ return ulongDescriptor;
+ }
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
