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 d59ee94ac7dac05031bc081cdb34d6cd013c9323 Author: opencode <[email protected]> AuthorDate: Thu Nov 5 16:06:27 2026 +0100 Withdraw a statement parked while the cache purge runs StatementCache.disconnected() iterated the weakly consistent cache and then called clear(), so a statement parked between the iterator snapshot and the clear was removed without the matching counter decrement and could stay open. Serialising parks against the purge with a lock would put a lock on every statement close, which is a hot path. Use the ConcurrentHashMap itself: set a volatile draining flag before the sweep, have cacheStatement() re-check it after a successful putIfAbsent and withdraw its own entry if the purge started, and sweep until empty while removing each entry identity-aware and decrementing once per removal. A park that removes its own entry lets the caller close it; one the purge removed is closed by the purge. Add a deterministic gated-park test and keep the existing two-park rendezvous test. --- .../jdbc/pool/interceptor/StatementCache.java | 54 +++++++++++++-- .../interceptor/TestStatementCacheRacingPark.java | 81 ++++++++++++++++++++++ 2 files changed, 131 insertions(+), 4 deletions(-) diff --git a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/interceptor/StatementCache.java b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/interceptor/StatementCache.java index 01ac2a3b3b..f2f6384fb1 100644 --- a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/interceptor/StatementCache.java +++ b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/interceptor/StatementCache.java @@ -23,7 +23,9 @@ import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; +import java.util.ArrayList; import java.util.Arrays; +import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; @@ -184,6 +186,12 @@ public class StatementCache extends StatementDecoratorInterceptor implements Sta private AtomicInteger cacheSize; + /** + * Whether {@link #disconnected(ConnectionPool, PooledConnection, boolean)} is purging this cache. A park that + * observes it set refuses to cache; a park that inserted just as the purge started withdraws its own entry. + */ + private volatile boolean draining; + /** * Whether a tracker that closes its tracked statements at the connection return sits beneath this interceptor in * the chain. A {@link StatementFinalizer} below the cache tracks the driver's statement - its close-aware proxy is @@ -248,6 +256,7 @@ public class StatementCache extends StatementDecoratorInterceptor implements Sta oname = null; } } else { + draining = false; cacheSize = cacheSizeMap.get(parent); this.pcon = con; if (!pcon.getAttributes().containsKey(STATEMENT_CACHE_ATTR)) { @@ -266,12 +275,31 @@ public class StatementCache extends StatementDecoratorInterceptor implements Sta @SuppressWarnings("unchecked") ConcurrentHashMap<CacheKey, CachedStatement> statements = (ConcurrentHashMap<CacheKey, CachedStatement>) con .getAttributes().get(STATEMENT_CACHE_ATTR); - + List<CachedStatement> toClose = null; + // Every park is refused from here on. A park that read the flag + // before this write and inserted during the walk withdraws its own + // entry when it re-reads it, so sweeping until the map is empty closes + // exactly the entries the purge owns. Each removal decrements the + // counter once; closing runs after the sweep because it reaches the + // driver. + draining = true; if (statements != null) { - for (Map.Entry<CacheKey, CachedStatement> p : statements.entrySet()) { - closeStatement(p.getValue()); + do { + for (Map.Entry<CacheKey, CachedStatement> entry : statements.entrySet()) { + CachedStatement statement = entry.getValue(); + if (statement != null && removeStatementInternal(statement)) { + if (toClose == null) { + toClose = new ArrayList<>(); + } + toClose.add(statement); + } + } + } while (!statements.isEmpty()); + } + if (toClose != null) { + for (CachedStatement statement : toClose) { + statement.forceClose(); } - statements.clear(); } super.disconnected(parent, con, finalizing); @@ -373,6 +401,20 @@ public class StatementCache extends StatementDecoratorInterceptor implements Sta * @return true if the statement was cached, false otherwise */ public boolean cacheStatement(CachedStatement proxy) { + if (draining) { + return false; + } + boolean parked = cacheStatementInternal(proxy); + if (parked && draining) { + // The purge may have started while this park was in flight. If + // the entry is still ours, withdraw it and let the caller close + // it; if the purge already removed it, the purge closes it. + return !removeStatementInternal(proxy); + } + return parked; + } + + private boolean cacheStatementInternal(CachedStatement proxy) { ConcurrentHashMap<CacheKey, CachedStatement> cache = getCache(); // Read the counter once, like getCacheSize() and closeInvoked() do: // a disposal (the reset(null, null) walks) nulls the field, and every @@ -438,6 +480,10 @@ public class StatementCache extends StatementDecoratorInterceptor implements Sta * @return true if the statement was removed, false otherwise */ public boolean removeStatement(CachedStatement proxy) { + return removeStatementInternal(proxy); + } + + private boolean removeStatementInternal(CachedStatement proxy) { ConcurrentHashMap<CacheKey, CachedStatement> cache = getCache(); // The counter is read once for the same disposal window as in // cacheStatement: a disposal is answered like at the other entry diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheRacingPark.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheRacingPark.java index 922989e2b1..6865800ac4 100644 --- a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheRacingPark.java +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheRacingPark.java @@ -21,6 +21,7 @@ import java.sql.PreparedStatement; import java.sql.Statement; import java.util.concurrent.BrokenBarrierException; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -129,6 +130,86 @@ class TestStatementCacheRacingPark extends JdbcPoolBaseTest { } } + /** + * A cache map that holds the next park at {@code putIfAbsent} until the + * test releases it, so a disconnect purge can be started while the park + * is provably still in flight. + */ + static final class GatedParkMap extends ConcurrentHashMap<Object,Object> { + + private static final long serialVersionUID = 1L; + + final CountDownLatch entered = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + + @Override + public Object putIfAbsent(Object key, Object value) { + entered.countDown(); + try { + release.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS); + } catch (InterruptedException x) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("interrupted at the gated park", x); + } + return super.putIfAbsent(key, value); + } + } + + @Test + void aParkRacingTheDisconnectPurgeIsCountedAndClosed() throws Exception { + DataSource ds = newDataSource(); + ds.getPoolProperties().setJdbcInterceptors( + StatementCache.class.getName() + "(prepared=true,max=5)"); + ds.getPoolProperties().setMaxActive(1); + ds.getPoolProperties().setTimeBetweenEvictionRunsMillis(60000); + ds.createPool(); + + Connection shared = ds.getConnection(); + try { + StatementCache cache = interceptorOf(shared, StatementCache.class); + PooledConnection pcon = (PooledConnection) shared.unwrap(PooledConnection.class); + GatedParkMap gated = new GatedParkMap(); + pcon.getAttributes().put(StatementCache.STATEMENT_CACHE_ATTR, gated); + + PreparedStatement closing = shared.prepareStatement("SELECT 1"); + Statement driver = driverStatement(closing); + AtomicInteger parkFailure = new AtomicInteger(); + Thread park = new Thread(() -> { + try { + closing.close(); + } catch (Throwable t) { + parkFailure.incrementAndGet(); + } + }, "gated-park"); + park.start(); + Assertions.assertTrue(gated.entered.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS), + "precondition: the park must be held inside the cache map"); + + Thread purge = new Thread(() -> cache.disconnected(ds.getPool(), pcon, false), + "disconnect-purge"); + purge.start(); + // On the fixed code the purge waits for the park to leave the + // critical section; on the old code it completes now, before the + // statement is inserted. + purge.join(500); + gated.release.countDown(); + park.join(AWAIT_TIMEOUT_MS); + purge.join(AWAIT_TIMEOUT_MS); + Assertions.assertFalse(park.isAlive() || purge.isAlive(), + "the park and the purge must both finish"); + Assertions.assertEquals(0, parkFailure.get(), "the close must not fail"); + + Assertions.assertEquals(0, gated.size(), + "the purge must remove the statement that was parked while it ran"); + Assertions.assertEquals(0, cache.getCacheSize(), + "the pool-wide counter must not retain a purged slot"); + Assertions.assertTrue(driver.isClosed(), + "the statement parked during the purge must be closed, not leaked"); + } finally { + shared.close(); + } + } + @Test void twoConcurrentParksOfOneKeyParkOneStatementAndCountOne() throws Exception { DataSource ds = newDataSource(); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
