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/main/java/org/opends/server/backends/jdbc/JDBCStorage.java | 856 ++++++++++++++++++++++++++++++++++++++++++++++++++++----
1 files changed, 792 insertions(+), 64 deletions(-)
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();
}
--
Gitblit v1.10.0