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 22d63d4902..d59e1f4b22 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 @@ public class CachedConnection implements Connection { /** 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. + *

+ * 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 @@ public class CachedConnection implements Connection { */ private static final Map 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 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> cached = Caffeine.newBuilder() - .expireAfterAccess(Duration.ofMillis(getCacheTtlMillis())) - .removalListener((String key, BlockingDeque 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 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. + *

+ * 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 @@ private static long getCacheTtlMillis() { * value the unit conversion saturates on from leaving every connection of the pool trusted for * the life of the server. *

- * 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 @@ static long getAliveBypassMillis() { 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. + *

+ * 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. + *

+ * 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. + *

+ * 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 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. + *

+ * 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 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. + *

+ * 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 static boolean isBound(String value) { */ 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. + *

+ * 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 @@ static Connection getConnection(String connectionString) throws Exception { * 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 @@ static Connection getConnection(String connectionString, boolean trusted) throws // 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 @@ static long attemptSeconds(long connectTimeoutSeconds, long deadline) { 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 @@ private static boolean isKnownAlive(CachedConnection con) { 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 @@ private static int boundValidation(Connection con) { 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 @@ static CachedConnection connect(String connectionString, ConnectDialect dialect, 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 @@ static boolean isWorthRetrying(SQLException e, ConnectDialect dialect) { 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 int indexOfAny(String url, String separators, int from) { 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 @@ public void rollback() throws SQLException { @Override public void close() throws SQLException { + // 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; + } + 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(); - } 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); - return; + 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); + } } - // 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); } @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 9f437283c2..a0059e5a07 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 @@ -697,6 +697,17 @@ Connection getConnection() throws Exception { return getConnection(true); } + /** + * 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 @@ -713,17 +724,89 @@ Connection getValidatedConnection() throws Exception { // 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 @@ public void close() { // 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 @@ boolean isMysqlBackslashEscape(Connection con) throws SQLException { 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 @@ private static boolean saysTheConnectionIsGone(SQLException e) { * 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 @@ final class ImporterImpl implements Importer { * 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); + } + } + // 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); } - this.con=con; - this.isOpen=isOpen; - txr=new ReadableTransactionImpl(con, StatementBound.BULK); - txw=new WriteableTransactionTransactionImpl(con, StatementBound.BULK); } @Override @@ -2601,6 +2738,34 @@ public void aborted() { 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 @@ public void aborted() { @Override public void close() { try { + SQLException failure=null; try { con.commit(); if (aborted) { @@ -2616,15 +2782,27 @@ public void close() { }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; + } + // 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); } - } catch (SQLException e) { - throw new StorageRuntimeException(e); } finally { if (!isOpen) { JDBCStorage.this.close(); @@ -2663,52 +2841,11 @@ public SequentialCursor openCursor(TreeName treeName) { //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 c98f280910..3cc49b86c9 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 validateEveryBorrow() { 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 @@ -126,6 +137,569 @@ public void clearProperties() { CachedConnection.aliveBypassNanos = CONFIGURED_ALIVE_BYPASS_NANOS; } + /** + * 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 startBorrow(String url) { + final FutureTask 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 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 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 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 @@ -620,7 +1194,7 @@ public void testAPooledConnectionWhoseDriverAnswersANegativeBoundIsPutBackUnboun 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 @@ public void testConnectionThatCannotBeRolledBackIsClosed() throws Exception { 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 @@ public void testAPooledConnectionIsNotDiscardedUnvalidatedAtTheDeadline() throws 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 @@ public void testAConnectionStillCarryingTheBoundOfItsLoginIsNotPooled() throws E .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 @@ public void testAConnectionDroppedInsideTheWindowIsHandedOutOnceAndThenValidated * 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 @@ private static final class StubDriver implements Driver { 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 @@ public Connection connect(String url, Properties info) throws SQLException { 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 7c0228b8d8..32ad5a525d 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 @@ public void testEveryStatementOfAnImportIsBulk() throws Exception { 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 @@ public void testAConnectionThatFailedTheBackstopIsGivenBackWhatItCarried() throw @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 @@ public StorageStatus getStorageStatus() { // 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 @@ public void close() { // 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 3f2f14e82b..9cc4d976ad 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 @@ public void testLoginBoundDoesNotOutliveTheLogin() throws Exception { } // 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"); }