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 52a7db6b79183b0e164654c43cef3a48fc53d75b Author: opencode <[email protected]> AuthorDate: Fri Oct 9 14:54:26 2026 +0200 Fix in-flight counter leak in ThroughputInterceptor after send failure The send path reports throughput when the in-flight access counter returns to zero. On a ChannelException the counter was only decremented when an unsynchronized check saw it at exactly one. If another send was still in flight, the decrement was skipped and the counter leaked: it never returned to zero again, so the send-side report and the txStart refresh silently stopped for the life of the channel. The check was racy the other way as well, two failing sends could both see one and double-decrement. Decrement the counter exactly once on the failure path, matching the success path, before rethrowing. The error accounting itself is unchanged: failed sends count towards msgTxErr only. --- .../group/interceptors/ThroughputInterceptor.java | 8 +- .../interceptors/TestThroughputInterceptor.java | 147 +++++++++++++++++++++ 2 files changed, 152 insertions(+), 3 deletions(-) diff --git a/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java index f063b0d0be..ee3e9bfe6c 100644 --- a/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java +++ b/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java @@ -75,9 +75,11 @@ public class ThroughputInterceptor extends ChannelInterceptorBase implements Thr super.sendMessage(destination, msg, payload); } catch (ChannelException x) { msgTxErr.addAndGet(1); - if (access.get() == 1) { - access.addAndGet(-1); - } + // Release the in-flight slot exactly once. The previous check-then-act on + // access could skip the decrement when another send was still in flight, + // permanently keeping the counter above zero and stopping the reports, or + // let two failing sends decrement the same slot. + access.addAndGet(-1); throw x; } mbTx += (bytes * destination.length) / (1024d * 1024d); diff --git a/test/org/apache/catalina/tribes/group/interceptors/TestThroughputInterceptor.java b/test/org/apache/catalina/tribes/group/interceptors/TestThroughputInterceptor.java new file mode 100644 index 0000000000..1605cc6357 --- /dev/null +++ b/test/org/apache/catalina/tribes/group/interceptors/TestThroughputInterceptor.java @@ -0,0 +1,147 @@ +/* + * 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.catalina.tribes.group.interceptors; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Assert; +import org.junit.Test; + +import org.apache.catalina.tribes.ChannelException; +import org.apache.catalina.tribes.ChannelMessage; +import org.apache.catalina.tribes.Member; +import org.apache.catalina.tribes.group.ChannelInterceptorBase; +import org.apache.catalina.tribes.group.GroupChannel; +import org.apache.catalina.tribes.group.InterceptorPayload; +import org.apache.catalina.tribes.io.ChannelData; +import org.apache.catalina.tribes.io.XByteBuffer; +import org.apache.catalina.tribes.membership.MemberImpl; +import org.apache.catalina.tribes.util.UUIDGenerator; + +public class TestThroughputInterceptor { + + /* + * Chain end that either blocks until released or fails the send, so that a send can + * be made to fail deterministically while another send is still in flight. + */ + private static class GatedFailureChain extends ChannelInterceptorBase { + + private final CountDownLatch slowStarted = new CountDownLatch(1); + private final CountDownLatch releaseSlow = new CountDownLatch(1); + + private volatile boolean failNext = false; + + @Override + public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) + throws ChannelException { + if (failNext) { + throw new ChannelException("Induced send failure"); + } + slowStarted.countDown(); + try { + releaseSlow.await(10, TimeUnit.SECONDS); + } catch (InterruptedException x) { + Thread.currentThread().interrupt(); + } + } + } + + private static ChannelData createMessage(Member source) { + ChannelData data = new ChannelData(false); + data.setUniqueId(UUIDGenerator.randomUUID(false)); + data.setAddress(source); + data.setMessage(new XByteBuffer(64, false)); + return data; + } + + @Test + public void testConcurrentSendFailureReleasesInFlightSlot() throws Exception { + ThroughputInterceptor interceptor = new ThroughputInterceptor(); + GatedFailureChain chain = new GatedFailureChain(); + interceptor.setChannel(new GroupChannel()); + interceptor.setNext(chain); + + MemberImpl member = new MemberImpl("localhost", 4000, -1); + AtomicReference<Throwable> slowFailure = new AtomicReference<>(); + AtomicReference<Throwable> failingOutcome = new AtomicReference<>(); + + Thread slowSender = new Thread(() -> { + try { + interceptor.sendMessage(new Member[] { member }, createMessage(member), null); + } catch (ChannelException x) { + slowFailure.set(x); + } + }); + slowSender.start(); + Assert.assertTrue("Blocking send did not start", chain.slowStarted.await(10, TimeUnit.SECONDS)); + + // A second send fails while the first one is still in flight + Thread failingSender = new Thread(() -> { + chain.failNext = true; + try { + interceptor.sendMessage(new Member[] { member }, createMessage(member), null); + failingOutcome.set(new IllegalStateException("The second send should have failed")); + } catch (ChannelException expected) { + // Expected + } + }); + failingSender.start(); + failingSender.join(10000); + Assert.assertFalse("Failing send thread did not finish", failingSender.isAlive()); + + chain.releaseSlow.countDown(); + slowSender.join(10000); + Assert.assertFalse("Slow send thread did not finish", slowSender.isAlive()); + + Assert.assertNull("The slow send should not have failed", slowFailure.get()); + Assert.assertNull("The failing send should have thrown ChannelException", failingOutcome.get()); + Assert.assertEquals("A failed send must release the in-flight counter slot", + 0, interceptor.access.get()); + Assert.assertEquals("The failed send must be counted as an error", + 1, interceptor.getMsgTxErr().get()); + } + + @Test + public void testSuccessfulSendReleasesInFlightSlot() throws Exception { + ThroughputInterceptor interceptor = new ThroughputInterceptor(); + GatedFailureChain chain = new GatedFailureChain(); + interceptor.setChannel(new GroupChannel()); + interceptor.setNext(chain); + + MemberImpl member = new MemberImpl("localhost", 4000, -1); + AtomicReference<Throwable> sendFailure = new AtomicReference<>(); + + Thread sender = new Thread(() -> { + try { + interceptor.sendMessage(new Member[] { member }, createMessage(member), null); + } catch (ChannelException x) { + sendFailure.set(x); + } + }); + sender.start(); + Assert.assertTrue("Blocking send did not start", chain.slowStarted.await(10, TimeUnit.SECONDS)); + chain.releaseSlow.countDown(); + sender.join(10000); + Assert.assertFalse("Send thread did not finish", sender.isAlive()); + + Assert.assertNull("The send should not have failed", sendFailure.get()); + Assert.assertEquals("A successful send must release the in-flight counter slot", + 0, interceptor.access.get()); + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
