This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 10.1.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 530225cfc6d7bba42c9970903b6bfa3163003eba Author: opencode <[email protected]> AuthorDate: Sat Oct 31 10:04:48 2026 +0100 Accept or refuse a pool notification as a whole The pool MBean's notify() handed one notification over in more than one submission to the bounded delivery queue: the JDK broadcaster's sendNotification puts one task per selected JMX listener, its loop unprotected against the queue refusing one of them, and the local listeners followed as a separate task of their own. Once saturation filled the queue partway through a notification, the later submissions of that same notification were refused while its earlier ones stood: notify() logged the notification dropped and answered false although it had reached some listeners, the local listeners never got their task, and a listener registered on both sides could be served by one half and skipped by the other. Hand the notification over as exactly one task, making the delivery queue the single gate that accepts or refuses it whole: inside the task, the JMX listeners are reached first and the local ones next, and an accepted notification can no longer be refused again on the way. The broadcaster now runs the per-listener hand-offs it builds inline rather than queueing them behind the task already running them: they could only ever run on that one thread anyway, and queuing them re-added one task per listener to the very queue whose bound split the notification apart. A refusal now means nothing was scheduled - no listener will receive the notification the log says was dropped - and an acceptance means every selected listener is reached once the queue drains, which the rewritten saturation test audits listener by listener. Nothing of the delivery, not even a JMX listener's filter, runs on the thread that reports the notification. --- .../tomcat/jdbc/pool/jmx/ConnectionPool.java | 72 ++++-- .../jdbc/pool/jmx/TestJmxNotificationDelivery.java | 243 +++++++++++++++++++++ 2 files changed, 295 insertions(+), 20 deletions(-) diff --git a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/jmx/ConnectionPool.java b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/jmx/ConnectionPool.java index 41517fedfb..07ade4194d 100644 --- a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/jmx/ConnectionPool.java +++ b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/jmx/ConnectionPool.java @@ -100,9 +100,18 @@ public class ConnectionPool extends NotificationBroadcasterSupport implements Co * @param pool the connection pool to wrap */ public ConnectionPool(org.apache.tomcat.jdbc.pool.ConnectionPool pool) { - // Hand the JMX listener delivery to the notifier thread as well; - // see notify(String,String) - super(NOTIFIER); + // Run the per-listener hand-offs that sendNotification builds on the + // thread that calls sendNotification instead of queueing one task per + // listener: notify() hands a notification to the notifier as a single + // task and calls sendNotification on that thread, so a queued + // hand-off would in the end only ever run on the very thread calling + // it - and a delivery queue that stayed full could refuse some of + // those re-queued tasks after the queue had accepted the notification + // as a whole, splitting one notification across its own listeners. + // Run inline, the bounded queue sees one task per notification and + // accepts or refuses the notification as a whole - see + // NOTIFY_QUEUE_CAPACITY and notify(String,String) + super(command -> command.run()); this.pool = pool; } @@ -197,32 +206,55 @@ public class ConnectionPool extends NotificationBroadcasterSupport implements Co /** * Reports a condition to the pool's notification listeners. * <p> - * The delivery is asynchronous: both the listeners reached through JMX - delivered by - * {@link #sendNotification(Notification)} - and the ones registered locally with - * {@link #addListener(NotificationListener)} receive the notification on the pool's notification thread, not on the - * thread that calls this method. The pool raises several of its notifications from the connection lifecycle, some - * of them while holding the lock of the connection concerned, and a monitoring listener that is slow or blocked - * must not reach those paths. The notification itself is still built on the calling thread, so it carries the time, - * sequence number and serializable source of the condition it reports - see {@link #getNotificationSource()} - and, - * since the notifier is a single thread, listeners observe notifications in the order the pool reported them. + * The delivery is asynchronous: the notification is handed to the pool's notification thread as one task, and that + * task delivers it - first through {@link #sendNotification(Notification)} to the listeners reached through JMX, + * then to the ones registered locally with {@link #addListener(NotificationListener)}. Nothing of the delivery runs + * on the thread that calls this method, not even the JMX listeners' filters: the pool raises several of its + * notifications from the connection lifecycle, some of them while holding the lock of the connection concerned, + * and a monitoring listener - or a filter - that is slow or blocked must not reach those paths. The notification + * itself is still built on the calling thread, so it carries the time, sequence number and serializable source of + * the condition it reports - see {@link #getNotificationSource()} - and, since the notifier is a single thread and + * delivers a notification to all its listeners in one go, every listener observes the notifications in the order + * the pool reported them and a listener registered both through JMX and locally is reached through JMX before it + * is reached locally, as it was when delivery was synchronous. * * @param type The notification type * @param message The message * - * @return true if the notification was accepted for delivery, false if it was dropped - which happens when the - * delivery queue is full or the notification could not be built. The result reports the handoff only: - * it says nothing about whether any listener was reached or what a listener did with the notification. + * @return true if the notification was accepted for delivery as a whole, false if it was dropped as a whole - + * which happens when the delivery queue is full or the notification could not be built. A false means + * nothing was scheduled: no listener will receive this notification. The result otherwise reports the + * handoff only: it says nothing about whether any listener was reached or what a listener did with + * the notification. */ public boolean notify(final String type, String message) { try { Notification n = new Notification(type, getNotificationSource(), sequence.incrementAndGet(), System.currentTimeMillis(), "[" + type + "] " + message); - // Queues the notification for the JMX listeners on the notifier - sendNotification(n); - // And the same for the local listeners, behind the JMX handoff so - // that per notification the listener order stays as it was when - // delivery was synchronous - NOTIFIER.execute(() -> deliverLocally(n, type)); + // Hand the notification over as one task: the delivery queue is + // then the single gate that accepts or refuses it, so the result + // and the drop log below stay true to what was scheduled - a + // refusal never means the notification partly reached its + // listeners, it means nothing was queued at all. On the notifier + // thread the JMX listeners are reached first - the constructor + // runs their hand-offs inline, on this thread - and the local + // ones next, so a listener registered on both sides is reached + // through JMX before it is reached locally, as it was when + // delivery was synchronous. + NOTIFIER.execute(() -> { + try { + sendNotification(n); + } catch (Exception x) { + // sendNotification contains what a listener or a filter + // throws, job per job; this catch is a defence against a + // failure escaping it, which must not skip the local + // listeners of this notification + if (log.isDebugEnabled()) { + log.debug("JMX notification hand-off failed. Type=" + type, x); + } + } + deliverLocally(n, type); + }); return true; } catch (RejectedExecutionException x) { // The delivery queue is full; dropping the notification is the diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/jmx/TestJmxNotificationDelivery.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/jmx/TestJmxNotificationDelivery.java index 5d3d88a31a..85869b9e2b 100644 --- a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/jmx/TestJmxNotificationDelivery.java +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/jmx/TestJmxNotificationDelivery.java @@ -17,7 +17,10 @@ package org.apache.tomcat.jdbc.pool.jmx; import java.sql.Connection; +import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -66,6 +69,12 @@ class TestJmxNotificationDelivery extends JdbcPoolBaseTest { /* How many notifications are pushed at a queue nobody is draining. */ private static final int FLOOD_SIZE = 4000; + /* + * How many notifications are pushed at the queue in the split test below. + * Above the queue bound (2000 tasks), so the queue certainly overflows. + */ + private static final int SPLIT_FLOOD_SIZE = 2500; + /** * Every notification type the pool MBean advertises, in the order the * burst test reports them. @@ -106,6 +115,49 @@ class TestJmxNotificationDelivery extends JdbcPoolBaseTest { } } + /* + * A listener that records the notifications it receives and blocks the + * notifier thread on the first one until the test opens the gate - for at + * most STUCK_LISTENER_TIMEOUT_S, bounding the stuck state as the listener + * above does - and from then on only records. + */ + private static final class GateListener implements NotificationListener { + private final CountDownLatch gate; + private final CountDownLatch entered = new CountDownLatch(1); + private final List<Notification> received = new CopyOnWriteArrayList<>(); + private boolean firstHolds = true; + + GateListener(CountDownLatch gate) { + this.gate = gate; + } + + @Override + public void handleNotification(Notification notification, Object handback) { + received.add(notification); + if (firstHolds) { + firstHolds = false; + entered.countDown(); + try { + gate.await(STUCK_LISTENER_TIMEOUT_S, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + } + + /* + * A listener that only records the notifications it receives. + */ + private static final class RecordingListener implements NotificationListener { + private final List<Notification> received = new CopyOnWriteArrayList<>(); + + @Override + public void handleNotification(Notification notification, Object handback) { + received.add(notification); + } + } + @Test void aStuckLocalListenerDoesNotStallTheThreadThatReports() throws Exception { DataSource ds = newDataSource(); @@ -278,4 +330,195 @@ class TestJmxNotificationDelivery extends JdbcPoolBaseTest { "the first notification must have been delivered, not dropped"); jmxPool.removeListener(stuck); } + + /* + * The overflow drop is a per-notification decision, never a per-listener + * one. The JMX hand-off used to run on the thread that reported the + * condition: the JDK broadcaster hands the notification to the delivery + * queue once per selected JMX listener and lets a full queue throw from + * the middle of that loop, so on the notification that filled the queue + * between the JMX hand-off and the local one, notify() reported the + * notification dropped although some JMX listeners were already + * scheduled to receive it - and the local listeners never were. Delivery + * is now one task per notification, whose JMX hand-offs run inline on the + * notifier thread: the queue accepts or refuses the whole notification, + * so a notification reported dropped reaches no listener at all, and one + * reported accepted reaches every JMX listener and every local one once + * the queue has drained. + */ + @Test + void aSaturatedQueueRefusesOrAcceptsTheWholeNotificationNotPartOfIt() + throws Exception { + DataSource ds = newDataSource(); + ds.getPoolProperties().setName("JmxDeliverySplit-" + System.nanoTime()); + ds.createPool(); + // an unregistered MBean is enough: a listener can join its JMX + // listener list directly, no MBean server is involved + ConnectionPool jmxPool = new ConnectionPool(ds.getPool()); + try { + /* + * The notifier and its queue are shared JVM-wide, so first let + * whatever an earlier test queued on it drain: once a probe + * notification is raised on its own wrapper and received, the + * notifier thread has finished everything that was queued before + * it. The probe wrapper keeps the probe deliveries off the + * listener lists of the wrapper under test. + */ + ConnectionPool probePool = new ConnectionPool(ds.getPool()); + RecordingListener probe = new RecordingListener(); + probePool.addListener(probe); + awaitUntil(() -> { + probePool.notify(ConnectionPool.POOL_EMPTY, "drain-probe"); + return !probe.received.isEmpty(); + }); + probePool.removeListener(probe); + + CountDownLatch gate = new CountDownLatch(1); + GateListener jmxOne = new GateListener(gate); + RecordingListener jmxTwo = new RecordingListener(); + RecordingListener jmxThree = new RecordingListener(); + RecordingListener local = new RecordingListener(); + // three JMX listeners: the old calling-thread fan-out queued four + // hand-offs for every notification (one per JMX listener plus the + // local one), so with the three hand-offs of the gate still + // queued the bound was reached partway through a later + // notification - some of its listeners scheduled although + // notify() reported it dropped - rather than cleanly between two + // notifications; that split is what the audit below trips on + jmxPool.addNotificationListener(jmxOne, null, null); + jmxPool.addNotificationListener(jmxTwo, null, null); + jmxPool.addNotificationListener(jmxThree, null, null); + jmxPool.addListener(local); + List<Boolean> accepted = new ArrayList<>(); + try { + // the gate notification gets the notifier thread out of its + // idle state; it is then stuck delivering this very + // notification, and by the time it signals that it got in, + // the shared queue holds nothing but what this test queues + Assertions.assertTrue( + jmxPool.notify(ConnectionPool.POOL_EMPTY, "gate"), + "the notification that gates this test must be accepted"); + Assertions.assertTrue(jmxOne.entered.await(5, TimeUnit.SECONDS), + "the notifier thread must be stuck on the gate notification"); + for (int i = 0; i < SPLIT_FLOOD_SIZE; i++) { + accepted.add(jmxPool.notify(ConnectionPool.POOL_EMPTY, "split-" + i)); + } + } finally { + gate.countDown(); + } + int handedOff = 0; + int dropped = 0; + for (boolean reported : accepted) { + if (reported) { + handedOff++; + } else { + dropped++; + } + } + Assertions.assertTrue(dropped > 0, + "a queue that nothing drains must start dropping: all " + SPLIT_FLOOD_SIZE + + " notifications were accepted, so the queue is unbounded"); + Assertions.assertTrue(handedOff > 0, + "before the queue is full, the notifications must be handed off"); + + // drain what the queue took; the local listener is delivered in + // the notification's own task, so once it has every accepted + // notification, every task the queue took has run + int expected = 1 + handedOff; + awaitUntil(() -> local.received.size() >= expected); + Assertions.assertEquals(expected, local.received.size(), + "every accepted notification must have reached the local listener, and no" + + " refused one may reach it"); + + /* + * Let the queue drain completely before reading the JMX + * listeners' books: a per-listener hand-off that the old calling- + * thread fan-out had already queued when the notification as a + * whole was refused is itself queued behind the last accepted + * notification and must have run by the time the checks below + * look for it. + */ + RecordingListener drained = new RecordingListener(); + probePool.addListener(drained); + awaitUntil(() -> { + probePool.notify(ConnectionPool.POOL_EMPTY, "drained-probe"); + return !drained.received.isEmpty(); + }); + probePool.removeListener(drained); + + /* + * Every notification the queue accepted must have reached every + * listener, and one it refused must have reached none: the + * accepted task hands the notification to the JMX listeners + * inline before the local ones, so no per-listener hand-off is + * ever put at the queue's bound a second time, and the drain + * above has by now run every task the queue took. + */ + Set<String> byJmxOne = new HashSet<>(messages(jmxOne.received)); + Set<String> byJmxTwo = new HashSet<>(messages(jmxTwo.received)); + Set<String> byJmxThree = new HashSet<>(messages(jmxThree.received)); + Set<String> byLocal = new HashSet<>(messages(local.received)); + for (int i = 0; i < accepted.size(); i++) { + String message = "[" + ConnectionPool.POOL_EMPTY + "] split-" + i; + if (accepted.get(i)) { + Assertions.assertTrue(byJmxOne.contains(message), + "the notification reported as sent must reach every JMX listener: split-" + i + + " was accepted but the first JMX listener did not receive it"); + Assertions.assertTrue(byJmxTwo.contains(message), + "the notification reported as sent must reach every JMX listener: split-" + i + + " was accepted but the second JMX listener did not receive it"); + Assertions.assertTrue(byJmxThree.contains(message), + "the notification reported as sent must reach every JMX listener: split-" + i + + " was accepted but the third JMX listener did not receive it"); + Assertions.assertTrue(byLocal.contains(message), + "the notification reported as sent must reach the local listener: split-" + i + + " was accepted but the local listener did not receive it"); + } else { + Assertions.assertFalse(byJmxOne.contains(message), + "the notification reported as dropped must reach no listener: the first JMX listener" + + " received split-" + i + " although notify() reported it dropped"); + Assertions.assertFalse(byJmxTwo.contains(message), + "the notification reported as dropped must reach no listener: the second JMX listener" + + " received split-" + i + " although notify() reported it dropped"); + Assertions.assertFalse(byJmxThree.contains(message), + "the notification reported as dropped must reach no listener: the third JMX listener" + + " received split-" + i + " although notify() reported it dropped"); + Assertions.assertFalse(byLocal.contains(message), + "the notification reported as dropped must reach no listener: the local listener" + + " received split-" + i + " although notify() reported it dropped"); + } + } + Assertions.assertTrue(jmxPool.notify(ConnectionPool.POOL_EMPTY, "final"), + "a notification reported after the queue drained must be accepted"); + String finalMessage = "[" + ConnectionPool.POOL_EMPTY + "] final"; + awaitUntil(() -> containsMessage(jmxOne.received, finalMessage) + && containsMessage(jmxTwo.received, finalMessage) + && containsMessage(jmxThree.received, finalMessage) + && containsMessage(local.received, finalMessage)); + + jmxPool.removeNotificationListener(jmxOne); + jmxPool.removeNotificationListener(jmxTwo); + jmxPool.removeNotificationListener(jmxThree); + jmxPool.removeListener(local); + } finally { + ds.close(); + } + } + + private static List<String> messages(List<Notification> received) { + List<String> messages = new ArrayList<>(received.size()); + for (Notification n : received) { + messages.add(n.getMessage()); + } + return messages; + } + + private static boolean containsMessage(List<Notification> received, String message) { + for (Notification n : received) { + if (message.equals(n.getMessage())) { + return true; + } + } + return false; + } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
