mirror of https://github.com/OpenIdentityPlatform/OpenDJ.git
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
@@ -2897,8 +2897,8 @@ * listed, uncommitted and unowned, so the request is what brings them back, and * this thread is the one there to run it (issue #954). A give-back which threw * before it released anything left the parked changes as they were, owned by this * thread and handed out by getNextUpdate() to whichever thread clears what they * wait for: nothing here can do better for those. * thread and handed out by getNextUpdate() to whichever thread calls it first once * what they wait for is gone: nothing here can do better for those. * * Not run on a domain whose session has an owner, the way no road of a failed * replay runs it there: the restart is refused where it runs and the request @@ -4168,7 +4168,8 @@ * <p> * 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 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
@@ -94,8 +94,9 @@ * 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 - 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). * <p> @@ -551,10 +552,11 @@ * 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 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). * <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, @@ -582,11 +584,12 @@ * <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). * 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). * <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 @@ -607,8 +610,9 @@ * * @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<CSN> releaseParkedChangesOwnedByCurrentThread() { @@ -669,9 +673,11 @@ * 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. */ @@ -765,11 +771,13 @@ } /* * 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). */ opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
@@ -96,8 +96,9 @@ /* * 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 opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
@@ -3199,10 +3199,22 @@ * 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, 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. * <p> * 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 @@ 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 @@ * 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 @@ -3324,6 +3345,7 @@ 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 @@ { 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 @@ { 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 @@ "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 @@ + " 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 @@ * 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 @@ + " 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 @@ * <p> * 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 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
@@ -529,12 +529,13 @@ } /** * 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 @@ -562,7 +563,8 @@ 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<>(); @@ -671,12 +673,13 @@ /** * 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 the give-back * of the change a replay was unwound on 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) - which is why the parked ones are given back on a road of their * own, {@link RemotePendingChanges#releaseParkedChangesOwnedByCurrentThread()}, where * both happen in one step (issue #954). * 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) - which is * why the parked ones are given back on a road of their own, * {@link RemotePendingChanges#releaseParkedChangesOwnedByCurrentThread()}, where both * happen in one step (issue #954). * <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 @@ -719,7 +722,7 @@ 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 @@ 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 @@ * when the replay which parked it is unwound. * <p> * 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 @@ 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); }
