* A parked change is handed out again by {@code getNextUpdate()} alone, which every * replay loop of this 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 whose + * replayed by whichever thread calls it first once the change it was waiting for is + * gone - the thread which cleared it, as a rule. A thread whose * replay was unwound is not on that road anymore - it takes the next delivery off the * replay queue - so a change it parked would be left owned by a thread which is not * coming back to it, while every redelivery of it is refused as a duplicate. On a domain diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java index 62629a0637..5835b12e91 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java @@ -94,8 +94,9 @@ final class RemotePendingChanges * this issue is about (issue #922). *
* 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 - a parked change is not the one - * this thread is replaying, and giving it back is + * 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, which enters it here again, and giving it back is * {@link #releaseParkedChangesOwnedByCurrentThread()}, which reads the changes which are * waiting rather than this index (issue #954). *
@@ -551,10 +552,11 @@ public boolean markInProgress(LDAPUpdateMsg msg) * Returns the CSN of the change the calling thread is replaying, when it still owns one. *
* 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 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). + * 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 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). *
* 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, @@ -582,11 +584,12 @@ CSN getChangeOwnedByCurrentThread() *
* 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). + * whichever thread calls it first once the change it was waiting for is gone - the + * thread which cleared it, as a rule - 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). *
* 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
@@ -607,8 +610,9 @@ CSN getChangeOwnedByCurrentThread()
*
* @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
+ * parked them, until {@link #getNextUpdate()} hands them to whichever thread
+ * calls it first once what they wait for is gone - the thread which cleared it,
+ * as a rule - or they are given back here
*/
List
* 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.
*/
@@ -765,11 +771,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 of the change a replay was
- * unwound on leaves it alone (issue #922). What hands a parked change back is
+ * 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 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).
*/
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
index 27cffe67d5..6c5e357226 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
@@ -96,8 +96,9 @@ public void run()
/*
* 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
+ * with a change: a parked change is replayed by whichever thread calls it first once
+ * the change it was waiting for is gone - the thread which cleared it, as a rule. 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
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
index 325212b7ae..46df38b748 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
@@ -3199,10 +3199,22 @@ 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.
*
- * 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, as a rule, 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.
+ *
+ * The pin this gives issue #922 holds on that first arm alone: nothing here orders the
+ * parker's own {@code getNextUpdate()} call against the parent being let go, and a parker
+ * delayed past the parent's release, commit and ack takes the child back on its own way
+ * out instead - measured with a mutant, 500 ms after {@code checkDependencies()} parks the
+ * child. A revert of the fix this test is for goes uncaught on that arm: the wait above
+ * sees the child parked either way, and only the parent's road runs the code #922 is about.
*/
@Test
public void theChangesParkedBehindAChangeWhoseAckFailedAreReplayed() throws Exception
@@ -3244,25 +3256,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),
@@ -3279,19 +3296,23 @@ public void theChangesParkedBehindAChangeWhoseAckFailedAreReplayed() throws Exce
* replay it was handed to is unwound.
*
* 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.
*
- * 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
@@ -3324,6 +3345,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)));
/*
@@ -3340,7 +3362,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"),
@@ -3349,12 +3377,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()
@@ -3370,10 +3405,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
@@ -3406,7 +3437,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
@@ -3591,7 +3622,8 @@ public void call() throws Exception
* A change which waits for another one is parked and stays owned by the replay thread
* which parked it, while that thread goes on to the changes which follow: it is handed
* out again by {@code getNextUpdate()}, which every replay loop of this domain runs once
- * it is done, so it is replayed by whichever thread clears the change it was waiting for.
+ * it is done, so it is replayed by whichever thread calls it first once the change it was
+ * waiting for is gone - the thread which cleared it, as a rule.
* A replay which is unwound leaves the thread which parked it without that road - it
* takes the next delivery off the shared queue instead, and never comes back to the
* change it parked - and every redelivery of a change a replay thread owns is refused as
@@ -3727,8 +3759,8 @@ public void aChangeParkedByAnUnwoundReplayIsDeliveredAgain() throws Exception
+ " it restarts the session for the changes it gave back");
/*
- * The barrier is lifted, which lets the ServerState past it and hands the parked
- * change to the thread which cleared it.
+ * The barrier is lifted, which lets the ServerState past it and has the parked
+ * change handed out.
*/
giveUpOn(domain, failing, waitedOn, failingMods, waitedOnUUID);
checkEntryHasAttributeValue(waitedOn, "description", parkedDescription, 30,
@@ -3814,13 +3846,14 @@ public void call() throws Exception
*
* A parked change stays owned by the replay thread which parked it while that thread
* goes back to the pool and takes the changes which follow: {@code getNextUpdate()} is
- * what hands it out again, to whichever replay thread clears the change it was waiting
- * for. Changing the number of replay threads stops the whole pool and creates another
- * one, so a thread which parked a change and went back to the queue is joined while it
- * is idle, and it would end still recorded as the owner of that change - a thread which
- * does not exist anymore, 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 this
- * replica's ServerState, and every change behind it from every master, stops.
+ * what hands it out again, to whichever thread calls it first once the change it was
+ * waiting for is gone - the thread which cleared it, as a rule. Changing the number of
+ * replay threads stops the whole pool and creates another one, so a thread which parked
+ * a change and went back to the queue is joined while it is idle, and it would end still
+ * recorded as the owner of that change - a thread which does not exist anymore, 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 this replica's ServerState, and
+ * every change behind it from every master, stops.
*/
@Test
public void aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain() throws Exception
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
index 30107ac3b0..f99cca8495 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
@@ -529,12 +529,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).
*
* 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
@@ -562,7 +563,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
* 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
@@ -719,7 +722,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");
@@ -797,7 +800,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");
}
@@ -807,13 +810,13 @@ public void aChangeWhichWasWaitingIsHandedOutOnce() throws Exception
* when the replay which parked it is unwound.
*
* A parked change stays owned by the thread which parked it, and is handed out again by
- * {@link RemotePendingChanges#getNextUpdate()} to whichever thread clears the change it
- * waits for. A replay which is unwound - a JVM out of memory, a throw from what the replay
- * runs once the ack of its delivery is out - leaves that thread without a road back to the
- * change: it takes the next delivery off the queue instead. Nothing else asks for the
- * change either, since every redelivery of a change a replay thread owns is refused as a
- * duplicate, so this domain's ServerState would stay behind it until some other change is
- * replayed on this domain.
+ * {@link RemotePendingChanges#getNextUpdate()} to whichever thread calls it first once the
+ * change it waits for is gone. A replay which is unwound - a JVM out of memory, a throw
+ * from what the replay runs once the ack of its delivery is out - leaves that thread
+ * without a road back to the change: it takes the next delivery off the queue instead.
+ * Nothing else asks for the change either, since every redelivery of a change a replay
+ * thread owns is refused as a duplicate, so this domain's ServerState would stay behind it
+ * until some other change is replayed on this domain.
*/
@Test
public void aChangeParkedByAReplayWhichIsUnwoundIsGivenBack() throws Exception
@@ -936,7 +939,7 @@ public void run()
assertFalse(pendingChanges.putRemoteUpdate(renameIntoDeletedEntry(renamed)),
"a change a replay thread owns must not be taken over (OPENDJ-1115)");
- // It is still handed to whichever thread clears the change it was waiting for.
+ // It is still handed out once the change it was waiting for is gone.
pendingChanges.commit(deleted);
assertSame(pendingChanges.getNextUpdate(), rename);
}