diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index b94a4dc8ba..9a496c0b58 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -2606,20 +2606,25 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) * failing and is eventually given up on: handing it back bare would have this domain * 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. + * The changes this thread parked as waiting for another change are given back too, + * and before the road above runs: that road restarts the session, and a change which + * is still owned when the replication server sends it again over it is turned down as + * a duplicate - the one delivery which could have taken it over (issue #954). * - * 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, - * so a lookup which threw in its turn - on the road out of a JVM which has just - * refused an allocation - would leave the change listed, uncommitted and owned by a - * thread which is about to end, which is the state this whole issue is about. + * Which change this thread owns is read first of all, and that read takes no lock and + * allocates nothing: everything below is gated on the answer, so a lookup which threw + * in its turn - on the road out of a JVM which has just refused an allocation - would + * leave the change listed, uncommitted and owned by a thread which is about to end, + * which is the state this whole issue is about. It is read before the parked changes + * are given back rather than after, because that give-back allocates and can throw on + * the same road, and the last resort below can only hand back a change it was told + * about. */ CSN owned = null; try { owned = remotePendingChanges.getChangeOwnedByCurrentThread(); + final boolean parkedGivenBack = giveBackParkedChanges(); if (owned != null) { if (replayThreadShutdown.get() || shutdown.get() || disabled) @@ -2650,6 +2655,17 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) recoverFromReplayFailure(owned, replayThreadShutdown, t instanceof OutOfMemoryError); } } + if (parkedGivenBack && !shutdown.get() && !disabled) + { + /* + * The road the change this thread was replaying took may have run the restart the + * give-back asked for - they ask for the same one - and it may have had none to + * run: this thread owned no change, or the change it owned was given up on. Run + * what is still requested, or a domain which then goes quiet would leave the + * changes which were handed back waiting for a delivery nobody asks for. + */ + runRequestedSessionRestarts(!replayThreadShutdown.get()); + } } catch (Throwable recoveryFailure) { @@ -2682,19 +2698,25 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) { suppress(recoveryFailure, reportFailure); } - try - { - runRequestedSessionRestarts(false); - } - catch (Throwable restartFailure) - { - /* - * Nothing is left to try: the change is listed, uncommitted and unowned, so any - * later session restart of this domain delivers it again. This goes with the - * throwable which is rethrown below rather than being reported on its own. - */ - suppress(recoveryFailure, restartFailure); - } + } + try + { + /* + * Outside the guard above, since the give-back of the changes this thread parked + * asks for the same restart and may be what threw: a change which nobody owns + * anymore is one only a new delivery brings back, whichever of the two roads + * released it (issue #954). + */ + runRequestedSessionRestarts(false); + } + catch (Throwable restartFailure) + { + /* + * Nothing is left to try: the changes are listed, uncommitted and unowned, so any + * later session restart of this domain delivers them again. This goes with the + * throwable which is rethrown below rather than being reported on its own. + */ + suppress(recoveryFailure, restartFailure); } // The error which unwound the replay is the one reported, whatever the give-back // ran into on top of it. @@ -3173,12 +3195,14 @@ else if (op instanceof ModifyDNOperation) * an OutOfMemoryError of its own, still owns its change: it is given back * counted, and the thread ends on this error rather than on the one it stepped * over. A replay which committed owns nothing anymore - commit() cleared the - * owner, and the index the give-back reads, in the same step - so the give-back - * is a no-op, and rightly so: a change which is in the data is not one to ask - * for again. What that road steps over is getNextUpdate() below, so the changes - * parked behind the committed change wait for the next replay of this domain to - * hand them out. That is the trade #923 asks for: a thread which met this error - * is not to carry on, not even for them. + * owner, and the index the give-back reads, in the same step - so the change it + * was replaying is not given back, and rightly so: a change which is in the data + * is not one to ask for again. What that road steps over is getNextUpdate() + * below, which hands out the changes parked behind the committed change: the + * ones this thread parked are given back on the way out of replay() and the + * session is restarted for them (issue #954), the ones other threads parked wait + * for the next replay of this domain to hand them out. That is the trade #923 + * asks for: a thread which met this error is not to carry on, not even for them. */ throw e; } @@ -3672,6 +3696,55 @@ private void runRequestedSessionRestarts(boolean wait) } } + /** + * Gives back the changes this replay thread parked as waiting for another change, on the + * way out of a replay which was unwound. + *

+ * 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 + * 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 + * which then goes quiet that change is where this replica's ServerState, and every change + * behind it from every master, stops (issue #954). + *

+ * They are handed back without a failure being counted against them: they were never + * applied here, so the give-up budget which decides when this replica skips a change it + * can not apply is not this delivery's to spend, the way it is not for a change abandoned + * by a replay thread which is stopping. + *

+ * The delivery which carried one published no ack - the ack of a parked change is + * published by the delivery which replays it - so it is counted as processed here, the + * way a delivery which is dropped rather than replayed is: that count is of the + * deliveries this replica took off the session, and these are over. The window they hold + * is not given back either, and does not need to be: the session they came over is about + * to be restarted, and a session which starts is given its receive window anew. + * + * @return whether any change was handed back, so that the caller restarts the session for + * them: a change which nobody owns is one only a new delivery brings back + */ + private boolean giveBackParkedChanges() + { + final List parked = remotePendingChanges.releaseParkedChangesOwnedByCurrentThread(); + if (parked.isEmpty()) + { + return false; + } + /* + * Asked for before the changes are reported: a throw out of the report - the JVM which + * unwound this replay is out of memory - must not lose the restart which is what brings + * them back. + */ + sessionRestartRequested.set(true); + for (CSN csn : parked) + { + incProcessedUpdates(); + logger.info(NOTE_REPLAY_PARKED_CHANGE_GIVEN_BACK, csn, getBaseDN()); + } + return true; + } + /** * Gives a change back to the replication server when this replay thread stops before it * could apply it. 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 ec3ef7b756..b16ada2f3e 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 @@ -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). *

* 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). *

* 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. *

* 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). *

* 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,72 @@ 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. + *

+ * 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). + *

+ * 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). + *

+ * 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. + *

+ * 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 has + * no parked change left, which is what every replay which was not unwound while + * it held one leaves behind + */ + List releaseParkedChangesOwnedByCurrentThread() + { + final Thread current = Thread.currentThread(); + pendingChangesWriteLock.lock(); + dependentChangesLock.lock(); + try + { + if (dependentChanges.isEmpty()) + { + // Nothing is waiting, which is the state every replay but a handful leaves behind. + return emptyList(); + } + final List released = new ArrayList<>(); + final Iterator 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(); + pendingChangesWriteLock.unlock(); + } + } + /** * Get the first update in the list that have some dependencies cleared. *

@@ -672,8 +745,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()); } diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties index d5a7e69970..efd51f64c5 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -668,6 +668,9 @@ ERR_ACK_NOT_PUBLISHED_316=Could not complete the delivery of change %s in domain ERR_REPLAY_GIVE_BACK_FAILED_317=Could not give change %s of domain "%s" back to the replication \ server after the replay which owned it was unwound: %s. The change has been released without its \ failure being counted, and the session is being restarted 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 which parked it was unwound. 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 \ 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 b791c2a7da..c4d37d3259 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 @@ -3293,6 +3293,218 @@ public void call() throws Exception } } + /** + * Test case for [Issue 954]: a change parked as waiting for another one is given back + * when the replay which parked it is unwound. + *

+ * 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. + * 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 + * 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 aChangeParkedByAnUnwoundReplayIsDeliveredAgain() throws Exception + { + testSetUp("aChangeParkedByAnUnwoundReplayIsDeliveredAgain"); + logger.error(LocalizableMessage.raw( + "Starting replication test : aChangeParkedByAnUnwoundReplayIsDeliveredAgain")); + + final DN waitedOn = addEntryForChange("user.954.1"); + final String waitedOnUUID = getEntry(waitedOn, 1, true).parseAttribute("entryuuid").asString(); + final DN other = addEntryForChange("user.954.2"); + final String otherUUID = getEntry(other, 1, true).parseAttribute("entryuuid").asString(); + + final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); + domain.resetUnreplayedChangeAlertThrottle(); + final long inProgress = getMonitorAttrValue(baseDN, "changes-in-progress-size"); + assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0, + "no change of this domain is waiting for another one when this test starts"); + + final CSNGenerator gen = new CSNGenerator(24, TimeThread.getTime()); + final CSN failing = gen.newCSN(); + final CSN parked = gen.newCSN(); + final CSN unwound = gen.newCSN(); + final List failingMods = generatemods("description", "the replay of this change fails"); + final String parkedDescription = "the change which was parked as a dependency"; + final List parkedMods = generatemods("description", parkedDescription); + final List unwoundMods = + generatemods("description", "the replay of this change is unwound"); + + /* + * The change whose replay fails is the barrier the parked change waits behind, so its + * budget must not be spent while this test is setting up: it is shortened once the + * change which was parked has been given back, and put back in the finally below. + */ + setReplayGiveUpDelay("unlimited"); + /* + * One replay thread, so that the change which is parked and the replay which is + * unwound after it are the same thread's: a parked change is left owned by the thread + * which parked it, and this is about a thread which does not come back to it. + */ + setNumUpdateReplayThreads(1); + try + { + /* + * A change whose replay failed stays listed and uncommitted - it is what holds this + * domain's ServerState back - and stays among the changes the newer ones are checked + * against, so a change which follows it on the same entry has to wait for it. It has + * to be a change which is asked for again rather than stepped over: one whose + * operation is built and then refused, since #928 has a modify whose entry DN does + * not parse reported once and recorded as replayed. + */ + deliverUntilMonitorReaches(domain, "changes-in-progress-size", inProgress + 1, + () -> new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID), + "the change whose replay fails must stay listed as one which is not in the data"); + + // The change which is parked as waiting for it by the replay thread it was given to. + deliverUntilMonitorReaches(domain, "dependent-changes-size", 1, + () -> new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID), + "a change which waits for one that is not in the data must be parked"); + + /* + * The replay which is unwound while that same thread still holds the parked change. + * It is a change whose replay is unwound once the ack of its delivery is out: what + * the replay runs from there - the give-back of the change it was replaying, and the + * hand-out of the changes which were waiting for that one - is past every catch the + * replay itself has, so a throw there is what leaves replay() by the way this issue + * is about. An Error met replaying a change does not, and neither does a throw from + * publishing the ack: both are reported and take the ordinary road of a failed replay + * (issue #922). + * + * It is made on another entry, so that it is replayed rather than parked in its turn, + * and it is waited for on the changes being replayed rather than on the failure it + * reports: what this test is about happens on the way out of that replay. + */ + deliverUntilMonitorReaches(domain, "changes-in-progress-size", inProgress + 3, + () -> new ModifyMsgWhoseReplayIsUnwoundAfterItsAck(unwound, other, unwoundMods, otherUUID), + "the replay of the change which follows the parked one must have been reached"); + + assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", 0, + "a change parked by a replay which was unwound must be given back"); + + /* + * Nothing sends these changes again - they never travelled a session - so the + * deliveries which take over from the ones which were unwound are made here. The + * change no delivery can replay is given up on, which lets the ServerState past it, + * and the two which never failed are applied. + */ + setReplayGiveUpDelay(TEST_GIVE_UP_DELAY); + TestTimer timer = new TestTimer.Builder() + .maxSleep(120, SECONDS) + .sleepTimes(500, MILLISECONDS) + .toTimer(); + timer.repeatUntilSuccess(new CallableVoid() + { + @Override + public void call() throws Exception + { + final ServerState state = domain.getServerState(); + if (!state.cover(failing)) + { + domain.processUpdate( + new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID)); + } + if (!state.cover(parked)) + { + domain.processUpdate(new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID)); + } + if (!state.cover(unwound)) + { + domain.processUpdate(new ModifyMsg(unwound, other, unwoundMods, otherUUID)); + } + assertTrue(state.cover(parked), + "the change which was given back must be replayed by the delivery which takes it over"); + assertTrue(state.cover(unwound), + "the change whose replay was unwound must be replayed by the delivery which takes it over"); + } + }); + checkEntryHasAttributeValue(waitedOn, "description", parkedDescription, 30, + "the change which was parked must be applied by the delivery which took it over"); + } + finally + { + resetReplayGiveUpDelay(); + resetNumUpdateReplayThreads(); + } + } + + /** + * Delivers a change until a monitor attribute of the domain reaches the expected value. + *

