This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 31f772f5291f0a413c18359d9a17a1af836eeb1b Author: opencode <[email protected]> AuthorDate: Wed Oct 28 15:27:17 2026 +0100 Terminate the asynchronous borrows of a closing JDBC connection pool An asynchronous borrow registers a waiter on the fair idle queue, and nothing in the pool's shutdown ever woke or failed those waiters: an async waiter is not counted in the pool wait count, so the release of a connection sends it no wake-up placeholder, and a closed pool never hands another connection to the queue. A caller parked in the get() of the future returned by getConnectionAsync() therefore waited forever on a pool that could no longer serve it, and could not even observe the closed state from inside the queue wait. As closing a data source is a normal lifecycle step rather than a caller's choice, this leaked a thread on every undeploy that left an async poll outstanding. Make closure terminal for asynchronous borrows. Closing the pool now cancels the polls that are still waiting through a new waiters hook of the fair queue, so their get() reports a CancellationException and a poll whose connection had already been delivered keeps it. An asynchronous borrow that registers after the closure is rejected with an SQLException, like a synchronous borrow on a closed pool, while one whose waiter was registered just before the closure ran gets the cancellation instead. A cancellation of a poll that the closure has already cancelled now reports that nothing was cancelled, instead of running a useless cleanup that logged the normal case as an error. The wake-up of the queue waiters is limited to the asynchronous ones, so the shutdown drain of the pool keeps its established behavior. Cover it with a pool-level test that a closure releases a caller blocked in get(), and queue-level tests for the wake-up of blocked pollers, for the refusal of new ones, and for leaving a delivered item with its future. --- .../apache/tomcat/jdbc/pool/ConnectionPool.java | 35 ++++++- .../apache/tomcat/jdbc/pool/FairBlockingQueue.java | 91 ++++++++++++++++- .../apache/tomcat/jdbc/pool/TestAsyncBorrow.java | 108 ++++++++++++++++++++- .../tomcat/jdbc/pool/TestFairBlockingQueue.java | 63 +++++++++++- webapps/docs/changelog.xml | 14 +++ 5 files changed, 304 insertions(+), 7 deletions(-) diff --git a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/ConnectionPool.java b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/ConnectionPool.java index c84d571ba1..c20cac57ab 100644 --- a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/ConnectionPool.java +++ b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/ConnectionPool.java @@ -30,6 +30,7 @@ import java.util.Set; import java.util.Timer; import java.util.TimerTask; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; @@ -167,6 +168,9 @@ public class ConnectionPool { * until a connection has become available. * If a connection is not retrieved, the Future must be cancelled in order for the connection to be returned * to the pool. + * Closing the pool cancels a future that has not received a connection yet, so a thread blocked in + * {@link Future#get()} observes a {@link java.util.concurrent.CancellationException} instead of waiting for + * a connection that can no longer be handed out. * @return a Future containing a reference to the connection or the future connection * @throws SQLException Cannot use asynchronous connect */ @@ -189,6 +193,15 @@ public class ConnectionPool { //we can only retrieve a future if the underlying queue supports it. if (idle instanceof FairBlockingQueue<?>) { Future<PooledConnection> pcf = ((FairBlockingQueue<PooledConnection>)idle).pollAsync(); + if (isClosed() && pcf.isCancelled()) { + //The pool was closed between the borrow attempt above and the + //registration of the waiter: close() fails every waiter the + //queue holds and a later poll cannot register one, so this + //future will never hand out a connection. Reject the borrow the + //same way a synchronous borrow on a closed pool does, rather than + //handing out a future that only ever fails with a cancellation. + throw new SQLException("Connection pool closed."); + } return new ConnectionFuture(pcf); } else { throw new SQLException("Connection pool is misconfigured, doesn't support async retrieval. Set the 'fair' property to 'true'"); @@ -441,6 +454,15 @@ public class ConnectionPool { //stop background thread terminatePoolCleaner(); + //An asynchronous borrow waits on the idle queue, where a thread cannot + //observe the closed flag: this pool never offers a connection to that + //queue once it is closed, so a caller parked in the Future.get() of + //getConnectionAsync() would stay parked forever. Fail the waiters that + //are registered on the queue, so that the futures handed out terminate. + if (idle instanceof FairBlockingQueue<?>) { + ((FairBlockingQueue<PooledConnection>)idle).cancelWaiters(); + } + /* release all idle connections */ BlockingQueue<PooledConnection> pool = (!idle.isEmpty())?idle:(force?busy:idle); boolean interrupted = false; @@ -1636,8 +1658,12 @@ public class ConnectionPool { //On the async path the pc field is never assigned; once a //connection has been produced it is held in the result field, //so a non-null result means the future is complete and it is - //too late to cancel - if (pc!=null || result!=null) { + //too late to cancel. A queue future the pool closure already + //cancelled is complete in the same way and leaves nothing to + //deregister and no connection to clean up, so report nothing + //cancelled without arming the deferred cleanup task, which a + //java.util.concurrent.Future must not run for a completed task. + if (pc!=null || result!=null || pcFuture.isCancelled()) { return false; } else if (cancelled.compareAndSet(false, true)) { if (!pcFuture.cancel(mayInterruptIfRunning)) { @@ -1727,6 +1753,11 @@ public class ConnectionPool { } }catch (ExecutionException ex) { //we can ignore this + }catch (CancellationException ex) { + //the pool closure cancelled the queued wait between the + //registration of this cleanup task and its execution; this + //future never produced a connection, so there is nothing + //to close }catch (Exception x) { log.error("Unable to cancel ConnectionFuture.",x); } diff --git a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/FairBlockingQueue.java b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/FairBlockingQueue.java index c76ea430e6..4e66eb5af4 100644 --- a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/FairBlockingQueue.java +++ b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/FairBlockingQueue.java @@ -16,9 +16,11 @@ */ package org.apache.tomcat.jdbc.pool; +import java.util.ArrayList; import java.util.Collection; import java.util.Iterator; import java.util.LinkedList; +import java.util.List; import java.util.NoSuchElementException; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CancellationException; @@ -73,6 +75,14 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { */ final LinkedList<ExchangeCountDownLatch<E>> waiters; + /** + * Set once {@link #cancelWaiters()} has failed the waiters of this queue. + * The caller that cancels the waiters (a closing pool) stops offering + * items to this queue afterwards, so an asynchronous poll that registers + * a waiter on a cancelled queue could never be served. + */ + private boolean waitersCancelled = false; + /** * Creates a new fair blocking queue. */ @@ -199,6 +209,9 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { * item has not already been handed over to the future. A future that is no * longer needed must therefore be cancelled, so that a subsequent offer * does not hand an item over to a future nobody is waiting on. + * Once {@link #cancelWaiters()} has been called, no waiter is registered + * anymore: the returned future is cancelled right away, as the caller + * that cancelled the waiters offers no further item that could serve it. * @return - a future pending the result from the queue poll request */ public Future<E> pollAsync() { @@ -212,9 +225,21 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { if (item==null) { //queue is empty, add ourselves as waiters ExchangeCountDownLatch<E> c = new ExchangeCountDownLatch<>(1); - waiters.addLast(c); //return a future that will wait for the object - result = new ItemFuture<>(c); + ItemFuture<E> future = new ItemFuture<>(c); + //tell the latch which future its item is meant for, before the + //waiter becomes visible to cancelWaiters() + c.setFuture(future); + if (waitersCancelled) { + //the caller that cancelled the waiters offers no further + //item, so registering a waiter here would park its caller + //forever - hand out a future that is cancelled from the + //start instead + future.cancel(false); + } else { + waiters.addLast(c); + } + result = future; } else { //return a future with the item result = new ItemFuture<>(item); @@ -225,6 +250,46 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { return result; } + /** + * Cancel the futures of all the asynchronous polls that are waiting for an + * item on this queue. A thread that is blocked in one of the + * {@link Future#get()} methods of such a future observes a + * {@link CancellationException} instead of waiting for an item that will + * never be offered, and later calls to {@link #pollAsync()} do not register + * a waiter at all. A future an item has already been handed over to is not + * cancelled, it completes with that item as usual. + * <br> + * This makes a queue that is never fed again terminate the callers that + * wait on it, instead of leaving them parked forever. + */ + public void cancelWaiters() { + final ReentrantLock lock = this.lock; + List<ItemFuture<E>> futures; + lock.lock(); + try { + //the caller that cancels the waiters offers no further item from + //this point on, so a waiter that registers after it can never be + //served either + waitersCancelled = true; + futures = new ArrayList<>(waiters.size()); + for (ExchangeCountDownLatch<E> c : waiters) { + ItemFuture<E> future = c.getFuture(); + if (future != null) { + //only the asynchronous polls have a future to cancel, the + //waiters of poll(long,TimeUnit) time out on their own + futures.add(future); + } + } + } finally { + lock.unlock(); + } + //cancel outside of the lock, as the cancellation wakes up the callers + //and deregisters their waiter itself + for (ItemFuture<E> future : futures) { + future.cancel(false); + } + } + @Override public boolean remove(Object e) { final ReentrantLock lock = this.lock; @@ -562,6 +627,12 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { protected class ExchangeCountDownLatch<T> extends CountDownLatch { /** The item to exchange. */ protected volatile T item; + /** + * The future the item is exchanged with, set by + * {@link FairBlockingQueue#pollAsync()} and null for the waiters of + * {@link FairBlockingQueue#poll(long, TimeUnit)}. + */ + protected volatile ItemFuture<T> future; /** * Creates a new ExchangeCountDownLatch with the given count. * @@ -586,6 +657,22 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { public void setItem(T item) { this.item = item; } + /** + * Returns the future the exchanged item is meant for. + * + * @return the future of an asynchronous poll, or null + */ + public ItemFuture<T> getFuture() { + return future; + } + /** + * Sets the future the exchanged item is meant for. + * + * @param future the future of an asynchronous poll + */ + public void setFuture(ItemFuture<T> future) { + this.future = future; + } } //------------------------------------------------------------------ diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestAsyncBorrow.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestAsyncBorrow.java index d6112e43df..7ab8d2c2ea 100644 --- a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestAsyncBorrow.java +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestAsyncBorrow.java @@ -19,9 +19,12 @@ package org.apache.tomcat.jdbc.pool; import java.sql.Connection; import java.sql.ResultSet; import java.sql.SQLException; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -32,8 +35,9 @@ import org.apache.tomcat.jdbc.JdbcPoolBaseTest; * Expected functionality of the asynchronous borrow API * ({@code getConnectionAsync()}): the same accounting, the same * database-side truth and the same delivery guarantees as the blocking - * API, and a clear failure when the pool is configured with a queue that - * cannot hand out futures. + * API, a clear failure when the pool is configured with a queue that + * cannot hand out futures, and the termination of an outstanding poll + * once the pool is closed. * * <p>Scenario harvested from the retired legacy classes * {@code Bug50805} (async borrow accounting) and @@ -220,4 +224,104 @@ class TestAsyncBorrow extends JdbcPoolBaseTest { "the pool must keep at most maxActive connections"); awaitSessions(2); } + + @Test + void closingThePoolTerminatesAnAsyncPollThatIsBlockedInGet() throws Exception { + DataSource ds = newDataSource(); + ds.getPoolProperties().setFairQueue(true); + ds.getPoolProperties().setInitialSize(0); + ds.getPoolProperties().setMinIdle(0); + ds.getPoolProperties().setMaxActive(1); + ds.getPoolProperties().setMaxIdle(1); + ds.createPool(); + ConnectionPool pool = ds.getPool(); + + ds.getConnection(); // takes the only slot, and the pool is at maxActive + + // the borrow below cannot be served by a new connection, so it has to + // register a waiter on the idle queue and wait for a returned one + Future<Connection> future = ds.getConnectionAsync(); + Assertions.assertFalse(future.isDone(), + "the poll must still be waiting for a connection"); + + CountDownLatch parked = new CountDownLatch(1); + CountDownLatch finished = new CountDownLatch(1); + AtomicReference<Throwable> failure = new AtomicReference<>(); + Thread caller = new Thread(() -> { + parked.countDown(); + try { + future.get(); // the documented, untimed usage + } catch (Throwable e) { + failure.set(e); + } finally { + finished.countDown(); + } + }, "blocked-async-borrower"); + caller.setDaemon(true); + caller.start(); + Assertions.assertTrue(parked.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS)); + sleep(200); // let the caller reach the wait on the idle queue + + ds.close(true); + + // The caller cannot observe the closed pool itself - it is inside the + // queue wait, and a closed pool never offers a connection to that queue + // again. Closing the pool has to terminate the future instead. + Assertions.assertTrue(finished.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS), + "closing the pool must release a caller blocked in get()"); + Assertions.assertTrue(failure.get() instanceof CancellationException, + "a pool closure must fail an outstanding poll with a " + + "cancellation, got " + failure.get()); + Assertions.assertTrue(future.isDone()); + Assertions.assertTrue(future.isCancelled(), + "an outstanding poll must be cancelled by the pool closure"); + Assertions.assertThrows(CancellationException.class, future::get, + "a completed poll must keep reporting its cancellation"); + + // and the borrow path itself reports the closure the same way a + // synchronous borrow does + Assertions.assertThrows(SQLException.class, pool::getConnectionAsync, + "an async borrow on a closed pool must be rejected"); + awaitSessions(0); + } + + @Test + void cancellingAnAsyncPollAfterThePoolClosedReportsNothingCancelled() throws Exception { + DataSource ds = newDataSource(); + ds.getPoolProperties().setFairQueue(true); + ds.getPoolProperties().setInitialSize(0); + ds.getPoolProperties().setMinIdle(0); + ds.getPoolProperties().setMaxActive(1); + ds.getPoolProperties().setMaxIdle(1); + ds.createPool(); + ConnectionPool pool = ds.getPool(); + + ds.getConnection(); // takes the only slot, so the poll stays queued + Future<Connection> future = ds.getConnectionAsync(); + + ds.close(true); + + // the closure has already cancelled the queued poll + Assertions.assertTrue(future.isCancelled()); + Assertions.assertTrue(future.isDone()); + + // a cancel() after the fact must report nothing cancelled, as the + // future is already complete, and must not arm the deferred cleanup + // task, which would only run to observe that same cancellation + Assertions.assertFalse(future.cancel(true), + "a poll the pool closure already cancelled must not report " + + "a further cancellation"); + Assertions.assertTrue(future.isCancelled()); + Assertions.assertTrue(future.isDone()); + Assertions.assertThrows(CancellationException.class, future::get, + "the completed cancellation must keep being reported"); + + // the refused cancel must not act on any connection: beyond what the + // closure itself already released, the pool released nothing + long releasedAfterClose = pool.getReleasedCount(); + sleep(200); // a cleanup task scheduled by cancel() would have run + Assertions.assertEquals(releasedAfterClose, pool.getReleasedCount(), + "a refused cancel must not retrieve or release any connection"); + awaitSessions(0); + } } diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestFairBlockingQueue.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestFairBlockingQueue.java index c32daf8f92..de905e1775 100644 --- a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestFairBlockingQueue.java +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestFairBlockingQueue.java @@ -34,7 +34,8 @@ import org.junit.jupiter.api.Test; * the fair queue is enabled: unbounded offering, most-recently-used * hand-off, blocking retrieval with timeout, fairness towards the * longest waiting consumer, deregistration of cancelled asynchronous - * polls, and the collection operations the queue deliberately leaves + * polls, the termination of every outstanding poll once the waiters are + * cancelled, and the collection operations the queue deliberately leaves * unsupported. */ class TestFairBlockingQueue { @@ -284,6 +285,66 @@ class TestFairBlockingQueue { Assertions.assertEquals(0, queue.size()); } + @Test + void cancellingTheWaitersReleasesBlockedPollersAndRefusesNewOnes() throws Exception { + final FairBlockingQueue<String> queue = new FairBlockingQueue<>(); + final Future<String> pending = queue.pollAsync(); + final AtomicReference<Throwable> error = new AtomicReference<>(); + final CountDownLatch started = new CountDownLatch(1); + final CountDownLatch done = new CountDownLatch(1); + Thread caller = new Thread(() -> { + started.countDown(); + try { + pending.get(); + } catch (Throwable t) { + error.set(t); + } finally { + done.countDown(); + } + }); + caller.start(); + Assertions.assertTrue(started.await(5, TimeUnit.SECONDS)); + Thread.sleep(100); // let the caller block inside get() + + queue.cancelWaiters(); + + Assertions.assertTrue(done.await(5, TimeUnit.SECONDS), + "cancelling the waiters must release a caller blocked in get()"); + Assertions.assertTrue(error.get() instanceof CancellationException, + "a cancelled waiter must fail get() with CancellationException, got " + + error.get()); + Assertions.assertTrue(pending.isCancelled()); + Assertions.assertTrue(pending.isDone()); + caller.join(1000); + + // a poll that registers on a cancelled queue can never be served, so it + // must not stay registered + Future<String> late = queue.pollAsync(); + Assertions.assertTrue(late.isCancelled(), + "a poll registered after the cancellation must be cancelled"); + Assertions.assertThrows(CancellationException.class, late::get); + Assertions.assertTrue(queue.offer("keep")); + Assertions.assertEquals(1, queue.size(), + "a cancelled poll must not consume an offered element"); + // the queue still hands out what it holds + Assertions.assertEquals("keep", queue.poll(200, TimeUnit.MILLISECONDS)); + } + + @Test + void cancellingTheWaitersLeavesADeliveredItemWithItsFuture() throws Exception { + FairBlockingQueue<String> queue = new FairBlockingQueue<>(); + Future<String> future = queue.pollAsync(); + Assertions.assertTrue(queue.offer("handed"), + "the element must be handed to the registered waiter"); + queue.cancelWaiters(); + Assertions.assertFalse(future.isCancelled(), + "a future that already received its element must not be cancelled"); + Assertions.assertTrue(future.isDone()); + Assertions.assertEquals("handed", future.get(5, TimeUnit.SECONDS), + "the element belongs to the future, the cancellation must not take it back"); + Assertions.assertEquals(0, queue.size()); + } + @Test void timedOutAsyncPollIsDeregistered() throws Exception { FairBlockingQueue<String> queue = new FairBlockingQueue<>(); diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index db43e5e2bd..75d7f7390f 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -1245,6 +1245,20 @@ constants from the interface; the notification types and the strings they carry on the wire are unchanged. (remm) </fix> + <fix> + Closing the connection pool now also terminates the asynchronous + borrows that were still waiting for a connection. A caller blocked in + the <code>get()</code> of the future that + <code>getConnectionAsync()</code> had returned waited forever on the + closed pool: a closed pool never hands another connection to the queue + that future waits on, and the waiter itself cannot observe the closed + state. The pool now cancels those outstanding polls, so their + <code>get()</code> reports a <code>CancellationException</code>. An + asynchronous borrow that races the closure is rejected with an + <code>SQLException</code>, like a synchronous borrow on a closed pool; + a borrow that had registered its waiter just before the closure ran + gets a future that reports that cancellation instead. (remm) + </fix> </changelog> </subsection> <subsection name="Other"> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
