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 ca5e231ce8c1e906f3e6f380db8f543ece632601 Author: remm <[email protected]> AuthorDate: Wed Oct 28 17:15:37 2026 +0100 Fail a timed-out async poll of the fair JDBC pool queue terminally An ItemFuture whose timed get() timed out deregistered its waiter from the queue's waiter list and threw TimeoutException, but left the future in a state no terminal event could ever reach: not cancelled, never completable again since the queue could no longer see the latch, and reporting isDone() false forever. A second get(), which the java.util.concurrent.Future contract permits until it succeeds, then parked its caller on the deregistered latch indefinitely, even while the pool kept handing connections to other borrowers. The public getConnectionAsync() path exposed this directly: an async borrow that timed out once could not be abandoned through a later untimed get() at all, and the deferred cleanup armed by a cancel() after such a timeout parked in the same way. Decide the timeout atomically under the queue lock: re-check for an item a concurrent offer committed to the latch, re-check for a concurrent cancellation, and otherwise deregister the waiter and mark the future abandoned in the same lock hold. The abandoned future completes exceptionally: further timed retrievals report a TimeoutException right away, the untimed get() reports that timeout as its ExecutionException cause, isDone() is true, isCancelled() stays false for a future nobody cancelled, and cancel() reports nothing left to cancel. A caller already blocked in a get() when a concurrent timed retrieval times out is released with the same failure, and an item committed to the latch just as the timeout fires is delivered rather than lost, as the delivered-or-still-in-queue conservation of the queue is preserved throughout. Cover it with queue-level tests for the terminal failure, the release of a concurrently blocked getter and the offer race, and a pool-level test that a timed-out async borrow fails further getters instead of blocking them. --- .../apache/tomcat/jdbc/pool/FairBlockingQueue.java | 111 +++++++++++++-- .../apache/tomcat/jdbc/pool/TestAsyncBorrow.java | 66 ++++++++- .../tomcat/jdbc/pool/TestFairBlockingQueue.java | 150 ++++++++++++++++++++- webapps/docs/changelog.xml | 5 + 4 files changed, 318 insertions(+), 14 deletions(-) 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 4e66eb5af4..8fa4e0ddcb 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 @@ -209,6 +209,10 @@ 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. + * A {@link java.util.concurrent.Future#get(long, TimeUnit)} call on the + * future that times out deregisters the waiter in the same way and fails + * the future terminally, so that every later retrieval from it fails right + * away instead of waiting for an item that can no longer be handed over. * 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. @@ -488,6 +492,15 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { * A future used to check if a connection has been made available. * Cancelling the future deregisters its waiter from the queue, as long * as the item has not already been handed over to the future. + * <br> + * A {@link #get(long, TimeUnit)} call that times out deregisters the + * waiter in the same way, and additionally fails the future terminally: + * no offer can reach the deregistered latch anymore, so the future + * reports {@link #isDone()} true without ever being cancelled, keeps + * failing every later {@link #get(long, TimeUnit)} with a + * {@link TimeoutException} and every later {@link #get()} with an + * {@link ExecutionException} caused by one, and can no longer be + * cancelled. * * @param <T> the type of the item */ @@ -498,6 +511,14 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { protected volatile ExchangeCountDownLatch<T> latch = null; /** Whether this future has been cancelled. */ protected volatile boolean canceled = false; + /** + * Whether a {@link #get(long, TimeUnit)} call on this future timed + * out. Its waiter was then deregistered from the queue, so no item + * can ever be handed over to this future anymore; the future is + * terminally failed and its get() methods fail fast instead of + * parking on a latch that nothing can count down again. + */ + protected volatile boolean abandoned = false; /** * Creates an ItemFuture with the given item already available. @@ -532,7 +553,10 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { try { //the item is handed over under the queue lock, so a non-null //item means an offer() has already committed to this future - if (l.getItem()!=null || canceled) { + if (l.getItem()!=null || canceled || abandoned) { + //the future is complete: it received its item, it was + //cancelled earlier, or a timed out get() call abandoned + //its waiter, so there is nothing left to cancel return false; } canceled = true; @@ -556,10 +580,19 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { if (canceled) { throw new CancellationException(); } + if (abandoned) { + throw abandonedFailure(); + } latch.await(); + //the latch is counted down by the offer() that hands over + //the item, by cancel(), or by a timed get() call that + //abandoned this future - check for each of them if (canceled) { throw new CancellationException(); } + if (abandoned) { + throw abandonedFailure(); + } return latch.getItem(); } else { throw new ExecutionException("ItemFuture incorrectly instantiated. Bug in the code?", new Exception()); @@ -574,27 +607,59 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { if (canceled) { throw new CancellationException(); } + if (abandoned) { + throw abandonedTimeout(); + } boolean timedout = !latch.await(timeout, unit); if (canceled) { throw new CancellationException(); } + if (abandoned) { + //another timed get() call abandoned this future while + //this one was waiting, so this wait can no longer be + //served either + throw abandonedTimeout(); + } if (timedout) { - //an abandoned timed-out waiter would otherwise keep - //consuming offered items, so deregister it as well - - //this mirrors what the timed poll() method does final ReentrantLock lock = FairBlockingQueue.this.lock; + T delivered; + boolean lostToCancel; lock.lock(); try { - waiters.remove(latch); + //the item is handed over under the queue lock, so an + //offer() that committed to this latch concurrently + //with the timeout is visible here + delivered = latch.getItem(); + lostToCancel = canceled; + if (delivered==null && !lostToCancel) { + //an abandoned timed-out waiter would otherwise keep + //consuming offered items, so deregister it as well - + //this mirrors what the timed poll() method does + waiters.remove(latch); + //no offer() can reach the deregistered latch + //anymore and nothing else can complete the future, + //so fail it terminally: a later get() must not + //park on a latch that will never be counted down + abandoned = true; + } } finally { lock.unlock(); } - //an item may have been handed over concurrently with the - //timeout; in that case it is delivered rather than lost - T delivered = latch.getItem(); if (delivered!=null) { + //an item may have been handed over concurrently with + //the timeout; in that case it is delivered rather than + //lost return delivered; } + if (lostToCancel) { + //a cancel() won the race against this timeout; its + //callers are told about the cancellation + throw new CancellationException(); + } + //wake up any other caller already blocked in a get() + //method, so it observes the abandonment instead of + //waiting on a latch nothing will count down again + latch.countDown(); throw new TimeoutException(); } else { return latch.getItem(); @@ -611,7 +676,35 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { @Override public boolean isDone() { - return (item!=null || canceled || (latch!=null && latch.getItem()!=null)); + return (item!=null || canceled || abandoned + || (latch!=null && latch.getItem()!=null)); + } + + /** + * The failure for a caller of {@link #get(long, TimeUnit)} of a + * future whose waiter was abandoned by an earlier timeout. The wait + * cannot ever succeed anymore, so the timeout of this call is what + * the caller gets to hear about, right away. + * + * @return the timeout failure to throw + */ + private TimeoutException abandonedTimeout() { + return new TimeoutException("The waiter of this future timed out on an earlier" + + " get() call and was deregistered from the queue; the future can no" + + " longer receive an item"); + } + + /** + * The failure for a caller of the untimed {@link #get()} of a future + * whose waiter was abandoned by an earlier timeout. The get() method + * cannot declare a {@link TimeoutException}, so, the same way + * {@link ConnectionPool.ConnectionFuture#get()} does, it reports the + * timeout as the cause of an ExecutionException. + * + * @return the execution failure to throw + */ + private ExecutionException abandonedFailure() { + return new ExecutionException(abandonedTimeout()); } } 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 7ab8d2c2ea..b6110e7711 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 @@ -21,8 +21,10 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -36,8 +38,9 @@ import org.apache.tomcat.jdbc.JdbcPoolBaseTest; * ({@code getConnectionAsync()}): the same accounting, the same * database-side truth and the same delivery guarantees as the blocking * 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. + * cannot hand out futures, the termination of an outstanding poll once + * the pool is closed, and the immediate failure of a further retrieval + * from a poll whose timed get already timed out. * * <p>Scenario harvested from the retired legacy classes * {@code Bug50805} (async borrow accounting) and @@ -285,6 +288,65 @@ class TestAsyncBorrow extends JdbcPoolBaseTest { awaitSessions(0); } + @Test + void aTimedOutAsyncBorrowFailsFurtherGettersInsteadOfBlockingThem() + 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(); + + Connection holder = ds.getConnection(); // takes the only slot + Future<Connection> future = ds.getConnectionAsync(); // queued poll + + Assertions.assertThrows(TimeoutException.class, + () -> future.get(100, TimeUnit.MILLISECONDS), + "the borrow cannot be served while the pool holds its only slot"); + + // the timed-out poll left the idle queue, so the connection that + // comes back is not stranded on it but goes to the next borrower + holder.close(); + try (Connection served = ds.getConnection()) { + Assertions.assertNotNull(served, + "a returned connection must not be consumed by the abandoned poll"); + } + + // the poll can never receive a connection anymore - it must say so, + // instead of leaving a later getter parked on the abandoned waiter + AtomicReference<Throwable> failure = new AtomicReference<>(); + CountDownLatch finished = new CountDownLatch(1); + Thread caller = new Thread(() -> { + try { + future.get(); // the documented, untimed usage + } catch (Throwable e) { + failure.set(e); + } finally { + finished.countDown(); + } + }, "timed-out-async-borrower"); + caller.setDaemon(true); + caller.start(); + Assertions.assertTrue(finished.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS), + "an untimed get() after the timeout must not block forever"); + Assertions.assertTrue(failure.get() instanceof ExecutionException + && failure.get().getCause() instanceof TimeoutException, + "the untimed get() must report the earlier timeout as its cause, got " + + failure.get()); + + Assertions.assertTrue(future.isDone(), + "a poll whose timed get() timed out is complete"); + Assertions.assertFalse(future.isCancelled(), + "nobody cancelled the poll, a timeout must not report a cancellation"); + Assertions.assertThrows(TimeoutException.class, + () -> future.get(100, TimeUnit.MILLISECONDS), + "a repeated timed get() must keep reporting the timeout"); + + ds.close(); + } + @Test void cancellingAnAsyncPollAfterThePoolClosedReportsNothingCancelled() throws Exception { DataSource ds = newDataSource(); 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 de905e1775..0a30d26a11 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 @@ -21,9 +21,11 @@ import java.util.Collections; import java.util.List; import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Assertions; @@ -34,9 +36,10 @@ 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, the termination of every outstanding poll once the waiters are - * cancelled, and the collection operations the queue deliberately leaves - * unsupported. + * polls, the terminal failure of an asynchronous poll whose timed get + * times out, the termination of every outstanding poll once the waiters + * are cancelled, and the collection operations the queue deliberately + * leaves unsupported. */ class TestFairBlockingQueue { @@ -357,4 +360,145 @@ class TestFairBlockingQueue { "an element offered after the timeout must stay in the queue"); Assertions.assertEquals("keep", queue.poll()); } + + @Test + void timedOutAsyncPollTerminallyFailsItsFuture() throws Exception { + final FairBlockingQueue<String> queue = new FairBlockingQueue<>(); + final Future<String> future = queue.pollAsync(); + Assertions.assertThrows(TimeoutException.class, + () -> future.get(200, TimeUnit.MILLISECONDS)); + // the timed-out waiter must not consume a subsequently offered element + Assertions.assertTrue(queue.offer("keep")); + Assertions.assertEquals(1, queue.size(), + "an element offered after the timeout must stay in the queue"); + Assertions.assertEquals("keep", queue.poll()); + + // a Future may be retrieved any number of times until it succeeds, so + // the untimed get() after the timeout must terminate - the waiter is + // deregistered and nothing can ever complete the future, which must + // therefore fail fast instead of parking on the dead latch + final AtomicReference<Throwable> error = new AtomicReference<>(); + final CountDownLatch done = new CountDownLatch(1); + Thread caller = new Thread(() -> { + try { + future.get(); + } catch (Throwable t) { + error.set(t); + } finally { + done.countDown(); + } + }); + caller.setDaemon(true); + caller.start(); + Assertions.assertTrue(done.await(5, TimeUnit.SECONDS), + "an untimed get() after the timeout must terminate, not block forever"); + Assertions.assertTrue(error.get() instanceof ExecutionException + && error.get().getCause() instanceof TimeoutException, + "the untimed get() must report the earlier timeout as its cause, got " + + error.get()); + caller.join(1000); + + // the state of the terminally failed future is truthful + Assertions.assertTrue(future.isDone(), + "a future whose waiter was abandoned can never receive an element"); + Assertions.assertFalse(future.isCancelled(), + "a timeout must not report a cancellation nobody asked for"); + + // a repeated timed get reports the failure right away + long start = System.currentTimeMillis(); + Assertions.assertThrows(TimeoutException.class, + () -> future.get(10, TimeUnit.SECONDS), + "a get() after the timeout must keep reporting the timeout"); + Assertions.assertTrue(System.currentTimeMillis() - start < 5000, + "the abandoned future must fail fast instead of serving a fresh timeout"); + + // a future that already failed terminally cannot be cancelled, and a + // refused cancel must not make its cancellation state untrue + Assertions.assertFalse(future.cancel(true), + "a terminally failed future must not report a cancellation"); + Assertions.assertFalse(future.isCancelled()); + Assertions.assertTrue(future.isDone()); + } + + @Test + void timedOutAsyncPollReleasesAConcurrentlyBlockedGetter() throws Exception { + // one caller is already waiting untimed on the poll when a timed get + // times out and abandons the waiter: the blocked caller must observe + // the abandonment instead of parking on a latch nothing counts down + final FairBlockingQueue<String> queue = new FairBlockingQueue<>(); + final Future<String> future = queue.pollAsync(); + final AtomicReference<Throwable> error = new AtomicReference<>(); + final CountDownLatch started = new CountDownLatch(1); + final CountDownLatch done = new CountDownLatch(1); + Thread blocked = new Thread(() -> { + started.countDown(); + try { + future.get(); + } catch (Throwable t) { + error.set(t); + } finally { + done.countDown(); + } + }); + blocked.setDaemon(true); + blocked.start(); + Assertions.assertTrue(started.await(5, TimeUnit.SECONDS)); + Thread.sleep(100); // let the caller block inside get() + + Assertions.assertThrows(TimeoutException.class, + () -> future.get(200, TimeUnit.MILLISECONDS)); + + Assertions.assertTrue(done.await(5, TimeUnit.SECONDS), + "the abandonment must release a caller blocked in get()"); + Assertions.assertTrue(error.get() instanceof ExecutionException + && error.get().getCause() instanceof TimeoutException, + "the blocked caller must observe the abandonment, got " + error.get()); + blocked.join(1000); + } + + @Test + void timedOutAsyncPollRacingAnOfferDoesNotStrandElements() throws Exception { + // an offer() that commits its element to a latch concurrently with a + // timeout is delivered rather than lost, and every timed-out waiter + // really leaves the waiters list: across this race, every element + // must end up either delivered to a future or still in the queue + final FairBlockingQueue<String> queue = new FairBlockingQueue<>(); + final int rounds = 200; + final AtomicInteger delivered = new AtomicInteger(); + Thread offerer = new Thread(() -> { + for (int i = 0; i < rounds; i++) { + queue.offer("e" + i); + } + }); + offerer.setDaemon(true); + offerer.start(); + for (int i = 0; i < rounds; i++) { + Future<String> future = queue.pollAsync(); + try { + String element = future.get(1, TimeUnit.MILLISECONDS); + Assertions.assertNotNull(element, + "a get() that returns must deliver the element it waited for"); + delivered.incrementAndGet(); + } catch (TimeoutException e) { + // the waiter was abandoned; the element stays in the queue + } + } + offerer.join(5000); + Assertions.assertFalse(offerer.isAlive(), "the offerer must have finished"); + + Assertions.assertEquals(rounds, delivered.get() + queue.size(), + "no element may be lost between a timing out get() and a racing offer()"); + + // and no timed-out future stayed registered: further elements are + // stored rather than handed to a phantom waiter + queue.offer("extra"); + Assertions.assertEquals(rounds - delivered.get() + 1, queue.size(), + "a timed-out waiter must not consume a further element"); + while (queue.poll() != null) { + // drain the queue + } + Assertions.assertTrue(queue.offer("last")); + Assertions.assertEquals(1, queue.size()); + Assertions.assertEquals("last", queue.poll()); + } } diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index 3c9500a22a..23b78e344a 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -1281,6 +1281,11 @@ <code>CancellationException</code> to its callers instead of a closed connection. (remm) </fix> + <fix> + Make a timed-out retrieval from the JDBC pool's fair queue fail + terminally instead of leaving its future in a state nobody can + complete. (remm) + </fix> </changelog> </subsection> <subsection name="Other"> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