+ * A delivery is dropped rather than queued while the listener thread is down, which it + * is for as long as a recovery is restarting the session, so a change which has to reach + * a replay thread is delivered until it does. A delivery of a change a replay thread + * owns is refused as the duplicate it is, so the deliveries which follow the one that + * was taken cost nothing. + */ + private void deliverUntilMonitorReaches(final LDAPReplicationDomain domain, + final String attributeName, final long expected, + final Supplier delivery, final String message) throws Exception + { + TestTimer timer = new TestTimer.Builder() + .maxSleep(20, SECONDS) + .sleepTimes(500, MILLISECONDS) + .toTimer(); + timer.repeatUntilSuccess(new CallableVoid() + { + @Override + public void call() throws Exception + { + if (getMonitorAttrValue(baseDN, attributeName) != expected) + { + domain.processUpdate(delivery.get()); + } + assertEquals(getMonitorAttrValue(baseDN, attributeName), expected, message); + } + }); + } + + /** + * Sets how many replay threads this server runs, the way an administrator would: the + * pool is stopped and created again with that number. + */ + private static void setNumUpdateReplayThreads(int threads) throws Exception + { + assertEquals(TestCaseUtils.applyModifications(true, + "dn: " + SYNCHRO_PLUGIN_DN, + "changetype: modify", + "replace: ds-cfg-num-update-replay-threads", + "ds-cfg-num-update-replay-threads: " + threads), 0, + "the number of replay threads could not be changed"); + } + + /** + * Puts the number of replay threads back to what this server computes for itself, which + * is what it runs with when the configuration carries no number of its own. + */ + private static void resetNumUpdateReplayThreads() throws Exception + { + assertEquals(TestCaseUtils.applyModifications(true, + "dn: " + SYNCHRO_PLUGIN_DN, + "changetype: modify", + "delete: ds-cfg-num-update-replay-threads"), 0, + "the number of replay threads could not be put back"); + } + + /** Adds the entry a change of these tests is made on. */ + private DN addEntryForChange(String uid) throws Exception + { + return TestCaseUtils.addEntry( + "dn: uid=" + uid + "," + baseDN, + "objectClass: top", + "objectClass: person", + "objectClass: organizationalPerson", + "objectClass: inetOrgPerson", + "uid: " + uid, + "cn: Aaccf Amar", + "sn: Amar").getName(); + } + /** A delivery of a change whose replay does not run to its end. */ private interface FailingDelivery { @@ -3313,16 +3525,7 @@ private interface FailingDelivery private void assertChangeIsDeliveredAgainAfter( FailingDelivery delivery, int serverId, String uid, final String description) throws Exception { - Entry tmp = TestCaseUtils.addEntry( - "dn: uid=" + uid + "," + baseDN, - "objectClass: top", - "objectClass: person", - "objectClass: organizationalPerson", - "objectClass: inetOrgPerson", - "uid: " + uid, - "cn: Aaccf Amar", - "sn: Amar"); - final DN dn = tmp.getName(); + final DN dn = addEntryForChange(uid); final String uuid = getEntry(dn, 1, true).parseAttribute("entryuuid").asString(); final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); 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 e91292f43e..30107ac3b0 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 @@ -15,6 +15,7 @@ */ package org.opends.server.replication.plugin; +import static java.util.Collections.*; import static org.testng.Assert.*; import java.util.NoSuchElementException; @@ -670,10 +671,12 @@ 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 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). *

