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 5358be3a72efbc5fe1870578d3dd01dac714266f Author: opencode <[email protected]> AuthorDate: Mon Oct 26 16:43:21 2026 +0100 Make async polls of the fair pool queue cancellable The Future returned by FairBlockingQueue.pollAsync() advertised cancel() but ignored it: cancel() was hard-wired to return false, isCancelled() was hard-wired to false, and the ExchangeCountDownLatch registered on the queue as the waiter was never removed. A future that was abandoned through cancel() therefore kept its queue slot forever, and the next offer() handed its item to that stale waiter unconditionally - at pool level, a returned connection was delivered to a Future nobody read and was lost from the pool. The pool worked around this only partially: ConnectionFuture.cancel() never touched the queue future, it always scheduled the 'cancellator' task that waits for the connection and closes it back, which recovers the connection only while that single-threaded task is healthy and never releases the queue waiter. FairBlockingQueue.ItemFuture.cancel() now deregisters the waiter under the queue lock, as long as the item has not already been committed to the future by an offer() (the item hand-off happens under the same lock, so the two cannot interleave). The canceled flag is set and the latch counted down, so a thread blocked in get() wakes up and both get() methods fail with CancellationException, matching the Future contract; isCancelled() reports the flag and isDone() is true once cancelled. cancel() still returns false when the item has already been handed over or the future is already cancelled. A timed get() that times out now also deregisters its waiter, mirroring the cleanup the timed poll() method already does, so an abandoned timed-out future cannot consume a later offer either. ConnectionPool.ConnectionFuture.cancel() first cancels the queue-side future, which releases the queue waiter; the cancellator fallback is only scheduled when the pooled connection is already in flight, and it returns false - rather than unconditionally true as before - when the future is already complete (a non-null result now also counts, closing the hazard of cancelling a completed future and having the cancellator close a connection the application is using) or already cancelled. Adds six regression tests to TestFairBlockingQueue covering cancellation before hand-off, cancel after delivery, cancellation of a blocked get(), a cancelled waiter not stealing the hand-off of the next waiter, cancel of an immediately completed future, and deregistration after a timed-out get(). --- .../apache/tomcat/jdbc/pool/ConnectionPool.java | 21 +++- .../apache/tomcat/jdbc/pool/FairBlockingQueue.java | 74 ++++++++++++-- .../tomcat/jdbc/pool/TestFairBlockingQueue.java | 108 ++++++++++++++++++++- webapps/docs/changelog.xml | 16 +++ 4 files changed, 206 insertions(+), 13 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 aead7cd96a..ef3b8e1c98 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 @@ -1597,13 +1597,24 @@ public class ConnectionPool { */ @Override public boolean cancel(boolean mayInterruptIfRunning) { - if (pc!=null) { + //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) { return false; - } else if ((!cancelled.get()) && cancelled.compareAndSet(false, true)) { - //cancel by retrieving the connection and returning it to the pool - ConnectionPool.this.cancellator.execute(this); + } else if (cancelled.compareAndSet(false, true)) { + if (!pcFuture.cancel(mayInterruptIfRunning)) { + //The pooled connection was already handed over to the + //queue future (or is being handed over right now), so + //cancel by retrieving the connection and returning it + //to the pool + ConnectionPool.this.cancellator.execute(this); + } + return true; } - return true; + //already cancelled + return false; } /** 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 9dd41e5c1e..c76ea430e6 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 @@ -21,6 +21,7 @@ import java.util.Iterator; import java.util.LinkedList; import java.util.NoSuchElementException; 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; @@ -192,7 +193,12 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { } /** - * Request an item from the queue asynchronously + * Request an item from the queue asynchronously. + * If no item is available, the returned future registers a waiter on the + * queue. Cancelling the future deregisters that waiter again, provided the + * 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. * @return - a future pending the result from the queue poll request */ public Future<E> pollAsync() { @@ -411,10 +417,12 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { //------------------------------------------------------------------ - // Non cancellable Future used to check and see if a connection has been made available + // Future used to check and see if a connection has been made available //------------------------------------------------------------------ /** - * A non-cancellable future used to check if a connection has been made available. + * 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. * * @param <T> the type of the item */ @@ -446,7 +454,33 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { @Override public boolean cancel(boolean mayInterruptIfRunning) { - return false; //don't allow cancel for now + if (item!=null) { + //the item is already available, too late to cancel + return false; + } + ExchangeCountDownLatch<T> l = latch; + if (l==null) { + return false; + } + final ReentrantLock lock = FairBlockingQueue.this.lock; + lock.lock(); + 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) { + return false; + } + canceled = true; + //deregister the waiter, so that a subsequent offer() does not + //hand an item over to a future nobody is waiting on anymore + waiters.remove(l); + } finally { + lock.unlock(); + } + //wake up any thread blocked in one of the get() methods, so that + //it can observe the cancellation instead of waiting forever + l.countDown(); + return true; } @Override @@ -454,7 +488,13 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { if (item!=null) { return item; } else if (latch!=null) { + if (canceled) { + throw new CancellationException(); + } latch.await(); + if (canceled) { + throw new CancellationException(); + } return latch.getItem(); } else { throw new ExecutionException("ItemFuture incorrectly instantiated. Bug in the code?", new Exception()); @@ -466,8 +506,30 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { if (item!=null) { return item; } else if (latch!=null) { + if (canceled) { + throw new CancellationException(); + } boolean timedout = !latch.await(timeout, unit); + if (canceled) { + throw new CancellationException(); + } 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; + lock.lock(); + try { + waiters.remove(latch); + } 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) { + return delivered; + } throw new TimeoutException(); } else { return latch.getItem(); @@ -479,12 +541,12 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { @Override public boolean isCancelled() { - return false; + return canceled; } @Override public boolean isDone() { - return (item!=null || latch.getItem()!=null); + return (item!=null || canceled || (latch!=null && latch.getItem()!=null)); } } 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 1a2368b874..c32daf8f92 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 @@ -19,8 +19,11 @@ package org.apache.tomcat.jdbc.pool; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Assertions; @@ -30,8 +33,9 @@ import org.junit.jupiter.api.Test; * Expected functionality of the idle-connection queue the pool uses when * the fair queue is enabled: unbounded offering, most-recently-used * hand-off, blocking retrieval with timeout, fairness towards the - * longest waiting consumer, and the collection operations the queue - * deliberately leaves unsupported. + * longest waiting consumer, deregistration of cancelled asynchronous + * polls, and the collection operations the queue deliberately leaves + * unsupported. */ class TestFairBlockingQueue { @@ -192,4 +196,104 @@ class TestFairBlockingQueue { Assertions.assertEquals(1, queue.size(), "the unsupported operations must not mutate the queue"); } + + @Test + void cancelledAsyncPollIsDeregistered() throws Exception { + FairBlockingQueue<String> queue = new FairBlockingQueue<>(); + Future<String> future = queue.pollAsync(); + Assertions.assertTrue(future.cancel(true), + "an async poll that has not received its element must be cancellable"); + Assertions.assertFalse(future.cancel(true), + "a second cancel must report nothing cancelled"); + Assertions.assertTrue(future.isCancelled()); + Assertions.assertTrue(future.isDone()); + Assertions.assertThrows(CancellationException.class, future::get); + // the cancelled waiter must not consume a subsequently offered element + Assertions.assertTrue(queue.offer("keep")); + Assertions.assertEquals(1, queue.size(), + "an element offered after the cancellation must stay in the queue"); + Assertions.assertEquals("keep", queue.poll()); + } + + @Test + void cancelAfterDeliveryReturnsFalse() 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"); + Assertions.assertFalse(future.cancel(true), + "a future that already received its element must not be cancellable"); + Assertions.assertFalse(future.isCancelled()); + Assertions.assertTrue(future.isDone()); + Assertions.assertEquals("handed", future.get(5, TimeUnit.SECONDS)); + // the element belongs to the future, not to the queue + Assertions.assertEquals(0, queue.size()); + } + + @Test + void getIsReleasedWithCancellationWhenCancelledWhileBlocked() throws Exception { + 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 caller = new Thread(() -> { + started.countDown(); + try { + future.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() + Assertions.assertTrue(future.cancel(true)); + Assertions.assertTrue(done.await(5, TimeUnit.SECONDS), + "cancellation must release a caller blocked in get()"); + Assertions.assertTrue(error.get() instanceof CancellationException, + "a cancelled future must fail get() with CancellationException, got " + + error.get()); + caller.join(1000); + } + + @Test + void cancelledWaiterDoesNotStealTheHandoffFromTheNextWaiter() throws Exception { + FairBlockingQueue<String> queue = new FairBlockingQueue<>(); + Future<String> cancelled = queue.pollAsync(); // first in line + Future<String> active = queue.pollAsync(); // second in line + Assertions.assertTrue(cancelled.cancel(true)); + Assertions.assertTrue(queue.offer("next")); + Assertions.assertEquals("next", active.get(5, TimeUnit.SECONDS), + "the element must go to the next waiter in line"); + Assertions.assertEquals(0, queue.size()); + } + + @Test + void cancelOfImmediatelyCompletedAsyncPollReturnsFalse() throws Exception { + FairBlockingQueue<String> queue = new FairBlockingQueue<>(); + queue.offer("item"); + Future<String> future = queue.pollAsync(); // completes right away + Assertions.assertFalse(future.cancel(true), + "an already completed future must not be cancellable"); + Assertions.assertTrue(future.isDone()); + Assertions.assertFalse(future.isCancelled()); + Assertions.assertEquals("item", future.get()); + Assertions.assertEquals(0, queue.size()); + } + + @Test + void timedOutAsyncPollIsDeregistered() throws Exception { + FairBlockingQueue<String> queue = new FairBlockingQueue<>(); + 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()); + } } diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index d7746dec31..82f748e057 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -841,6 +841,22 @@ missing configuration, and the documentation now states the requirement. (remm) </fix> + <fix> + Make the asynchronous poll of the JDBC pool's fair queue + cancellable. Cancelling the future returned by + <code>pollAsync()</code> was a no-op, so the waiter it had + registered stayed on the queue indefinitely and a later + <code>offer()</code> handed an item – in the pool, a returned + connection – to a future nobody was reading, losing it from the + pool. Cancellation now deregisters the waiter as long as the + item has not already been handed over, a cancelled future fails + its <code>get()</code> methods with + <code>CancellationException</code> instead of blocking forever, + and the pool's own connection future uses this to release its + queue waiter, only falling back to retrieving and closing an + already in-flight connection when the cancellation comes too + late. (remm) + </fix> </changelog> </subsection> <subsection name="Other"> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
