From a1b8537e1e0cd10e4d614dd9c6bdda541a614572 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 10 Sep 2026 11:56:05 +0000
Subject: [PATCH] [#891] Give the trees of an import connections of their own (#940)
---
opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java | 41 +
opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStatementBoundTestCase.java | 8
opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java | 71 +
opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/ImportConnectionsTestCase.java | 1348 ++++++++++++++++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java | 856 ++++++++++++++++++++-
5 files changed, 2,251 insertions(+), 73 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java
index c03b8f6..fc3881b 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
@@ -359,16 +359,28 @@
}
/**
- * The moment a borrow of this length gives up, or {@link Long#MAX_VALUE} where it gives up
- * never - a property of 0, and a value so large that the milliseconds of it would overflow.
+ * Whether a wait of this many seconds is one with no bound at all.
* <p>
- * The sum is guarded and not only the product: a value under the clamp above but large enough
+ * Two spellings of it, and one predicate for both so that they cannot drift: zero, which is how
+ * {@value #POOL_TIMEOUT_PROPERTY} says "wait for as long as it takes", and a number so large
+ * that the milliseconds it stands for do not fit in a {@code long} - a deadline computed from
+ * one of those overflows into the past, which is the opposite of what it asked for.
+ */
+ static boolean isUnboundedWait(long seconds) {
+ return seconds == 0 || seconds >= Long.MAX_VALUE / 1000;
+ }
+
+ /**
+ * The moment a borrow of this length gives up, or {@link Long#MAX_VALUE} where it gives up
+ * never - the two readings {@link #isUnboundedWait} names.
+ * <p>
+ * The sum is guarded and not only the product: a value that predicate lets through but large enough
* that the moment it names is past the end of the epoch would wrap to a deadline already behind
* us, and a borrow configured to wait practically forever would give up on its first retryable
* failure - the opposite of what was asked for.
*/
static long deadlineOf(long startedAt, long poolTimeoutSeconds) {
- if (poolTimeoutSeconds == 0 || poolTimeoutSeconds >= Long.MAX_VALUE / 1000) {
+ if (isUnboundedWait(poolTimeoutSeconds)) {
return Long.MAX_VALUE;
}
final long deadline = startedAt + poolTimeoutSeconds * 1000;
@@ -522,8 +534,10 @@
* <p>
* A lower bound than that is what is reported, not every way past it: the replay threads of
* replication default to the same count again and borrow on top of the workers, and an import or
- * a rebuild borrows besides. So this names one difference the operator can act on rather than
- * standing for the whole demand on the pool.
+ * a rebuild borrows besides - one connection per tree it writes at a time, up to the bound of
+ * {@code JDBCStorage.IMPORT_CONNECTIONS_PROPERTY}, and it holds them for its whole duration
+ * (#891). So this names one difference the operator can act on rather than standing for the
+ * whole demand on the pool.
* <p>
* Nothing fails for the difference alone: the surplus waits for a connection to be returned,
* which is what the bound is there for. But every one of those waits is paid on an operation,
@@ -809,8 +823,15 @@
* 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
* value the message names is not mistaken for another.
+ * <p>
+ * Package private rather than private so that the bound on the connections of an import
+ * ({@code JDBCStorage.IMPORT_CONNECTIONS_PROPERTY}) can be read through it too, the way the
+ * bounds of the pool are: an operator who mistypes one of those is told rather than left with a
+ * default. Not every number this backend takes from a property comes through here - the
+ * statistics timeout and the fetch sizes of {@code JDBCStorage} are read with
+ * {@code Integer.getInteger}, which replaces a value it cannot parse in silence.
*/
- private static long getNonNegativeProperty(String name, long defaultValue, String unit) {
+ static long getNonNegativeProperty(String name, long defaultValue, String unit) {
final String value = System.getProperty(name);
if (value != null) {
try {
@@ -1260,11 +1281,45 @@
* 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 {
+ return getConnection(connectionString, trusted, 0);
+ }
+
+ /**
+ * The wait a borrow that carries a bound of its own actually gets: the shorter of what the
+ * deployment asked for and what the caller can afford, and the caller's where the deployment
+ * asked for no bound at all.
+ *
+ * @param maxWaitSeconds 0 for a caller that has no bound of its own, which takes the wait of
+ * the deployment whatever it is
+ */
+ static long boundedWait(long poolTimeoutSeconds, long maxWaitSeconds) {
+ if (isUnboundedWait(maxWaitSeconds)) {
+ return poolTimeoutSeconds;
+ }
+ return isUnboundedWait(poolTimeoutSeconds)
+ ? maxWaitSeconds : Math.min(poolTimeoutSeconds, maxWaitSeconds);
+ }
+
+ /**
+ * Borrows a connection, waiting at the bound of the pool no longer than the given number of
+ * seconds however long {@value #POOL_TIMEOUT_PROPERTY} says to wait.
+ *
+ * @param maxWaitSeconds the longest this borrow may wait at the bound of the pool, or 0 to wait
+ * as the property says. For a caller whose own connections are what the pool is full of - an
+ * import holds one per tree it writes until it ends (#891) - a wait with no bound is a deadlock
+ * rather than a queue: nothing is going to return the connection it is waiting for but itself.
+ * A caller that names one has somewhere to go when it runs out, so the wait of the deployment
+ * is capped rather than merely replaced where it is unbounded: an import that has to ask the
+ * pool once per tree would otherwise pay a long {@value #POOL_TIMEOUT_PROPERTY} over again for
+ * every tree the pool has nothing to spare for.
+ */
+ static Connection getConnection(String connectionString, boolean trusted, long maxWaitSeconds)
+ throws Exception {
final Pool pool = poolOf(connectionString);
final ConnectDialect dialect = ConnectDialect.of(connectionString);
reportUnknownDialect(connectionString, dialect);
final long connectTimeoutSeconds = getConnectTimeoutSeconds();
- final long poolTimeoutSeconds = getPoolTimeoutSeconds();
+ final long poolTimeoutSeconds = boundedWait(getPoolTimeoutSeconds(), maxWaitSeconds);
final long ttlMillis = getCacheTtlMillis();
final long startedAt = System.currentTimeMillis();
final long deadline = deadlineOf(startedAt, poolTimeoutSeconds);
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 8b711a0..a2bbc4f 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
@@ -43,9 +43,12 @@
import java.sql.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Predicate;
import static org.opends.server.backends.pluggable.spi.StorageUtils.addErrorMessage;
@@ -600,12 +603,15 @@
* The socket read timeout of one connection, and the statements running on it. This second
* layer of the bound is a property of the socket rather than of a statement, so it cannot be
* armed and put back per statement wherever a connection carries more than one at a time: an
- * {@code ImporterImpl} holds a single connection for the whole of an import and writes to it
- * from every phase-one worker and every phase-two task, and there the first statement to finish
- * would take the backstop away from every statement still in flight - while a statement whose
- * class carries no bound at all would run under whatever value a concurrent one happened to
- * arm, dying at it with nothing to say which property cut it, since such a statement never
- * reaches {@link #timedOut}.
+ * {@code ImporterImpl} held a single connection for the whole of an import and wrote to it from
+ * every phase-one worker and every phase-two task until the trees of an import were given
+ * connections of their own (#891), and there the first statement to finish would take the
+ * backstop away from every statement still in flight - while a statement whose class carries no
+ * bound at all would run under whatever value a concurrent one happened to arm, dying at it
+ * with nothing to say which property cut it, since such a statement never reaches
+ * {@link #timedOut}. The arbitration stays now that no path of this class puts two threads on
+ * one connection: what it holds is a property of the socket, so a connection that carries two
+ * statements again must not have either of them cut by the bound of the other.
* <p>
* So the value armed is the loosest of the bounds of the statements in flight, and a statement
* with no bound of its own takes it off for as long as it runs: this backstop exists to end a
@@ -949,7 +955,19 @@
// 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(poolKey(), trusted);
+ return getConnection(trusted, 0);
+ }
+
+ /**
+ * The borrow every path of this class makes, and the seam a test stands in for the pool at.
+ *
+ * @param maxWaitSeconds the longest this borrow may wait at the bound of the pool, whatever the
+ * deployment asked for - 0 to wait as it says. Only the connections an import takes after its
+ * first pass a number here: they are held until the import ends, so a pool full of them has
+ * nothing to return to the thread waiting for one (#891).
+ */
+ Connection getConnection(boolean trusted, long maxWaitSeconds) throws Exception {
+ return CachedConnection.getConnection(poolKey(), trusted, maxWaitSeconds);
}
@@ -4886,10 +4904,168 @@
return true;
}
+ /**
+ * How many connections one import may write through; unset for the default of
+ * {@link #importConnections()}, and 0 or 1 for the single connection an import had before #891 -
+ * serialized, and the least a database sees of one.
+ */
+ static final String IMPORT_CONNECTIONS_PROPERTY = "org.openidentityplatform.opendj.jdbc.import.connections";
+
+ /**
+ * How many connections one import writes through, which is how many of its threads can write
+ * at the same time.
+ * <p>
+ * More than one because {@code Importer} is thread-safe by contract and a
+ * {@code java.sql.Connection} is not: phase two of {@code OnDiskMergeImporter} runs a thread
+ * per tree and phase one clears the trees of a container while another thread writes id2entry,
+ * so a shared connection has two threads issuing statements on it at once. The drivers differ
+ * in what they make of that - pgjdbc and Connector/J serialize the work of a connection behind
+ * a lock of their own - but the sql server driver keeps the reconnect listeners of a
+ * connection in a plain {@code ArrayList} that every {@code prepareStatement()} and every
+ * statement {@code close()} mutates, and two import threads walked it past the end of its
+ * array (issue #891).
+ * <p>
+ * Bounded because an import holds what it takes for its whole duration, out of a pool that is
+ * bounded since #878 and shared with every other backend on that database: a default backend
+ * has trees enough - one per index, three per attribute index - for a connection per tree to
+ * empty a default pool and leave the LDAP traffic of an online rebuild waiting at its bound.
+ * <p>
+ * Half the bound of the pool by default, and no more than half of what the default bound would
+ * be: a deployment that turned the bound off ({@code pool.max=0}) asked for its operations not
+ * to queue behind one another, not for an import to open a connection per tree.
+ * <p>
+ * A number the operator set is taken as given, up to the bound of the pool: an import-ldif of a
+ * backend that is offline has no traffic of its own to leave room for, and whoever raises this
+ * on a server that is answering is making that trade knowingly - the default is where the
+ * caution belongs.
+ * <p>
+ * What this bounds is how many connections one import holds, which is not the same as staying
+ * inside the bound of the pool: a thread that already holds one connection of a pool is exempt
+ * from waiting at its bound (see {@code CachedConnection.Pool}), and every thread of an import
+ * takes a connection per tree it touches - so the borrows after the first go unmetered where the
+ * pool stands at its bound, and are destroyed rather than pooled when they come back. That
+ * exemption is what keeps the threads of an import from waiting on each other: they hold their
+ * connections until the import ends, so a pool full of them has nothing left to return, and a
+ * bound they had to wait at would be a deadlock rather than a queue.
+ */
+ int importConnections() {
+ final int poolMax=poolMax();
+ final long byDefault=Math.max(1, Math.min(poolMax, CachedConnection.DEFAULT_POOL_MAX)/2);
+ // read the way every other bound of this backend is: a value that is not a non-negative
+ // number is reported to the operator rather than quietly replaced. The default is handed to
+ // it rather than a zero, so that the number the warning names is the number the import goes
+ // on to use.
+ final long configured=CachedConnection.getNonNegativeProperty(IMPORT_CONNECTIONS_PROPERTY, byDefault, "connections");
+ // clamped to the pool: connections past its bound are not there to be had, so a number above
+ // it buys nothing and costs the borrow deadline of every tree over the bound
+ // (CachedConnection.POOL_TIMEOUT_PROPERTY) before the fallback of connectionOf() takes over
+ return (int) Math.max(1, Math.min(configured, poolMax));
+ }
+
+ /**
+ * The bound of the pool this storage borrows from. A seam of its own, like
+ * {@link #getConnection(boolean)}: a test that stands in for the pool must not have this reach
+ * past it into the static registry, which would intern a pool for its connection string.
+ */
+ int poolMax() {
+ return CachedConnection.poolOf(poolKey()).max();
+ }
+
final class ImporterImpl implements Importer {
- final Connection con;
- final ReadableTransactionImpl txr;
- final WriteableTransactionTransactionImpl txw;
+ /**
+ * One connection of an import, the two transactions over it and the monitor that keeps one
+ * thread at a time on it. Every statement of an import is issued under that monitor: what
+ * the threads of an import must not do is share a connection, and the trees of an import
+ * outnumber the connections it may take.
+ */
+ final class ImportConnection {
+ final Connection con;
+ final ReadableTransactionImpl txr;
+ final WriteableTransactionTransactionImpl txw;
+ /**
+ * When this connection was last written, taken from {@link ImporterImpl#writes}, and
+ * zero while it has nothing to commit. Written and read under the monitor of this
+ * object, like the statements it counts: read anywhere else it says what the connection
+ * held rather than what it holds, because it is set once the write it stands for has
+ * come back - and a write still in flight is exactly the one a commit point must not
+ * pass over.
+ */
+ long lastWrite;
+
+ ImportConnection(Connection con) {
+ this.con=con;
+ this.txr=new ReadableTransactionImpl(con, StatementBound.BULK);
+ this.txw=new WriteableTransactionTransactionImpl(con, StatementBound.BULK);
+ // the mode this import was started with rather than the one the storage carries now:
+ // the transaction reads that mutable field as it is built, and these are built as the
+ // trees of an import are first touched - so a storage reopened read-only under a
+ // running import would have the connections it opened before that keep writing while
+ // every one after it refused, halfway through and with the clears already committed
+ this.txw.isReadOnly=false;
+ }
+
+ /** Called under this monitor by whoever writes through this connection. */
+ void written() {
+ lastWrite=writes.incrementAndGet();
+ }
+ }
+
+ /** The connections this import has taken, by the index the trees are handed out against. */
+ private final ConcurrentMap<Integer,ImportConnection> connections = new ConcurrentHashMap<>();
+ /** The connection index each tree is written through: a tree keeps the one it was first given. */
+ private final ConcurrentMap<TreeName,Integer> connectionOfTree = new ConcurrentHashMap<>();
+ /** Handed out round-robin, so that the first trees of an import get connections of their own. */
+ private final AtomicInteger nextConnection = new AtomicInteger();
+ /** One per index, so that the connection of an index is borrowed once however many trees want it. */
+ private final ConcurrentMap<Integer,Object> borrowing = new ConcurrentHashMap<>();
+ /**
+ * The indexes no connection could be borrowed for, whose trees write through the first
+ * connection of this import instead. Remembered rather than asked again per tree: there are
+ * at most {@link #maxConnections} of them, and a tree that lands on one would otherwise pay
+ * the borrow deadline of the pool over again for the answer the tree before it already got.
+ */
+ private final Set<Integer> sharedIndexes = ConcurrentHashMap.newKeySet();
+ /**
+ * Read once rather than per statement: this is a property of the whole import, and reading
+ * it per record would be a property lookup per entry of an import-ldif.
+ */
+ final int maxConnections;
+
+ /**
+ * The connection the constructor borrows. Also the one a borrow that cannot be made falls
+ * back to, so it is the one connection of an import that is always there.
+ */
+ private static final int FIRST_CONNECTION = 0;
+
+ /**
+ * Set by {@code close()} before it commits, after which this import has no transaction left
+ * for a write to belong to. Phase two gives its threads five seconds to answer an interrupt
+ * and closes the importer whether they answered or not
+ * ({@code OnDiskMergeImporter.invokeParallel}), so a write can arrive with the connections
+ * of the import already committed and back in the pool.
+ */
+ private volatile boolean closed;
+
+ /**
+ * Numbers the writes of this import, so that {@code close()} can commit its connections in
+ * the order they were last written to.
+ * <p>
+ * The connections of an import are transactions of their own, so {@code close()} cannot make
+ * them durable as one - and the last thing an import writes is the flag that says the rest of
+ * it is good: {@code afterPhaseTwo} sets the trust flag of every index of a container once
+ * phase two has written them all, one write per base DN. Committed in that order, those flags
+ * go last, and an earlier commit that fails takes them down with it: a failed import cannot
+ * leave an index marked trusted over data that never got there.
+ * <p>
+ * That the flags are the last writes is the caller's doing rather than this class's:
+ * {@code OnDiskMergeImporter} waits for every phase-two task before it runs
+ * {@code afterPhaseTwo}, and nothing here refuses a write that arrives after the flags while
+ * the importer is still open. An importer whose writes did not end there would order its
+ * commits by what its last writes actually were, which is what this counter says and all it
+ * says.
+ */
+ private final AtomicLong writes = new AtomicLong();
+
// The trees this import wrote: close() refreshes the statistics of these and only these,
// so rebuilding a single index does not gather statistics for the whole backend. A full
// import legitimately covers every tree - AbstractTwoPhaseImportStrategy.beforePhaseOne
@@ -4943,10 +5119,21 @@
if (!accessMode.isWriteable()) {
throw new ReadOnlyStorageException();
}
+ // Inside the try like the refusal above: the pool this asks about is the one the open
+ // just registered with, so a failure here has the storage this constructor opened to
+ // give back as well.
+ maxConnections=importConnections();
+ // What an import takes out of the pool is worth reading when a borrow of one fails at
+ // the bound: the pool names the property that bounds it, and this names the one that
+ // bounds the demand.
+ logger.debug(LocalizableMessage.raw("jdbc: import writes through up to %d connections (%s)",
+ maxConnections, IMPORT_CONNECTIONS_PROPERTY));
borrowed=getValidatedConnection();
- txr =new ReadableTransactionImpl(borrowed, StatementBound.BULK);
- txw =new WriteableTransactionTransactionImpl(borrowed, StatementBound.BULK);
- con = borrowed;
+ // The first connection is taken here rather than on the first tree, as it was before
+ // the connections of an import became several (#891): an import of a database that
+ // takes no connection is refused where it is started, and a storage this constructor
+ // opened is given back by the catch below rather than by a put() far from it.
+ connections.put(FIRST_CONNECTION, new ImportConnection(borrowed));
borrowed=null;
}catch (Throwable e){
// Throwable rather than Exception, the way close() below catches it and for the same
@@ -4983,73 +5170,518 @@
}
/**
- * Hands the connection back to the pool and closes the sessions the transaction opened
- * beside it - the stamp one and the catalog one (#888), both outside the pool and neither
- * outliving the import that opened it - 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.
+ * The connection the given tree is written through, borrowed from the pool the first time
+ * this import touches the tree.
+ * <p>
+ * Bound to the tree rather than to the thread, although it is the threads of an import that
+ * must not share one: the connections of an import are transactions of their own, so two of
+ * them writing one row would have the second wait for the first to commit - and an import
+ * commits at {@code close()}, when every thread of it is long done. That is not a shape to
+ * leave lying about: {@code setTrust()} writes the state tree of a container from
+ * {@code beforePhaseOne} on an import thread and again from {@code afterPhaseTwo} on the
+ * thread that closes the importer, and bound to the thread those two writes would be two
+ * transactions waiting for each other with nothing left to break the wait - the bulk class
+ * carries no bound, and the default lock wait is forever on three of the four engines.
+ * Bound to the tree they are one transaction that waits for nothing, and phase two - which
+ * runs a thread per tree - still gets the connection per thread this is all about.
*/
- private SQLException releaseConnection(SQLException failure) {
+ ImportConnection connectionOf(TreeName treeName) {
+ // The lookup before the assignment, not for want of a computeIfAbsent: this runs once per
+ // record of an import-ldif, and the mapping function below allocates a capture of this
+ // importer on every call however long the tree has had a connection.
+ final Integer assigned=connectionOfTree.get(treeName);
+ if (assigned!=null) {
+ final ImportConnection open=connections.get(assigned);
+ if (open!=null) {
+ return open;
+ }
+ }
+ checkOpen();
+ // floorMod rather than %: the counter is shared by every thread of the import and an
+ // index of its own is all a tree needs, so it is never reset - and a negative index
+ // would be one no connection is ever opened for
+ final Integer index=connectionOfTree.computeIfAbsent(treeName,
+ tree -> Math.floorMod(nextConnection.getAndIncrement(), maxConnections));
+ final ImportConnection open=connections.get(index);
+ if (open!=null) {
+ return open;
+ }
+ final ImportConnection taken=sharedIndexes.contains(index) ? sharedConnection() : openConnection(index);
+ if (connections.get(index)!=taken) {
+ // this tree was given a connection of another index, the pool having none to spare:
+ // pointed at it, every record of the tree takes the lookup at the top of this method
+ // rather than this path, which allocates a capture of this importer per call
+ connectionOfTree.put(treeName, FIRST_CONNECTION);
+ }
+ return taken;
+ }
+
+ /**
+ * Takes the connection of an index, or the one this import already has where the pool has
+ * none left to give.
+ * <p>
+ * The borrow is made outside the map rather than in a {@code computeIfAbsent}: it waits for a
+ * connection to be returned where the pool stands at its bound - up to
+ * {@code CachedConnection.POOL_TIMEOUT_PROPERTY} - and a wait of that length inside a mapping
+ * function holds the bin of that key against every other index that hashes to it, which the
+ * contract of {@code ConcurrentHashMap} says not to do. One borrow per index all the same:
+ * two threads first touching trees of one index would otherwise each take a connection, and
+ * the loser's would be a borrow of the pool made and given back for nothing.
+ * <p>
+ * A connection that does not end up in the map is given back here rather than left behind:
+ * only its {@code close()} returns the permit it took, and a pool is never removed from the
+ * map, so one lost here would be lost for the life of the server (#878).
+ */
+ private ImportConnection openConnection(Integer index) {
+ synchronized (borrowing.computeIfAbsent(index, i -> new Object())) {
+ final ImportConnection opened=connections.get(index);
+ if (opened!=null) {
+ return opened; // another thread of this import got here first
+ }
+ if (sharedIndexes.contains(index)) {
+ // ... and found the pool with nothing to spare: this thread has the same answer
+ // waiting for it, and would pay the borrow deadline of the pool again to get it
+ return sharedConnection();
+ }
+ return openConnectionOnce(index);
+ }
+ }
+
+ private ImportConnection openConnectionOnce(Integer index) {
+ final Connection borrowed=borrowedOrShared(index);
+ if (borrowed==null) { // shared, see borrowedOrShared()
+ return sharedConnection();
+ }
+ final ImportConnection built;
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 {
+ // nothing holds the borrow until this returns: new WriteableTransactionTransactionImpl
+ // runs a StampSession in a field initializer, and an Error out of a bulk import - an
+ // OutOfMemoryError is the one to expect - would otherwise leave it out of the pool
+ built=new ImportConnection(borrowed);
+ }catch (Throwable e) {
try {
- txw.stampSession.close();
- } finally {
- txw.catalogSession.close();
+ borrowed.close();
+ }catch (Throwable e2) {
+ // suppressed rather than logged, the way the constructor of this importer joins the
+ // same pair: 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);
}
+ throw e;
+ }
+ // Entered under the monitor of the map, which close() takes to mark this import closed and
+ // to take its connections away: without it a borrow in flight could be put back after
+ // close() had walked the map, leaving a connection nothing would commit or return - or,
+ // worse, be released twice, the second time onto a connection the pool had already handed
+ // to somebody else.
+ boolean tooLate=false;
+ synchronized (connections) {
+ if (closed) {
+ tooLate=true;
+ }else {
+ connections.put(index, built);
+ }
+ }
+ if (tooLate) {
+ // outside the monitor: a return is a round trip, and close() must not wait behind it
+ releaseUnwatched(built);
+ throw importIsClosed();
+ }
+ return built;
+ }
+
+ /**
+ * A connection of the pool for the given index, or null for a tree that is to share the
+ * connection this import already has.
+ * <p>
+ * The pool having none left is not a reason to fail an import: what an import must not do is
+ * put two threads on one connection, and the monitor of an {@link ImportConnection} sees to
+ * that whether one tree writes through it or five. So a tree that cannot be given a
+ * connection of its own is given the first one instead - the import runs with less of the
+ * parallelism it asked for, rather than stopping halfway through with its clears already
+ * committed. Before #891 an import held one connection for all of its trees and this is what
+ * that looked like.
+ * <p>
+ * Every answer of the pool and of the database is taken this way, not only the one that ran
+ * out of time. The pool at its bound, and a database that {@code CachedConnection} waited out
+ * for the whole deadline, are reported as a {@code SQLTimeoutException} - but a limit of the
+ * database that its dialect table does not recognize is raised at once instead: a mysql
+ * account with a {@code MAX_USER_CONNECTIONS} of its own answers 1226 on SQLState 42000, and
+ * a driver of no known dialect has no vendor code read at all (issue #1011). The type is no
+ * rule either: a failure whose chain names the credentials of the backend is rebuilt as a
+ * plain {@code SQLException} whatever the driver threw. Sorting them here would be that
+ * classification written out a second time, with the failure of a multi-hour import as the
+ * cost of getting it wrong (issue #1013).
+ * <p>
+ * Nothing is hidden by taking them all. The credentials, the driver and the database were
+ * proved by the borrow this importer was built on, so a later one fails for a reason of the
+ * database or of the network - and where the database really is gone, the connection this
+ * import falls back to is gone with it and the next statement fails with what actually
+ * happened, which is a better report than the failure of a borrow.
+ * <p>
+ * An interrupt is not one of those answers: it is how phase two stops an import
+ * ({@code OnDiskMergeImporter.invokeParallel} gives its threads five seconds to answer one),
+ * and a thread that met it by writing the tree through another connection would be carrying
+ * on with the work it was told to drop. It goes back on the thread as well - the wait of a
+ * borrow clears it - since what reads it next is the executor of phase two.
+ */
+ private Connection borrowedOrShared(Integer index) {
+ try {
+ // bounded, and by the default wait of a borrow however much longer the deployment
+ // made that wait: the connections this one waits for are held by this import until
+ // it ends, so an unbounded wait here is a thread waiting for itself - and a long one
+ // is paid over again for every index the pool has no connection to spare for
+ return getConnection(false, CachedConnection.DEFAULT_POOL_TIMEOUT_SECONDS);
+ }catch (SQLException e) {
+ sharedIndexes.add(index);
+ if (e instanceof SQLTimeoutException) {
+ logger.debug(LocalizableMessage.raw("jdbc: the pool has no connection to spare for a tree of this import,"
+ + " which writes it through one it already holds: %s", stackTraceToSingleLineString(e)));
+ }else {
+ // louder than the wait above: that one is the deployment's own bound being
+ // reached, while this is a database refusing a connection outright - the import
+ // goes on with less parallelism than it asked for, and an operator reading a
+ // long import has nothing else to tell them why
+ logger.warn(LocalizableMessage.raw("jdbc: no further connection is given to this import, which"
+ + " writes the tree through one it already holds: %s", stackTraceToSingleLineString(e)));
+ }
+ return null;
+ }catch (InterruptedException e) {
+ // put back where the borrow found it, before the wrapper leaves this class
+ Thread.currentThread().interrupt();
+ throw new StorageRuntimeException(e);
+ }catch (Exception e) {
+ throw e instanceof StorageRuntimeException ? (StorageRuntimeException) e : new StorageRuntimeException(e);
+ }
+ }
+
+ /**
+ * The connection every tree of this import falls back to, which is the one its constructor
+ * borrowed - the only connection an import is sure to have.
+ */
+ private ImportConnection sharedConnection() {
+ final ImportConnection shared=connections.get(FIRST_CONNECTION);
+ if (shared==null) {
+ throw importIsClosed(); // close() took the connections away
+ }
+ return shared;
+ }
+
+ /**
+ * Refuses what arrives after {@code close()}: there is no transaction of this import left to
+ * join, and the connection a write would go to is committed and back in the pool, serving
+ * whoever borrowed it next.
+ * <p>
+ * Asked again under the monitor of the connection by everything that issues a statement. The
+ * flag alone is a moment in time: a thread that read it before {@code close()} raised it, and
+ * reached the connection after, would write on a connection of another borrower. Read under
+ * the monitor it cannot: {@code close()} takes that same monitor to commit and to return the
+ * connection, so either this thread is in front of the commit and part of the import, or it
+ * is behind the return and refused.
+ * <p>
+ * Reasoned rather than pinned by a test: what a test would have to do is park a thread
+ * between the read of the flag and the monitor, and the monitor is the only boundary there
+ * is to park at. That a write after {@code close()} is refused at all is asserted, on one
+ * thread, by {@code ImportConnectionsTestCase}.
+ */
+ private void checkOpen() {
+ if (closed) {
+ throw importIsClosed();
+ }
+ }
+
+ private StorageRuntimeException importIsClosed() {
+ return new StorageRuntimeException(new IllegalStateException(
+ "this import is closed: its connections are committed and back in the pool"));
+ }
+
+ /**
+ * Hands one connection back to the pool and closes the sessions its transaction opened
+ * beside it - the stamp one and the catalog one (#888), both outside the pool and neither
+ * outliving the import that opened it - under the monitor of the connection: the return
+ * rolls back and hands the connection to the next borrower, so a statement of a straggling
+ * import thread must not still be in flight on it - that is the
+ * two-threads-on-one-connection of #891 with another borrower's operation on the other side
+ * of it.
+ * <p>
+ * The failure of a return is returned rather than thrown: a throw out of here would leave
+ * with the failure the caller actually came for - the commit of {@code close()}, and in its
+ * {@code Throwable} branch the {@code Error} that branch exists to preserve - dropped on the
+ * floor, and the connections after this one unreturned (#878).
+ * <p>
+ * By the time {@code close()} reaches this, no statement of the import can be in flight on
+ * the connection anyway: every connection of that list has been through {@code commit()}
+ * under this same monitor, and a write arriving after {@code close()} raised its flag is
+ * refused under it by {@code checkOpen()}. So this monitor is not what the guarantee rests
+ * on today and no test can tell it apart from the commit in front of it - it is here so that
+ * the guarantee does not depend on that ordering, and it is all there is on the paths that
+ * reach here with no commit in front of them ({@link #releaseUnwatched(ImportConnection)}).
+ *
+ * @return what went wrong on the way back, or null
+ */
+ private SQLException release(ImportConnection connection) {
+ synchronized (connection) {
+ try {
+ connection.con.close();
+ return null;
+ } catch (Throwable e) {
+ // Throwable rather than SQLException: this close() is the return to the pool, whose
+ // rollback a driver is free to fail unchecked.
+ return e instanceof SQLException ? (SQLException) e
+ : new SQLException("a connection of the import could not be returned to the pool", e);
+ } finally {
+ // under the monitor with the return itself: each of these sessions holds a connection
+ // of its own in a plain field that its close() reads and nulls without one
+ try {
+ connection.txw.stampSession.close();
+ } finally {
+ connection.txw.catalogSession.close();
+ }
+ }
+ }
+ }
+
+ /**
+ * The return of a connection no caller is waiting on - the loser of a race to open one, and
+ * one borrowed into an import that closed underneath it - which has no failure of an
+ * operation for a failure of the return to ride along with.
+ */
+ private void releaseUnwatched(ImportConnection connection) {
+ final SQLException failure=release(connection);
+ if (failure!=null) {
+ logger.trace(LocalizableMessage.raw("jdbc: unable to return a connection of the import: %s",
+ stackTraceToSingleLineString(failure)));
+ }
+ }
+
+ /** The return of one connection, joined to the failure the caller is going to report. */
+ private SQLException release(ImportConnection connection, SQLException failure) {
+ final SQLException reported=release(connection);
+ if (reported==null) {
+ return failure;
+ }
+ if (failure==null) {
+ return reported;
+ }
+ failure.addSuppressed(reported);
+ return failure;
+ }
+
+ /**
+ * Commits one connection of this import under its monitor, which is what keeps the commit
+ * from being the second statement in flight on it: an import thread that did not answer the
+ * interrupt of phase two can still be inside a statement while {@code close()} runs
+ * ({@code OnDiskMergeImporter.invokeParallel} waits five seconds for its threads and closes
+ * the importer whether they stopped or not), and two threads on one connection is the whole
+ * of #891.
+ * <p>
+ * So a close waits for a statement of a straggler to finish, and a bulk statement carries no
+ * bound of its own. That wait is not new: pgjdbc and Connector/J serialize the work of a
+ * connection behind a lock of their own, so a commit issued beside a statement in flight
+ * already waited there - what is new is that it waits on every dialect, sql server included,
+ * rather than corrupting the driver's state on the one that does not lock.
+ * <p>
+ * A connection with nothing written since its last commit is left alone: the clears of
+ * {@code beforePhaseOne} pass through here for every tree of a container, and a commit of an
+ * empty transaction is a round trip to the database for nothing. Asked here rather than by
+ * the caller, and under this monitor: read in front of it, {@code lastWrite} says what the
+ * connection held rather than what it holds - a write in flight has not counted itself yet.
+ * <p>
+ * On both paths that reach this the wait is already paid in front of it: {@code close()}
+ * reads {@code lastWrite} of every connection of the import under this same monitor before
+ * it commits any of them, and {@code commitPeersOf()} holds it over this call - so no test
+ * can tell this monitor from the ones ahead of it, and it is here so that what keeps two
+ * threads off one connection does not rest on the order another method happens to work in.
+ */
+ private void commit(ImportConnection connection) throws SQLException {
+ synchronized (connection) {
+ if (connection.lastWrite==0) {
+ return;
+ }
+ connection.con.commit();
+ connection.lastWrite=0;
+ }
+ }
+
+ /**
+ * Commits the other connections of this import, which is what the one connection an import
+ * used to hold did of its own accord: {@code clearTree()} ends in a commit, and that commit
+ * made durable every write the import had made so far. Split over several connections and
+ * left to {@code close()}, those writes would stay uncommitted while the tables they are
+ * about were emptied and committed one after another.
+ * <p>
+ * What that is worth depends on the order the caller writes in, and the two strategies of
+ * {@code OnDiskMergeImporter} differ. A {@code rebuild-index} writes the
+ * {@code setTrust(false)} of {@code RebuildIndexStrategy.beforePhaseOne} before it empties
+ * the trees that flag describes, so this makes the flag durable in front of the clear it
+ * belongs to: a server that stops in between comes back to an index that is empty and says
+ * so. An {@code import-ldif} takes {@code AbstractTwoPhaseImportStrategy.beforePhaseOne},
+ * which empties every tree of the container first and writes the flags after - so there this
+ * bounds how much of an import stays uncommitted (the flags of one container are made
+ * durable by the clears of the next), rather than closing that window. Not a regression of
+ * the connections an import now takes: the one connection it held before #891 committed in
+ * exactly the same places, because the caller writes in exactly the same order.
+ * <p>
+ * Run in front of the clear rather than after it, so that a peer whose commit fails leaves
+ * the import with the tree not yet emptied: the destructive half of a clear is the one thing
+ * that must not be durable while a write it is meant to invalidate is not.
+ * <p>
+ * Which peers hold something to commit is decided under the monitor of each of them - by
+ * {@link #commit(ImportConnection)}, which passes over a connection with nothing written
+ * since its last commit - rather than by a read of {@code lastWrite} taken in front of that
+ * monitor. A peer whose first write is still in flight has not set it yet ({@code put()}
+ * counts a write once the statement has come back), and that is precisely the write a
+ * cheaper test would pass over: {@code beforePhaseOne} runs on the import threads, one
+ * container at a time per thread and several containers at once, so the flags of a container
+ * being written are a peer of the clears of the next. The one connection an import held
+ * before #891 carried that write into the commit of the clear - it was the same transaction,
+ * and a clear issued beside it waited for it in the driver - so passing over it here would
+ * be a commit point the single connection did not have.
+ * <p>
+ * What that costs is the wait: a clear now waits for a statement in flight on each peer, as
+ * every write of an import waited for the one connection it all went through.
+ * <p>
+ * One connection is held at a time, so no thread of an import ever holds two of these
+ * monitors and two threads clearing at once cannot wait for each other.
+ */
+ private void commitPeersOf(ImportConnection cleared) {
+ for (final ImportConnection peer : connections.values()) {
+ if (peer==cleared) {
+ continue;
+ }
+ try {
+ synchronized (peer) {
+ // under the monitor, like every other statement of an import: a clear that
+ // arrives once close() has committed and returned this connection would
+ // otherwise commit the transaction of whoever borrowed it next
+ checkOpen();
+ commit(peer);
+ }
+ }catch (SQLException e) {
+ // reported the way the commit of the clear itself is: what these make durable is
+ // the work the clear is the commit point for
+ throw new StorageRuntimeException(e);
+ }
+ }
+ }
+
+ /**
+ * Commits the given connections in the order they were last written to - see {@link #writes}.
+ * <p>
+ * The order is taken from a copy of what each connection carries rather than read as the sort
+ * goes: a straggling import thread that got in front of {@code close()} can raise the number
+ * of the connection it holds while this runs, and a sort whose keys move under it is reported
+ * by {@code TimSort} as a comparator that violates its contract rather than as the race it is.
+ */
+ private void commitAll(List<ImportConnection> taken) throws SQLException {
+ final Map<ImportConnection,Long> lastWrites=new IdentityHashMap<>();
+ for (final ImportConnection connection : taken) {
+ synchronized (connection) {
+ lastWrites.put(connection, connection.lastWrite);
+ }
+ }
+ final List<ImportConnection> byLastWrite=new ArrayList<>(taken);
+ byLastWrite.sort(Comparator.comparingLong(lastWrites::get));
+ for (final ImportConnection connection : byLastWrite) {
+ commit(connection);
+ }
+ }
+
+ /**
+ * Hands every connection this import took back to the pool, 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.
+ * <p>
+ * Every connection is released even when one of them fails on the way: what a connection
+ * left behind holds is a permit of the pool, and a pool is never removed from the map.
+ */
+ private SQLException releaseConnections(List<ImportConnection> taken, SQLException failure) {
+ for (final ImportConnection connection : taken) {
+ failure=release(connection, failure);
}
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
- // way there - and a commit that throws on the way would otherwise leave the connection
- // out of the pool for good, holding the transaction and the locks of that import.
+ // The connections go back whatever the commit does, and the storage this importer opened
+ // is closed whatever they do: 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 way there -
+ // and a commit that throws on the way would otherwise leave the connections out of the
+ // pool for good, holding the transactions and the locks of that import.
+ //
+ // Every connection is committed, not only the one the constructor borrowed: each is a
+ // transaction of its own, and what one of them holds uncommitted is the work of every tree
+ // it was given (#891).
@Override
public void close() {
try {
+ // Taken out of the map rather than walked in it, under the monitor that guards it:
+ // from here this import has no transaction left for a write to belong to, a borrow
+ // still in flight is refused rather than left behind, and a second close() finds
+ // nothing to commit or return - one that walked the map again would roll back and
+ // re-pool connections the pool had already handed to somebody else.
+ final List<ImportConnection> taken;
+ final ImportConnection describing;
+ synchronized (connections) {
+ closed=true;
+ describing=connections.get(FIRST_CONNECTION);
+ taken=new ArrayList<>(connections.values());
+ connections.clear();
+ }
SQLException failure=null;
try {
- con.commit();
- if (aborted) {
- logger.debug(LocalizableMessage.raw("jdbc: import aborted: statistics of the trees it wrote are left alone"));
- }else {
- updateTableStatistics(con, writtenTrees);
- }
+ commitAll(taken);
} 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
+ // a driver is supposed to throw: nothing else holds these connections, and only
+ // their close() gives back the permits they 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);
+ final SQLException onTheWayOut=releaseConnections(taken, null);
if (onTheWayOut!=null) {
t.addSuppressed(onTheWayOut);
}
throw t;
}
- // Back to the pool even when the commit failed: nothing else holds this connection,
+ // Everything but the connection that describes goes back before the statistics are
+ // gathered: that is one statement per tree the import wrote, a full scan of the table
+ // on oracle and bounded by a property of its own, and the connections of an import are
+ // the pool's to hand to the operations of the server as soon as they are committed.
+ for (final ImportConnection connection : taken) {
+ if (connection!=describing) {
+ failure=release(connection, failure);
+ }
+ }
+ try {
+ if (aborted) {
+ logger.debug(LocalizableMessage.raw("jdbc: import aborted: statistics of the trees it wrote are left alone"));
+ }else if (describing!=null && failure==null) {
+ // On the connection the constructor borrowed, under its monitor like every
+ // other statement of an import: the statements below describe a table to the
+ // optimizer rather than read one, and they run once every connection of this
+ // import is committed - so there is no work of another one left for them to
+ // miss.
+ synchronized (describing) {
+ updateTableStatistics(describing.con, writtenTrees);
+ }
+ }
+ } catch (Throwable t) {
+ if (describing!=null) {
+ final SQLException onTheWayOut=release(describing, null);
+ if (onTheWayOut!=null) {
+ t.addSuppressed(onTheWayOut);
+ }
+ }
+ throw t;
+ }
+ // Back to the pool even when a commit failed: nothing else holds this connection,
// so leaving it behind would leak it along with the failure.
- failure=releaseConnection(failure);
+ if (describing!=null) {
+ failure=release(describing, failure);
+ }
if (failure!=null) {
throw new StorageRuntimeException(failure);
}
@@ -5062,21 +5694,36 @@
@Override
public void clearTree(TreeName name) {
- txw.clearTree(name);
+ final ImportConnection connection=connectionOf(name);
+ commitPeersOf(connection); // in front of the clear, see there
+ synchronized (connection) {
+ checkOpen();
+ connection.txw.clearTree(name);
+ connection.lastWrite=0; // the clear ends in a commit of this connection
+ }
writtenTrees.add(name);
}
@Override
public void put(TreeName treeName, ByteSequence key, ByteSequence value) {
- txw.put(treeName, key, value);
+ final ImportConnection connection=connectionOf(treeName);
+ synchronized (connection) {
+ checkOpen();
+ connection.txw.put(treeName, key, value);
+ connection.written();
+ }
writtenTrees.add(treeName);
}
-
+
@Override
public ByteString read(TreeName treeName, ByteSequence key) {
- return txr.read(treeName, key);
+ final ImportConnection connection=connectionOf(treeName);
+ synchronized (connection) {
+ checkOpen();
+ return connection.txr.read(treeName, key);
+ }
}
-
+
// Bulk like every other statement of an import, by the class of the transaction it comes
// from: this walks a whole tree with no client waiting on it - phase one of a rebuild-index
// reads every record of id2entry through this cursor (OnDiskMergeImporter.ID2EntrySource) -
@@ -5084,7 +5731,88 @@
// rather than a step along an index.
@Override
public SequentialCursor<ByteString, ByteString> openCursor(TreeName treeName) {
- return txr.openCursor(treeName);
+ final ImportConnection connection=connectionOf(treeName);
+ synchronized (connection) {
+ checkOpen();
+ return new ImportCursor(connection, connection.txr.openCursor(treeName));
+ }
+ }
+
+ /**
+ * A cursor of an import, every method of which runs under the monitor of the connection it
+ * walks. An import has more trees than connections, so the tree this cursor walks shares
+ * its connection with the trees written through it, and a batch of this cursor must not be
+ * in flight there beside a statement of one of them.
+ * <p>
+ * The methods that issue no statement take the monitor as well, rather than being excused
+ * on the ground that the thread which opened the cursor is the only one to call them:
+ * nothing here enforces that, and what they read - the batch the last {@code next()} left
+ * in {@link CursorImpl}, an {@code ArrayDeque} and four plain fields - is written under
+ * that monitor and carries no memory barrier of its own. An uncontended monitor is what
+ * that costs.
+ */
+ private final class ImportCursor implements SequentialCursor<ByteString, ByteString> {
+ private final ImportConnection connection;
+ private final SequentialCursor<ByteString, ByteString> cursor;
+
+ ImportCursor(ImportConnection connection, SequentialCursor<ByteString, ByteString> cursor) {
+ this.connection=connection;
+ this.cursor=cursor;
+ }
+
+ @Override
+ public boolean next() {
+ synchronized (connection) {
+ checkOpen();
+ return cursor.next();
+ }
+ }
+
+ // A cursor of an import is opened on the read transaction of its connection, whose
+ // isReadOnly CursorImpl carries, so this forwards the refusal that transaction answers
+ // with rather than a delete. Nothing is recorded against the connection for that reason:
+ // a mark here would be a write number this cursor never made, and the highest one at
+ // that - it would move its connection to the end of the order close() commits in, which
+ // is where the trust flags of afterPhaseTwo belong. An importer that ever opens a
+ // writeable cursor has to record the write there, next to the delete that made it.
+ @Override
+ public void delete() {
+ synchronized (connection) {
+ checkOpen();
+ cursor.delete();
+ }
+ }
+
+ @Override
+ public boolean isDefined() {
+ synchronized (connection) {
+ return cursor.isDefined();
+ }
+ }
+
+ @Override
+ public ByteString getKey() {
+ synchronized (connection) {
+ return cursor.getKey();
+ }
+ }
+
+ @Override
+ public ByteString getValue() {
+ synchronized (connection) {
+ return cursor.getValue();
+ }
+ }
+
+ // Not refused after close() the way a read or a write of the importer is: this frees what
+ // the last batch left in the cursor and reaches no connection, and the try-with-resources
+ // of a cancelled phase-two task closes its cursor after the importer it walked.
+ @Override
+ public void close() {
+ synchronized (connection) {
+ cursor.close();
+ }
+ }
}
}
@@ -5092,9 +5820,9 @@
@Override
public Importer startImport() throws ConfigException, StorageRuntimeException {
// 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.
+ // borrowing the first of the connections 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();
}
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 3e834f3..5d22954 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,8 +15,10 @@
*/
package org.opends.server.backends.jdbc;
+import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
import org.opends.server.DirectoryServerTestCase;
+import org.opends.server.backends.pluggable.spi.TreeName;
import org.opends.server.backends.pluggable.spi.AccessMode;
import org.opends.server.backends.pluggable.spi.Importer;
import org.testng.annotations.AfterClass;
@@ -36,6 +38,7 @@
import java.sql.Driver;
import java.sql.DriverManager;
import java.sql.DriverPropertyInfo;
+import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.SQLTimeoutException;
import java.util.ArrayDeque;
@@ -61,6 +64,7 @@
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
@@ -685,6 +689,10 @@
final String url = StubDriver.PREFIX + "import-unchecked-commit";
final Connection parent = mock(Connection.class);
when(parent.isValid(anyInt())).thenReturn(true);
+ // the statement of the write below, which is what gives the commit of close() something to do:
+ // a connection an import wrote nothing through is not committed at all (#891), so an import
+ // that writes nothing would reach neither the Error this test injects nor the return it is about
+ when(parent.prepareStatement(anyString())).thenReturn(mock(PreparedStatement.class));
doThrow(new Error("out of memory while importing")).when(parent).commit();
stub.answerWith(parent);
final JDBCBackendCfg cfg = mock(JDBCBackendCfg.class);
@@ -692,12 +700,17 @@
final JDBCStorage storage = new JDBCStorage(cfg, null);
storage.open(AccessMode.READ_WRITE);
final Importer importer = storage.startImport();
+ importer.put(new TreeName("dc=example,dc=com", "id2entry"),
+ ByteString.valueOfUtf8("key"), ByteString.valueOfUtf8("value"));
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
+ // reported to the caller, which is what an Error out of an import has to be. Asserted
+ // rather than accepted whole: an AssertionError of the fail() above is an Error too, and
+ // would otherwise be caught here and read as the injected one
+ assertEquals(expected.getMessage(), "out of memory while importing");
}
assertEquals(CachedConnection.poolOf(url).idleCount(), 1, "the import kept the connection of the pool");
@@ -1695,6 +1708,32 @@
assertEquals(CachedConnection.deadlineOf(startedAt, 60), startedAt + 60_000);
}
+ /**
+ * A borrow that carries a bound of its own waits for the shorter of the two, whichever way the
+ * deployment spelled a wait with no bound at all.
+ * <p>
+ * Only an import carries one: it holds a connection per tree it writes until it ends (#891), so
+ * the pool it waits at may be full of nothing but its own connections and the wait would be a
+ * deadlock rather than a queue. It has somewhere to go when the wait runs out - the tree is
+ * written through a connection the import already holds - so a long wait of the deployment is
+ * capped rather than merely replaced: paid over again for every tree the pool has nothing to
+ * spare for, it would be the duration of an import rather than a bound on it.
+ */
+ @Test
+ public void testABorrowWithABoundOfItsOwnWaitsForTheShorterOfTheTwo() {
+ assertEquals(CachedConnection.boundedWait(30, 60), 30, "the wait of the deployment was inside the bound");
+ assertEquals(CachedConnection.boundedWait(600, 60), 60, "a long finite wait was not capped");
+ assertEquals(CachedConnection.boundedWait(0, 60), 60, "0 stands for a wait with no bound");
+ // the other spelling of a wait with no bound: seconds enough that the milliseconds they
+ // stand for do not fit in a long, which the deadline of the borrow reads as forever
+ assertEquals(CachedConnection.boundedWait(Long.MAX_VALUE / 1000, 60), 60,
+ "a wait whose milliseconds overflow a long is unbounded too");
+ assertEquals(CachedConnection.boundedWait(Long.MAX_VALUE, 60), 60);
+ // ... and a borrow that carries no bound of its own takes the wait of the deployment whole
+ assertEquals(CachedConnection.boundedWait(600, 0), 600);
+ assertEquals(CachedConnection.boundedWait(0, 0), 0);
+ }
+
/** The connection string holds the credentials of the backend: a stall report must not carry them. */
@Test
public void testLoggedConnectionStringCarriesNoCredentials() throws Exception {
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/ImportConnectionsTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/ImportConnectionsTestCase.java
new file mode 100644
index 0000000..201ad8a
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/ImportConnectionsTestCase.java
@@ -0,0 +1,1348 @@
+/*
+ * The contents of this file are subject to the terms of the Common Development and
+ * Distribution License (the License). You may not use this file except in compliance with the
+ * License.
+ *
+ * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
+ * specific language governing permission and limitations under the License.
+ *
+ * When distributing Covered Software, include this CDDL Header Notice in each file and include
+ * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
+ * Header, with the fields enclosed by brackets [] replaced by your own identifying
+ * information: "Portions Copyright [year] [name of copyright owner]".
+ *
+ * Copyright 2026 3A Systems, LLC.
+ */
+package org.opends.server.backends.jdbc;
+
+import org.forgerock.opendj.ldap.ByteString;
+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.opends.server.backends.pluggable.spi.StorageRuntimeException;
+import org.opends.server.backends.pluggable.spi.TreeName;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.sql.Connection;
+import java.sql.Driver;
+import java.sql.DriverManager;
+import java.sql.DriverPropertyInfo;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+import java.util.logging.Logger;
+
+import static org.forgerock.opendj.config.ConfigurationMock.mockCfg;
+import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+
+/**
+ * The {@link Importer} of the JDBC backend is used by several threads at once - phase two of an
+ * import runs one thread per tree - and {@code java.sql.Connection} is not thread-safe. Sharing
+ * one connection between those threads corrupts the driver rather than merely serializing the
+ * work: the sql server driver keeps the reconnect listeners of a connection in a plain
+ * {@code ArrayList} that every {@code prepareStatement()} and every statement {@code close()}
+ * mutates, and two import threads on one connection walked it past the end of its array
+ * (issue #891).
+ * <p>
+ * Needs no database: the connections are handed out by a driver of this test, which records every
+ * thread inside a statement of a connection - and which thread commits, rolls back or returns one
+ * while another is in there - so what it records is the defect itself, two threads on one
+ * connection at the same time, rather than an exception of one driver. It therefore holds for every
+ * dialect this backend takes.
+ * <p>
+ * Two ways of getting the threads to overlap, because trees that get connections of their own and
+ * trees that share one cannot be asked the same question. Where the connections differ, the driver
+ * holds the first statement of each thread at a rendezvous until its peers have one in flight of
+ * their own. Where they do not, a second thread never reaches such a rendezvous - the monitor of
+ * the connection is what this is about - so the threads line up before they write instead, and the
+ * first one in holds the connection for a moment to see whether the other turns up beside it.
+ */
+@SuppressWarnings("javadoc")
+@Test(groups = { "precommit", "jdbc" }, sequential = true)
+public class ImportConnectionsTestCase extends DirectoryServerTestCase {
+
+ /** Long enough for a peer thread to reach a statement on a loaded machine, and no longer. */
+ private static final long RENDEZVOUS_SECONDS = 30;
+
+ /**
+ * How long the first thread inside a statement holds the connection it is on. Paid in full by
+ * every run where the importer serializes its threads as it should - which is the point: a peer
+ * that is let in arrives at once, having been waiting on the monitor since the starting line.
+ */
+ private static final long HOLD_SECONDS = 2;
+
+ /** How long a close of an import may take before a thread of the test is taken to be stuck. */
+ private static final long CLOSE_SECONDS = 30;
+
+ /**
+ * How long a thread parked inside a statement stays in there before it gives up on the test
+ * that parked it. Paid by no passing run - the test lets it go as soon as it has what it came
+ * for - and it is bounded so that a run which never gets there is a red test rather than a
+ * build that hangs on a monitor nobody is going to release.
+ */
+ private static final long PARK_SECONDS = 30;
+
+ private static final TreeName ID2ENTRY = new TreeName("dc=example,dc=com", "id2entry");
+ private static final TreeName DN2ID = new TreeName("dc=example,dc=com", "dn2id");
+ private static final TreeName STATE = new TreeName("dc=example,dc=com", "state");
+
+ private final StubDriver stub = new StubDriver();
+ /**
+ * What the driver of this test answers a connect with, or null for a connection of its own.
+ * A failure per connect rather than one instance thrown over and over: a caller is free to
+ * give what it caught a cause and a suppressed exception of its own.
+ */
+ private final AtomicReference<Supplier<SQLException>> refusal = new AtomicReference<>();
+
+ /** The connections an import issued a statement on, in no particular order. */
+ private final Set<Connection> used =
+ Collections.synchronizedSet(Collections.newSetFromMap(new IdentityHashMap<Connection, Boolean>()));
+ /** The same, in the order the import first issued a statement on them. */
+ private final List<Connection> usedInOrder = Collections.synchronizedList(new ArrayList<Connection>());
+ /** Every commit of the import, in the order the connections took them. */
+ private final List<Connection> commits = Collections.synchronizedList(new ArrayList<Connection>());
+ /**
+ * The threads inside a statement of a connection right now - every one of them, rather than the
+ * first to arrive: a thread that finds a peer there records the pair and stays out of the map,
+ * and the peer's exit would then leave the map saying the connection is free while it is not.
+ * What comes after - another statement, a commit, the rollback of a return - would be recorded
+ * against nobody, which is the one thing this suite is here to notice.
+ */
+ private final Map<Connection, Set<Thread>> inStatement =
+ Collections.synchronizedMap(new IdentityHashMap<Connection, Set<Thread>>());
+ /** Every pair of threads that was inside a statement of one connection at the same time. */
+ private final List<String> shared = Collections.synchronizedList(new ArrayList<String>());
+
+ /** Where the threads of a test meet, so that their statements are in flight at the same time. */
+ private volatile CyclicBarrier rendezvous;
+ /** Whether this thread has already met its peers: a thread waits there once, whatever it issues after. */
+ private final ThreadLocal<Boolean> met = ThreadLocal.withInitial(() -> Boolean.FALSE);
+
+ /**
+ * Where the threads of a test line up before they write, which is in front of every monitor the
+ * importer takes - unlike {@link #rendezvous}, which they reach once they are already inside a
+ * statement, and which two threads sharing a connection therefore cannot both reach.
+ */
+ private volatile CyclicBarrier startingLine;
+ /**
+ * Counted down by every thread that gets inside a statement, and waited on by the first of them:
+ * it holds the connection long enough for a peer to get in there beside it, if anything lets it.
+ */
+ private volatile CountDownLatch insideAStatement;
+ /** Whether this thread has already held a connection: it holds the first statement it issues. */
+ private final ThreadLocal<Boolean> held = ThreadLocal.withInitial(() -> Boolean.FALSE);
+
+ /**
+ * Counted down by the first thread of a test to get inside a statement, and awaited by the test
+ * itself: what a write in flight looks like to the rest of the import - a connection whose
+ * statement has not come back, and which therefore has nothing recorded against it yet.
+ */
+ private volatile CountDownLatch insideAStatementNow;
+ /** Counted down by the test to let the parked write finish, once it has seen what it came to see. */
+ private volatile CountDownLatch letTheStatementFinish;
+ /** Whether this thread has already been parked: a thread is parked in the first statement it issues. */
+ private final ThreadLocal<Boolean> parked = ThreadLocal.withInitial(() -> Boolean.FALSE);
+
+ private ExecutorService threads;
+ /** The pool bound and its borrow deadline as this JVM had them, put back after every test. */
+ private String poolMaxOfTheJvm;
+ private String poolTimeoutOfTheJvm;
+
+ @BeforeClass
+ public void registerStubDriver() throws Exception {
+ DriverManager.registerDriver(stub);
+ }
+
+ /**
+ * The bound of the pools of this suite, pinned: how many connections an import takes is clamped
+ * to it, so a value another suite of this package left set - CachedConnectionTestCase varies it
+ * - would decide the counts asserted here. A pool reads it once, when the first borrow of a
+ * connection string creates it, so it has to stand before the storage of a test is opened.
+ */
+ @BeforeMethod
+ public void pinThePoolBound() {
+ poolMaxOfTheJvm = System.getProperty(CachedConnection.POOL_MAX_PROPERTY);
+ poolTimeoutOfTheJvm = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
+ System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "16");
+ }
+
+ @AfterClass
+ public void deregisterStubDriver() throws Exception {
+ DriverManager.deregisterDriver(stub);
+ }
+
+ @AfterMethod
+ public void forgetWhatTheTestRecorded() {
+ System.clearProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY);
+ // put back rather than cleared: a sibling suite of this package varies this property, and a
+ // pool is created with the value that stands when its connection string is first borrowed on
+ putBack(CachedConnection.POOL_MAX_PROPERTY, poolMaxOfTheJvm);
+ putBack(CachedConnection.POOL_TIMEOUT_PROPERTY, poolTimeoutOfTheJvm);
+ // the rendezvous of the next test is not the one this thread already went to
+ met.remove();
+ held.remove();
+ parked.remove();
+ // let go of anything this test left parked before its threads are shut down: a write held
+ // inside a statement holds the monitor of a connection, and the next test of this suite
+ // starts by opening a storage on that pool
+ if (letTheStatementFinish != null) {
+ letTheStatementFinish.countDown();
+ }
+ if (threads != null) {
+ threads.shutdownNow();
+ threads = null;
+ }
+ refusal.set(null);
+ rendezvous = null;
+ startingLine = null;
+ insideAStatement = null;
+ insideAStatementNow = null;
+ letTheStatementFinish = null;
+ forgetTheConnections();
+ inStatement.clear();
+ shared.clear();
+ }
+
+ private static void putBack(String property, String value) {
+ if (value == null) {
+ System.clearProperty(property);
+ } else {
+ System.setProperty(property, value);
+ }
+ }
+
+ private void forgetTheConnections() {
+ used.clear();
+ usedInOrder.clear();
+ commits.clear();
+ }
+
+ /**
+ * Two trees written at the same time are written through connections of their own. This is the
+ * contract {@code Importer} states and the defect of #891: every thread of phase two wrote
+ * through the one connection the importer borrowed in its constructor.
+ */
+ @Test(timeOut = 120000)
+ public void testTwoTreesAreWrittenThroughConnectionsOfTheirOwn() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
+ final String url = StubDriver.PREFIX + "importer-parallel";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ importing(importer, () -> {
+ // both threads have to be inside a statement at once for the record below to say
+ // anything: a run where one finished before the other started shares no connection
+ // however the importer borrows them
+ rendezvous = new CyclicBarrier(2);
+ threads = daemonThreads(2, "import-writer");
+ final Future<?> id2entry = put(importer, ID2ENTRY);
+ final Future<?> dn2id = put(importer, DN2ID);
+ id2entry.get(RENDEZVOUS_SECONDS * 2, TimeUnit.SECONDS);
+ dn2id.get(RENDEZVOUS_SECONDS * 2, TimeUnit.SECONDS);
+
+ // belt and braces here rather than the cover of this suite: with a connection each
+ // the record below is empty however the import behaves, and a run where the two
+ // trees did share one fails at the rendezvous instead - the second thread would be
+ // on the monitor of the connection and never reach it. What pins the monitor is
+ // testTwoTreesOnOneConnectionAreNotWrittenAtTheSameTime
+ assertEquals(shared, Collections.emptyList(),
+ "two import threads issued a statement on one connection at the same time");
+ assertEquals(used.size(), 2, "the two trees of the import did not get a connection each");
+ });
+ } finally {
+ storage.close();
+ }
+ }
+
+ /**
+ * Two trees that share one connection are not written at the same time. The bound of an import,
+ * and a pool with nothing to spare, both put several trees on one connection - and what #891 is
+ * about is two threads issuing statements on one connection, not how many connections an import
+ * holds: the monitor of the connection is what has to keep the second thread out.
+ * <p>
+ * The two threads cannot be brought together at a barrier inside their statements the way the
+ * test above does it - the second one never gets in to reach it - so the first one in holds the
+ * connection for {@link #HOLD_SECONDS} instead, having lined up with its peer beforehand. A
+ * second thread let in beside it arrives at once, since it has been waiting on that monitor
+ * since the starting line, and both are then recorded in {@link #shared}.
+ */
+ @Test(timeOut = 120000)
+ public void testTwoTreesOnOneConnectionAreNotWrittenAtTheSameTime() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "1"); // both trees on one connection
+ final String url = StubDriver.PREFIX + "importer-serialized";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ importing(importer, () -> {
+ startingLine = new CyclicBarrier(2);
+ insideAStatement = new CountDownLatch(2);
+ threads = daemonThreads(2, "import-writer");
+ final Future<?> id2entry = put(importer, ID2ENTRY);
+ final Future<?> dn2id = put(importer, DN2ID);
+ id2entry.get(RENDEZVOUS_SECONDS * 2, TimeUnit.SECONDS);
+ dn2id.get(RENDEZVOUS_SECONDS * 2, TimeUnit.SECONDS);
+
+ assertEquals(used.size(), 1, "the two trees of this import did not share one connection");
+ assertEquals(shared, Collections.emptyList(),
+ "two import threads issued a statement on one connection at the same time");
+ });
+ } finally {
+ storage.close();
+ }
+ }
+
+ /**
+ * ... and no more connections than that: an import takes them for its whole duration, and the
+ * pool they come from is bounded and shared with the operations of every other backend on that
+ * database (#878). One thread per tree is what phase two runs, and a default backend has trees
+ * enough to empty a default pool.
+ */
+ @Test(timeOut = 120000)
+ public void testTreesShareTheConnectionsOfABoundAboveOne() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
+ final String url = StubDriver.PREFIX + "importer-bounded-two";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ // five trees over two connections: the bound is what the round robin wraps at
+ importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(new TreeName("dc=example,dc=com", "id2childrencount"),
+ ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(new TreeName("dc=example,dc=com", "referral"),
+ ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+
+ assertEquals(used.size(), 2, "the import did not spread its trees over the connections of its bound");
+ } finally {
+ importer.close();
+ storage.close();
+ }
+ }
+
+ /**
+ * The bound an import takes where the operator set none, which is the branch every real install
+ * takes: half the bound of the pool, and no more than half of what the default bound would be.
+ * An import holds what it takes for its whole duration out of a pool that is shared with the
+ * LDAP traffic of every backend on that database (#878), so the deployment that raised the
+ * bound of its pool did not thereby ask for an import to take half of it.
+ */
+ @Test(timeOut = 120000)
+ public void testTheBoundOfAnImportDefaultsToHalfThePool() throws Exception {
+ // set by no test of this suite, and cleared after every one of them: this is the default
+ System.clearProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY);
+ // a bound of the pool that is not the default one: DEFAULT_POOL_MAX is
+ // max(16, processors * 2), so the 16 pinThePoolBound() puts on the pools of this suite is
+ // the default itself on a machine of eight cores or fewer - and half the pool and half the
+ // default are then the same number, which is no test of which of the two an import halves
+ System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "6");
+ final JDBCStorage half = importingStorage(StubDriver.PREFIX + "importer-default-bound");
+ try {
+ assertEquals(half.importConnections(), 3, "an import did not default to half the bound of the pool");
+ } finally {
+ half.close();
+ }
+ System.setProperty(CachedConnection.POOL_MAX_PROPERTY, String.valueOf(CachedConnection.DEFAULT_POOL_MAX * 4));
+ final JDBCStorage capped = importingStorage(StubDriver.PREFIX + "importer-default-bound-large-pool");
+ try {
+ assertEquals(capped.importConnections(), CachedConnection.DEFAULT_POOL_MAX / 2,
+ "an import of a pool larger than the default took more than half of what the default bound would be");
+ } finally {
+ capped.close();
+ }
+ }
+
+ /** ... and a bound the pool cannot honour is the pool's, not the property's. */
+ @Test(timeOut = 120000)
+ public void testTheBoundOfAnImportIsClampedToTheBoundOfThePool() throws Exception {
+ System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "2");
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "8");
+ final String url = StubDriver.PREFIX + "importer-clamped";
+ final JDBCStorage storage = importingStorage(url);
+ try {
+ assertEquals(storage.importConnections(), 2, "the bound of an import was not clamped to the pool");
+ } finally {
+ storage.close();
+ }
+ }
+
+ @Test(timeOut = 120000)
+ public void testTheConnectionsOfAnImportAreBounded() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "1");
+ final String url = StubDriver.PREFIX + "importer-bounded";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+
+ assertEquals(used.size(), 1, "the import took more connections than the bound allows");
+ } finally {
+ importer.close();
+ storage.close();
+ }
+ }
+
+ /**
+ * The trees decide which connection a write goes through, not the threads that write them. The
+ * connections of an import are transactions of their own, so two of them writing one row would
+ * have the second wait for the first to commit - which an import does in {@code close()}, when
+ * the thread that would have to release the row is long done, and the bulk class carries no
+ * bound to break the wait. {@code setTrust()} writes the state tree of a container from an
+ * import thread in {@code beforePhaseOne} and from the closing thread in {@code afterPhaseTwo},
+ * which is exactly that shape.
+ */
+ @Test(timeOut = 120000)
+ public void testOneTreeIsWrittenThroughOneConnectionWhoeverWritesIt() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "4");
+ final String url = StubDriver.PREFIX + "importer-one-tree";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ importing(importer, () -> {
+ // three threads, one tree: an import thread of phase one, another of phase two, and
+ // the thread that closes the importer, which is the one afterPhaseTwo runs on
+ putOnAThreadOfItsOwn(importer, STATE);
+ putOnAThreadOfItsOwn(importer, STATE);
+ importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+
+ assertEquals(used.size(), 1, "one tree was written through more than one connection");
+ });
+ } finally {
+ storage.close();
+ }
+ }
+
+ /**
+ * Every connection an import took is committed and given back, not only the one its
+ * constructor borrowed: what a connection left behind holds is a transaction of the import and
+ * a permit of the pool, and a pool is never removed from the map - so a permit lost to an
+ * import is lost for the life of the server (#878).
+ */
+ @Test(timeOut = 120000)
+ public void testEveryConnectionOfAnImportIsCommittedAndReturned() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "3");
+ final String url = StubDriver.PREFIX + "importer-returned";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+
+ importer.close();
+
+ assertEquals(used.size(), 3, "the three trees of the import did not get a connection each");
+ for (final Connection con : connectionsUsed()) {
+ verify(con, times(1)).commit();
+ }
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 3, "an import kept a connection of the pool");
+ } finally {
+ storage.close();
+ }
+ assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "an import kept a permit of the pool");
+ }
+
+ /**
+ * A clear commits every connection of the import, not only the one whose tree it clears, and it
+ * does so before it empties anything. The one connection an import used to hold made that so of
+ * its own accord - {@code clearTree()} ends in a commit, and that commit made durable every
+ * write the import had made so far. What that is worth to a {@code rebuild-index} is the
+ * {@code setTrust(false)} of {@code RebuildIndexStrategy.beforePhaseOne}, which is written just
+ * in front of the clear of the tree it describes: left to {@code close()} that flag would still
+ * be uncommitted while the table was emptied and committed, and a server that stopped in
+ * between would come back to an index that is empty and marked trusted.
+ */
+ @Test(timeOut = 120000)
+ public void testAClearCommitsTheOtherConnectionsOfTheImport() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
+ final String url = StubDriver.PREFIX + "importer-clear-commits";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ // the state tree stands for the trust flag: written, and left uncommitted until
+ // something commits the connection it went to
+ importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.clearTree(ID2ENTRY);
+
+ // asserted in order rather than counted: the destructive half of a clear is the one
+ // thing that must not be durable while a write it is meant to invalidate is not, so the
+ // commit of the peer belongs in front of the one the clear itself ends in - a count of
+ // two is the same whichever way round they happened
+ assertEquals(commits, Arrays.asList(usedInOrder.get(0), usedInOrder.get(1)),
+ "the clear did not commit the other connection of the import before it emptied its own tree");
+ } finally {
+ importer.close();
+ storage.close();
+ }
+ }
+
+ /**
+ * ... including a peer whose first write is still in flight, which is not a connection with
+ * nothing to commit but one whose write has not come back yet.
+ * <p>
+ * An import counts a write once its statement returns, so a connection written for the first
+ * time has nothing recorded against it for exactly as long as that statement takes. That window
+ * is reached by more than one thread at once: {@code beforePhaseOne} runs on the import threads
+ * - {@code OnDiskMergeImporter.processEntry} guards it with a latch per container, not one for
+ * all of them - so on an {@code import-ldif} of two base DNs the {@code setTrust(false)} of one
+ * container is written while the clears of the next are running. The one connection an import
+ * held before #891 carried such a write into the commit of the clear, being the same
+ * transaction and issued in front of it in the driver, so a clear that passed over it here
+ * would be a commit point the single connection did not have.
+ */
+ @Test(timeOut = 120000)
+ public void testAClearCommitsAPeerWhoseWriteIsStillInFlight() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
+ final String url = StubDriver.PREFIX + "importer-clear-commits-in-flight";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ importing(importer, () -> {
+ insideAStatementNow = new CountDownLatch(1);
+ letTheStatementFinish = new CountDownLatch(1);
+ // the state tree stands for the trust flag of beforePhaseOne: written on a thread of
+ // the import, and held inside the statement it issues
+ final AtomicReference<Throwable> flagFailure = new AtomicReference<>();
+ final Thread writing = daemon(() -> {
+ try {
+ importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ } catch (Throwable t) {
+ flagFailure.set(t);
+ }
+ }, "import-flag");
+ // the clear of the next container, on a connection of its own
+ final AtomicReference<Throwable> clearFailure = new AtomicReference<>();
+ final Thread clearing = daemon(() -> {
+ try {
+ importer.clearTree(ID2ENTRY);
+ } catch (Throwable t) {
+ clearFailure.set(t);
+ }
+ }, "import-clear");
+ try {
+ writing.start();
+ assertTrue(insideAStatementNow.await(RENDEZVOUS_SECONDS, TimeUnit.SECONDS),
+ "the write of the trust flag never got inside a statement");
+ // where the clear gets to on its own: waiting on the monitor of the peer, which
+ // is the commit point being paid for - a clear that decided the peer had nothing
+ // to commit runs through to the end instead
+ clearing.start();
+ untilBlockedOrDone(clearing);
+ } finally {
+ letTheStatementFinish.countDown();
+ }
+ writing.join(TimeUnit.SECONDS.toMillis(RENDEZVOUS_SECONDS));
+ clearing.join(TimeUnit.SECONDS.toMillis(RENDEZVOUS_SECONDS));
+ reportFailureOf("the write of the trust flag", flagFailure);
+ reportFailureOf("the clear", clearFailure);
+ assertFalse(writing.isAlive(), "the write of the trust flag did not finish");
+ assertFalse(clearing.isAlive(), "the clear did not finish");
+
+ assertEquals(usedInOrder.size(), 2, "the two trees of this import did not get a connection each");
+ assertEquals(commits, Arrays.asList(usedInOrder.get(0), usedInOrder.get(1)),
+ "the clear passed over a peer whose write was still in flight");
+ });
+ } finally {
+ storage.close();
+ }
+ }
+
+ /**
+ * The connection of the last write is committed last. The connections of an import are
+ * transactions of their own, so {@code close()} cannot commit them as one - and the last thing
+ * an import writes is the flag that says the rest of it is good ({@code setTrust(true)} of
+ * {@code afterPhaseTwo}). Committed last, that flag is rolled back with the connection it is on
+ * whenever an earlier commit fails, so a failed import cannot leave an index marked trusted
+ * over data that never got there.
+ */
+ @Test(timeOut = 120000)
+ public void testTheConnectionsAreCommittedInTheOrderTheyWereLastWrittenTo() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "3");
+ final String url = StubDriver.PREFIX + "importer-last-write";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ // the two writes that stand for the trust flags afterPhaseTwo writes, one per base DN:
+ // both have to be committed after the connection that holds nothing but data
+ importer.put(ID2ENTRY, ByteString.valueOfUtf8("k2"), ByteString.valueOfUtf8("v"));
+ importer.put(DN2ID, ByteString.valueOfUtf8("k2"), ByteString.valueOfUtf8("v"));
+
+ importer.close();
+
+ assertEquals(usedInOrder.size(), 3, "the three trees of the import did not get a connection each");
+ assertEquals(commits, Arrays.asList(usedInOrder.get(2), usedInOrder.get(0), usedInOrder.get(1)),
+ "the connections were not committed in the order they were last written to");
+ } finally {
+ storage.close();
+ }
+ }
+
+ /**
+ * {@code close()} waits for a statement of a straggler to finish rather than committing the
+ * connection beside it.
+ * <p>
+ * Phase two gives its threads five seconds to answer an interrupt and closes the importer
+ * whether they answered or not ({@code OnDiskMergeImporter.invokeParallel}), so a write of an
+ * import can still be in flight when its connections are committed and handed back. A commit
+ * issued beside that statement is two threads on one connection - the whole of #891, with the
+ * closing thread on one side of it. Every other test of this suite joins its writers before it
+ * closes, so this is the only one where anything of a close happens beside a statement, and it
+ * is what the other half of the detector is for: the one that watches a commit, a rollback or a
+ * return land while a thread is inside a statement.
+ * <p>
+ * The straggler is let go once the close has got as far as it can on its own, and its write is
+ * then committed rather than lost: a thread that was inside the import when {@code close()}
+ * began is part of it. That is the assertion which bites - a close that reads the state of a
+ * connection without waiting for the statement in flight on it finds nothing written and
+ * commits nothing, and the write of the straggler is rolled back by the return instead.
+ */
+ @Test(timeOut = 120000)
+ public void testACloseWaitsForAStatementOfAStragglerRatherThanCommittingBesideIt() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
+ final String url = StubDriver.PREFIX + "importer-close-waits";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ importing(importer, () -> {
+ insideAStatementNow = new CountDownLatch(1);
+ letTheStatementFinish = new CountDownLatch(1);
+ final AtomicReference<Throwable> stragglerFailure = new AtomicReference<>();
+ final Thread straggling = daemon(() -> {
+ try {
+ importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ } catch (Throwable t) {
+ stragglerFailure.set(t);
+ }
+ }, "import-straggler");
+ final AtomicReference<Throwable> closeFailure = new AtomicReference<>();
+ final Thread closing = daemon(() -> {
+ try {
+ importer.close();
+ } catch (Throwable t) {
+ closeFailure.set(t);
+ }
+ }, "import-close-beside-a-statement");
+ try {
+ straggling.start();
+ assertTrue(insideAStatementNow.await(RENDEZVOUS_SECONDS, TimeUnit.SECONDS),
+ "the write of the straggler never got inside a statement");
+ // where the close gets to on its own: waiting on the monitor of the connection
+ // the straggler is writing through
+ closing.start();
+ untilBlockedOrDone(closing);
+ } finally {
+ letTheStatementFinish.countDown();
+ }
+ straggling.join(TimeUnit.SECONDS.toMillis(RENDEZVOUS_SECONDS));
+ closing.join(TimeUnit.SECONDS.toMillis(CLOSE_SECONDS));
+ reportFailureOf("the write of the straggler", stragglerFailure);
+ reportFailureOf("the close of the import", closeFailure);
+ assertFalse(straggling.isAlive(), "the write of the straggler did not finish");
+ assertFalse(closing.isAlive(), "the import did not close");
+
+ assertEquals(shared, Collections.emptyList(),
+ "the close touched a connection while a thread of the import was inside a statement of it");
+ assertEquals(usedInOrder.size(), 1, "the write of the straggler did not reach a connection");
+ verify(usedInOrder.get(0), times(1)).commit();
+ });
+ } finally {
+ storage.close();
+ }
+ assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "the import kept a permit of the pool");
+ }
+
+ /**
+ * A write that arrives after {@code close()} takes no connection. Phase two gives its threads
+ * five seconds to answer an interrupt and then closes the importer whether they did or not, so
+ * a write can reach it once its connections are back in the pool - and one that borrowed there
+ * would take a permit nothing is left to give back, out of a pool that is never removed from
+ * the map (#878).
+ */
+ @Test(timeOut = 120000)
+ public void testAWriteAfterCloseTakesNoConnection() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "4");
+ final String url = StubDriver.PREFIX + "importer-after-close";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.close();
+
+ try {
+ importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ fail("a write that arrives after the import is closed must be refused");
+ } catch (StorageRuntimeException expected) {
+ // the import is over: there is no transaction left for this write to belong to
+ }
+ try {
+ // the same for a clear, which commits the other connections of the import before it
+ // empties a tree: those are back in the pool, serving whoever borrowed them next
+ importer.clearTree(ID2ENTRY);
+ fail("a clear that arrives after the import is closed must be refused");
+ } catch (StorageRuntimeException expected) {
+ // as above
+ }
+
+ assertEquals(used.size(), 1, "a write after close() took a connection of the pool");
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 1, "a write after close() kept a connection");
+ } finally {
+ storage.close();
+ }
+ assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "a write after close() kept a permit");
+ }
+
+ /**
+ * A commit that fails still gives every connection of the import back, and takes the ones after
+ * it down with it. What a connection left behind holds is a permit of a pool that is never
+ * removed from the map, so a permit lost to a failed import is lost for the life of the server
+ * (#878) - and what a connection committed after the failure of an earlier one would hold is
+ * the trust flags of {@code afterPhaseTwo}, which {@code close()} commits last for exactly this
+ * reason: a failed import must not leave an index marked trusted over data that never got there.
+ */
+ @Test(timeOut = 120000)
+ public void testAFailedCommitStillReturnsEveryConnection() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "3");
+ final String url = StubDriver.PREFIX + "importer-failed-commit";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ doThrow(new SQLException("the socket went away")).when(usedInOrder.get(1)).commit();
+
+ try {
+ importer.close();
+ fail("the failure of the commit was not reported");
+ } catch (StorageRuntimeException expected) {
+ assertEquals(expected.getCause().getMessage(), "the socket went away");
+ }
+
+ // the connections are committed in the order they were last written to, so this one
+ // stands where the trust flags of afterPhaseTwo stand: behind the connection that just
+ // failed, and rolled back by the return rather than committed on top of a failed import
+ verify(usedInOrder.get(2), never()).commit();
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 3, "a failed import kept a connection of the pool");
+ } finally {
+ storage.close();
+ }
+ assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "a failed import kept a permit of the pool");
+ }
+
+ /**
+ * A tree that cannot be given a connection of its own is written through one the import already
+ * holds, rather than failing the import. The connections are taken as the trees are first
+ * touched, so a pool at its bound would otherwise stop an import halfway through - with the
+ * clears of {@code beforePhaseOne} already committed - where the one connection an import held
+ * before #891 would have carried it to the end.
+ */
+ @Test(timeOut = 120000)
+ public void testATreeSharesAConnectionWhenThePoolHasNoneToSpare() throws Exception {
+ System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "2");
+ System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "1"); // nothing is going to be returned
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
+ final String url = StubDriver.PREFIX + "importer-pool-full";
+ final JDBCStorage storage = importingStorage(url);
+ try {
+ final Importer importer = storage.startImport(); // one of the two connections of the pool
+ // and the other one to an operation of the server, so that the pool has none to spare
+ final Connection heldByAnOperation = CachedConnection.getConnection(url);
+ try {
+ importing(importer, () -> {
+ importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ // on a thread of its own: a thread that already holds a connection of the pool is
+ // exempt from waiting at its bound, and every thread of phase two starts holding
+ // none
+ putOnAThreadOfItsOwn(importer, DN2ID);
+
+ assertEquals(used.size(), 1, "the tree the pool had no connection for did not share one");
+ });
+ } finally {
+ heldByAnOperation.close();
+ }
+ } finally {
+ storage.close();
+ }
+ }
+
+ /**
+ * A second {@code close()} touches nothing: the connections of the import are back in the pool
+ * by then, and the pool hands them out again - so a close that walked its map a second time
+ * would roll back and re-pool a connection another borrower is holding.
+ */
+ @Test(timeOut = 120000)
+ public void testASecondCloseTouchesNothingTheImportGaveBack() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
+ final String url = StubDriver.PREFIX + "importer-second-close";
+ final JDBCStorage storage = importingStorage(url);
+ try {
+ final Importer importer = storage.startImport();
+ importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ importer.close();
+
+ // the pool hands one of them straight back out, the way an LDAP operation would
+ final Connection borrowedAgain = CachedConnection.getConnection(url);
+ try {
+ importer.close();
+ assertEquals(CachedConnection.poolOf(url).idleCount(), 1,
+ "a second close() returned a connection the pool had already handed to somebody else");
+ } finally {
+ borrowedAgain.close();
+ }
+ } finally {
+ storage.close();
+ }
+ assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "a connection of the import was lost");
+ }
+
+ /**
+ * A tree the database refuses a connection for is written through one the import already holds,
+ * the way a tree the pool has nothing to spare for is (#1013).
+ * <p>
+ * The two ends of a borrow that can run out are not reported alike: the bound of the pool is
+ * answered with a {@code SQLTimeoutException}, while a database at a limit of its own is
+ * answered with whatever its driver says - and a code {@code CachedConnection.isWorthRetrying}
+ * does not recognize is raised at once rather than waited out. A mysql account with a
+ * {@code MAX_USER_CONNECTIONS} of its own answers 1226 on SQLState 42000, which is such a code
+ * (#1011). An import that failed on it would stop halfway through, with the clears of
+ * {@code beforePhaseOne} already committed, while every connection it holds still works.
+ */
+ @Test(timeOut = 120000)
+ public void testATreeSharesAConnectionWhenTheDatabaseRefusesOne() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
+ final String url = StubDriver.PREFIX + "importer-refused";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ importing(importer, () -> {
+ // the first tree takes the connection the constructor borrowed, so that the second is
+ // one the import has to go to the database for
+ importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ // armed only now: what this is about is the borrows an import makes as it goes, not
+ // the one it is started with - an import of a database that takes no connection at
+ // all is refused where it is started
+ refuseFurtherConnects("User 'opendj' has exceeded the 'max_user_connections' resource"
+ + " (current value: 1)", "42000", 1226);
+ // on a thread of its own, the way every thread of phase two starts: one that already
+ // holds a connection of the pool takes another path through the borrow
+ putOnAThreadOfItsOwn(importer, DN2ID);
+
+ assertEquals(used.size(), 1, "the tree the database refused a connection for did not share one");
+ });
+ } finally {
+ storage.close();
+ }
+ assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "a refused borrow kept a permit of the pool");
+ }
+
+ /**
+ * ... and an import whose borrow is interrupted fails instead, keeping the interrupt.
+ * <p>
+ * The connection this import cannot have is the loss of a parallelism it asked for, and that is
+ * what sharing is the answer to. An interrupt is not that: phase two interrupts the threads of
+ * an import to stop it ({@code OnDiskMergeImporter.invokeParallel} gives them five seconds to
+ * answer), and a thread that answered by writing the tree through another connection would be
+ * carrying on with the work it was told to drop. The interrupt goes back on the thread as well:
+ * a borrow that throws {@code InterruptedException} has cleared it, and what reads it next is
+ * the executor of phase two.
+ */
+ @Test(timeOut = 120000)
+ public void testAnInterruptedBorrowFailsTheImportRatherThanSharingAConnection() throws Exception {
+ System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
+ final String url = StubDriver.PREFIX + "importer-interrupted";
+ final JDBCStorage storage = importingStorage(url);
+ final Importer importer = storage.startImport();
+ try {
+ importing(importer, () -> {
+ importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+
+ final AtomicReference<Throwable> failure = new AtomicReference<>();
+ final AtomicBoolean interruptKept = new AtomicBoolean();
+ final Thread writer = daemon(() -> {
+ // set before the write rather than raced with it: the pool waits for an idle
+ // connection interruptibly, so a thread that carries an interrupt into the borrow
+ // gets the same InterruptedException as one interrupted while parked there
+ Thread.currentThread().interrupt();
+ try {
+ importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ } catch (Throwable t) {
+ failure.set(t);
+ } finally {
+ interruptKept.set(Thread.currentThread().isInterrupted());
+ }
+ }, "import-interrupted");
+ writer.start();
+ writer.join(TimeUnit.SECONDS.toMillis(RENDEZVOUS_SECONDS));
+ assertFalse(writer.isAlive(), "the interrupted write did not finish");
+
+ final Throwable reported = failure.get();
+ assertTrue(reported instanceof StorageRuntimeException,
+ "an interrupted borrow was not reported as a failure of the import: " + reported);
+ assertTrue(reported.getCause() instanceof InterruptedException,
+ "an interrupted borrow was reported as something else: " + reported.getCause());
+ assertTrue(interruptKept.get(), "the borrow swallowed the interrupt of an import thread");
+ });
+ } finally {
+ storage.close();
+ }
+ }
+
+ /** A storage open for writing over the driver of this test, which is what an import needs. */
+ private JDBCStorage importingStorage(String url) throws Exception {
+ // mockCfg rather than a bare mock: it answers every getter with the value declared in
+ // JDBCBackendConfiguration.xml, so a storage that comes to read a setting this test never
+ // thought of gets the default instead of a null - the way the sibling suites of this package
+ // build their configurations
+ final JDBCBackendCfg cfg = mockCfg(JDBCBackendCfg.class);
+ when(cfg.getDBDirectory()).thenReturn(url);
+ final JDBCStorage storage = new JDBCStorage(cfg, null);
+ storage.open(AccessMode.READ_WRITE);
+ // the connection the open borrowed and gave back is not one of the import's: only what a
+ // statement was issued on is recorded, and the open issues none
+ forgetTheConnections();
+ return storage;
+ }
+
+ /**
+ * Makes the driver of this test answer every further connect with a failure of the given shape,
+ * the way a database at a limit of its own does.
+ */
+ private void refuseFurtherConnects(final String message, final String sqlState, final int vendorCode) {
+ refusal.set(() -> new SQLException(message, sqlState, vendorCode));
+ }
+
+ private Future<?> put(final Importer importer, final TreeName treeName) {
+ return threads.submit(() -> {
+ toTheStartingLine();
+ importer.put(treeName, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ });
+ }
+
+ /** A write of a thread that has written nothing before, the way every thread of an import starts. */
+ private void putOnAThreadOfItsOwn(final Importer importer, final TreeName treeName) throws Exception {
+ final AtomicReference<Throwable> failure = new AtomicReference<>();
+ final Thread thread = daemon(() -> {
+ try {
+ importer.put(treeName, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ }, "import-" + treeName.getIndexId());
+ thread.start();
+ thread.join(TimeUnit.SECONDS.toMillis(RENDEZVOUS_SECONDS));
+ if (failure.get() != null) {
+ throw new IllegalStateException("the write of " + treeName + " failed", failure.get());
+ }
+ // asserted rather than left to the join: a write still blocked - on the monitor of a
+ // connection, or on a borrow - would otherwise leave the assertions of the test standing on
+ // a run that never made the state they are about
+ assertFalse(thread.isAlive(), "the write of " + treeName + " did not finish");
+ }
+
+ /** Raises what a thread of this test caught, under a name that says which thread it was. */
+ private static void reportFailureOf(String what, AtomicReference<Throwable> failure) {
+ final Throwable caught = failure.get();
+ if (caught != null) {
+ throw new IllegalStateException(what + " failed", caught);
+ }
+ }
+
+ /**
+ * A thread of this suite. Daemon, every one of them: a write left standing on the monitor of a
+ * connection by a failing run must not keep the JVM of the build alive once the test that
+ * started it has been reported.
+ */
+ private static Thread daemon(Runnable body, String name) {
+ final Thread thread = new Thread(body, name);
+ thread.setDaemon(true);
+ return thread;
+ }
+
+ /** The threads a test writes through, so that a stuck one is not a build that never ends. */
+ private static ExecutorService daemonThreads(int count, final String name) {
+ final AtomicInteger numbered = new AtomicInteger();
+ return Executors.newFixedThreadPool(count,
+ runnable -> daemon(runnable, name + "-" + numbered.incrementAndGet()));
+ }
+
+ /** The body of a test that writes through an import - see {@link #importing(Importer, ImportBody)}. */
+ private interface ImportBody {
+ void run() throws Exception;
+ }
+
+ /**
+ * Runs the body of a test against an import and closes that import afterwards, whatever the
+ * body did - without the close having the last word on what went wrong.
+ * <p>
+ * The close used to stand in a {@code finally} of the test, and it is here instead because the
+ * two things that can go wrong there are not disjoint: {@code close()} commits and returns every
+ * connection under the monitor of that connection, so a write of this test left standing on one
+ * is exactly what makes the close wait - and a close that gave up waiting and threw out of that
+ * {@code finally} replaced the failure the test came for with one naming the monitor. Here the
+ * body has the first say and the trouble of the close rides along with it as suppressed.
+ */
+ private void importing(Importer importer, ImportBody body) throws Exception {
+ Throwable found = null;
+ try {
+ body.run();
+ } catch (Throwable t) {
+ found = t;
+ }
+ final Throwable reported = closeWithoutHanging(importer, found);
+ if (reported instanceof Error) {
+ throw (Error) reported;
+ }
+ if (reported instanceof Exception) {
+ throw (Exception) reported;
+ }
+ if (reported != null) {
+ throw new IllegalStateException("the import failed to close", reported);
+ }
+ }
+
+ /**
+ * Closes an import without waiting for a straggling thread of the test forever, and returns
+ * what is to be reported: what the body of the test found where it found something, with the
+ * trouble of the close - a close that never finished, or one that failed - suppressed into it.
+ */
+ private Throwable closeWithoutHanging(Importer importer, Throwable found) {
+ final AtomicReference<Throwable> failure = new AtomicReference<>();
+ final Thread closing = daemon(() -> {
+ try {
+ importer.close();
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ }, "import-close");
+ closing.start();
+ Throwable trouble;
+ try {
+ closing.join(TimeUnit.SECONDS.toMillis(CLOSE_SECONDS));
+ trouble = closing.isAlive()
+ ? new IllegalStateException("the import did not close within " + CLOSE_SECONDS
+ + "s: a thread of this test is holding the monitor of one of its connections")
+ : failure.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ trouble = new IllegalStateException("interrupted while closing the import", e);
+ }
+ if (found == null) {
+ return trouble;
+ }
+ if (trouble != null) {
+ found.addSuppressed(trouble);
+ }
+ return found;
+ }
+
+ private List<Connection> connectionsUsed() {
+ synchronized (used) {
+ return new ArrayList<>(used);
+ }
+ }
+
+ /**
+ * A connection of this test: it answers every statement with one mock of its own and records
+ * the thread that asked for it while that thread is inside the call.
+ */
+ private Connection newConnection() throws SQLException {
+ final Connection con = org.mockito.Mockito.mock(Connection.class);
+ final PreparedStatement statement = org.mockito.Mockito.mock(PreparedStatement.class);
+ when(con.isValid(anyInt())).thenReturn(true);
+ // the statement names the connection it runs on, as CachedConnection.prepareStatement() has
+ // it: the bound of a statement is arbitrated on the connection, which is read off the statement
+ when(statement.getConnection()).thenReturn(con);
+ // a thread is inside a statement of a connection for longer than the prepareStatement that
+ // hands it one: what the sql server driver walks past the end of its array is the list of
+ // listeners that the execution and the close of a statement mutate as readily (#891), so
+ // those are recorded here too - the window this suite watches is the whole of a statement
+ doAnswer(invocation -> {
+ enterAStatement(con);
+ try {
+ return 0; // what a mock of this returns anyway: no row of this test is counted
+ } finally {
+ leaveAStatement(con);
+ }
+ }).when(statement).executeUpdate();
+ doAnswer(invocation -> {
+ enterAStatement(con);
+ try {
+ return Boolean.FALSE; // as above: nothing of this test reads a result set
+ } finally {
+ leaveAStatement(con);
+ }
+ }).when(statement).execute();
+ doAnswer(invocation -> {
+ enterAStatement(con);
+ try {
+ return null;
+ } finally {
+ leaveAStatement(con);
+ }
+ }).when(statement).close();
+ // the commits of the import in the order they happen: which connection is committed when is
+ // what keeps a failed import from leaving an index marked trusted over data that is not there
+ doAnswer(invocation -> {
+ recordAThreadInsideAStatement(con, "commit");
+ commits.add(con);
+ return null;
+ }).when(con).commit();
+ // the return of a connection rolls it back and hands it to the next borrower, so neither may
+ // happen while a statement of an import thread is in flight on it
+ doAnswer(invocation -> {
+ recordAThreadInsideAStatement(con, "rollback");
+ return null;
+ }).when(con).rollback();
+ doAnswer(invocation -> {
+ recordAThreadInsideAStatement(con, "close");
+ return null;
+ }).when(con).close();
+ when(con.prepareStatement(anyString())).thenAnswer(invocation -> {
+ if (used.add(con)) {
+ usedInOrder.add(con);
+ }
+ enterAStatement(con);
+ try {
+ // the hold outside the monitor of the map above, and after this thread has entered
+ // it: what it is waiting for is a peer thread getting in here and recording itself
+ parkInsideTheStatement();
+ holdTheConnection();
+ meetPeers();
+ return statement;
+ } finally {
+ leaveAStatement(con);
+ }
+ });
+ return con;
+ }
+
+ /** Records this thread as inside a statement of the given connection, with whoever is already there. */
+ private void enterAStatement(Connection con) {
+ synchronized (inStatement) {
+ final Set<Thread> inside = inStatement.get(con);
+ final Set<Thread> threadsInside = inside != null ? inside : new LinkedHashSet<Thread>();
+ for (final Thread peer : threadsInside) {
+ if (peer != Thread.currentThread()) {
+ shared.add(peer.getName() + " and " + Thread.currentThread().getName());
+ }
+ }
+ threadsInside.add(Thread.currentThread());
+ inStatement.put(con, threadsInside);
+ }
+ }
+
+ private void leaveAStatement(Connection con) {
+ synchronized (inStatement) {
+ final Set<Thread> threadsInside = inStatement.get(con);
+ if (threadsInside != null && threadsInside.remove(Thread.currentThread()) && threadsInside.isEmpty()) {
+ inStatement.remove(con);
+ }
+ }
+ }
+
+ /** Records a connection touched from one thread while another is inside a statement on it. */
+ private void recordAThreadInsideAStatement(Connection con, String what) {
+ synchronized (inStatement) {
+ final Set<Thread> threadsInside = inStatement.get(con);
+ if (threadsInside == null) {
+ return;
+ }
+ for (final Thread inside : threadsInside) {
+ if (inside != Thread.currentThread()) {
+ shared.add(what + " of " + Thread.currentThread().getName() + " while " + inside.getName()
+ + " was inside a statement");
+ }
+ }
+ }
+ }
+
+ /**
+ * Parks a thread inside the first statement it issues and tells the test it is in there, until
+ * the test lets it go.
+ * <p>
+ * A write in flight, as the rest of the import sees one: the connection holds a statement that
+ * has not come back, so nothing is recorded against it yet - {@code put()} counts a write once
+ * the statement returns - and its monitor is held for as long as this lasts. What a test does
+ * from there is ask another thread of the import a question about that connection.
+ */
+ private void parkInsideTheStatement() {
+ final CountDownLatch letGo = letTheStatementFinish;
+ if (letGo == null || parked.get()) {
+ return;
+ }
+ parked.set(Boolean.TRUE);
+ insideAStatementNow.countDown();
+ try {
+ letGo.await(PARK_SECONDS, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Waits until a thread of this test has got as far as it is going to get on its own: either it
+ * is blocked on a monitor - the thing this suite is about - or it is done.
+ * <p>
+ * Waited for rather than slept past: what the tests below ask is whether a thread stops at a
+ * monitor or walks through it, and a fixed pause would answer that with the load of the machine
+ * on a slow run.
+ */
+ private static void untilBlockedOrDone(Thread thread) throws InterruptedException {
+ final long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(RENDEZVOUS_SECONDS);
+ while (System.currentTimeMillis() < deadline) {
+ final Thread.State state = thread.getState();
+ if (state == Thread.State.BLOCKED || state == Thread.State.TERMINATED) {
+ return;
+ }
+ Thread.sleep(10);
+ }
+ }
+
+ /**
+ * Holds the first statement of a thread on the connection it is on, until a peer thread has one
+ * of its own in flight there or the hold runs out.
+ * <p>
+ * How a test asks whether two threads can be inside a statement of one connection at the same
+ * time: they cannot be brought together at a barrier the way the threads of two connections are
+ * - a second thread held on the monitor of the connection never reaches one - so the first
+ * thread in waits there instead, and the peer either turns up beside it, which is #891, or does
+ * not, which is the hold being paid in full.
+ */
+ private void holdTheConnection() {
+ final CountDownLatch inside = insideAStatement;
+ if (inside == null || held.get()) {
+ return;
+ }
+ held.set(Boolean.TRUE);
+ inside.countDown();
+ try {
+ inside.await(HOLD_SECONDS, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /** Where the threads of a test start together, in front of the monitors of the importer. */
+ private void toTheStartingLine() {
+ final CyclicBarrier line = startingLine;
+ if (line == null) {
+ return;
+ }
+ try {
+ line.await(RENDEZVOUS_SECONDS, TimeUnit.SECONDS);
+ } catch (InterruptedException | TimeoutException | BrokenBarrierException e) {
+ throw new IllegalStateException("the threads of this test did not reach the starting line", e);
+ }
+ }
+
+ /**
+ * Holds the first statement of a thread until every peer of the rendezvous has one in flight
+ * too. Reported as a failure of the statement rather than waited out: a rendezvous nobody else
+ * reaches is an import that serialized its threads, which is what the parallel test is about.
+ */
+ private void meetPeers() throws SQLException {
+ final CyclicBarrier barrier = rendezvous;
+ if (barrier == null || met.get()) {
+ return;
+ }
+ met.set(Boolean.TRUE);
+ try {
+ barrier.await(RENDEZVOUS_SECONDS, TimeUnit.SECONDS);
+ } catch (TimeoutException | BrokenBarrierException e) {
+ throw new SQLException("no peer thread had a statement of its own in flight within "
+ + RENDEZVOUS_SECONDS + "s: the import did not let its threads write at the same time", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new SQLException("interrupted while waiting for the peer threads of the import", e);
+ }
+ }
+
+ /** A driver of this test, so that the connections of an import need no database behind them. */
+ private final class StubDriver implements Driver {
+ static final String PREFIX = "jdbc:opendj-import-stub:";
+
+ @Override
+ public Connection connect(String url, Properties info) throws SQLException {
+ if (!acceptsURL(url)) {
+ return null;
+ }
+ final Supplier<SQLException> refused = refusal.get();
+ if (refused != null) {
+ throw refused.get();
+ }
+ return newConnection();
+ }
+
+ @Override
+ public boolean acceptsURL(String url) {
+ return url != null && url.startsWith(PREFIX);
+ }
+
+ @Override
+ public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) {
+ return new DriverPropertyInfo[0];
+ }
+
+ @Override
+ public int getMajorVersion() {
+ return 1;
+ }
+
+ @Override
+ public int getMinorVersion() {
+ return 0;
+ }
+
+ @Override
+ public boolean jdbcCompliant() {
+ return false;
+ }
+
+ @Override
+ public Logger getParentLogger() {
+ return Logger.getLogger(StubDriver.class.getName());
+ }
+ }
+}
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 0430507..8fe246e 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
@@ -1081,6 +1081,14 @@
}
@Override
+ int poolMax() {
+ // stood in for like the borrow above: read for real, this would intern a pool of the
+ // static registry for the url of the mock configuration, and start its sweeper, from
+ // a test that has no pool at all (#891)
+ return 1;
+ }
+
+ @Override
public StorageStatus getStorageStatus() {
return StorageStatus.working(); // open already, so the importer borrows and no more
}
--
Gitblit v1.10.0