From cf2068420f92f25985c22a6cdb16c17d9ceb1efb Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Sat, 05 Sep 2026 18:16:09 +0000
Subject: [PATCH] [#878] Bound the JDBC connection pool and expire its connections one by one (#884)
---
opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java | 2
opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java | 631 ++++++++++++++++++++++
opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStatementBoundTestCase.java | 36 +
opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java | 682 +++++++++++++++++++++---
opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java | 285 +++++++--
5 files changed, 1,462 insertions(+), 174 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java
index 22d63d4..d59e1f4 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java
@@ -15,14 +15,12 @@
*/
package org.opends.server.backends.jdbc;
-import com.github.benmanes.caffeine.cache.Caffeine;
-import com.github.benmanes.caffeine.cache.LoadingCache;
-import com.github.benmanes.caffeine.cache.RemovalCause;
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
+import org.opends.server.api.WorkQueue;
+import org.opends.server.core.DirectoryServer;
import java.sql.*;
-import java.time.Duration;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
@@ -36,6 +34,8 @@
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -121,6 +121,32 @@
/** 57P03, cannot_connect_now: postgresql starting up, shutting down or in recovery. */
private static final String NOT_ACCEPTING_YET_SQL_STATE = "57P03";
+ /**
+ * The greatest number of connections one pool holds to one database; 0 for no bound. Read once
+ * per pool, when the first borrow of a connection string creates it, unlike the bounds of a
+ * borrow above: a pool is never removed from the map, and the permits of one already created
+ * are not resized, so this one takes a restart of the server to change.
+ */
+ static final String POOL_MAX_PROPERTY = "org.openidentityplatform.opendj.jdbc.pool.max";
+ /**
+ * Sized like the worker thread pool the server sizes for itself
+ * ({@code Platform.computeNumberOfThreads(16, 2)}), since an operation borrows one connection for
+ * its duration: the bound is there to keep a burst from opening as many connections as the
+ * database will accept, not to throttle steady traffic.
+ * <p>
+ * That formula is only what {@code WorkQueue.computeNumWorkerThreads} falls back to. A configured
+ * {@code ds-cfg-num-worker-threads} replaces it outright, and there is no default that can follow
+ * it: this bound belongs to a database that two backends may share, while that count belongs to
+ * the server. So an installation that raised it is told at open where the two stand, by
+ * {@link #reportBoundBelowBorrowers}, rather than left to find the wait in a latency graph.
+ */
+ static final int DEFAULT_POOL_MAX = Math.max(16, Runtime.getRuntime().availableProcessors() * 2);
+
+ /** How long a borrow waits for a connection to be returned before looking at the pool again. */
+ private static final long POOL_FULL_POLL_MS = 250;
+ /** The sweep runs at half the TTL, and no more often than this. */
+ private static final long MIN_SWEEP_INTERVAL_MS = 1000;
+
static final long MAX_BACKOFF_MS = 1000;
static final long STALL_WARNING_AFTER_MS = 1000;
static final long STALL_WARNING_INTERVAL_MS = 10000;
@@ -171,36 +197,49 @@
*/
private static final Map<String, Long> poolDistrustedAt = new ConcurrentHashMap<>();
+ // Throttled like the stall warning above, and keyed the same way: the bound is one setting, so
+ // one line per interval says so - but it is a setting of one pool, and two backends standing
+ // full at once each have their own to report. A single timestamp would let the pool that
+ // reported first silence the other, whose operations are failing with nothing in the log
+ // naming the database behind them.
+ private static final Map<String, AtomicLong> lastPoolFullWarning = new ConcurrentHashMap<>();
+
final Connection parent;
- // A deque handed out from the end it is returned to: the connection borrowed next is the one
- // returned last, so under any load the pool keeps reusing its hottest connections instead of
- // walking round every one it ever opened. That is what gives the window above anything to
- // bypass - a connection reached only after a whole cycle of the pool has been idle far longer
- // than the window - and it leaves the connections nothing needs at the cold end of the deque,
- // where the per-connection idle expiry of #878 can find them. Until that lands, the cold end
- // is reached only when the whole pool expires, after DEFAULT_TTL_MS with the backend idle.
- // A deque takes one lock for both of its ends where the queue it replaces took one for each,
- // so a borrow and a return no longer proceed side by side - against the round trip the window
- // above saves, and the connect the reuse saves, that lock is not worth a FIFO handoff.
- static LoadingCache<String, BlockingDeque<CachedConnection>> cached = Caffeine.newBuilder()
- .expireAfterAccess(Duration.ofMillis(getCacheTtlMillis()))
- .removalListener((String key, BlockingDeque<CachedConnection> value, RemovalCause cause) -> {
- for (CachedConnection con : value) {
- try {
- if (!con.isClosed()) {
- con.parent.close();
- }
- } catch (SQLException e) {
- // ignore
- }
- }
- })
- .build(conStr -> new LinkedBlockingDeque<>());
+ /** The pool this connection belongs to, held directly so that the return needs no lookup. */
+ private final Pool pool;
+ /** Whether this connection holds a permit of its pool: a reentrant borrow does not. */
+ private final boolean metered;
+ /**
+ * The depth counter of the thread that borrowed it, lowered by the return. Held rather than the
+ * thread itself: a return made on another thread has to lower the depth of the borrower all the
+ * same, and a check of the returning thread against the borrowing one left that depth standing -
+ * the borrower was then taken for a nested borrow for the life of the server, exempt from the
+ * wait at the bound and opening an unmetered connection, destroyed on return, per operation
+ * (issue #878).
+ */
+ private volatile AtomicInteger depth;
+ /** When it was last returned to the pool, which is what the TTL is measured from. */
+ volatile long returnedAtMillis;
+ private final AtomicBoolean permitReleased = new AtomicBoolean();
+ /** Whether it has been handed back already: JDBC makes close() on a closed connection a no-op. */
+ private final AtomicBoolean returned = new AtomicBoolean();
+
+ /** The pool of every connection string in use, kept until the last storage using it closes. */
+ static final ConcurrentMap<String, Pool> pools = new ConcurrentHashMap<>();
+
+ /** The sweep that closes connections nothing has borrowed for the TTL, started with the first pool. */
+ private static volatile ScheduledExecutorService sweeper;
+
+ /** Where the sweep closes what it reaped, so that a close which does not return keeps it: see {@link Pool#sweep}. */
+ private static volatile Executor closer = DIRECT_EXECUTOR;
/**
* Returns the time after which an idle pooled connection is closed, as configured by the
* {@value #TTL_PROPERTY} system property. An invalid value is ignored in favor of the default.
+ * <p>
+ * Read on every borrow and every sweep rather than once, so that it can be changed on a running
+ * server the way the bounds of a borrow can.
*/
private static long getCacheTtlMillis() {
return getNonNegativeProperty(TTL_PROPERTY, DEFAULT_TTL_MS, "ms");
@@ -215,8 +254,10 @@
* value the unit conversion saturates on from leaving every connection of the pool trusted for
* the life of the server.
* <p>
- * Read at class initialization, like the ttl it is clamped to, so a value set after that
- * changes neither.
+ * Read at class initialization, so a value of this property set after that does not change the
+ * window. The ttl is not: {@link #getCacheTtlMillis()} is read on every borrow and every sweep,
+ * and the clamp above is not applied again - the window keeps the value it was computed with,
+ * so a ttl lowered on a running server does not lower the window with it.
*/
static long getAliveBypassMillis() {
long configured = getNonNegativeProperty(ALIVE_BYPASS_PROPERTY, DEFAULT_ALIVE_BYPASS_MS, "ms");
@@ -240,6 +281,386 @@
return configured;
}
+ /** The pool of a connection string, created on first use. */
+ static Pool poolOf(String connectionString) {
+ final Pool pool = pools.computeIfAbsent(connectionString, Pool::new);
+ startSweeper();
+ return pool;
+ }
+
+ private static void startSweeper() {
+ if (sweeper != null) {
+ return;
+ }
+ synchronized (pools) {
+ if (sweeper == null) {
+ // A thread per close in flight, and none while nothing is being closed. One thread
+ // shared by all of them would only move the head of the line, which is the point
+ // of not closing on the sweeper in the first place.
+ closer = Executors.newCachedThreadPool(runnable -> {
+ final Thread thread = new Thread(runnable, "JDBC backend connection pool closer");
+ thread.setDaemon(true);
+ return thread;
+ });
+ final ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(runnable -> {
+ final Thread thread = new Thread(runnable, "JDBC backend connection pool sweeper");
+ thread.setDaemon(true);
+ return thread;
+ });
+ sweeper = service;
+ // Rescheduled after each run rather than left at a fixed delay: the interval comes
+ // from the ttl, and the ttl is read on every borrow and every sweep so that it can
+ // be changed on a running server. A delay computed once would keep the sweeper of a
+ // lowered ttl waking as rarely as the old one, so connections would go on being
+ // reaped no sooner than the setting the operator replaced (issue #878).
+ scheduleNextSweep(service);
+ }
+ }
+ }
+
+ /** Half the ttl, and no more often than {@value #MIN_SWEEP_INTERVAL_MS} ms. */
+ private static long sweepIntervalMillis() {
+ return Math.max(MIN_SWEEP_INTERVAL_MS, getCacheTtlMillis() / 2);
+ }
+
+ /**
+ * Books the next sweep, and the one after it out of its own run. Every run books its successor
+ * in a finally: a sweep that ends in a Throwable the per-pool guard did not catch would
+ * otherwise stop the expiry of every pool in the JVM, the way a task thrown out of
+ * scheduleWithFixedDelay does.
+ */
+ private static void scheduleNextSweep(ScheduledExecutorService service) {
+ try {
+ service.schedule(() -> {
+ try {
+ sweep();
+ } finally {
+ scheduleNextSweep(service);
+ }
+ }, sweepIntervalMillis(), TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // the sweeper is shutting down: there is nothing left to book a run on
+ logger.traceException(e);
+ }
+ }
+
+ // Expiry has to happen without a borrow behind it. Caffeine was left without a scheduler, so an
+ // entry was only ever expired by a later cache operation - and a backend that has gone idle,
+ // the one case the TTL exists for, performs none (issue #878).
+ static void sweep() {
+ final long ttlMillis = getCacheTtlMillis();
+ final Executor closeOn = closer;
+ for (final Pool pool : pools.values()) {
+ try {
+ pool.sweep(ttlMillis, closeOn);
+ } catch (Throwable t) {
+ // Error included: scheduleWithFixedDelay cancels a task that throws, so anything
+ // escaping here would stop the expiry of every pool in the JVM for good - and
+ // silently, which is the failure mode the hand-off of the close exists to avoid.
+ logger.traceException(t);
+ }
+ }
+ }
+
+ /**
+ * Registers a storage as a user of the pool of a connection string. Reference counted because a
+ * pool belongs to a database rather than to a backend: two backends may address one database,
+ * and closing one of them must not take the connections of the other with it.
+ */
+ static void openPool(String connectionString) {
+ final Pool pool = poolOf(connectionString);
+ pool.addUser();
+ reportBoundBelowBorrowers(connectionString, pool);
+ }
+
+ /**
+ * Reports a bound smaller than the number of worker threads. An operation borrows one connection
+ * for its duration, so the worker threads are the borrowers this default is sized against - and
+ * it is sized against the count the server computes for itself, not against a
+ * {@code ds-cfg-num-worker-threads} the operator set, which replaces that count outright.
+ * <p>
+ * A lower bound than that is what is reported, not every way past it: the replay threads of
+ * replication default to the same count again and borrow on top of the workers, and an import or
+ * a rebuild borrows besides. So this names one difference the operator can act on rather than
+ * standing for the whole demand on the pool.
+ * <p>
+ * Nothing fails for the difference alone: the surplus waits for a connection to be returned,
+ * which is what the bound is there for. But every one of those waits is paid on an operation,
+ * and past {@value #POOL_TIMEOUT_PROPERTY} the operation fails - on a setting whose effect on
+ * this backend the operator had no reason to expect (issue #878).
+ */
+ private static void reportBoundBelowBorrowers(String connectionString, Pool pool) {
+ final WorkQueue<?> workQueue = DirectoryServer.getWorkQueue();
+ if (workQueue == null) {
+ // an offline tool, or the server before its work queue is up: no borrowers to count
+ return;
+ }
+ final int borrowers = workQueue.getNumWorkerThreads();
+ if (borrowers <= pool.max()) {
+ return;
+ }
+ final long poolTimeoutSeconds = getNonNegativeProperty(POOL_TIMEOUT_PROPERTY, DEFAULT_POOL_TIMEOUT_SECONDS, "s");
+ final String wait = poolTimeoutSeconds == 0
+ ? "waits for one to be returned for as long as that takes"
+ : "waits up to " + poolTimeoutSeconds + "s for one to be returned and fails if none is";
+ warnOnce(safeUrl(connectionString) + "|bound-below-borrowers",
+ "the connection pool of %s holds at most %d connections while %d worker threads may each borrow one:"
+ + " an operation finding it at its bound %s (raise %s to allow more connections, or lower"
+ + " ds-cfg-num-worker-threads)",
+ safeUrl(connectionString), pool.max(), borrowers, wait, POOL_MAX_PROPERTY);
+ }
+
+ /** Unregisters a storage; the connections are released once the last user is gone. */
+ static void closePool(String connectionString) {
+ final Pool pool = pools.get(connectionString);
+ if (pool != null) {
+ pool.removeUser();
+ }
+ }
+
+ /**
+ * The connections of one connection string.
+ * <p>
+ * This replaces the cache entry that used to hold them. That one carried the TTL on the pool
+ * rather than on a connection - {@code expireAfterAccess} keyed by the connection string, reset
+ * by every borrow and every return - so under continuous traffic nothing ever expired and the
+ * peak count of a burst stayed open for as long as the backend saw any traffic at all. It also
+ * had no bound, so the only ceiling on the connections of a backend was the {@code
+ * max_connections} of the database itself (issue #878).
+ */
+ static final class Pool {
+ final String connectionString;
+ /** Idle connections, most recently returned first: the ones a burst opened sink to the bottom, where the sweep finds them. */
+ private final LinkedBlockingDeque<CachedConnection> idle = new LinkedBlockingDeque<>();
+ /** One permit per live connection, borrowed or idle. Sized once: this is how large the pool may grow, not a rate. */
+ private final Semaphore permits;
+ private final int max;
+ /**
+ * How many connections of this pool the current thread holds. A borrow made while one is
+ * already held may exceed the bound, because the two are held at the same time and waiting
+ * for the first to be returned would wait for this very thread:
+ * {@code PersistentCompressedSchema.store()} opens a write of its own - the definition has
+ * to commit independently of the entry - and {@code EntryContainer.modifyDN} reaches it
+ * from inside a transaction, having encoded the entry there. The exemption is from the
+ * wait rather than from the pool: a nested borrow served out of the idle deque carries the
+ * permit that connection already holds and is pooled again on return like any other. Only
+ * one that had to establish a connection of its own, because the pool stood at its bound,
+ * holds no permit - and that one is closed rather than pooled when it comes back, so the
+ * pool does not grow past its bound.
+ * <p>
+ * Counted per pool rather than per thread, because that deadlock only exists within one
+ * pool: a count shared by all of them would judge a thread holding a connection to one
+ * database reentrant while it borrows from another, passing the bound of a pool it holds
+ * nothing of and destroying the connection instead of pooling it, on every operation.
+ */
+ private final ThreadLocal<AtomicInteger> held = ThreadLocal.withInitial(AtomicInteger::new);
+ /** Open storages using this pool, guarded by this. */
+ private int users;
+ /**
+ * Set when the last storage using this pool closed. A pool no storage ever registered with -
+ * a borrow made straight through {@link CachedConnection#getConnection}, as the tests do -
+ * is not closed and pools normally; only one that had a user and lost it stops keeping
+ * connections for a borrower that is not going to come.
+ */
+ private volatile boolean closed;
+
+ Pool(String connectionString) {
+ this.connectionString = connectionString;
+ final long configured = getNonNegativeProperty(POOL_MAX_PROPERTY, DEFAULT_POOL_MAX, "connections");
+ this.max = (configured == 0 || configured > Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int) configured;
+ this.permits = new Semaphore(max);
+ }
+
+ int max() {
+ return max;
+ }
+
+ /** Whether the calling thread already holds a connection of this pool. */
+ boolean heldByCurrentThread() {
+ return held.get().get() > 0;
+ }
+
+ /**
+ * Raises the depth of the borrowing thread and hands back the counter it was raised on, for
+ * the connection to lower on its return. The counter rather than the thread, because the
+ * return need not happen on the thread that borrowed - and the depth that has to come down
+ * is the borrower's either way. Read by that thread alone but written by whichever returns
+ * the connection, which is why it is an AtomicInteger and not an int.
+ */
+ AtomicInteger enter() {
+ final AtomicInteger depth = held.get();
+ depth.incrementAndGet();
+ return depth;
+ }
+
+ /** Lowers a depth this pool handed out, never below zero. */
+ static void leave(AtomicInteger depth) {
+ depth.updateAndGet(held -> held > 0 ? held - 1 : 0);
+ }
+
+ int idleCount() {
+ return idle.size();
+ }
+
+ /**
+ * The connections of this pool holding a permit, borrowed and idle together. Not every
+ * connection of the pool: a borrow nested in one this thread already holds goes on
+ * unmetered when the pool stands at its bound, so the connections this count misses are
+ * exactly the ones over the bound. They take no place in it and are closed rather than
+ * pooled when they come back, which makes this the count the bound is about - how much of
+ * it is taken - rather than the number of sockets open to the database.
+ */
+ int meteredCount() {
+ return max - permits.availablePermits();
+ }
+
+ synchronized void addUser() {
+ users++;
+ closed = false;
+ }
+
+ void removeUser() {
+ final boolean wasLast;
+ synchronized (this) {
+ wasLast = users > 0 && --users == 0;
+ if (wasLast) {
+ closed = true;
+ }
+ }
+ if (wasLast) {
+ // Outside the monitor: closing a connection is a round trip, and an open of the
+ // same database has no reason to wait behind it. The borrowed ones are not here to
+ // be closed - give() closes them when they come back, since a pool nobody uses must
+ // not keep them for a borrower that is not going to come.
+ logger.trace(LocalizableMessage.raw("releasing %d pooled connections of %s: its last user closed",
+ idle.size(), safeUrl(connectionString)));
+ drainIdle();
+ }
+ }
+
+ void drainIdle() {
+ for (CachedConnection con = idle.pollFirst(); con != null; con = idle.pollFirst()) {
+ destroy(con);
+ }
+ }
+
+ /**
+ * Takes a connection out of the pool, waiting up to waitMs for one to be returned, and
+ * discarding the ones that are broken or have been idle for longer than the TTL.
+ * <p>
+ * Bounded by the deadline of the borrow, and not only by waitMs: a poll of no duration
+ * still hands out whatever the deque holds, and discarding a connection whose socket is
+ * half-open costs the validation timeout apiece. The pool holds as many of those as its
+ * bound allows, so draining the deque overran the bound the operator set - by minutes on a
+ * large pool, before the connect that follows it had even started (issue #878).
+ */
+ CachedConnection pollIdle(long waitMs, long ttlMillis, long deadline, boolean trusted)
+ throws InterruptedException {
+ long remainingWait = waitMs;
+ while (true) {
+ final long polledAt = System.currentTimeMillis();
+ final CachedConnection con = idle.pollFirst(remainingWait, TimeUnit.MILLISECONDS);
+ if (con == null) {
+ return null;
+ }
+ if (System.currentTimeMillis() - con.returnedAtMillis <= ttlMillis && isUsable(con, trusted)) {
+ return con;
+ }
+ destroy(con);
+ final long remaining = deadline - System.currentTimeMillis();
+ if (remaining <= 0) {
+ return null;
+ }
+ // one more look, since a connection may have been returned in the meantime
+ remainingWait = Math.min(Math.max(0, remainingWait - (System.currentTimeMillis() - polledAt)), remaining);
+ }
+ }
+
+ /** Takes the right to hold one more connection, or reports that the pool is full. */
+ boolean tryReserve() {
+ return permits.tryAcquire();
+ }
+
+ void cancelReservation() {
+ permits.release();
+ }
+
+ /** Hands a connection back, closing it rather than pooling it when it may not be kept. */
+ void give(CachedConnection con) {
+ // An unmetered connection holds no permit, so pooling it would put the pool one over its
+ // bound for good; and a closed pool has nobody left to hand it to.
+ if (con.metered && !closed) {
+ addIdle(con);
+ if (closed) {
+ // The last user left while this one was on its way back, so it missed the drain.
+ drainIdle();
+ }
+ } else {
+ destroy(con);
+ }
+ }
+
+ /** Puts a connection into the pool. The caller must hold the right to keep it there. */
+ void addIdle(CachedConnection con) {
+ con.returnedAtMillis = System.currentTimeMillis();
+ idle.addFirst(con);
+ }
+
+ void destroy(CachedConnection con) {
+ try {
+ closeQuietly(con.parent);
+ } finally {
+ // However the close went, the pool holds one connection fewer. A permit not given
+ // back here is given back by nothing at all: only a live connection carries one,
+ // and this one is gone (issue #878).
+ con.releasePermit();
+ }
+ }
+
+ void sweep(long ttlMillis) {
+ sweep(ttlMillis, DIRECT_EXECUTOR);
+ }
+
+ /**
+ * Closes the connections nothing has borrowed for the TTL, handing each to the executor
+ * given rather than closing it here. The sweep of every pool shares one thread and
+ * {@code scheduleWithFixedDelay} never overlaps its runs, so one close that does not
+ * return would stop the expiry of every pool in the JVM - and silently, since only a
+ * thrown exception is logged. Oracle logs off over the network, and the read bound of the
+ * login has been lifted by then (issue #878).
+ */
+ void sweep(long ttlMillis, Executor closeOn) {
+ final long deadline = System.currentTimeMillis() - ttlMillis;
+ // From the tail: the least recently returned connection is the first to have expired,
+ // and once one has not, neither has anything in front of it.
+ for (CachedConnection con = idle.peekLast(); con != null; con = idle.peekLast()) {
+ if (con.returnedAtMillis > deadline) {
+ return;
+ }
+ if (!idle.removeLastOccurrence(con)) {
+ // A borrow took it between the two. What is behind it may still have expired,
+ // and ending the cycle here would leave every one of those open until the
+ // next sweep.
+ continue;
+ }
+ if (con.returnedAtMillis > deadline) {
+ // A borrow took it between the peek and the removal and gave it back, so the
+ // reading the decision was made on is not the one it carries now: closing it
+ // would cost the next borrow a connect over a connection a moment old. Back to
+ // the end it is returned to, where its refreshed reading belongs.
+ idle.addFirst(con);
+ return;
+ }
+ final CachedConnection expired = con;
+ try {
+ closeOn.execute(() -> destroy(expired));
+ } catch (RuntimeException e) { // no thread to close it on: here rather than nowhere
+ destroy(expired);
+ }
+ }
+ }
+ }
+
/**
* Returns the value of a numeric system property, ignoring a value that is not a non-negative
* number in favor of the default. The unit is the one the property is read in, so that the
@@ -596,17 +1017,42 @@
*/
private volatile long lastKnownAliveNanos;
+ /**
+ * A connection outside the accounting of its pool: it holds no permit and is never pooled - the
+ * flag says so as well as the accounting does, since a connection holding no permit is closed
+ * by {@link Pool#give} rather than kept whatever the flag says.
+ * <p>
+ * It still names a pool, because that is what closes it and what the sweep runs over, so the
+ * pool of this connection string is created here if it does not exist yet and the sweeper is
+ * started with it.
+ */
public CachedConnection(String connectionString, Connection parent) {
- this(connectionString, parent, true);
+ this(connectionString, parent, poolOf(connectionString), false, false);
}
- CachedConnection(String connectionString, Connection parent, boolean poolable) {
+ CachedConnection(String connectionString, Connection parent, Pool pool, boolean metered, boolean poolable) {
this.connectionString = connectionString;
this.parent = parent;
+ this.pool = pool;
+ this.metered = metered;
this.poolable = poolable;
this.lastKnownAliveNanos = System.nanoTime();
}
+ /** Gives back the right to hold this connection, once and only if it was taken. */
+ void releasePermit() {
+ if (metered && permitReleased.compareAndSet(false, true)) {
+ pool.cancelReservation();
+ }
+ }
+
+ /** Records that the borrowing thread holds this connection, so a borrow nested in it is recognized. */
+ private static CachedConnection borrowed(CachedConnection con) {
+ con.returned.set(false);
+ con.depth = con.pool.enter();
+ return con;
+ }
+
/**
* Borrows a connection: a usable one out of the pool, or a newly established one. Bounded in
* both phases - every operation of this backend, the open of a backend and the import
@@ -630,26 +1076,65 @@
* them is one borrow of a cold path, where the round trip the window saves is worth nothing.
*/
static Connection getConnection(String connectionString, boolean trusted) throws Exception {
+ final Pool pool = poolOf(connectionString);
final ConnectDialect dialect = ConnectDialect.of(connectionString);
reportUnknownDialect(connectionString, dialect);
final long connectTimeoutSeconds = Math.min(
getNonNegativeProperty(CONNECT_TIMEOUT_PROPERTY, DEFAULT_CONNECT_TIMEOUT_SECONDS, "s"),
Integer.MAX_VALUE / 1000);
final long poolTimeoutSeconds = getNonNegativeProperty(POOL_TIMEOUT_PROPERTY, DEFAULT_POOL_TIMEOUT_SECONDS, "s");
+ final long ttlMillis = getCacheTtlMillis();
final long startedAt = System.currentTimeMillis();
final long deadline = (poolTimeoutSeconds == 0 || poolTimeoutSeconds >= Long.MAX_VALUE / 1000)
? Long.MAX_VALUE : startedAt + poolTimeoutSeconds * 1000;
+ // A thread already holding a connection is not made to wait for one: the two are held at
+ // the same time, so waiting for the first to come back would wait for itself.
+ final boolean reentrant = pool.heldByCurrentThread();
long waitMs = 0;
long backoffMs = 0;
int attempts = 0;
while (true) {
- final CachedConnection pooled = poll(connectionString, waitMs, deadline, trusted);
+ final CachedConnection pooled = pool.pollIdle(waitMs, ttlMillis, deadline, trusted);
if (pooled != null) {
- return pooled;
+ return borrowed(pooled);
+ }
+ // Asked for whether this borrow is nested or not: the exemption a nested one carries is
+ // from the wait, not from the pool. A nested borrow made while the pool has room takes a
+ // permit like any other and is pooled again on return; only one that finds the pool at its
+ // bound goes on unmetered, and that one is closed rather than pooled when it comes back.
+ final boolean metered = pool.tryReserve();
+ if (!metered && !reentrant) {
+ // The pool holds as many connections as it may: only a returned one can serve this
+ // borrow now, and the deadline decides how long that is worth waiting for. This is
+ // the point of the bound - without it the borrow would open one more connection,
+ // and the only ceiling left would be the max_connections of the database itself.
+ final long remaining = deadline - System.currentTimeMillis();
+ if (remaining <= 0) {
+ // The restart is part of the remedy, so the message says so: the bound is read
+ // once, when the pool is created, and a pool is never removed from the map - so
+ // the property set on a running server changes nothing until it is read again.
+ final String message = "no connection to " + safeUrl(connectionString)
+ + " could be borrowed within " + poolTimeoutSeconds + "s: all " + pool.max()
+ + " connections of the pool are in use (raise " + POOL_MAX_PROPERTY
+ + " and restart the server to allow more)";
+ // The one failure the bound introduces has to reach the server log too: an
+ // installation whose peak sits above the default would otherwise see its
+ // operations fail with nothing in the log naming the pool behind it.
+ warnPoolFull(connectionString, message);
+ throw new SQLTimeoutException(message);
+ }
+ waitMs = Math.min(POOL_FULL_POLL_MS, remaining);
+ continue;
}
attempts++;
+ CachedConnection established = null;
+ boolean handedOff = false;
try {
- return connect(connectionString, dialect, attemptSeconds(connectTimeoutSeconds, deadline));
+ established = connect(connectionString, dialect,
+ attemptSeconds(connectTimeoutSeconds, deadline), pool, metered);
+ final CachedConnection con = borrowed(established);
+ handedOff = true;
+ return con;
} catch (SQLException e) {
// A database that takes no connection for the moment is the failure worth waiting
// out: it is at its connection limit, and one of ours is going to come back to the
@@ -680,6 +1165,20 @@
// a driver reporting a connect it will not make as an unchecked failure carries the
// connection string of the backend in its message as readily as a SQLException does
throw reportedUnchecked(e, connectionString);
+ } finally {
+ // What the attempt took is given back on every way out of it, not only on the
+ // SQLException a driver is supposed to throw. DriverManager catches SQLException
+ // alone, so an unchecked failure of a driver reaches here - Connector/J hands a url
+ // with a "%" in it to URLDecoder, and this backend keeps its credentials in the url
+ // - and a permit left behind is left behind for good: only a live connection
+ // carries one, and a failed attempt has none to give (issue #878).
+ if (!handedOff) {
+ if (established != null) {
+ pool.destroy(established); // the permit went with it, and comes back with it
+ } else if (metered) {
+ pool.cancelReservation();
+ }
+ }
}
}
}
@@ -728,33 +1227,6 @@
return Math.max(1, Math.min(bound, Integer.MAX_VALUE / 1000));
}
- /**
- * Takes a usable connection out of the pool, waiting up to waitMs for one to be returned to it.
- * The validation of a connection costs a round trip, and the pool has no upper bound on the
- * number of them it holds, so draining a pool the database no longer answers is given the
- * deadline of the borrow as well: past it, establishing a connection is the faster answer.
- * The connection in hand is always looked at first - trusted or validated, see
- * {@link #isKnownAlive} - whatever the deadline says: a database at its connection limit has
- * no other source of connections than the ones coming back, and one returned to the pool a
- * moment before the deadline is the very connection this borrow waited for. Only a connection
- * the database no longer answers is closed here.
- */
- private static CachedConnection poll(String connectionString, long waitMs, long deadline, boolean trusted)
- throws InterruptedException {
- CachedConnection con = cached.get(connectionString).pollFirst(waitMs, TimeUnit.MILLISECONDS);
- while (con != null) {
- if (isUsable(con, trusted)) {
- return con;
- }
- closeQuietly(con.parent);
- if (System.currentTimeMillis() >= deadline) {
- return null;
- }
- con = cached.get(connectionString).pollFirst();
- }
- return null;
- }
-
private static boolean isUsable(CachedConnection con, boolean trusted) {
if (trusted && isKnownAlive(con)) {
return true;
@@ -833,10 +1305,14 @@
if (distrusted != null && provenAt - distrusted <= 0) { // the overflow safe form of the comparison
return false;
}
- // What the validation this replaces also answered: the removalListener above closes every
- // connection it finds in the deque when the pool expires, and it iterates a weakly
- // consistent view, so a connection taken out by a borrow running at the same time can be
- // closed under it. Answered by the driver out of a flag of its own, not by a round trip.
+ // What the validation this replaces also answered, asked of the driver out of a flag of its
+ // own rather than by a round trip: a connection the driver has already given up on - the
+ // database dropped it and the driver noticed - is not one to hand out on the strength of a
+ // window. It no longer stands for a drain closing a connection under its borrower, the way
+ // it did while the pool was a cache entry whose removalListener iterated a weakly consistent
+ // view of the deque: every path that destroys an idle connection now takes it out of the
+ // deque first (pollIdle, drainIdle, and the removeLastOccurrence of the sweep), so what a
+ // borrow holds is not there to be found (issue #878).
return !isClosed(con.parent);
}
@@ -900,8 +1376,8 @@
return previous < 0 ? 0 : previous;
}
- static CachedConnection connect(String connectionString, ConnectDialect dialect, long connectTimeoutSeconds)
- throws SQLException {
+ static CachedConnection connect(String connectionString, ConnectDialect dialect, long connectTimeoutSeconds,
+ Pool pool, boolean metered) throws SQLException {
// A driver is free to write into the map it is handed, so it gets one of its own.
final Properties properties = new Properties();
final boolean readBoundSet = dialect != null && connectTimeoutSeconds > 0
@@ -926,7 +1402,7 @@
closeQuietly(conNew);
throw e;
}
- final CachedConnection established = new CachedConnection(connectionString, conNew, poolable);
+ final CachedConnection established = new CachedConnection(connectionString, conNew, pool, metered, poolable);
established.lastKnownAliveNanos = provenAt;
return established;
}
@@ -1000,6 +1476,19 @@
return false;
}
+ // The bound of the pool is a reason for an operation to fail that no version before it had,
+ // so it belongs in the server log as well as in the error the client is given. Throttled like
+ // the stall warning: every worker thread reaches it at once when the pool stands full.
+ private static void warnPoolFull(String connectionString, String message) {
+ final long now = System.currentTimeMillis();
+ final AtomicLong lastOfThisUrl =
+ lastPoolFullWarning.computeIfAbsent(safeUrl(connectionString), url -> new AtomicLong());
+ final long last = lastOfThisUrl.get();
+ if (now - last >= STALL_WARNING_INTERVAL_MS && lastOfThisUrl.compareAndSet(last, now)) {
+ logger.warn(LocalizableMessage.raw("%s", message));
+ }
+ }
+
// A stall has to reach the server log: without it a database accepting no further connection
// is indistinguishable from a hang. Throttled, since every operation of the backend borrows
// through here and would otherwise log a copy of its own.
@@ -1386,8 +1875,8 @@
private static void closeQuietly(Connection con) {
try {
con.close();
- } catch (SQLException e) {
- // ignore: it is on its way out anyway
+ } catch (SQLException | RuntimeException e) {
+ // ignore: it is on its way out anyway, and the caller has a permit to give back
}
}
@@ -1433,21 +1922,46 @@
@Override
public void close() throws SQLException {
- try {
- rollback();
- } catch (SQLException e) {
- // A connection that cannot be rolled back must not be handed to the next borrower -
- // and must not be dropped on the floor either: nothing else holds it any more.
- closeQuietly(parent);
- throw e;
- }
- if (!poolable) {
- closeQuietly(parent);
+ // JDBC makes close() on a closed connection a no-op, and this one has to be one: a second
+ // return would put the same connection into the pool twice, to be handed to two borrowers.
+ if (!returned.compareAndSet(false, true)) {
return;
}
- // Returned to the end the next borrow takes it from, so that the pool keeps reusing its
- // hottest connections rather than cycling through every one it ever opened.
- cached.get(connectionString).addFirst(this);
+ final AtomicInteger borrowerDepth = depth;
+ depth = null;
+ if (borrowerDepth != null) {
+ Pool.leave(borrowerDepth);
+ }
+ // Set before the hand-off rather than after it: from the moment give() is called the pool
+ // owns this connection, and a second destroy() of one that reached the idle deque would
+ // close a connection still waiting there to be handed out.
+ boolean handedToPool = false;
+ try {
+ rollback();
+ if (poolable) {
+ // Straight to the pool it came from rather than through a lookup of its connection
+ // string: the entry the lookup returned could be evicted between the two, leaving
+ // the connection in a queue nothing referred to any more - never handed out, never
+ // closed (issue #878).
+ handedToPool = true;
+ pool.give(this);
+ }
+ } finally {
+ // Every way out that is not a give(): the SQLException a rollback is supposed to throw,
+ // a connection that may not be pooled, and the unchecked failure a driver throws
+ // instead of a SQLException. The CAS above has already made this the one close() of
+ // this connection, so what leaves here through neither give() nor destroy() is closed
+ // by nothing at all - and its permit is released by nothing either, since destroy() is
+ // the only caller of releasePermit(). A pool is never removed from the static map, so
+ // that place in the bound would be gone for the life of the server, and enough of them
+ // leave every borrow to fail with a SQLTimeoutException (issue #878).
+ if (!handedToPool) {
+ // destroy() rather than a bare close: the permit this connection holds has to go
+ // back to the pool with it, or the bound loses a place for every connection kept
+ // out of it.
+ pool.destroy(this);
+ }
+ }
}
@Override
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java
index 9f43728..a0059e5 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java
@@ -698,6 +698,17 @@
}
/**
+ * The connection string this storage borrows on, distrusts and closes with: the one
+ * {@link #open(AccessMode)} registered with, and only failing that the one config names now. Every path
+ * that names a pool goes through here, for the reason given in {@link #getConnection(boolean)} - a
+ * db-directory changed on a running backend otherwise sends each of them to a different pool.
+ */
+ private String poolKey() {
+ final String registered=poolConnectionString;
+ return registered!=null ? registered : config.getDBDirectory();
+ }
+
+ /**
* Borrows a connection the pool validates whatever the alive window of
* {@link CachedConnection#ALIVE_BYPASS_PROPERTY} says, for the borrows this class compensates a dropped
* connection on in no other way: {@link #open(AccessMode)}, {@link #removeStorageFiles()} and the importer
@@ -713,17 +724,89 @@
// for the pool stands in for every path that takes a connection. A stand-in of the trusted
// borrow alone let the open, the import and the removal - the three that ask for a validated
// one - reach a real database instead.
+ //
+ // It names the pool this storage registered with in open(), not the one config names now.
+ // Nothing keeps db-directory from being changed on a running backend - applyConfigurationChange()
+ // takes it, isConfigurationChangeAcceptable() refuses nothing, and the component-restart admin
+ // action renders a message rather than holding the change back - so re-reading it here would
+ // borrow from a pool this storage never registered with, leaving the one it did register with
+ // holding a user that never borrows: the leak of #878 back through the configuration. And an
+ // unregistered pool is drained the moment another backend that did register with it closes,
+ // with this one still borrowing from it (issue #878).
Connection getConnection(boolean trusted) throws Exception {
- return CachedConnection.getConnection(config.getDBDirectory(), trusted);
+ return CachedConnection.getConnection(poolKey(), trusted);
}
AccessMode accessMode=AccessMode.READ_ONLY;
+
+ // Whether this storage counts as a user of the pool of its connection string. The pool belongs
+ // to the database rather than to this backend - two backends may address one database - so it
+ // is reference counted, and this flag keeps an open() or a close() that comes twice from
+ // counting twice (issue #878).
+ private final AtomicBoolean poolRegistered=new AtomicBoolean();
+
+ // The connection string open() registered with. applyConfigurationChange() replaces config, so
+ // reading db-directory again at close() could give back the pool of a database this storage
+ // never registered with - leaving the one it did with a user it never loses (issue #878).
+ private volatile String poolConnectionString;
+
@Override
public void open(AccessMode accessMode) throws Exception {
- try (final Connection con=getValidatedConnection()) {
- this.accessMode = accessMode;
- storageStatus = StorageStatus.working();
+ final boolean claimedHere=poolRegistered.compareAndSet(false, true);
+ // Raised once openPool() has returned, which is when a user has actually been added. The
+ // claim alone cannot answer for that: releasePool() on a claim openPool() never made would
+ // take a user off a pool this storage never added one to - and the pool of a database two
+ // backends share would lose the user of the other one, draining connections it is still
+ // borrowing.
+ boolean registeredHere=false;
+ try {
+ // Inside the try, so that the registration this call made is given back however the open
+ // ends - the registration is taken before the pool is of any use, and a pool holding a
+ // user that never borrows keeps its connections for a borrower that is not going to come.
+ if (claimedHere) {
+ poolConnectionString=config.getDBDirectory();
+ CachedConnection.openPool(poolConnectionString);
+ registeredHere=true;
+ }
+ // The validated borrow is the whole of the open, and nothing is taken from it here: the
+ // status is set below rather than inside the block, or a throw from the implicit close()
+ // - the rollback of the return goes to the database - would leave the storage reporting
+ // working() while this method fails and the catch takes its registration back. write()
+ // and ImporterImpl both skip the re-open when the status says working, so the pool would
+ // be left with no user at all: every connection returned to it destroyed on the spot,
+ // pooling off for that database for as long as the server runs (issue #878).
+ try (final Connection con=getValidatedConnection()) {
+ }
+ } catch (Throwable e) {
+ // Throwable rather than Exception: an Error out of the borrow - a NoClassDefFoundError
+ // from the static initializer of a driver is the one to expect here - would otherwise
+ // leave the pool holding a user that never leaves.
+ // Only what this call registered is given back: an open that found the registration
+ // already made took nothing, and giving it back would release a pool still in use.
+ if (registeredHere) {
+ releasePool();
+ } else if (claimedHere) {
+ // The claim was won but no user was added. The claim goes back on its own, without
+ // touching the pool: left standing it would send the close() of this storage to
+ // releasePool() for a registration it never made.
+ poolConnectionString=null;
+ poolRegistered.set(false);
+ }
+ throw e;
+ }
+ this.accessMode = accessMode;
+ storageStatus = StorageStatus.working();
+ }
+
+ /** Gives up the registration of this storage with the pool of the database it opened. */
+ private void releasePool() {
+ if (poolRegistered.compareAndSet(true, false)) {
+ final String registered=poolConnectionString;
+ poolConnectionString=null;
+ if (registered!=null) {
+ CachedConnection.closePool(registered);
+ }
}
}
@@ -740,6 +823,10 @@
// that it is not reissued for every tree on every open; disabling and re-enabling the
// backend is the way to try again once the privilege has been granted
unstampableTrees.clear();
+ // A closed backend has no use for its connections. They used to stay open - close() only
+ // flipped the status - so disabling or removing a JDBC backend left them behind, and with
+ // nothing left to expire the pool entry they could stay open for good (issue #878).
+ releasePool();
}
// The trees this storage has taken an interest in, and the tables they map to. listTrees() -
@@ -972,7 +1059,12 @@
Connection newStampConnection(Dialect dialect) throws SQLException {
final Properties properties=new Properties();
properties.putAll(dialect.connectProperties);
- final Connection con=DriverManager.getConnection(config.getDBDirectory(), properties);
+ // poolKey() rather than the configuration as it stands: this connection is not pooled, but it
+ // is a connection to the database of this storage, and db-directory may be changed on a
+ // running backend. Reading it again here would stamp the trees of this backend in whichever
+ // database the configuration names now, while every other connection of it stays with the
+ // one open() registered (issue #878).
+ final Connection con=DriverManager.getConnection(poolKey(), properties);
try {
con.setAutoCommit(false);
executeSessionStatement(con, dialect.lockTimeoutSql); // give up instead of waiting for another session
@@ -1762,7 +1854,9 @@
* connection established before it, and the pool has no other way of hearing about any of them.
*/
private void distrustPool() {
- CachedConnection.distrustPool(config.getDBDirectory());
+ // keyed like every other pool lookup of this storage: a drop reported against the string
+ // config names now would be filed on a pool holding none of this storage's connections
+ CachedConnection.distrustPool(poolKey());
}
/** Returns the randomized delay before the given attempt is replayed, doubling with each attempt up to a cap. */
@@ -2578,22 +2672,65 @@
* of an online import blocked by an LDAP write on the same table sat until the bound of an
* entry read and then failed the import.
*/
- ImporterImpl(Connection con, boolean isOpen) {
- // An import writes by definition, so a storage that is not writeable refuses one where the
- // importer is built - which is where it was refused until the write transaction of a read-only
- // storage became one that is granted and checks per operation (#874). Left to that check, an
- // import of such a storage would take a connection out of the pool, begin its transaction and
- // fail at the first tree it clears rather than at its start.
- // What arrives here read-only is a storage that was already open: import-ldif and
- // rebuild-index both close it first, and startImport() opens a closed one READ_WRITE - an
- // import of any storage of this server reopens it that way - so those two arrive writeable.
- if (!accessMode.isWriteable()) {
- throw new ReadOnlyStorageException();
+ public ImporterImpl() {
+ // The open belongs here with the borrow it precedes (#878): startImport() used to do both,
+ // and a failure between them had two owners to give back what each had taken.
+ isOpen=getStorageStatus().isWorking();
+ if (!isOpen) {
+ try {
+ open(AccessMode.READ_WRITE);
+ }catch (Exception e) {
+ throw new StorageRuntimeException(e);
+ }
}
- this.con=con;
- this.isOpen=isOpen;
- txr=new ReadableTransactionImpl(con, StatementBound.BULK);
- txw=new WriteableTransactionTransactionImpl(con, StatementBound.BULK);
+ // Nothing holds what this constructor takes until it returns: close() belongs to an
+ // object that was built, so a throw below would leave the connection borrowed and the
+ // storage this constructor opened open, with nobody left to give either back.
+ Connection borrowed=null;
+ try {
+ // An import writes by definition, so a storage that is not writeable refuses one where the
+ // importer is built - which is where it was refused until the write transaction of a read-only
+ // storage became one that is granted and checks per operation (#874). Left to that check, an
+ // import of such a storage would take a connection out of the pool, begin its transaction and
+ // fail at the first tree it clears rather than at its start.
+ // Inside the try and in front of the borrow: with the borrow moved in here (#878) the
+ // refusal now takes no connection at all, and the open above is still given back by the
+ // catch below - which is the half of it a storage that arrives closed and read-only needs.
+ if (!accessMode.isWriteable()) {
+ throw new ReadOnlyStorageException();
+ }
+ borrowed=getValidatedConnection();
+ txr =new ReadableTransactionImpl(borrowed, StatementBound.BULK);
+ txw =new WriteableTransactionTransactionImpl(borrowed, StatementBound.BULK);
+ con = borrowed;
+ borrowed=null;
+ }catch (Throwable e){
+ // Throwable rather than Exception, the way close() below catches it and for the same
+ // reason: the borrow is handed off to nothing until this constructor returns, and
+ // only its close() gives back the permit it took. new WriteableTransactionTransactionImpl
+ // runs a StampSession in a field initializer, so an Error out of a bulk import - an
+ // OutOfMemoryError is the one to expect - would leave the connection borrowed for the
+ // life of the server, and enough of them walk the bound of the pool down to nothing
+ // (issue #878).
+ if (borrowed!=null) {
+ try {
+ borrowed.close();
+ }catch (Throwable e2) {
+ // suppressed rather than dropped: the failure being unwound is the one the
+ // caller asked about, and a return that failed on top of it is worth reading
+ e.addSuppressed(e2);
+ }
+ }
+ if (!isOpen) {
+ JDBCStorage.this.close();
+ }
+ if (e instanceof Error) {
+ // on its way out as it is: an Error says the JVM is in no state to have this
+ // wrapped and reported as a failure of the storage
+ throw (Error) e;
+ }
+ throw e instanceof StorageRuntimeException ? (StorageRuntimeException) e : new StorageRuntimeException(e);
+ }
}
@Override
@@ -2601,6 +2738,34 @@
aborted = true;
}
+ /**
+ * Hands the connection back to the pool and closes the stamp session, whatever went before.
+ * Returns the failure the caller is to report: the return rolls back, and the rollback
+ * fails on exactly the connection whose commit just did, so the commit stays the exception
+ * the caller sees and this one rides along with it instead of replacing it.
+ */
+ private SQLException releaseConnection(SQLException failure) {
+ try {
+ con.close();
+ } catch (Throwable e) {
+ // Throwable rather than SQLException: this close() is the return to the pool, whose
+ // rollback a driver is free to fail unchecked. Reported rather than thrown, since a
+ // throw out of here would leave with the failure the caller actually came for - the
+ // commit above, and in the Throwable branch of close() the Error that branch exists
+ // to preserve - dropped on the floor (issue #878).
+ final SQLException reported=e instanceof SQLException ? (SQLException) e
+ : new SQLException("the connection of the import could not be returned to the pool", e);
+ if (failure==null) {
+ failure=reported;
+ }else {
+ failure.addSuppressed(reported);
+ }
+ } finally {
+ txw.stampSession.close();
+ }
+ return failure;
+ }
+
// The connection goes back whatever the commit does, and the storage this importer opened
// is closed whatever the connection does: an importer is closed on the way out of a failed
// import as readily as a finished one - a clearTree() that reaches the bulk bound is one
@@ -2609,6 +2774,7 @@
@Override
public void close() {
try {
+ SQLException failure=null;
try {
con.commit();
if (aborted) {
@@ -2616,15 +2782,27 @@
}else {
updateTableStatistics(con, writtenTrees);
}
- } finally { // the pooled connection must be returned even when the commit or a statistics statement throws
- try {
- con.close();
- } finally {
- txw.stampSession.close();
+ } catch (SQLException e) {
+ failure=e;
+ } catch (Throwable t) {
+ // Back to the pool whatever came out of the commit, not only on the SQLException
+ // a driver is supposed to throw: nothing else holds this connection, and only
+ // its close() gives back the permit it took. A pool is never removed from the
+ // map, so a permit lost to an Error out of a bulk import - or to a driver
+ // failing unchecked - is lost for the life of the server, and enough of them
+ // walk the bound down to nothing (issue #878).
+ final SQLException onTheWayOut=releaseConnection(null);
+ if (onTheWayOut!=null) {
+ t.addSuppressed(onTheWayOut);
}
+ throw t;
}
- } catch (SQLException e) {
- throw new StorageRuntimeException(e);
+ // Back to the pool even when the commit failed: nothing else holds this connection,
+ // so leaving it behind would leak it along with the failure.
+ failure=releaseConnection(failure);
+ if (failure!=null) {
+ throw new StorageRuntimeException(failure);
+ }
} finally {
if (!isOpen) {
JDBCStorage.this.close();
@@ -2663,52 +2841,11 @@
//import
@Override
public Importer startImport() throws ConfigException, StorageRuntimeException {
- final boolean wasOpen=getStorageStatus().isWorking();
- if (!wasOpen) {
- try {
- open(AccessMode.READ_WRITE);
- }catch (Exception e) {
- throw new StorageRuntimeException(e);
- }
- }
- final Connection con;
- try {
- con=getValidatedConnection();
- }catch (Exception e){
- // and the storage this method opened goes back with it: ImporterImpl.close() is what closes
- // it again when an import opened it, and no importer is going to be built to reach that
- if (!wasOpen) {
- close();
- }
- throw new StorageRuntimeException(e);
- }
- // outside the catch: the importer of a read-only storage throws ReadOnlyStorageException,
- // which a caller tells apart from any other failure of an import
- boolean built=false;
- try {
- final Importer importer=new ImporterImpl(con, wasOpen);
- built=true;
- return importer;
- }finally {
- // and the connection borrowed above goes back on every path that does not build an
- // importer to hold it: it is the one an import keeps for its whole duration, so leaving it
- // here takes it out of the pool for good, with the transaction it had already begun. A
- // finally rather than a catch, so that it covers what a catch has to name - an Error
- // leaves the pool one connection short exactly as ReadOnlyStorageException did.
- if (!built) {
- try {
- con.close();
- }catch (SQLException ignored) {
- // the importer was never built; the failure to report is the one on its way out
- }
- // and the storage this method opened goes back with the connection, for the reason the
- // borrow above gives: ImporterImpl.close() is what closes it again when an import
- // opened it, and there is no importer here to reach that
- if (!wasOpen) {
- close();
- }
- }
- }
+ // Everything this used to do before building the importer - opening a closed storage, and
+ // borrowing the connection an import keeps for its whole duration - is the importer's own now
+ // (#878). Split between the two, a failure in between had to be given back by whichever of them
+ // had taken what, and the constructor's own throw was covered by neither.
+ return new ImporterImpl();
}
//backup
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java
index c98f280..3cc49b8 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java
@@ -15,7 +15,10 @@
*/
package org.opends.server.backends.jdbc;
+import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
import org.opends.server.DirectoryServerTestCase;
+import org.opends.server.backends.pluggable.spi.AccessMode;
+import org.opends.server.backends.pluggable.spi.Importer;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
@@ -39,20 +42,27 @@
import java.util.Collections;
import java.util.Deque;
import java.util.IdentityHashMap;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.Executor;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Logger;
import org.mockito.InOrder;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.inOrder;
@@ -118,6 +128,7 @@
public void clearProperties() {
System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
System.clearProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
+ System.clearProperty(CachedConnection.POOL_MAX_PROPERTY);
System.clearProperty(CachedConnection.TTL_PROPERTY);
System.clearProperty(CachedConnection.ALIVE_BYPASS_PROPERTY);
// what has been reported once is remembered for the life of the jvm: left standing, the key
@@ -127,6 +138,569 @@
}
/**
+ * Nothing used to limit how many connections a backend opened: the pool was an unbounded queue
+ * behind a cache with no maximum size, so a burst of concurrent operations opened as many
+ * connections as there were threads asking, and the only ceiling left was the max_connections of
+ * the database itself (#878).
+ */
+ @Test(timeOut = 120000)
+ public void testThePoolDoesNotGrowPastItsBound() throws Exception {
+ final String url = StubDriver.PREFIX + "bounded";
+ System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "2");
+ System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "1");
+ stub.answerWith(null);
+
+ // One thread per borrow, as the worker threads of the server are: two borrows on one thread
+ // are nested by definition, and a nested one is allowed past the bound on purpose.
+ final Connection first = borrowOnAThreadOfItsOwn(url);
+ final Connection second = borrowOnAThreadOfItsOwn(url);
+ assertEquals(CachedConnection.poolOf(url).meteredCount(), 2);
+ try {
+ borrowOnAThreadOfItsOwn(url);
+ fail("a third connection was opened past the bound of two");
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof SQLTimeoutException, String.valueOf(e.getCause()));
+ assertTrue(e.getCause().getMessage().contains("all 2 connections"), e.getCause().getMessage());
+ }
+
+ // The bound waits for a returned connection rather than refusing outright: it is a ceiling
+ // on the connections held, not on the operations served.
+ first.close();
+ final Connection third = borrowOnAThreadOfItsOwn(url);
+ assertSame(third, first);
+ third.close();
+ second.close();
+ CachedConnection.poolOf(url).drainIdle();
+ }
+
+ /** Borrows the way the server does, one operation to a thread. */
+ private static Connection borrowOnAThreadOfItsOwn(String url) throws Exception {
+ return startBorrow(url).get(120, TimeUnit.SECONDS);
+ }
+
+ /** The same, left running: a borrow that waits has to be looked at while it does. */
+ private static FutureTask<Connection> startBorrow(String url) {
+ final FutureTask<Connection> borrow = new FutureTask<>(() -> CachedConnection.getConnection(url));
+ final Thread thread = new Thread(borrow, "borrow-" + url);
+ thread.setDaemon(true);
+ thread.start();
+ return borrow;
+ }
+
+ /**
+ * A borrow made while this thread already holds a connection must not wait for the bound: the
+ * two are held at once, so it would wait for itself. PersistentCompressedSchema.store() opens a
+ * write of its own and is reached from inside a transaction by EntryContainer.importEntry and
+ * EntryContainer.modifyDN, both of which encode the entry inside it.
+ */
+ @Test(timeOut = 120000)
+ public void testABorrowNestedInAnotherMayPassTheBound() throws Exception {
+ final String url = StubDriver.PREFIX + "reentrant";
+ System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "1");
+ System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "1");
+ stub.answerWith(null);
+
+ final Connection outer = CachedConnection.getConnection(url);
+ final Connection nested = CachedConnection.getConnection(url);
+ assertNotSame(nested, outer);
+
+ // It holds no permit of the pool, so pooling it would leave the pool one connection over
+ // its bound for good: it is closed instead.
+ nested.close();
+ verify(((CachedConnection) nested).parent).close();
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 0);
+
+ outer.close();
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 1);
+ CachedConnection.poolOf(url).drainIdle();
+ }
+
+ /**
+ * The TTL used to sit on the pool rather than on a connection - keyed by the connection string,
+ * and touched by every borrow and every return - so under continuous traffic nothing in it ever
+ * expired (#878).
+ */
+ @Test(timeOut = 120000)
+ public void testAnIdleConnectionIsClosedAfterItsTtl() throws Exception {
+ final String url = StubDriver.PREFIX + "ttl";
+ stub.answerWith(null);
+
+ final CachedConnection first = (CachedConnection) CachedConnection.getConnection(url);
+ first.close();
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 1);
+ first.returnedAtMillis = System.currentTimeMillis() - 60000;
+ System.setProperty(CachedConnection.TTL_PROPERTY, "1000");
+
+ final Connection second = CachedConnection.getConnection(url);
+
+ assertNotSame(second, first, "a connection idle far longer than the TTL was handed out");
+ verify(first.parent).close();
+ second.close();
+ CachedConnection.poolOf(url).drainIdle();
+ }
+
+ /**
+ * Expiry has to happen without a borrow behind it: the cache was built without a scheduler, so
+ * an entry was only ever expired by a later cache operation - and a backend that has gone idle,
+ * the one case the TTL exists for, performs none (#878). This is what the sweeper thread runs.
+ */
+ @Test(timeOut = 120000)
+ public void testTheSweepClosesAnIdleConnectionWithNoBorrowBehindIt() throws Exception {
+ final String url = StubDriver.PREFIX + "sweep";
+ stub.answerWith(null);
+ final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url);
+ con.close();
+ // The sweeper of the server is running while this case does, over every pool and reading
+ // the TTL as it goes: out of its reach, so that the sweep asserted here is the one below.
+ System.setProperty(CachedConnection.TTL_PROPERTY, "600000");
+ con.returnedAtMillis = System.currentTimeMillis() - 60000;
+ final CachedConnection.Pool pool = CachedConnection.poolOf(url);
+ assertEquals(pool.idleCount(), 1);
+
+ pool.sweep(1000);
+
+ assertEquals(pool.idleCount(), 0);
+ verify(con.parent).close();
+ assertEquals(pool.meteredCount(), 0, "a swept connection kept its place in the pool");
+ }
+
+ /** A closed backend has no use for its connections; they used to be left open (#878). */
+ @Test(timeOut = 120000)
+ public void testClosingTheLastUserReleasesTheConnections() throws Exception {
+ final String url = StubDriver.PREFIX + "release";
+ stub.answerWith(null);
+ CachedConnection.openPool(url);
+ final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url);
+ con.close();
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 1);
+
+ CachedConnection.closePool(url);
+
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 0);
+ verify(con.parent).close();
+ assertEquals(CachedConnection.poolOf(url).meteredCount(), 0);
+ }
+
+ /**
+ * A pool belongs to a database rather than to a backend: two backends may address one database,
+ * and closing one of them must not take the connections of the other with it.
+ */
+ @Test(timeOut = 120000)
+ public void testConnectionsSurviveWhileAnotherBackendStillUsesTheDatabase() throws Exception {
+ final String url = StubDriver.PREFIX + "shared";
+ stub.answerWith(null);
+ CachedConnection.openPool(url);
+ CachedConnection.openPool(url);
+ final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url);
+ con.close();
+
+ CachedConnection.closePool(url);
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 1, "the second backend lost its connections");
+
+ CachedConnection.closePool(url);
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 0);
+ verify(con.parent).close();
+ }
+
+ /** A connection out on loan when the last backend closed is closed when it comes back. */
+ @Test(timeOut = 120000)
+ public void testAConnectionReturnedAfterTheLastUserLeftIsClosed() throws Exception {
+ final String url = StubDriver.PREFIX + "return-after-close";
+ stub.answerWith(null);
+ CachedConnection.openPool(url);
+ final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url);
+
+ CachedConnection.closePool(url);
+ con.close();
+
+ verify(con.parent).close();
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 0);
+ }
+
+ /** A backend closed and opened again pools its connections as before: addUser() clears the flag. */
+ @Test(timeOut = 120000)
+ public void testABackendClosedAndOpenedAgainPoolsItsConnections() throws Exception {
+ final String url = StubDriver.PREFIX + "reopen";
+ stub.answerWith(null);
+ CachedConnection.openPool(url);
+ CachedConnection.getConnection(url).close();
+ CachedConnection.closePool(url);
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 0);
+
+ CachedConnection.openPool(url);
+ CachedConnection.getConnection(url).close();
+
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 1, "a reopened backend stopped pooling its connections");
+ CachedConnection.closePool(url);
+ }
+
+ /**
+ * A connect that fails with something other than a SQLException must not cost the pool a
+ * permit. DriverManager catches SQLException alone, so an unchecked failure of a driver reaches
+ * the borrow: Connector/J hands a url with a "%" in it to URLDecoder, and this backend keeps
+ * its credentials in the url. Only a live connection carries a permit, so one left behind is
+ * left behind for good - after as many failures as the bound the pool would report that every
+ * connection is in use while holding none (#878).
+ */
+ @Test(timeOut = 120000)
+ public void testAConnectFailingUncheckedCostsThePoolNothing() throws Exception {
+ final String url = StubDriver.PREFIX + "unchecked";
+ System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "2");
+ System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "1");
+ final CachedConnection.Pool pool = CachedConnection.poolOf(url);
+ stub.failWith(new IllegalArgumentException("URLDecoder: Illegal hex characters in escape (%) pattern"),
+ StubDriver.ALWAYS);
+
+ for (int i = 1; i <= 2 * pool.max(); i++) {
+ try {
+ CachedConnection.getConnection(url);
+ fail("the connect did not fail");
+ } catch (IllegalArgumentException expected) {
+ // reported to the caller, as a configuration error has to be
+ }
+ assertEquals(pool.meteredCount(), 0, "attempt " + i + " kept a permit of the pool");
+ }
+
+ // and the pool still serves, rather than reporting connections it does not hold as in use
+ stub.answerWith(null);
+ final Connection con = CachedConnection.getConnection(url);
+ assertNotNull(con);
+ con.close();
+ CachedConnection.poolOf(url).drainIdle();
+ }
+
+ /**
+ * The exemption of a nested borrow belongs to one pool: a thread holding a connection to one
+ * database holds nothing of another, so the bound of that other pool applies and its connection
+ * comes back to it rather than being closed.
+ */
+ @Test(timeOut = 120000)
+ public void testHoldingAConnectionToOneDatabaseDoesNotExemptABorrowFromAnother() throws Exception {
+ final String first = StubDriver.PREFIX + "held-first";
+ final String second = StubDriver.PREFIX + "held-second";
+ stub.answerWith(null);
+
+ final Connection held = CachedConnection.getConnection(first);
+ final Connection other = CachedConnection.getConnection(second);
+ assertEquals(CachedConnection.poolOf(second).meteredCount(), 1, "the borrow passed the bound of the other pool");
+ other.close();
+
+ assertEquals(CachedConnection.poolOf(second).idleCount(), 1, "the borrow was taken for a nested one and closed");
+ held.close();
+ CachedConnection.poolOf(first).drainIdle();
+ CachedConnection.poolOf(second).drainIdle();
+ }
+
+ /**
+ * A connection returned on a thread other than the one that borrowed it still lowers the depth
+ * of the borrower. The depth used to be lowered only where the returning thread was the
+ * borrowing one, and nulled either way, so a cross-thread return left the borrower standing at a
+ * depth it could never come down from: that thread was taken for a nested borrow for the life of
+ * the server, exempt from the wait at the bound, and every operation on it opened an unmetered
+ * connection that the return then closed - a physical connect apiece, past a bound the operator
+ * set (#878).
+ */
+ @Test(timeOut = 120000)
+ public void testAReturnOnAnotherThreadLowersTheDepthOfTheBorrower() throws Exception {
+ final String url = StubDriver.PREFIX + "cross-thread-return";
+ System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "1");
+ System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "1");
+ stub.answerWith(null);
+ final CachedConnection.Pool pool = CachedConnection.poolOf(url);
+ final ExecutorService borrower = Executors.newSingleThreadExecutor(runnable -> {
+ final Thread thread = new Thread(runnable, "cross-thread-borrower");
+ thread.setDaemon(true);
+ return thread;
+ });
+
+ try {
+ // borrowed there, returned here
+ final Connection borrowed = borrower.submit(() -> CachedConnection.getConnection(url))
+ .get(120, TimeUnit.SECONDS);
+ borrowed.close();
+ assertEquals(pool.idleCount(), 1, "the connection was not pooled by the return");
+
+ // the one place of the pool goes to somebody else, so the borrower thread has to wait
+ // for it - and, having no connection of its own any more, has to give up when it does
+ // not come
+ final Connection held = borrowOnAThreadOfItsOwn(url);
+ try {
+ borrower.submit(() -> CachedConnection.getConnection(url)).get(120, TimeUnit.SECONDS);
+ fail("the borrower thread was taken for a nested borrow and passed the bound of the pool");
+ } catch (ExecutionException expected) {
+ assertTrue(expected.getCause() instanceof SQLTimeoutException,
+ "the bound was passed rather than waited out: " + expected.getCause());
+ }
+ held.close();
+ } finally {
+ borrower.shutdownNow();
+ }
+ CachedConnection.poolOf(url).drainIdle();
+ }
+
+ /** JDBC makes close() on a closed connection a no-op; a second return would pool the same one twice. */
+ @Test(timeOut = 120000)
+ public void testASecondCloseDoesNotPoolTheConnectionTwice() throws Exception {
+ final String url = StubDriver.PREFIX + "double-close";
+ stub.answerWith(null);
+ final Connection con = CachedConnection.getConnection(url);
+ con.close();
+ con.close();
+
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 1, "one connection was pooled twice");
+ CachedConnection.poolOf(url).drainIdle();
+ }
+
+ /**
+ * What the sweeper runs hands the close elsewhere instead of running it. The sweep of every
+ * pool shares one thread and scheduleWithFixedDelay never overlaps its runs, so one close that
+ * does not return would stop the expiry of every pool in the JVM, silently (#878).
+ */
+ @Test(timeOut = 120000)
+ public void testTheSweepDoesNotCloseOnTheSweeperThread() throws Exception {
+ final String url = StubDriver.PREFIX + "sweep-elsewhere";
+ stub.answerWith(null);
+ final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url);
+ con.close();
+ System.setProperty(CachedConnection.TTL_PROPERTY, "600000"); // see the case above
+ con.returnedAtMillis = System.currentTimeMillis() - 60000;
+ final CachedConnection.Pool pool = CachedConnection.poolOf(url);
+ final List<Runnable> handedOff = new ArrayList<>();
+
+ pool.sweep(1000, handedOff::add);
+
+ assertEquals(pool.idleCount(), 0, "the expired connection kept its place in the pool");
+ verify(con.parent, never()).close();
+ assertEquals(handedOff.size(), 1);
+
+ handedOff.get(0).run();
+ verify(con.parent).close();
+ assertEquals(pool.meteredCount(), 0, "a swept connection kept its permit");
+ }
+
+ /**
+ * And the sweep the scheduled sweeper actually runs closes elsewhere too: the case above
+ * supplies an executor of its own, so it would pass just as well with the production one left
+ * closing inline.
+ */
+ @Test(timeOut = 120000)
+ public void testTheScheduledSweepClosesOnAThreadOfItsOwn() throws Exception {
+ final String url = StubDriver.PREFIX + "sweeper-thread";
+ stub.answerWith(null);
+ final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url);
+ con.close();
+ final AtomicReference<String> closedOn = new AtomicReference<>();
+ doAnswer(invocation -> {
+ closedOn.set(Thread.currentThread().getName());
+ return null;
+ }).when(con.parent).close();
+ con.returnedAtMillis = System.currentTimeMillis() - 60000;
+ System.setProperty(CachedConnection.TTL_PROPERTY, "1000");
+
+ CachedConnection.sweep(); // what the scheduled sweeper runs, with nothing supplied to it
+
+ for (int i = 0; i < 200 && closedOn.get() == null; i++) {
+ Thread.sleep(50);
+ }
+ assertNotNull(closedOn.get(), "the sweep never closed the expired connection");
+ assertFalse(closedOn.get().contains("sweeper"), "the close ran on the sweeper thread: " + closedOn.get());
+ assertTrue(closedOn.get().startsWith("JDBC backend connection pool closer"), closedOn.get());
+ }
+
+ /**
+ * A borrow may not outlast the deadline it was given while emptying the pool. A poll of no
+ * duration still hands out whatever the deque holds, and a connection whose socket is half-open
+ * - a moved VIP, a firewall that dropped the idle sockets - costs the validation timeout to
+ * discard, so draining a pool of its full bound overran the deadline by minutes, before the
+ * connect that follows it had even started (#878).
+ */
+ @Test(timeOut = 120000)
+ public void testABorrowStopsAtItsDeadlineRatherThanDrainingThePool() throws Exception {
+ final String url = StubDriver.PREFIX + "deadline-drain";
+ System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "6");
+ System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "1");
+ final CachedConnection.Pool pool = CachedConnection.poolOf(url);
+
+ // Fresh by the TTL, and each one a second to find broken: the pool a burst of traffic left
+ // behind, against a database that has stopped answering.
+ for (int i = 0; i < 6; i++) {
+ final Connection halfOpen = mock(Connection.class);
+ when(halfOpen.isValid(anyInt())).thenAnswer(invocation -> {
+ Thread.sleep(1000);
+ return false;
+ });
+ assertTrue(pool.tryReserve());
+ pool.addIdle(new CachedConnection(url, halfOpen, pool, true, true));
+ }
+ stub.answerWith(null);
+
+ final long startedAt = System.currentTimeMillis();
+ final Connection borrowed = CachedConnection.getConnection(url);
+ final long elapsed = System.currentTimeMillis() - startedAt;
+
+ assertNotNull(borrowed);
+ assertTrue(elapsed < 3500, "the borrow drained the pool past its deadline: " + elapsed + " ms");
+ borrowed.close();
+ CachedConnection.poolOf(url).drainIdle();
+ }
+
+ /** 0 means "no bound" for the size of the pool, and an invalid value means "the default". */
+ @Test(timeOut = 120000)
+ public void testTheBoundOfThePoolReadsItsBoundaryValues() throws Exception {
+ assertEquals(poolWithMax("unbounded", "0").max(), Integer.MAX_VALUE, "0 must mean no bound");
+ assertEquals(poolWithMax("negative", "-1").max(), CachedConnection.DEFAULT_POOL_MAX);
+ assertEquals(poolWithMax("not-a-number", "sixteen").max(), CachedConnection.DEFAULT_POOL_MAX);
+ }
+
+ private static CachedConnection.Pool poolWithMax(String name, String max) {
+ System.setProperty(CachedConnection.POOL_MAX_PROPERTY, max);
+ return CachedConnection.poolOf(StubDriver.PREFIX + "bound-" + name); // read when the pool is built
+ }
+
+ /** 0 means "wait without limit" for a borrow, rather than "give up at once". */
+ @Test(timeOut = 120000)
+ public void testABorrowWithNoDeadlineWaitsForAReturnedConnection() throws Exception {
+ final String url = StubDriver.PREFIX + "no-deadline";
+ System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "1");
+ System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0");
+ stub.answerWith(null);
+
+ final Connection held = borrowOnAThreadOfItsOwn(url);
+ final FutureTask<Connection> waiting = startBorrow(url);
+ try {
+ waiting.get(1500, TimeUnit.MILLISECONDS);
+ fail("the borrow gave up although it was given no deadline");
+ } catch (TimeoutException expected) {
+ // still waiting for the connection of the pool to come back, which is the point
+ }
+
+ held.close();
+ final Connection served = waiting.get(120, TimeUnit.SECONDS);
+ assertSame(served, held, "the borrow was served by something other than the returned connection");
+ served.close();
+ CachedConnection.poolOf(url).drainIdle();
+ }
+
+ /** 0 means "keep nothing" for the TTL: an idle connection is not handed out again. */
+ @Test(timeOut = 120000)
+ public void testAZeroTtlKeepsNoIdleConnection() throws Exception {
+ final String url = StubDriver.PREFIX + "zero-ttl";
+ stub.answerWith(null);
+ final CachedConnection first = (CachedConnection) CachedConnection.getConnection(url);
+ first.close();
+ first.returnedAtMillis = System.currentTimeMillis() - 5;
+ System.setProperty(CachedConnection.TTL_PROPERTY, "0");
+
+ final Connection second = CachedConnection.getConnection(url);
+
+ assertNotSame(second, first, "a connection was kept although the TTL keeps none");
+ verify(first.parent).close();
+ second.close();
+ CachedConnection.poolOf(url).drainIdle();
+ }
+
+ /**
+ * The storage borrows from the pool it registered with, and gives that registration back when
+ * it closes. db-directory may be changed on a running backend - applyConfigurationChange takes
+ * it and nothing refuses it - and a borrow that followed the change would leave the pool this
+ * storage registered with holding a user that never borrows, while the pool it borrowed from
+ * has none: the leak of #878 back through the configuration, and a pool another backend may
+ * drain while this one is still borrowing from it.
+ */
+ @Test(timeOut = 120000)
+ public void testTheStorageBorrowsFromThePoolItRegisteredWith() throws Exception {
+ final String registered = StubDriver.PREFIX + "storage-registered";
+ final String changed = StubDriver.PREFIX + "storage-changed";
+ stub.answerWith(null);
+ final JDBCBackendCfg cfg = mock(JDBCBackendCfg.class);
+ when(cfg.getDBDirectory()).thenReturn(registered);
+ final JDBCStorage storage = new JDBCStorage(cfg, null);
+ storage.open(AccessMode.READ_WRITE);
+
+ when(cfg.getDBDirectory()).thenReturn(changed); // the configuration changed under it
+ try (final Connection con = storage.getConnection()) {
+ assertEquals(((CachedConnection) con).connectionString, registered,
+ "the borrow left the pool this storage registered with");
+ }
+ assertEquals(CachedConnection.poolOf(changed).meteredCount(), 0, "a pool with no user was borrowed from");
+
+ storage.close();
+
+ assertEquals(CachedConnection.poolOf(registered).idleCount(), 0,
+ "close() left the connections of the pool it registered with behind");
+ }
+
+ /**
+ * An open that failed has to leave the storage saying so. The status used to be set inside the
+ * try-with-resources of the validating borrow, so a throw from the implicit close() - the return
+ * rolls back, and the rollback goes to the database - left the storage at working() while open()
+ * failed and gave the registration of the pool back. write() and ImporterImpl both skip the
+ * re-open when the status says working, so the pool was left with no user at all: every
+ * connection returned to it destroyed on the spot, pooling off for that database for as long as
+ * the server runs (#878).
+ */
+ @Test(timeOut = 120000)
+ public void testAnOpenThatFailsOnTheReturnLeavesTheStorageClosed() throws Exception {
+ final String url = StubDriver.PREFIX + "open-return-failure";
+ final Connection parent = mock(Connection.class);
+ when(parent.isValid(anyInt())).thenReturn(true);
+ doThrow(new SQLException("the socket went away")).when(parent).rollback();
+ stub.answerWith(parent);
+ final JDBCBackendCfg cfg = mock(JDBCBackendCfg.class);
+ when(cfg.getDBDirectory()).thenReturn(url);
+ final JDBCStorage storage = new JDBCStorage(cfg, null);
+
+ try {
+ storage.open(AccessMode.READ_WRITE);
+ fail("a validated borrow that could not be returned must be reported");
+ } catch (SQLException expected) {
+ assertEquals(expected.getMessage(), "the socket went away");
+ }
+ assertFalse(storage.getStorageStatus().isWorking(), "an open that failed left the storage reporting working");
+
+ // and the open that follows is not skipped: it registers with the pool again, which is what
+ // makes the connections returned to it pooled rather than destroyed on the spot
+ doNothing().when(parent).rollback();
+ storage.open(AccessMode.READ_WRITE);
+ assertTrue(storage.getStorageStatus().isWorking(), "the storage did not reopen");
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 1,
+ "the reopened storage stopped pooling its connections");
+ storage.close();
+ }
+
+ /**
+ * An import gives its connection back however its commit went. The commit used to be guarded
+ * against SQLException alone, so an Error out of a bulk import - or a driver failing unchecked
+ * - left the connection borrowed and its permit with it; a pool is never removed from the map,
+ * so that permit was gone for the life of the server and enough imports walked the bound of
+ * the pool down to nothing (#878).
+ */
+ @Test(timeOut = 120000)
+ public void testAnImportGivesItsConnectionBackWhenTheCommitFailsUnchecked() throws Exception {
+ final String url = StubDriver.PREFIX + "import-unchecked-commit";
+ final Connection parent = mock(Connection.class);
+ when(parent.isValid(anyInt())).thenReturn(true);
+ doThrow(new Error("out of memory while importing")).when(parent).commit();
+ stub.answerWith(parent);
+ final JDBCBackendCfg cfg = mock(JDBCBackendCfg.class);
+ when(cfg.getDBDirectory()).thenReturn(url);
+ final JDBCStorage storage = new JDBCStorage(cfg, null);
+ storage.open(AccessMode.READ_WRITE);
+ final Importer importer = storage.startImport();
+
+ try {
+ importer.close();
+ fail("the failure of the commit was not reported");
+ } catch (Error expected) {
+ // reported to the caller, which is what an Error out of an import has to be
+ }
+
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 1, "the import kept the connection of the pool");
+ storage.close();
+ assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "the import kept a permit of the pool");
+ }
+
+ /**
* A driver that is not on the classpath - the JDBC backend needs one dropped into
* lib/extensions by hand - is a configuration error the caller has to see. Retried, it is
* indistinguishable from a database that hangs.
@@ -620,7 +1194,7 @@
final Connection pooled = mock(Connection.class);
when(pooled.isValid(anyInt())).thenReturn(true);
when(pooled.getNetworkTimeout()).thenReturn(-1);
- CachedConnection.cached.get(url).add(new CachedConnection(url, pooled));
+ CachedConnection.poolOf(url).addIdle(new CachedConnection(url, pooled));
final Connection borrowed = CachedConnection.getConnection(url);
@@ -730,7 +1304,38 @@
assertEquals(expected.getMessage(), "connection is closed");
}
verify(parent).close();
- assertTrue(CachedConnection.cached.get(url).isEmpty(), "a connection that cannot be rolled back was pooled");
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 0, "a connection that cannot be rolled back was pooled");
+ }
+
+ /**
+ * The same for the unchecked failure a driver is free to throw instead of a SQLException. close()
+ * runs past the CAS that makes it the one return of this connection, so a rollback escaping it
+ * leaves the connection closed by nothing at all - and its permit released by nothing either,
+ * since only destroy() gives one back. A pool is never removed from the map, so that place in
+ * the bound would be gone for the life of the server, and enough of them leave every borrow to
+ * fail with a SQLTimeoutException while the pool holds no connection at all (#878).
+ */
+ @Test(timeOut = 120000)
+ public void testAConnectionWhoseRollbackFailsUncheckedIsClosed() throws Exception {
+ final String url = StubDriver.PREFIX + "rollback-unchecked";
+ final Connection parent = mock(Connection.class);
+ when(parent.isValid(anyInt())).thenReturn(true);
+ doThrow(new IllegalStateException("the connection handle is no longer valid")).when(parent).rollback();
+ stub.answerWith(parent);
+ final CachedConnection.Pool pool = CachedConnection.poolOf(url);
+
+ final Connection con = CachedConnection.getConnection(url);
+ assertEquals(pool.meteredCount(), 1, "the borrow took no permit of the pool");
+ try {
+ con.close();
+ fail("a rollback that failed unchecked must be reported");
+ } catch (IllegalStateException expected) {
+ assertEquals(expected.getMessage(), "the connection handle is no longer valid");
+ }
+
+ verify(parent).close();
+ assertEquals(pool.idleCount(), 0, "a connection that could not be rolled back was pooled");
+ assertEquals(pool.meteredCount(), 0, "the return kept a permit of the pool");
}
@Test
@@ -1307,7 +1912,8 @@
assertSame(((CachedConnection) borrowed).parent, fresh, "the drain must stop at the deadline");
verify(stale).close();
verify(good, never()).close();
- assertFalse(CachedConnection.cached.get(url).isEmpty(), "a connection the deadline was reached in front of was lost");
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 1,
+ "a connection the deadline was reached in front of was lost");
}
/**
@@ -1348,11 +1954,13 @@
.when(parent).setNetworkTimeout(any(Executor.class), eq(0));
stub.answerWith(parent);
- final CachedConnection borrowed = CachedConnection.connect(url, CachedConnection.ConnectDialect.MYSQL, 30);
+ final CachedConnection.Pool pool = CachedConnection.poolOf(url);
+ final CachedConnection borrowed = CachedConnection.connect(url, CachedConnection.ConnectDialect.MYSQL, 30,
+ pool, false);
borrowed.close();
verify(parent).close();
- assertTrue(CachedConnection.cached.get(url).isEmpty(),
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 0,
"a connection still carrying the read bound of its login went back into the pool");
}
@@ -1567,8 +2175,9 @@
* from - so that the connection named first here is the one the next borrow gets.
*/
private static void seedPool(String url, Connection... parents) {
+ final CachedConnection.Pool pool = CachedConnection.poolOf(url);
for (int i = parents.length - 1; i >= 0; i--) {
- CachedConnection.cached.get(url).addFirst(new CachedConnection(url, parents[i]));
+ pool.addIdle(new CachedConnection(url, parents[i]));
}
}
@@ -1799,11 +2408,12 @@
static final int ALWAYS = -1;
final AtomicInteger attempts = new AtomicInteger();
- private volatile SQLException failure;
+ /** A SQLException, or the unchecked failure a driver is free to throw at DriverManager instead. */
+ private volatile Throwable failure;
private volatile int failuresLeft;
private volatile Connection answer;
- void failWith(SQLException failure, int times) {
+ void failWith(Throwable failure, int times) {
this.failure = failure;
this.failuresLeft = times;
this.answer = null;
@@ -1827,7 +2437,10 @@
if (failuresLeft > 0) {
failuresLeft--;
}
- throw failure;
+ if (failure instanceof SQLException) {
+ throw (SQLException) failure;
+ }
+ throw (RuntimeException) failure;
}
if (answer != null) {
return answer;
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStatementBoundTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStatementBoundTestCase.java
index 7c0228b..32ad5a5 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStatementBoundTestCase.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStatementBoundTestCase.java
@@ -829,8 +829,22 @@
when(statement.executeQuery()).thenReturn(mock(ResultSet.class));
when(parent.prepareStatement(anyString())).thenReturn(statement);
storage.accessMode = AccessMode.READ_WRITE; // an import has the storage open for writing
- final JDBCStorage.ImporterImpl importer =
- storage.new ImporterImpl(new CachedConnection("jdbc:mock", parent), true);
+ // Borrowed through the seam rather than handed to the constructor: the importer takes its own
+ // connection now (#878). It is the same physical connection the entry read below runs on,
+ // which is what this test pins - the backstop is keyed on the connection, not on the storage.
+ final JDBCStorage importing = new JDBCStorage(mockCfg(JDBCBackendCfg.class), null) {
+ @Override
+ Connection getConnection(boolean trusted) {
+ return new CachedConnection("jdbc:mock", parent);
+ }
+
+ @Override
+ public StorageStatus getStorageStatus() {
+ return StorageStatus.working(); // open already, so the importer borrows and no more
+ }
+ };
+ importing.accessMode = AccessMode.READ_WRITE;
+ final JDBCStorage.ImporterImpl importer = importing.new ImporterImpl();
final TreeName tree = new TreeName("dc=example,dc=com", "id2entry");
// an entry read of a client arms the backstop on the very connection the import writes to,
@@ -1006,9 +1020,11 @@
@Test
public void testStartImportGivesTheConnectionBackWhenTheImporterCannotBeBuilt() throws Exception {
final Connection con = mock(Connection.class);
+ final AtomicInteger borrows = new AtomicInteger();
final JDBCStorage readOnly = new JDBCStorage(mockCfg(JDBCBackendCfg.class), null) {
@Override
Connection getConnection(boolean trusted) {
+ borrows.incrementAndGet();
return con;
}
@@ -1026,7 +1042,13 @@
// the designed path this test is about
}
- verify(con).close();
+ // Nothing to give back. With the borrow inside the importer's constructor (#878) the refusal
+ // stands in front of it, so an import of a read-only storage takes no connection at all
+ // rather than taking one and returning it. Pinned as never borrowed rather than dropped: the
+ // leak this covers - a connection out of the pool for good, holding a transaction it had
+ // already begun - is the same one, and never taking it is the state that cannot leak it.
+ assertEquals(borrows.get(), 0);
+ verify(con, never()).close();
}
/**
@@ -1074,9 +1096,11 @@
// the build failing after this method opened the storage, which is the path under test
}
- assertEquals(opens.get(), 1, "the storage was not opened by startImport(), so nothing was owed back");
- verify(con).close();
- assertEquals(closes.get(), 1, "the storage this method opened was left open");
+ assertEquals(opens.get(), 1, "the storage was not opened by the importer, so nothing was owed back");
+ // the connection is not owed back here either: the refusal stands in front of the borrow now
+ // (#878), so what this path has to give back is the storage alone
+ verify(con, never()).close();
+ assertEquals(closes.get(), 1, "the storage the importer opened was left open");
}
/**
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java
index 3f2f14e..9cc4d97 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java
@@ -169,7 +169,7 @@
}
// a pooled connection would be handed back without being established again
- CachedConnection.cached.invalidate(url);
+ CachedConnection.poolOf(url).drainIdle();
try (final Connection con = CachedConnection.getConnection(url)) {
assertEquals(con.getNetworkTimeout(), 0, "the read bound of the login is still in force");
}
--
Gitblit v1.10.0