[#986] Give back the changes a replay thread the pool stopped had parked

opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java

@@ -33,6 +33,7 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
 
 import org.forgerock.i18n.LocalizableMessage;
 import org.forgerock.i18n.slf4j.LocalizedLogger;
@@ -796,6 +797,77 @@ private void setState(State newState)
     }
   }
 
+  /**
+   * Runs an action on every replication domain of this server, each of them getting its
+   * turn whatever one of them threw.
+   * <p>
+   * The replay threads are shared by every domain of this server, so what concerns the
+   * pool is done over the domains rather than over the one a replay was last for: a thread
+   * which is stopping gives back what it parked in any of them (issue #986). A throw at one
+   * domain must not leave the ones after it as they were - with changes owned by a thread
+   * which does not exist anymore - so the first failure is thrown once the loop is over,
+   * the others suppressed under it where it records suppression: the error a JVM out of
+   * memory prepared beforehand does not - it was made without its constructor, so it keeps
+   * no list to record them in - and the JVM hands that one out as often as it is asked for
+   * one, so two domains can throw the same instance, and a throwable can not suppress
+   * itself. Recording a failure under the first allocates the list it goes in, so on the
+   * road this loop is for it can be refused in its turn: a failure which can not be
+   * recorded is dropped, and the domains after it still get their turn.
+   * <p>
+   * The iterator over the domains is the one allocation made before the first of them gets
+   * its turn: refused, on the way out of a thread an OutOfMemoryError is ending, it leaves
+   * every domain as it was, and it has no cheaper form. The action itself is not one: a
+   * caller on that road passes an instance it holds rather than one it makes there.
+   * <p>
+   * Not synchronized, and it must not become so: it is called by a replay thread on its way
+   * out, while {@link #stopReplayThreads()} holds the monitor of this class and waits for
+   * that thread to end.
+   *
+   * @param action what is done on each domain; it declares no checked exception, so what
+   *          it throws is an Error or a RuntimeException, and that is what is thrown here
+   */
+  static void forEachDomain(Consumer<LDAPReplicationDomain> action)
+  {
+    Throwable failure = null;
+    for (LDAPReplicationDomain domain : domains.values())
+    {
+      try
+      {
+        action.accept(domain);
+      }
+      catch (Throwable domainFailure)
+      {
+        if (failure == null)
+        {
+          failure = domainFailure;
+        }
+        else if (failure != domainFailure)
+        {
+          try
+          {
+            failure.addSuppressed(domainFailure);
+          }
+          catch (OutOfMemoryError recordRefused)
+          {
+            /*
+             * The list the record goes in could not be allocated: the failure is dropped
+             * rather than allowed to end the loop, since the first one is what is reported
+             * and the domains after this one are still to be visited.
+             */
+          }
+        }
+      }
+    }
+    if (failure instanceof Error)
+    {
+      throw (Error) failure;
+    }
+    if (failure instanceof RuntimeException)
+    {
+      throw (RuntimeException) failure;
+    }
+  }
+
   /**
    * Gets the number of handled domain objects.
    * @return The number of handled domain objects

opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java

@@ -17,7 +17,11 @@
  */
 package org.opends.server.replication.plugin;
 
+import static java.util.Collections.*;
+
+import java.util.ArrayList;
 import java.util.Iterator;
+import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.SortedMap;
 import java.util.SortedSet;
