This is an automated email from the ASF dual-hosted git repository.
markt-asf pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/11.0.x by this push:
new 4ba7294334 Review "Fix MessageDispatchInterceptor queue size
accounting drift"
4ba7294334 is described below
commit 4ba7294334589669d638afe197bcea4e39414250
Author: Mark Thomas <[email protected]>
AuthorDate: Wed Oct 21 12:03:01 2026 +0100
Review "Fix MessageDispatchInterceptor queue size accounting drift"
max queue size is now enforced
addToQueue increments queue size
---
.../interceptors/MessageDispatchInterceptor.java | 53 +++++--
.../TestMessageDispatchInterceptor.java | 173 +++++++++++++++++++--
2 files changed, 205 insertions(+), 21 deletions(-)
diff --git
a/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java
b/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java
index f214a38706..c189a541a7 100644
---
a/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java
+++
b/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java
@@ -103,14 +103,12 @@ public class MessageDispatchInterceptor extends
ChannelInterceptorBase implement
throws ChannelException {
boolean async = (msg.getOptions() & Channel.SEND_OPTIONS_ASYNCHRONOUS)
== Channel.SEND_OPTIONS_ASYNCHRONOUS;
if (async && run) {
- if ((getCurrentSize() + msg.getMessage().getLength()) >
maxQueueSize) {
- if (alwaysSend) {
- super.sendMessage(destination, msg, payload);
- return;
- } else {
- throw new
ChannelException(sm.getString("messageDispatchInterceptor.queue.full",
- Long.toString(maxQueueSize),
Long.toString(getCurrentSize())));
- }
+ ChannelMessage original = msg;
+ // Reject a message that cannot fit before paying for the clone.
This check is only an optimization, the
+ // atomic reservation below enforces the limit when senders race.
+ if (getCurrentSize() + msg.getMessage().getLength() >
maxQueueSize) {
+ rejectOrSend(destination, original, payload);
+ return;
}
// add to queue
if (useDeepClone) {
@@ -126,7 +124,10 @@ public class MessageDispatchInterceptor extends
ChannelInterceptorBase implement
// interceptors may replace or extend the message payload while
the message
// is being sent, so the accounting must use a single snapshot of
the length.
long queuedLength = msg.getMessage().getLength();
- addAndGetCurrentSize(queuedLength);
+ if (!reserveQueueSize(queuedLength)) {
+ rejectOrSend(destination, original, payload);
+ return;
+ }
if (!addToQueue(msg, destination, payload, queuedLength)) {
addAndGetCurrentSize(-queuedLength);
throw new
ChannelException(sm.getString("messageDispatchInterceptor.unableAdd.queue"));
@@ -136,16 +137,46 @@ public class MessageDispatchInterceptor extends
ChannelInterceptorBase implement
}
}
+ private void rejectOrSend(Member[] destination, ChannelMessage msg,
InterceptorPayload payload)
+ throws ChannelException {
+ if (alwaysSend) {
+ super.sendMessage(destination, msg, payload);
+ } else {
+ throw new
ChannelException(sm.getString("messageDispatchInterceptor.queue.full",
+ Long.toString(maxQueueSize),
Long.toString(getCurrentSize())));
+ }
+ }
+
+ private boolean reserveQueueSize(long length) {
+ long current;
+ long updated;
+ do {
+ current = currentSize.get();
+ updated = current + length;
+ if (updated < current || updated > maxQueueSize) {
+ return false;
+ }
+ } while (!currentSize.compareAndSet(current, updated));
+ return true;
+ }
+
/**
- * Add a message to the queue.
+ * Add a message to the queue. The length of the message is recorded in
the queue size until the message has been
+ * sent. The queue size limit is not applied.
* @param msg the message
* @param destination the destination
* @param payload the payload
* @return true if added
*/
public boolean addToQueue(final ChannelMessage msg, final Member[]
destination, final InterceptorPayload payload) {
- return addToQueue(msg, destination, payload,
msg.getMessage().getLength());
+ long queuedLength = msg.getMessage().getLength();
+ addAndGetCurrentSize(queuedLength);
+ if (addToQueue(msg, destination, payload, queuedLength)) {
+ return true;
+ }
+ addAndGetCurrentSize(-queuedLength);
+ return false;
}
diff --git
a/test/org/apache/catalina/tribes/group/interceptors/TestMessageDispatchInterceptor.java
b/test/org/apache/catalina/tribes/group/interceptors/TestMessageDispatchInterceptor.java
index dc823a34a2..206d90026d 100644
---
a/test/org/apache/catalina/tribes/group/interceptors/TestMessageDispatchInterceptor.java
+++
b/test/org/apache/catalina/tribes/group/interceptors/TestMessageDispatchInterceptor.java
@@ -19,7 +19,12 @@ package org.apache.catalina.tribes.group.interceptors;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Assert;
import org.junit.Test;
@@ -91,8 +96,7 @@ public class TestMessageDispatchInterceptor {
Member[] destination = new Member[] { new MemberImpl("localhost",
4000, -1) };
interceptor.sendMessage(destination, data, null);
- Assert.assertTrue("Dispatch thread did not start",
- collector.dispatchStarted.await(10, TimeUnit.SECONDS));
+ Assert.assertTrue("Dispatch thread did not start",
collector.dispatchStarted.await(10, TimeUnit.SECONDS));
// Simulate GroupChannel.send() returning the buffer to the pool
for reuse
buffer.clear();
buffer.append(reused, 0, reused.length);
@@ -100,10 +104,9 @@ public class TestMessageDispatchInterceptor {
Assert.assertTrue("Dispatch thread did not collect the message",
collector.collected.await(10, TimeUnit.SECONDS));
- Assert.assertNotSame("Queued message must not share the pooled
buffer",
- buffer, collector.capturedBuffer);
- Assert.assertArrayEquals("Queued message content was corrupted by
buffer reuse",
- original, collector.capturedData);
+ Assert.assertNotSame("Queued message must not share the pooled
buffer", buffer, collector.capturedBuffer);
+ Assert.assertArrayEquals("Queued message content was corrupted by
buffer reuse", original,
+ collector.capturedData);
} finally {
interceptor.stopQueue();
}
@@ -160,16 +163,166 @@ public class TestMessageDispatchInterceptor {
size = interceptor.getCurrentSize();
}
Assert.assertEquals(
- "Queue size accounting drifted when a downstream
interceptor changed the message length",
- 0, size);
+ "Queue size accounting drifted when a downstream
interceptor changed the message length", 0, size);
} finally {
interceptor.stopQueue();
}
}
+ @Test
+ public void testPublicAddToQueueAccountsForQueueSize() throws Exception {
+ GroupChannel channel = new GroupChannel();
+ MessageDispatchInterceptor interceptor = new
MessageDispatchInterceptor();
+ MutatingCollector collector = new MutatingCollector();
+ interceptor.setChannel(channel);
+ interceptor.setNext(collector);
+ interceptor.startQueue();
+ try {
+ Member destination = new MemberImpl("localhost", 4000, -1);
+ ChannelData data = new ChannelData(false);
+ data.setUniqueId(UUIDGenerator.randomUUID(false));
+ data.setAddress(destination);
+ data.setOptions(Channel.SEND_OPTIONS_ASYNCHRONOUS);
+ XByteBuffer buffer = new XByteBuffer(64, false);
+ byte[] payloadBytes = "PAYLOAD".getBytes(StandardCharsets.UTF_8);
+ buffer.append(payloadBytes, 0, payloadBytes.length);
+ data.setMessage(buffer);
+
+ Assert.assertTrue(interceptor.addToQueue(data, new Member[] {
destination }, null));
+
+ long deadline = System.currentTimeMillis() + 10000;
+ long size = interceptor.getCurrentSize();
+ while (size != 0 && System.currentTimeMillis() < deadline) {
+ Thread.sleep(50);
+ size = interceptor.getCurrentSize();
+ }
+ Assert.assertEquals("Queue size must return to zero, not drift
negative", 0, size);
+ } finally {
+ interceptor.stopQueue();
+ }
+ }
+
+ @Test
+ public void testPublicAddToQueueRollsBackWhenRejected() throws Exception {
+ GroupChannel channel = new GroupChannel();
+ MessageDispatchInterceptor interceptor = new
MessageDispatchInterceptor();
+ interceptor.setChannel(channel);
+ interceptor.setNext(new MutatingCollector());
+ interceptor.startQueue();
+ // A stopped queue rejects new tasks
+ interceptor.stopQueue();
+
+ Member destination = new MemberImpl("localhost", 4000, -1);
+ ChannelData data = new ChannelData(false);
+ XByteBuffer buffer = new XByteBuffer(64, false);
+ byte[] payloadBytes = "PAYLOAD".getBytes(StandardCharsets.UTF_8);
+ buffer.append(payloadBytes, 0, payloadBytes.length);
+ data.setMessage(buffer);
+
+ Assert.assertFalse(interceptor.addToQueue(data, new Member[] {
destination }, null));
+ Assert.assertEquals(0, interceptor.getCurrentSize());
+ }
+
+ @Test
+ public void testRejectedMessageIsNotCloned() throws Exception {
+ byte[] payload = "PAYLOAD".getBytes(StandardCharsets.UTF_8);
+ MessageDispatchInterceptor interceptor = new
MessageDispatchInterceptor();
+ interceptor.setAlwaysSend(false);
+ interceptor.setMaxQueueSize(payload.length - 1);
+ interceptor.setChannel(new GroupChannel());
+ interceptor.setNext(new MutatingCollector());
+ interceptor.startQueue();
+ try {
+ Member destination = new MemberImpl("localhost", 4000, -1);
+ AtomicInteger clones = new AtomicInteger();
+ ChannelData data = new ChannelData(false) {
+ @Override
+ public Object deepclone() {
+ clones.incrementAndGet();
+ return super.deepclone();
+ }
+ };
+ data.setUniqueId(UUIDGenerator.randomUUID(false));
+ data.setAddress(destination);
+ data.setOptions(Channel.SEND_OPTIONS_ASYNCHRONOUS);
+ data.setMessage(new XByteBuffer(payload, false));
+
+ try {
+ interceptor.sendMessage(new Member[] { destination }, data,
null);
+ Assert.fail("The message does not fit in the queue");
+ } catch (ChannelException expected) {
+ // Expected
+ }
+ Assert.assertEquals("A message that cannot fit must be rejected
before it is cloned", 0, clones.get());
+ Assert.assertEquals(0, interceptor.getCurrentSize());
+ } finally {
+ interceptor.stopQueue();
+ }
+ }
+
+ @Test
+ public void testConcurrentSendsCannotExceedMaxQueueSize() throws Exception
{
+ byte[] payload = "PAYLOAD".getBytes(StandardCharsets.UTF_8);
+ CyclicBarrier clonesReady = new CyclicBarrier(2);
+ MessageDispatchInterceptor interceptor = new
MessageDispatchInterceptor();
+ interceptor.setAlwaysSend(false);
+ interceptor.setMaxQueueSize(payload.length);
+ GatedCollector collector = new GatedCollector();
+ interceptor.setChannel(new GroupChannel());
+ interceptor.setNext(collector);
+ interceptor.startQueue();
+ ExecutorService senders = Executors.newFixedThreadPool(2);
+ try {
+ Member destination = new MemberImpl("localhost", 4000, -1);
+ Member[] destinations = new Member[] { destination };
+ Future<Boolean> first = senders.submit(() -> Boolean
+ .valueOf(send(interceptor, destinations,
createMessage(payload, destination, clonesReady))));
+ Future<Boolean> second = senders.submit(() -> Boolean
+ .valueOf(send(interceptor, destinations,
createMessage(payload, destination, clonesReady))));
+
+ Boolean firstSent = first.get(10, TimeUnit.SECONDS);
+ Boolean secondSent = second.get(10, TimeUnit.SECONDS);
+ Assert.assertNotEquals("Exactly one message must fit in the
queue", firstSent, secondSent);
+ Assert.assertEquals(payload.length, interceptor.getCurrentSize());
+ } finally {
+ collector.bufferReused.countDown();
+ senders.shutdownNow();
+ interceptor.stopQueue();
+ }
+ }
+
+ private static boolean send(MessageDispatchInterceptor interceptor,
Member[] destinations, ChannelData message) {
+ try {
+ interceptor.sendMessage(destinations, message, null);
+ return true;
+ } catch (ChannelException e) {
+ return false;
+ }
+ }
+
+ private static ChannelData createMessage(byte[] payload, Member
destination, CyclicBarrier clonesReady) {
+ ChannelData data = new ChannelData(false) {
+
+ @Override
+ public Object deepclone() {
+ try {
+ clonesReady.await(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ return super.deepclone();
+ }
+ };
+ data.setUniqueId(UUIDGenerator.randomUUID(false));
+ data.setAddress(destination);
+ data.setOptions(Channel.SEND_OPTIONS_ASYNCHRONOUS);
+ data.setMessage(new XByteBuffer(payload, false));
+ return data;
+ }
+
/*
- * Simulates interceptors such as EncryptInterceptor or
FragmentationInterceptor that replace or extend the
- * message payload while the message is being sent downstream.
+ * Simulates interceptors such as EncryptInterceptor or
FragmentationInterceptor that replace or extend the message
+ * payload while the message is being sent downstream.
*/
private static class MutatingCollector extends ChannelInterceptorBase {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
