[#1036] Assert that a parked change is handed out, not which thread hands it out

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

@@ -2726,8 +2726,8 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown)
        * ask for it, and restart its session for it, for as long as the server is up.
        *
        * The changes this thread parked as waiting for another change are left alone: they
-       * are handed to whichever thread clears the change they are waiting for, and that
-       * thread takes them over.
+       * are handed out by getNextUpdate() to whichever thread calls it first once the
+       * changes before them are gone, and that thread takes them over.
        *
        * Which change this thread owns is read before anything is done with it, and that
        * read takes no lock and allocates nothing: everything below is gated on the answer,

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

@@ -90,8 +90,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 one is handed out by
+   * {@link #getNextUpdate()} to whichever thread calls it first once the changes before it
+   * are gone, so it is not this one's to give back unless it is handed it again, which
+   * enters it here again.
    * <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 +547,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 handed out by {@link #getNextUpdate()}
+   * to whichever thread calls it first once the changes before them are gone, and that
+   * thread takes them over, so giving one back here would have the same change handed to
+   * two threads (issue #922).
    * <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,
@@ -573,9 +576,11 @@ CSN getChangeOwnedByCurrentThread()
    * Get the first update in the list that have some dependencies cleared.
    * <p>
    * The change is handed to the calling thread, which owns it from then on: it is
-   * replayed by whichever replay thread cleared the change it was waiting for rather than
-   * by the one which parked it, and a change is given back by the thread which owns it
-   * and by nobody else (issue #922).
+   * replayed by whichever replay thread calls this first once the changes before it have
+   * left - as a rule the one which cleared the change it was waiting for, though the one
+   * which parked it comes through here on its own way out and takes it back itself when
+   * the clearing happened in between - and a change is given back by the thread which
+   * owns it and by nobody else (issue #922).
    *
    * @return The LDAPUpdateMsg to be handled.
    */
@@ -669,11 +674,13 @@ private void addDependency(PendingChange dependentChange)
       }
       /*
        * Whichever of the two it was, this thread is not replaying that change anymore: a
-       * 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).
+       * parked one is handed to whichever thread calls getNextUpdate() first once the
+       * changes before it are gone - the clearing thread as a rule, this one when the
+       * clearing lands before it gets there - 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).
        */
       changeBeingReplayed.remove(Thread.currentThread(), dependentChange.getCSN());
     }

opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java

@@ -3175,10 +3175,15 @@ public void call() throws Exception
    * handed out: the changes parked behind it stayed parked, and the ServerState of this
    * domain stayed behind them until some other change was replayed here.
    * <p>
-   * What tells the two apart is which thread replays the parked change. It is handed to
-   * whichever thread cleared the change it was waiting for, so it is replayed by the very
-   * thread which has just committed the change whose ack threw - a change nobody handed out
-   * is replayed by no one at all, and the wait below is what says so.
+   * What tells the two apart is that the parked change is replayed at all. The child is
+   * seen parked before the parent is let go, so the replay queue is not where it can come
+   * from anymore: only {@code getNextUpdate()} hands it out, and with the parent held until
+   * then, the thread which committed the parent is the one left to call it - a change
+   * nobody handed out is replayed by no one at all, and the wait below is what says so.
+   * Which thread replays it is deliberately not asserted: {@code getNextUpdate()} hands a
+   * parked change to whichever thread calls it first once the changes before it are gone,
+   * and the thread which parked it calls it on its own way out, so a parker which is slow
+   * to get there takes the child back itself when the parent commits in between.
    */
   @Test
   public void theChangesParkedBehindAChangeWhoseAckFailedAreReplayed() throws Exception
@@ -3220,25 +3225,30 @@ public void theChangesParkedBehindAChangeWhoseAckFailedAreReplayed() throws Exce
         OperationType.ADD, "PreParse",
         op -> parentCsn.equals(OperationContext.getCSN(op))
             || childCsn.equals(OperationContext.getCSN(op)));
+    final long initialDependent = getMonitorAttrValue(baseDN, "dependent-changes-size");
     try
     {
       domain.processUpdate(new AddMsgWhoseAckThrows(parentCsn, parent.getName(), parentUUID,
           baseUUID, parent.getObjectClassAttribute(), parent.getAllAttributes()));
-      final Thread replayingParent = parked.awaitParked(60, SECONDS);
+      parked.awaitParked(60, SECONDS);
 
       domain.processUpdate(new AddMsg(childCsn, child.getName(), childUUID, parentUUID,
           child.getObjectClassAttribute(), child.getAllAttributes(), null));
+      /*
+       * Seen parked before the parent is let go. Released on the spot, the parent could be
+       * applied and committed before a replay thread has taken the child off the queue at
+       * all, and the child would then be replayed from the queue with nothing to wait for -
+       * a pass which says nothing about the hand-out this test is about.
+       */
+      assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", initialDependent + 1,
+          "the child must be parked behind the parent while the parent is held");
 
       /*
        * The parent is applied and its ack throws where it is published. The replay carries
-       * on all the same, and the child is the change it hands itself next.
+       * on all the same, and the child is the change it hands out next.
        */
       parked.release();
-      final Thread replayingChild = parked.awaitParked(60, SECONDS);
-      Assertions.assertThat(replayingChild)
-          .as("the change which was parked must be replayed by the thread which cleared what"
-              + " it was waiting for, rather than be left waiting")
-          .isSameAs(replayingParent);
+      parked.awaitParked(60, SECONDS);
       parked.release();
 
       assertNotNull(getEntry(child.getName(), 30000, true),
@@ -3255,19 +3265,23 @@ public void theChangesParkedBehindAChangeWhoseAckFailedAreReplayed() throws Exce
    * replay it was handed to is unwound.
    * <p>
    * A change which was parked behind another one is handed out by {@code getNextUpdate()}
-   * to the thread which cleared what it was waiting for, and that thread owns it from then
-   * on. The give-back on the way out of an unwound replay asks which change this thread
-   * owns, so the hand-out has to be recorded where that question is answered, not only on
-   * the change: left out, the change would stay owned by a thread which is not replaying
-   * it anymore, and every later delivery of it would be refused as a duplicate - the wedge
-   * of this issue, on the dependency road.
+   * to whichever thread calls it first once the changes before it are gone, and that
+   * thread owns it from then on. The give-back on the way out of an unwound replay asks
+   * which change this thread owns, so the hand-out has to be recorded where that question
+   * is answered, not only on the change: left out, the change would stay owned by a thread
+   * which is not replaying it anymore, and every later delivery of it would be refused as
+   * a duplicate - the wedge of this issue, on the dependency road.
    * <p>
-   * The parent is held at the pre-parse plugin point while the child is delivered, so the
-   * child is parked behind a change in flight, and the thread which is thrown out of the
-   * child is read at the same plugin point: it must be the one which committed the parent,
-   * which is what says the child was handed out rather than taken off the queue. The child
-   * is thrown out of once, inside the replay, and then unwound past its ack, on the road
-   * every catch of the replay has already run on.
+   * The parent is held at the pre-parse plugin point while the child is delivered, and it
+   * is let go only once the child is seen parked behind it: that is what says the child was
+   * handed out rather than taken off the queue, since a parked change leaves by
+   * {@code getNextUpdate()} and by no other road. The thread which is thrown out of the
+   * child is read at the same plugin point, for the assertion that the Error did not end
+   * it; which thread it is says nothing about the hand-out and is not asserted - the parker
+   * calls {@code getNextUpdate()} on its own way out, so it takes the child back itself
+   * when the parent commits before it gets there. The child is thrown out of once, inside
+   * the replay, and then unwound past its ack, on the road every catch of the replay has
+   * already run on.
    */
   @Test
   public void aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound() throws Exception
@@ -3300,6 +3314,7 @@ public void aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound() thr
     final CSN parentCsn = gen.newCSN();
     final CSN childCsn = gen.newCSN();
     final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+    final long initialDependent = getMonitorAttrValue(baseDN, "dependent-changes-size");
     final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations(
         OperationType.ADD, "PreParse", op -> parentCsn.equals(OperationContext.getCSN(op)));
     /*
@@ -3316,7 +3331,13 @@ public void aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound() thr
           {
             return false;
           }
-          replayingChild.set(Thread.currentThread());
+          /*
+           * The first replay of the child is the one which meets the Error, and it is the
+           * one kept: this is evaluated ahead of the budget of the throw, so it runs on the
+           * by-hand redelivery below too, which must not overwrite the thread the
+           * assertion of issue #923 is about.
+           */
+          replayingChild.compareAndSet(null, Thread.currentThread());
           return true;
         },
         () -> new LinkageError("the replay of the change which was handed out meets an Error"),
@@ -3325,12 +3346,19 @@ public void aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound() thr
     {
       domain.processUpdate(new AddMsg(parentCsn, parent.getName(), parentUUID, baseUUID,
           parent.getObjectClassAttribute(), parent.getAllAttributes(), null));
-      final Thread replayingParent = parked.awaitParked(60, SECONDS);
+      parked.awaitParked(60, SECONDS);
 
       domain.processUpdate(new AddMsgWhoseReplayIsUnwoundAfterItsAck(childCsn, child.getName(),
           childUUID, parentUUID, child.getObjectClassAttribute(), child.getAllAttributes()));
+      /*
+       * Seen parked before the parent is let go: a child taken off the queue once the
+       * parent has committed is replayed with nothing to wait for, and the give-back it
+       * would then exercise is the one of the ordinary road rather than of the hand-out.
+       */
+      assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", initialDependent + 1,
+          "the child must be parked behind the parent while the parent is held");
 
-      // The parent is applied, and the child is the change its thread hands itself next.
+      // The parent is applied, and the child is the change handed out next.
       parked.release();
 
       TestTimer timer = new TestTimer.Builder()
@@ -3346,10 +3374,6 @@ public void call() throws Exception
               "the change which was handed out must have been thrown out of");
         }
       });
