This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 11.0.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 1966edc76e454935ed7b323fcc7e1f7e3be06540 Author: opencode <[email protected]> AuthorDate: Fri Oct 9 14:17:11 2026 +0200 Make FragmentationInterceptor reassembly thread safe The fragment storage of a FragCollection and the completion sequence in defrag() - add fragment, check for completion, remove the collection, assemble and deliver - were not synchronized. Fragments of the same message processed concurrently could both observe the collection as complete and deliver the assembled message twice, or assemble against a partially populated array. With the standard TCP topology the race is latent because all fragments of a message travel over one connection served by one worker, but the class provided no guarantee. Synchronise the fragment operations on the collection, add a one-shot assembleIfComplete() so completion is claimed exactly once, remove the collection from the map conditionally so a collection recreated after a heartbeat expiry is not evicted, and keep the deep clone of the incoming fragment and the delivery of the assembled message outside the monitor. --- .../interceptors/FragmentationInterceptor.java | 44 ++++++++++++--- .../interceptors/TestFragmentationInterceptor.java | 64 ++++++++++++++++++++++ 2 files changed, 100 insertions(+), 8 deletions(-) diff --git a/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java index 509942496b..51d79a91ff 100644 --- a/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java +++ b/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java @@ -141,11 +141,21 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements ChannelMessage complete = null; try { FragCollection coll = getFragCollection(key, msg); - coll.addMessage((ChannelMessage) msg.deepclone()); - - if (coll.complete()) { - removeFragCollection(key); - complete = coll.assemble(); + // The deep clone does not touch shared state, so it happens before the + // monitor is entered. The fragment storage of a collection is shared + // state. Synchronise on the collection so that adding a fragment and + // checking for completion form a single atomic sequence even when + // fragments arrive on multiple threads. + ChannelMessage stored = (ChannelMessage) msg.deepclone(); + synchronized (coll) { + coll.addMessage(stored); + complete = coll.assembleIfComplete(); + if (complete != null) { + // Remove only this collection. A heartbeat expiry may already + // have removed it and a late fragment may have installed a new + // collection under the same key, which must not be evicted. + fragpieces.remove(key, coll); + } } } catch (IllegalArgumentException | IllegalStateException e) { // Fragment metadata received as part of the message data failed validation @@ -256,6 +266,7 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements private final long received = System.currentTimeMillis(); private final ChannelMessage msg; private final XByteBuffer[] frags; + private boolean assembled = false; /** * Creates a new fragment collection for the given message. @@ -283,7 +294,7 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements * * @param msg The fragment message to add */ - public void addMessage(ChannelMessage msg) { + public synchronized void addMessage(ChannelMessage msg) { // At least the fragment number and the fragment count must be present. // Check before trimming, trimming more bytes than are available throws // an ArrayIndexOutOfBoundsException. @@ -311,7 +322,7 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements * * @return {@code true} if all fragments are present */ - public boolean complete() { + public synchronized boolean complete() { boolean result = true; for (int i = 0; (i < frags.length) && (result); i++) { result = (frags[i] != null); @@ -326,7 +337,7 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements * * @throws IllegalStateException if not all fragments have been received */ - public ChannelMessage assemble() { + public synchronized ChannelMessage assemble() { if (!complete()) { throw new IllegalStateException(sm.getString("fragmentationInterceptor.fragments.missing")); } @@ -346,6 +357,23 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements return msg; } + /** + * Assembles the complete message if all fragments have been received and this + * collection has not been assembled before. The one-shot guarantee makes the + * completion of a message delivered exactly once, even when the last fragments + * are processed concurrently on multiple threads. + * + * @return The assembled channel message, or <code>null</code> if the message is + * incomplete or was assembled by an earlier call + */ + public synchronized ChannelMessage assembleIfComplete() { + if (!assembled && complete()) { + assembled = true; + return assemble(); + } + return null; + } + /** * Checks if this fragment collection has expired. * diff --git a/test/org/apache/catalina/tribes/group/interceptors/TestFragmentationInterceptor.java b/test/org/apache/catalina/tribes/group/interceptors/TestFragmentationInterceptor.java index 6fa862d429..2f85b42e3e 100644 --- a/test/org/apache/catalina/tribes/group/interceptors/TestFragmentationInterceptor.java +++ b/test/org/apache/catalina/tribes/group/interceptors/TestFragmentationInterceptor.java @@ -20,6 +20,12 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.List; +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; @@ -143,6 +149,64 @@ public class TestFragmentationInterceptor { } } + @Test + public void testAssembleIfCompleteIsOneShotUnderConcurrency() throws Exception { + Member source = new MemberImpl("localhost", 4000, -1); + byte[] uniqueId = UUIDGenerator.randomUUID(false); + + XByteBuffer init = new XByteBuffer(16, false); + init.append("xx".getBytes(StandardCharsets.UTF_8), 0, 2); + init.append(2); + ChannelData initMessage = createMessage(new byte[0], source, uniqueId); + initMessage.setMessage(init); + FragmentationInterceptor.FragCollection coll = + new FragmentationInterceptor.FragCollection(initMessage); + + XByteBuffer fragData0 = new XByteBuffer(16, false); + fragData0.append("a".getBytes(StandardCharsets.UTF_8), 0, 1); + fragData0.append(0); + fragData0.append(2); + ChannelData frag0 = createMessage(new byte[0], source, uniqueId); + frag0.setMessage(fragData0); + + XByteBuffer fragData1 = new XByteBuffer(16, false); + fragData1.append("b".getBytes(StandardCharsets.UTF_8), 0, 1); + fragData1.append(1); + fragData1.append(2); + ChannelData frag1 = createMessage(new byte[0], source, uniqueId); + frag1.setMessage(fragData1); + + CyclicBarrier barrier = new CyclicBarrier(2); + AtomicInteger assembled = new AtomicInteger(); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + List<Future<?>> futures = new ArrayList<>(); + for (ChannelMessage frag : Arrays.asList(frag0, frag1)) { + futures.add(executor.submit(() -> { + coll.addMessage(frag); + try { + // Both threads have added their fragment once the barrier trips, + // so both observe the collection as complete + barrier.await(10, TimeUnit.SECONDS); + } catch (Exception e) { + throw new RuntimeException(e); + } + if (coll.assembleIfComplete() != null) { + assembled.incrementAndGet(); + } + })); + } + for (Future<?> future : futures) { + future.get(10, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + } + + Assert.assertEquals("A concurrently completed message must be assembled exactly once", + 1, assembled.get()); + } + @Test public void testFragmentRoundTrip() throws Exception { Member destination = new MemberImpl("localhost", 4000, -1); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
