You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@tomcat.apache.org by ma...@apache.org on 2026/10/21 13:27:02 UTC
(tomcat) branch main updated: Review "Fix in-flight counter leak in ThroughputInterceptor..."
This is an automated email from the ASF dual-hosted git repository.
markt-asf pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/main by this push:
new 0d33ddff82 Review "Fix in-flight counter leak in ThroughputInterceptor..."
0d33ddff82 is described below
commit 0d33ddff821f16c86a7f0f290f43f1399597e2b0
Author: Mark Thomas <ma...@apache.org>
AuthorDate: Wed Oct 21 14:26:28 2026 +0100
Review "Fix in-flight counter leak in ThroughputInterceptor..."
Additional robustness improvements
Co-authored by: CoPilot / GPT-5.6 Sol
---
.../group/interceptors/ThroughputInterceptor.java | 29 +++++++++++-----------
.../interceptors/TestThroughputInterceptor.java | 21 ++++++++++++++++
2 files changed, 35 insertions(+), 15 deletions(-)
diff --git a/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java
index c97886dfae..d6d87507fb 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java
@@ -73,26 +73,25 @@ public class ThroughputInterceptor extends ChannelInterceptorBase implements Thr
if (access.addAndGet(1) == 1) {
txStart = System.currentTimeMillis();
}
- long bytes = XByteBuffer.getDataPackageLength(((ChannelData) msg).getDataPackageLength());
+ boolean completed = false;
try {
+ long bytes = XByteBuffer.getDataPackageLength(((ChannelData) msg).getDataPackageLength());
super.sendMessage(destination, msg, payload);
+ mbTx += (bytes * destination.length) / (1024d * 1024d);
+ mbAppTx += bytes / (1024d * 1024d);
+ completed = true;
} catch (ChannelException x) {
msgTxErr.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);
- mbAppTx += bytes / (1024d * 1024d);
- if (access.addAndGet(-1) == 0) {
- long stop = System.currentTimeMillis();
- timeTx += (stop - txStart) / 1000d;
- if ((msgTxCnt.get() / (double) interval) >= lastCnt) {
- lastCnt++;
- report(timeTx);
+ } finally {
+ // Release the in-flight slot exactly once for every exit path.
+ if (access.addAndGet(-1) == 0 && completed) {
+ long stop = System.currentTimeMillis();
+ timeTx += (stop - txStart) / 1000d;
+ if ((msgTxCnt.get() / (double) interval) >= lastCnt) {
+ lastCnt++;
+ report(timeTx);
+ }
}
}
msgTxCnt.addAndGet(1);
diff --git a/test/org/apache/catalina/tribes/group/interceptors/TestThroughputInterceptor.java b/test/org/apache/catalina/tribes/group/interceptors/TestThroughputInterceptor.java
index 1f6fd62836..b26e649d66 100644
--- a/test/org/apache/catalina/tribes/group/interceptors/TestThroughputInterceptor.java
+++ b/test/org/apache/catalina/tribes/group/interceptors/TestThroughputInterceptor.java
@@ -148,6 +148,27 @@ public class TestThroughputInterceptor {
0, interceptor.access.get());
}
+ @Test
+ public void testPackageLengthFailureReleasesInFlightSlot() throws Exception {
+ ThroughputInterceptor interceptor = new ThroughputInterceptor();
+ interceptor.setChannel(new GroupChannel());
+ interceptor.setNext(new ChannelInterceptorBase() {
+ // NO-OP
+ });
+
+ ChannelData invalid = new ChannelData(false);
+ invalid.setMessage(new XByteBuffer(64, false));
+ try {
+ interceptor.sendMessage(new Member[0], invalid, null);
+ Assert.fail("A message without a unique ID cannot be packaged");
+ } catch (NullPointerException expected) {
+ // Expected
+ }
+
+ Assert.assertEquals("A package length failure must release the in-flight counter slot",
+ 0, interceptor.access.get());
+ }
+
@Test
public void testConcurrentSendReceiveAndReport() throws Exception {
ThroughputInterceptor interceptor = new ThroughputInterceptor();
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org