* 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 @@ -799,6 +802,145 @@ public void aChangeWhichWasWaitingIsHandedOutOnce() throws Exception "a change which has been handed out must not be handed out again"); } + /** + * Test case for [Issue 954]: a change parked as waiting for another one is given back + * 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. + */ + @Test + public void aChangeParkedByAReplayWhichIsUnwoundIsGivenBack() throws Exception + { + final ServerState state = new ServerState(); + final RemotePendingChanges pendingChanges = new RemotePendingChanges(state); + final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0); + final CSN deleted = generator.newCSN(); + final CSN renamed = generator.newCSN(); + + final DeleteMsg delete = deleteMsg(deleted, "uuid-1"); + assertTrue(pendingChanges.putRemoteUpdate(delete)); + assertTrue(pendingChanges.markInProgress(delete)); + + // A rename into the DN that delete is on: it can only be replayed once the delete has been. + final ModifyDNMsg rename = renameIntoDeletedEntry(renamed); + assertTrue(pendingChanges.putRemoteUpdate(rename)); + assertTrue(pendingChanges.markInProgress(rename)); + assertTrue(pendingChanges.checkDependencies(rename)); + + // The replay which parked it is unwound, so it gives back what it still owns. + assertEquals(pendingChanges.releaseParkedChangesOwnedByCurrentThread(), + singletonList(renamed), "the change this thread parked must be given back"); + + assertEquals(pendingChanges.getDependentChangesSize(), 0, + "a change which was given back must not be left waiting for a thread to hand it out"); + assertEquals(pendingChanges.getQueueSize(), 2, "the change must stay listed as pending"); + assertTrue(state.isEmpty(), "a change which was not replayed must not be recorded as replayed"); + assertEquals(pendingChanges.changesInProgressSize(), 2, + "a change which is not in the data yet must stay a dependency of the changes which follow it"); + + assertTrue(pendingChanges.putRemoteUpdate(renameIntoDeletedEntry(renamed)), + "the change the unwound replay gave back must be taken over by the next delivery"); + } + + /** + * Test case for [Issue 954]: a change which was given back is handed out by the next + * delivery of it and by nothing else. + *