@@ -90,8 +94,10 @@ final class RemotePendingChanges
    * this issue is about (issue #922).
    * <p>
    * A thread is entered here when it takes a change over and removed when it gives it back,
-   * applies it, or parks it as waiting for another change - the parked ones are handed to
-   * whichever thread clears what they wait for, so they are not this one's to give back.
+   * applies it, or parks it as waiting for another change - a parked change is not the one
+   * this thread is replaying, and giving it back is
+   * {@link #releaseParkedChangesOwnedByCurrentThread()}, which reads the changes which are
+   * waiting rather than this index (issue #954).
    * <p>
    * The entry of a thread is written by that thread and by nobody else, and that - not the
    * lock - is what keeps the writes apart: the park in {@link #addDependency(PendingChange)}
@@ -545,9 +551,10 @@ public boolean markInProgress(LDAPUpdateMsg msg)
    * Returns the CSN of the change the calling thread is replaying, when it still owns one.
    * <p>
    * A thread owns the change it is replaying and the ones it parked as waiting for another
-   * change. The parked ones are left out: they are handed to whichever thread clears the
-   * change they are waiting for, and that thread takes them over, so giving one back here
-   * would have the same change handed to two threads (issue #922).
+   * change. The parked ones are left out: they are not the change this thread is replaying,
+   * and giving one back is more than dropping its owner - it has to be unparked in the same
+   * step, or it would be handed out by two roads at once, which is what
+   * {@link #releaseParkedChangesOwnedByCurrentThread()} does (issues #922 and #954).
    * <p>
    * It is a plain read of {@link #changeBeingReplayed}: no lock is taken and nothing is
    * allocated. This is what the give-back on the way out of an unwound replay asks first,
@@ -569,6 +576,95 @@ CSN getChangeOwnedByCurrentThread()
     return changeBeingReplayed.get(Thread.currentThread());
   }
 
+  /**
+   * Gives back the changes the calling thread parked as waiting for another change, and
+   * takes them out of the changes which are waiting in the same step.
+   * <p>
+   * A parked change stays owned by the thread which parked it while that thread goes on
+   * to the changes which follow: {@link #getNextUpdate()} is what hands it out again, to
+   * whichever replay thread clears the change it was waiting for, and that thread takes it
+   * over. A replay which is unwound leaves the thread which parked it without that road -
+   * it takes the next delivery off the replay queue instead - so the change would be left
+   * owned by a thread which is never coming back to it, and every redelivery of a change a
+   * replay thread owns is refused as a duplicate (issue #954).
+   * <p>
+   * Unparking a change and giving it back is one step, under both locks, so that only one
+   * road can hand it out: a change which was released while it is still listed as waiting
+   * would be handed to the thread {@link #getNextUpdate()} gives it to and to the thread
+   * which takes over the delivery which follows - the double replay the ownership is there
+   * to prevent (OPENDJ-1115).
+   * <p>
+   * The changes stay listed and uncommitted, and stay among the changes the newer ones are
+   * checked against, the way a change whose replay failed does: they are not in the data,
+   * so they hold this domain's ServerState back and the changes which follow them keep
+   * waiting for them.
+   * <p>
+   * The changes another thread parked are left alone: a change is given back by the thread
+   * which owns it and by nobody else (issue #922). That thread may be inside the dependency
+   * checks which parked it - they park a change once per dependency it has - so a change
+   * released under it would be listed as waiting again a moment later, and handed out while
+   * the delivery which took it over is being replayed.
+   *
+   * @return the CSNs of the changes it gave back, oldest first; empty when this thread owns
+   *         no parked change - the changes a thread parked stay its own, whichever replay
+   *         parked them, until {@link #getNextUpdate()} hands them to the thread which
+   *         cleared what they wait for or they are given back here
+   */
+  List<CSN> releaseParkedChangesOwnedByCurrentThread()
+  {
+    final Thread current = Thread.currentThread();
+    /*
+     * The second lock is taken inside the try of the first: taking a lock which is held
+     * by another thread allocates the node this one waits on, and this runs on the road
+     * out of a JVM which has just refused an allocation. A throw out of the second lock
+     * would otherwise unwind past the first with that one held by a thread which is
+     * ending, and every road which lists, commits or gives back a change would wait for
+     * it for good.
+     */
+    pendingChangesWriteLock.lock();
+    try
+    {
+      dependentChangesLock.lock();
+      try
+      {
+        if (dependentChanges.isEmpty())
+        {
+          // Nothing is waiting, which is the state every replay but a handful leaves behind.
+          return emptyList();
+        }
+        /*
+         * Sized for every change which is waiting, so that the one allocation of this
+         * method past the locks is made before anything is taken out of the set. The rule
+         * getNextUpdate() states for itself holds here: an allocation which fails once a
+         * change has been unparked and released - and this runs on the road out of a JVM
+         * which has just refused one - would have moved that change out of the hands which
+         * hand it out again, with the caller never told that it did.
+         */
+        final List<CSN> released = new ArrayList<>(dependentChanges.size());
+        final Iterator<PendingChange> it = dependentChanges.iterator();
+        while (it.hasNext())
+        {
+          final PendingChange change = it.next();
+          if (change.isOwnedBy(current))
+          {
+            it.remove();
+            change.setOwner(null);
+            released.add(change.getCSN());
+          }
+        }
+        return released;
+      }
+      finally
+      {
+        dependentChangesLock.unlock();
+      }
+    }
+    finally
+    {
+      pendingChangesWriteLock.unlock();
+    }
+  }
+
   /**
    * Get the first update in the list that have some dependencies cleared.
    * <p>
@@ -672,8 +768,10 @@ private void addDependency(PendingChange dependentChange)
        * parked one is handed to the thread which clears what it waits for, and one which is
        * not listed here anymore is gone with the pending changes of a domain which was
        * disabled. The owner stays as it is - it is what has getNextUpdate() hand the change
-       * over rather than leave it to nobody - and the give-back on the way out of an
-       * unwound replay leaves it alone (issue #922).
+       * over rather than leave it to nobody - and the give-back of the change a replay was
+       * unwound on leaves it alone (issue #922). What hands a parked change back is
+       * releaseParkedChangesOwnedByCurrentThread(), which unparks it in the same step so
+       * that the two roads can not hand it out at once (issue #954).
        */
       changeBeingReplayed.remove(Thread.currentThread(), dependentChange.getCSN());
     }

opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java

@@ -24,6 +24,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
 
 import org.opends.server.api.DirectoryThread;
 import org.forgerock.i18n.slf4j.LocalizedLogger;
@@ -39,6 +40,15 @@
 public class ReplayThread extends DirectoryThread
 {
   private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
+  /**
+   * The give-back a thread runs on every domain of this server on its way out, held here
+   * rather than written where it is run: a method reference is linked, and its instance
+   * made, where it is first run, and this one is first run on the way out of a thread -
+   * which an OutOfMemoryError may be ending, on the road this give-back is there for. Made
+   * when this class is loaded instead, on a thread which can allocate (issue #986).
+   */
+  private static final Consumer<LDAPReplicationDomain> GIVE_BACK_PARKED_CHANGES =
+      LDAPReplicationDomain::giveBackChangesParkedByStoppingThread;
 
   private final BlockingQueue<UpdateToReplay> updateToReplayQueue;
   private final ReentrantLock switchQueueLock;
@@ -77,6 +87,53 @@ public void run()
       logger.trace("Replication Replay thread starting.");
     }
 
+    try
+    {
+      replayUntilStopped();
+    }
+    finally
+    {
+      /*
+       * The changes this thread parked as waiting for another change are handed out again
+       * by getNextUpdate() alone, which every replay loop of a domain runs once it is done
+       * with a change: a parked change is replayed by whichever thread clears the change it
+       * was waiting for. A thread which is stopping is not on that road anymore, so what it
+       * parked would be left owned by a thread which does not exist, while every redelivery
+       * of a change a replay thread owns is refused as a duplicate: on a domain which then
+       * goes quiet that change is where the ServerState of this replica, and every change
+       * behind it from every master, stops (issue #986).
+       *
+       * Given back by the thread which owns them, so that the rule every road which reads
+       * ownership follows holds on this one as well: a change is given back by the thread it
+       * was handed to and by nobody else (issue #922). It is also the one place which sees
+       * them all - the pool is shared by every domain of this server, while a replay knows
+       * only the domain it was replaying for.
+       *
+       * The session which brings them back is asked for and left standing, in every domain
+       * which got something back, and the state checkpointer of each of them runs it within
+       * its tick: a thread on its way out is not held for a session - the threads of the
+       * pool are stopped one after the other and joined, and each running a restart of its
+       * own would have the configuration change which is stopping them wait for one restart
+       * per thread - and a change delivered again before the pool which replaces this one
+       * is up waits in the replay queue for it. A thread which an OutOfMemoryError is ending
+       * gives back here what it parked in the domains it was not replaying for, on the same
+       * terms; the change it was replaying, and what it had parked in that same domain, were
+       * given back and asked for again on its way out of replay().
+       */
+      giveBackParkedChanges();
+    }
+    if (logger.isTraceEnabled())
+    {
+      logger.trace("Replication Replay thread stopping.");
+    }
+  }
+
+  /**
+   * Takes the deliveries of the domains of this server off the shared replay queue and
+   * replays them, until this thread is stopped.
+   */
+  private void replayUntilStopped()
+  {
     while (!shutdown.get())
     {
       try
@@ -145,9 +202,31 @@ public void run()
         logger.error(ERR_EXCEPTION_REPLAYING_REPLICATION_MESSAGE, stackTraceToSingleLineString(t));
       }
     }
-    if (logger.isTraceEnabled())
-    {
-      logger.trace("Replication Replay thread stopping.");
-    }
+  }
+
+  /**
+   * Gives back the changes this thread parked as waiting for another change, in every
+   * domain of this server.
+   * <p>
+   * A change which is given back stays listed and uncommitted, the way a change whose replay
+   * failed does: it is not in the data, so it holds the ServerState of its domain back and
+   * the changes which follow it keep waiting for it, until the delivery which takes it over
+   * replays it.
+   * <p>
+   * Every domain gets its turn whatever one of them threw: what can throw here is an
+   * allocation, on the way out of a thread an OutOfMemoryError may be ending - the iterator
+   * over the domains, before any of them is reached, then for each of them the list of what
+   * it released, made before anything is released, or the report of a change once it is,
+   * and between two domains the list a second failure is recorded in under the first, which
+   * the loop guards on its own - and the domains which follow would otherwise be left with
+   * changes owned by a thread which does not exist anymore, the state this give-back is
+   * for. A domain which threw past the release has asked for its restart already: the
+   * request is made before the report. The first failure is thrown once the loop is over,
+   * so that the uncaught exception handler of {@link DirectoryThread} writes the line and
+   * raises the alert.
+   */
+  private void giveBackParkedChanges()
+  {
+    MultimasterReplication.forEachDomain(GIVE_BACK_PARKED_CHANGES);
   }
 }

opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java

@@ -58,6 +58,21 @@ enum SessionRestart
   private final AtomicReference<SessionRestart> requested =
       new AtomicReference<>(SessionRestart.NONE);
 
+  SessionRestartRequests()
+  {
+    /*
+     * Every request is made on a replay-failure road, where an allocation may be what has
+     * just failed, and the first execution of merge() in a JVM allocates: the call site
+     * of its lambda and the VarHandle site inside accumulateAndGet() are linked when they
+     * are first run, and nothing runs them before a replay fails. Run once here, on a
+     * thread which can allocate, so that a request made on the road out of an
+     * OutOfMemoryError asks for nothing the JVM has just refused (issue #954). The same
+     * goes for what takes a request, which is run on the same roads.
+     */
+    merge(SessionRestart.NONE);
+    take();
+  }
+
   /**
    * Asks this domain to restart its session.
    *

opendj-server-legacy/src/messages/org/opends/messages/replication.properties

@@ -670,6 +670,9 @@ ERR_REPLAY_GIVE_BACK_FAILED_317=Could not give change %s of domain "%s" back to
  server after the replay which owned it was unwound: %s. The change has been released without its \
  failure being counted, and a restart of the session is asked for so that the change is delivered \
  again
+NOTE_REPLAY_PARKED_CHANGE_GIVEN_BACK_318=Change %s in domain "%s" was waiting for another change \
+ to be replayed when the replay thread which parked it went away. The change has not been recorded \
+ as replayed and is given back to the replication server, which still owns it and sends it again
 WARN_REPLAY_NOT_DRAINED_319=Domain "%s" is going down and gave up on waiting up to %d ms for \
  the replay of one of its changes to finish. A change which reaches the backend from now on \
  is not recorded in the ServerState being saved, so the replication server sends it again \