-      Assertions.assertThat(replayingChild.get())
-          .as("the change which was parked must be replayed by the thread which cleared what"
-              + " it was waiting for: that is the hand-out this test is about")
-          .isSameAs(replayingParent);
 
       /*
        * The child was thrown out of and its replay was then unwound, so it is not in the
@@ -3382,7 +3406,7 @@ public void call() throws Exception
               + " from the one which was unwound");
       assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-failed"), initialFailures,
           "a change which was delivered again must not be counted as one this replica gave up on");
-      assertTrue(replayingParent.isAlive(),
+      assertTrue(replayingChild.get().isAlive(),
           "an Error which unwinds a replay must not end the thread which met it (issue #923)");
     }
     finally

opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java

@@ -528,12 +528,13 @@ public void run()
   }
 
   /**
-   * A change which was parked because it depends on another one is handed to whichever
-   * replay thread clears the change it was waiting for, rather than replayed by the
-   * thread which parked it. The thread it is handed to is the one which owns it from
-   * then on: the failure of the replay it is about to be given is reported by that
-   * thread, and a give-back which comes from a thread the change was never handed to is
-   * ignored (issue #922).
+   * A change which was parked because it depends on another one is handed out by
+   * {@code getNextUpdate()} to whichever replay thread calls it first once the change it
+   * was waiting for is gone, rather than replayed by the thread which parked it as a
+   * matter of course. The thread it is handed to is the one which owns it from then on:
+   * the failure of the replay it is about to be given is reported by that thread, and a
+   * give-back which comes from a thread the change was never handed to is ignored (issue
+   * #922).
    * <p>
    * The hand-out is what the give-back on the way out of an unwound replay must read as
    * well as the owner: a change handed out by {@code getNextUpdate()} whose replay is then
@@ -561,7 +562,8 @@ public void aChangeTakenAsADependencyIsOwnedByTheThreadWhichTakesIt() throws Exc
     assertTrue(pendingChanges.checkDependencies(rename),
         "the rename must wait for the delete of the entry it renames into");
 
-    // The delete has been replayed, so the rename is handed to the thread which replayed it.
+    // The delete has been replayed, so the rename is handed out to whichever thread calls
+    // getNextUpdate() next - here, one which replayed nothing.
     pendingChanges.commit(deleted);
 
     final AtomicReference<LDAPUpdateMsg> taken = new AtomicReference<>();
@@ -670,10 +672,10 @@ public void run()
 
   /**
    * The change a thread parked as waiting for another one is not the change it is
-   * replaying: it is handed to whichever thread clears what it waits for, so a give-back on
-   * the way out of an unwound replay must leave it alone. Releasing it without taking it out
-   * of the changes which are waiting would have the same change handed to two threads
-   * (issue #922).
+   * replaying: it is handed out by {@code getNextUpdate()} to whichever thread calls it
+   * first once what it waits for is gone, so a give-back on the way out of an unwound
+   * replay must leave it alone. Releasing it without taking it out of the changes which
+   * are waiting would have the same change handed to two threads (issue #922).
    * <p>
    * The deliveries are taken in the order a replay thread takes them: one at a time, off
    * the queue the pool shares. So the change which is parked here is parked by the thread
@@ -716,7 +718,7 @@ public void run()
 
     assertNull(pendingChanges.getChangeOwnedByCurrentThread(),
         "a change this thread parked as waiting for another one is not one it gives back:"
-            + " it is handed to whichever thread clears what it waits for");
+            + " it is handed out by getNextUpdate() once what it waits for is gone");
 
     // The delivery this thread took once the change it parked was out of its hands.
     final DeleteMsg next = deleteMsg(taken, "uuid-3");
@@ -794,7 +796,7 @@ public void aChangeWhichWasWaitingIsHandedOutOnce() throws Exception
     pendingChanges.commit(deleted);
 
     assertSame(pendingChanges.getNextUpdate(), rename,
-        "the change which was waiting must be handed to the thread which cleared it");
+        "the change which was waiting must be handed out once what it waited for is gone");
     assertNull(pendingChanges.getNextUpdate(),
         "a change which has been handed out must not be handed out again");
   }