+ * Unparking it and giving it back is one step under both locks for that reason: a + * change which was released while it is still listed as waiting would be replayed by + * the thread {@link RemotePendingChanges#getNextUpdate()} hands it to and by the thread + * which takes over the delivery which follows - the double replay the ownership is + * there to prevent (OPENDJ-1115). + */ + @Test + public void aParkedChangeWhichWasGivenBackIsNotHandedOutAsADependency() throws Exception + { + final ServerState state = new ServerState(); + final RemotePendingChanges pendingChanges = new RemotePendingChanges(state); + final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0); + final CSN deleted = generator.newCSN(); + final CSN renamed = generator.newCSN(); + + final DeleteMsg delete = deleteMsg(deleted, "uuid-1"); + assertTrue(pendingChanges.putRemoteUpdate(delete)); + assertTrue(pendingChanges.markInProgress(delete)); + + final ModifyDNMsg rename = renameIntoDeletedEntry(renamed); + assertTrue(pendingChanges.putRemoteUpdate(rename)); + assertTrue(pendingChanges.markInProgress(rename)); + assertTrue(pendingChanges.checkDependencies(rename)); + pendingChanges.releaseParkedChangesOwnedByCurrentThread(); + + // The change it was waiting for is replayed, which is what used to hand it out. + pendingChanges.commit(deleted); + + assertNull(pendingChanges.getNextUpdate(), + "a change which was given back must not also be handed out as a dependency"); + + // It is replayed by the delivery which takes it over, and by that one only. + final ModifyDNMsg nextDelivery = renameIntoDeletedEntry(renamed); + assertTrue(pendingChanges.putRemoteUpdate(nextDelivery)); + assertTrue(pendingChanges.markInProgress(nextDelivery)); + pendingChanges.commit(renamed); + + assertTrue(state.cover(renamed)); + assertEquals(pendingChanges.getQueueSize(), 0); + } + + /** + * Test case for [Issue 954]: the changes another thread parked are left alone. + *

+ * A change is given back by the thread which owns it and by nobody else, here as + * everywhere else (issue #922). The thread which parked a change is the one which may + * still 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. + */ + @Test + public void theParkedChangesOfAnotherThreadAreLeftAlone() throws Exception + { + final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState()); + final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0); + final CSN deleted = generator.newCSN(); + final CSN renamed = generator.newCSN(); + + final DeleteMsg delete = deleteMsg(deleted, "uuid-1"); + assertTrue(pendingChanges.putRemoteUpdate(delete)); + assertTrue(pendingChanges.markInProgress(delete)); + + final ModifyDNMsg rename = renameIntoDeletedEntry(renamed); + assertTrue(pendingChanges.putRemoteUpdate(rename)); + assertTrue(pendingChanges.markInProgress(rename)); + assertTrue(pendingChanges.checkDependencies(rename)); + + // Another replay thread is unwound while this one holds the change it parked. + runAndJoin(new Runnable() + { + @Override + public void run() + { + assertEquals(pendingChanges.releaseParkedChangesOwnedByCurrentThread(), emptyList(), + "a change another thread parked is not this one's to give back"); + } + }); + + assertEquals(pendingChanges.getDependentChangesSize(), 1, + "the change must stay listed as waiting for the one it depends on"); + 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. + pendingChanges.commit(deleted); + assertSame(pendingChanges.getNextUpdate(), rename); + } + /** A rename of an entry into the DN {@code deleteMsg(csn, "uuid-1")} deletes. */ private ModifyDNMsg renameIntoDeletedEntry(CSN csn) throws Exception {