import javax.annotation.concurrent.NotThreadSafe;
import org.opendaylight.controller.cluster.access.client.ConnectionEntry;
import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.ClosedTransactionException;
import org.opendaylight.controller.cluster.access.commands.IncrementTransactionSequenceRequest;
import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
}
}
- // Generic state base class. Direct instances are used for fast paths, sub-class is used for successor transitions
+ /**
+ * Base class for representing logical state of this proxy. See individual instantiations and {@link SuccessorState}
+ * for details.
+ */
private static class State {
private final String string;
}
}
- // State class used when a successor has interfered. Contains coordinator latch, the successor and previous state
- private static final class SuccessorState extends State {
+ /**
+ * State class used when a successor has interfered. Contains coordinator latch, the successor and previous state.
+ * This is a temporary state introduced during reconnection process and is necessary for correct state hand-off
+ * between the old connection (potentially being accessed by the user) and the new connection (being cleaned up
+ * by the actor.
+ *
+ * <p>
+ * When a user operation encounters this state, it synchronizes on the it and wait until reconnection completes,
+ * at which point the request is routed to the successor transaction. This is a relatively heavy-weight solution
+ * to the problem of state transfer, but the user will observe it only if the race condition is hit.
+ */
+ private static class SuccessorState extends State {
private final CountDownLatch latch = new CountDownLatch(1);
private AbstractProxyTransaction successor;
private State prevState;
+ // SUCCESSOR + DONE
+ private boolean done;
+
SuccessorState() {
- super("successor");
+ super("SUCCESSOR");
}
// Synchronize with succession process and return the successor
}
State getPrevState() {
- return prevState;
+ return Verify.verifyNotNull(prevState, "Attempted to access previous state, which was not set");
}
void setPrevState(final State prevState) {
- Verify.verify(this.prevState == null);
+ Verify.verify(this.prevState == null, "Attempted to set previous state to %s when we already have %s",
+ prevState, this.prevState);
this.prevState = Preconditions.checkNotNull(prevState);
+ // We cannot have duplicate successor states, so this check is sufficient
+ this.done = DONE.equals(prevState);
}
// To be called from safe contexts, where successor is known to be completed
}
void setSuccessor(final AbstractProxyTransaction successor) {
- Verify.verify(this.successor == null);
+ Verify.verify(this.successor == null, "Attempted to set successor to %s when we already have %s",
+ successor, this.successor);
this.successor = Preconditions.checkNotNull(successor);
}
+
+ boolean isDone() {
+ return done;
+ }
+
+ void setDone() {
+ done = true;
+ }
}
private static final Logger LOG = LoggerFactory.getLogger(AbstractProxyTransaction.class);
AtomicIntegerFieldUpdater.newUpdater(AbstractProxyTransaction.class, "sealed");
private static final AtomicReferenceFieldUpdater<AbstractProxyTransaction, State> STATE_UPDATER =
AtomicReferenceFieldUpdater.newUpdater(AbstractProxyTransaction.class, State.class, "state");
- private static final State OPEN = new State("open");
- private static final State SEALED = new State("sealed");
- private static final State FLUSHED = new State("flushed");
+
+ /**
+ * Transaction has been open and is being actively worked on.
+ */
+ private static final State OPEN = new State("OPEN");
+
+ /**
+ * Transaction has been sealed by the user, but it has not completed flushing to the backed, yet. This is
+ * a transition state, as we are waiting for the user to initiate commit procedures.
+ *
+ * <p>
+ * Since the reconnect mechanics relies on state replay for transactions, this state needs to be flushed into the
+ * queue to re-create state in successor transaction (which may be based on different messages as locality may have
+ * changed). Hence the transition to {@link #FLUSHED} state needs to be handled in a thread-safe manner.
+ */
+ private static final State SEALED = new State("SEALED");
+
+ /**
+ * Transaction state has been flushed into the queue, i.e. it is visible by the successor and potentially
+ * the backend. At this point the transaction does not hold any state besides successful requests, all other state
+ * is held either in the connection's queue or the successor object.
+ *
+ * <p>
+ * Transition to this state indicates we have all input from the user we need to initiate the correct commit
+ * protocol.
+ */
+ private static final State FLUSHED = new State("FLUSHED");
+
+ /**
+ * Transaction state has been completely resolved, we have received confirmation of the transaction fate from
+ * the backend. The only remaining task left to do is finishing up the state cleanup, which is done via purge
+ * request. We need to hang on to the transaction until that is done, as we have to make sure backend completes
+ * purging its state -- otherwise we could have a leak on the backend.
+ */
+ private static final State DONE = new State("DONE");
// Touched from client actor thread only
private final Deque<Object> successfulRequests = new ArrayDeque<>();
* variable. It uses pre-allocated objects for fast paths (i.e. no successor present) and a per-transition object
* for slow paths (when successor is injected/present).
*/
- private volatile int sealed = 0;
- private volatile State state = OPEN;
+ private volatile int sealed;
+ private volatile State state;
- AbstractProxyTransaction(final ProxyHistory parent) {
+ AbstractProxyTransaction(final ProxyHistory parent, final boolean isDone) {
this.parent = Preconditions.checkNotNull(parent);
+ if (isDone) {
+ state = DONE;
+ // DONE implies previous seal operation completed
+ sealed = 1;
+ } else {
+ state = OPEN;
+ }
}
final void executeInActor(final Runnable command) {
if (t instanceof TransactionCommitSuccess) {
ret.set(Boolean.TRUE);
} else if (t instanceof RequestFailure) {
- ret.setException(((RequestFailure<?, ?>) t).getCause().unwrap());
+ final Throwable cause = ((RequestFailure<?, ?>) t).getCause().unwrap();
+ if (cause instanceof ClosedTransactionException) {
+ // This is okay, as it indicates the transaction has been completed. It can happen
+ // when we lose connectivity with the backend after it has received the request.
+ ret.set(Boolean.TRUE);
+ } else {
+ ret.setException(cause);
+ }
} else {
ret.setException(new IllegalStateException("Unhandled response " + t.getClass()));
}
}
LOG.debug("Transaction {} doCommit completed", this);
+
+ // Needed for ProxyHistory$Local data tree rebase points.
+ parent.completeTransaction(this);
+
enqueuePurge();
});
}
}
final void enqueuePurge(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
- enqueueRequest(purgeRequest(), resp -> {
- LOG.debug("Transaction {} purge completed", this);
- parent.completeTransaction(this);
+ LOG.debug("{}: initiating purge", this);
+
+ final State prev = state;
+ if (prev instanceof SuccessorState) {
+ ((SuccessorState) prev).setDone();
+ } else {
+ final boolean success = STATE_UPDATER.compareAndSet(this, prev, DONE);
+ if (!success) {
+ LOG.warn("{}: moved from state {} while we were purging it", this, prev);
+ }
+ }
+
+ successfulRequests.clear();
+
+ enqueueRequest(new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor()), resp -> {
+ LOG.debug("{}: purge completed", this);
+ parent.purgeTransaction(this);
+
if (callback != null) {
callback.accept(resp);
}
}, enqueuedTicks);
}
- private TransactionPurgeRequest purgeRequest() {
- successfulRequests.clear();
- return new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor());
- }
-
// Called with the connection unlocked
final synchronized void startReconnect() {
// At this point canCommit/directCommit are blocked, we assert a new successor state, retrieving the previous
}
// Called with the connection locked
- final void replayMessages(final AbstractProxyTransaction successor,
- final Iterable<ConnectionEntry> enqueuedEntries) {
+ final void replayMessages(final ProxyHistory successorHistory, final Iterable<ConnectionEntry> enqueuedEntries) {
final SuccessorState local = getSuccessorState();
+ final State prevState = local.getPrevState();
+
+ final AbstractProxyTransaction successor = successorHistory.createTransactionProxy(getIdentifier(),
+ isSnapshotOnly(), local.isDone());
+ LOG.debug("{} created successor {}", this, successor);
local.setSuccessor(successor);
// Replay successful requests first
for (Object obj : successfulRequests) {
if (obj instanceof TransactionRequest) {
LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
- successor.replayRequest((TransactionRequest<?>) obj, resp -> { }, now);
+ successor.doReplayRequest((TransactionRequest<?>) obj, resp -> { }, now);
} else {
Verify.verify(obj instanceof IncrementSequence);
final IncrementSequence increment = (IncrementSequence) obj;
- successor.replayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
+ successor.doReplayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
increment.getSequence(), localActor(), isSnapshotOnly(), increment.getDelta()), resp -> { },
now);
LOG.debug("Incrementing sequence {} to successor {}", obj, successor);
if (getIdentifier().equals(req.getTarget())) {
Verify.verify(req instanceof TransactionRequest, "Unhandled request %s", req);
LOG.debug("Replaying queued request {} to successor {}", req, successor);
- successor.replayRequest((TransactionRequest<?>) req, e.getCallback(), e.getEnqueuedTicks());
+ successor.doReplayRequest((TransactionRequest<?>) req, e.getCallback(), e.getEnqueuedTicks());
it.remove();
}
}
* reconnecting have been forced to slow paths, which will be unlocked once we unblock the state latch
* at the end of this method.
*/
- final State prevState = local.getPrevState();
if (SEALED.equals(prevState)) {
LOG.debug("Proxy {} reconnected while being sealed, propagating state to successor {}", this, successor);
flushState(successor);
* @param callback Callback to be invoked once the request completes
* @param enqueuedTicks ticker-based time stamp when the request was enqueued
*/
- private void replayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
+ private void doReplayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
final long enqueuedTicks) {
if (request instanceof AbstractLocalTransactionRequest) {
handleReplayedLocalRequest((AbstractLocalTransactionRequest<?>) request, callback, enqueuedTicks);
}
}
+ final void replayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
+ final long enqueuedTicks) {
+ getSuccessorState().getSuccessor().doReplayRequest(request, callback, enqueuedTicks);
+ }
+
abstract boolean isSnapshotOnly();
abstract void doDelete(YangInstanceIdentifier path);
Code Review / controller.git / blobdiff