Skip to content

[#954] Give back the changes a replay which is unwound parked as dependencies #985

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vharseko wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[#954] Give back the changes a replay which is unwound parked as dependencies #985

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
127 changes: 100 additions & 27 deletions ...rver-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
{
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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.
* <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
* 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).
* <p>
* 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.
* <p>
* 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<CSN> 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.
Expand Down
89 changes: 82 additions & 7 deletions ...erver-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -90,8 +94,10 @@ final class RemotePendingChanges
* this issue is about (issue #922).
* <p>
* A thread is entered here when it takes a change over and removed when it gives it back,
* applies it, or parks it as waiting for another change - the parked ones are handed to
* whichever thread clears what they wait for, so they are not this one's to give back.
* applies it, or parks it as waiting for another change - a parked change is not the one
* this thread is replaying, and giving it back is
* {@link #releaseParkedChangesOwnedByCurrentThread()}, which reads the changes which are
* waiting rather than this index (issue #954).
* <p>
* The entry of a thread is written by that thread and by nobody else, and that - not the
* lock - is what keeps the writes apart: the park in {@link #addDependency(PendingChange)}
Expand Down Expand Up @@ -545,9 +551,10 @@ public boolean markInProgress(LDAPUpdateMsg msg)
* Returns the CSN of the change the calling thread is replaying, when it still owns one.
* <p>
* A thread owns the change it is replaying and the ones it parked as waiting for another
* change. The parked ones are left out: they are handed to whichever thread clears the
* change they are waiting for, and that thread takes them over, so giving one back here
* would have the same change handed to two threads (issue #922).
* change. The parked ones are left out: they are not the change this thread is replaying,
* and giving one back is more than dropping its owner - it has to be unparked in the same
* step, or it would be handed out by two roads at once, which is what
* {@link #releaseParkedChangesOwnedByCurrentThread()} does (issues #922 and #954).
* <p>
* It is a plain read of {@link #changeBeingReplayed}: no lock is taken and nothing is
* allocated. This is what the give-back on the way out of an unwound replay asks first,
Expand All @@ -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.
* <p>
* A parked change stays owned by the thread which parked it while that thread goes on
* to the changes which follow: {@link #getNextUpdate()} is what hands it out again, to
* whichever replay thread clears the change it was waiting for, and that thread takes it
* over. A replay which is unwound leaves the thread which parked it without that road -
* it takes the next delivery off the replay queue instead - so the change would be left
* owned by a thread which is never coming back to it, and every redelivery of a change a
* replay thread owns is refused as a duplicate (issue #954).
* <p>
* Unparking a change and giving it back is one step, under both locks, so that only one
* road can hand it out: a change which was released while it is still listed as waiting
* would be handed to the thread {@link #getNextUpdate()} gives it to and to the thread
* which takes over the delivery which follows - the double replay the ownership is there
* to prevent (OPENDJ-1115).
* <p>
* The changes stay listed and uncommitted, and stay among the changes the newer ones are
* checked against, the way a change whose replay failed does: they are not in the data,
* so they hold this domain's ServerState back and the changes which follow them keep
* waiting for them.
* <p>
* The changes another thread parked are left alone: a change is given back by the thread
* which owns it and by nobody else (issue #922). That thread may be inside the dependency
* checks which parked it - they park a change once per dependency it has - so a change
* released under it would be listed as waiting again a moment later, and handed out while
* the delivery which took it over is being replayed.
*
* @return the CSNs of the changes it gave back, oldest first; empty when this thread has
* no parked change left, which is what every replay which was not unwound while
* it held one leaves behind
*/
List<CSN> 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<CSN> released = new ArrayList<>();
final Iterator<PendingChange> it = dependentChanges.iterator();
while (it.hasNext())
{
final PendingChange change = it.next();
if (change.isOwnedBy(current))
{
it.remove();
change.setOwner(null);
released.add(change.getCSN());
}
}
return released;
}
finally
{
dependentChangesLock.unlock();
pendingChangesWriteLock.unlock();
}
}

/**
* Get the first update in the list that have some dependencies cleared.
* <p>
Expand Down Expand Up @@ -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());
}
Expand Down
3 changes: 3 additions & 0 deletions opendj-server-legacy/src/messages/org/opends/messages/replication.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 \
Expand Down
Loading
Loading