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 deb7b489b6fe3802d14b11c0dcb2a8c9401b7d6c Author: opencode <[email protected]> AuthorDate: Fri Oct 9 15:22:18 2026 +0200 Make ThroughputInterceptor statistics reporting thread safe The interceptor used a shared DecimalFormat instance field to format the periodic report. DecimalFormat is not thread safe and report() can be entered concurrently by sender threads (in-flight counter reaching zero) and receiver threads (message count crossing the interval), which could produce corrupted log output or throw exceptions from inside the shared formatter's internal state. Drop the shared formatter and create a local DecimalFormat per report call instead. The statistics fields mbTx, mbAppTx, mbRx, timeTx, lastCnt, txStart and rxStart are made volatile so that concurrent reads and writes are atomic and visible. No locking is added: the interceptor sits in the shared channel interceptor chain, so the message path must not pay for a contended monitor. The counters stay best effort and may lose updates under concurrency, which is acceptable for statistics. Add a concurrency stress test that drives sends, receives and reports from multiple threads and verifies no exception is raised and the in-flight counter returns to zero. --- .../group/interceptors/ThroughputInterceptor.java | 22 +++++++---- .../interceptors/TestThroughputInterceptor.java | 45 ++++++++++++++++++++++ 2 files changed, 59 insertions(+), 8 deletions(-) diff --git a/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java index ee3e9bfe6c..c97886dfae 100644 --- a/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java +++ b/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java @@ -49,19 +49,22 @@ public class ThroughputInterceptor extends ChannelInterceptorBase implements Thr */ protected static final StringManager sm = StringManager.getManager(ThroughputInterceptor.class); - double mbTx = 0; - double mbAppTx = 0; - double mbRx = 0; - double timeTx = 0; - double lastCnt = 0; + // Statistics are best effort: the increments below may lose updates when several + // threads update them at once. Volatile is used to keep individual reads and writes + // atomic and visible without adding any locking to the message path, which is + // shared by every interceptor in the channel. + volatile double mbTx = 0; + volatile double mbAppTx = 0; + volatile double mbRx = 0; + volatile double timeTx = 0; + volatile double lastCnt = 0; final AtomicLong msgTxCnt = new AtomicLong(1); final AtomicLong msgRxCnt = new AtomicLong(0); final AtomicLong msgTxErr = new AtomicLong(0); int interval = 10000; final AtomicInteger access = new AtomicInteger(0); - long txStart = 0; - long rxStart = 0; - final DecimalFormat df = new DecimalFormat("#0.00"); + volatile long txStart = 0; + volatile long rxStart = 0; @Override @@ -113,6 +116,9 @@ public class ThroughputInterceptor extends ChannelInterceptorBase implements Thr @Override public void report(double timeTx) { if (log.isInfoEnabled()) { + // Local formatter: DecimalFormat is not thread safe and this method can be + // entered concurrently by send and receive threads. + DecimalFormat df = new DecimalFormat("#0.00"); double txRate = timeTx > 0 ? mbTx / timeTx : 0; double appTxRate = timeTx > 0 ? mbAppTx / timeTx : 0; double rxElapsed = (System.currentTimeMillis() - rxStart) / 1000d; diff --git a/test/org/apache/catalina/tribes/group/interceptors/TestThroughputInterceptor.java b/test/org/apache/catalina/tribes/group/interceptors/TestThroughputInterceptor.java index 1605cc6357..1f6fd62836 100644 --- a/test/org/apache/catalina/tribes/group/interceptors/TestThroughputInterceptor.java +++ b/test/org/apache/catalina/tribes/group/interceptors/TestThroughputInterceptor.java @@ -62,6 +62,9 @@ public class TestThroughputInterceptor { } } + private static class PassThrough extends ChannelInterceptorBase { + } + private static ChannelData createMessage(Member source) { ChannelData data = new ChannelData(false); data.setUniqueId(UUIDGenerator.randomUUID(false)); @@ -144,4 +147,46 @@ public class TestThroughputInterceptor { Assert.assertEquals("A successful send must release the in-flight counter slot", 0, interceptor.access.get()); } + + @Test + public void testConcurrentSendReceiveAndReport() throws Exception { + ThroughputInterceptor interceptor = new ThroughputInterceptor(); + interceptor.setChannel(new GroupChannel()); + interceptor.setNext(new PassThrough()); + // Small interval so the periodic report paths inside sendMessage and + // messageReceived are exercised in addition to the direct calls below + interceptor.setInterval(10); + + MemberImpl member = new MemberImpl("localhost", 4000, -1); + int threadCount = 4; + int iterations = 250; + CountDownLatch done = new CountDownLatch(threadCount); + AtomicReference<Throwable> failure = new AtomicReference<>(); + + for (int i = 0; i < threadCount; i++) { + Thread thread = new Thread(() -> { + try { + for (int j = 0; j < iterations; j++) { + interceptor.sendMessage(new Member[] { member }, createMessage(member), null); + interceptor.messageReceived(createMessage(member)); + interceptor.report(interceptor.getTimeTx()); + } + } catch (Throwable x) { + failure.compareAndSet(null, x); + } finally { + done.countDown(); + } + }); + thread.start(); + } + + Assert.assertTrue("Traffic threads did not finish", done.await(30, TimeUnit.SECONDS)); + Assert.assertNull("Concurrent traffic and reporting must not throw: " + failure.get(), failure.get()); + Assert.assertEquals("Every completed send must release the in-flight counter slot", + 0, interceptor.access.get()); + Assert.assertEquals("All successful sends must be counted", + threadCount * iterations, interceptor.getMsgTxCnt().get() - 1); + Assert.assertEquals("All receives must be counted", + (long) threadCount * iterations, interceptor.getMsgRxCnt().get()); + } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
