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;
}
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, "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
}
/**
- * Seal this transaction before it is either committed or aborted.
+ * Seal this transaction before it is either committed or aborted. This method should only be invoked from
+ * application thread.
*/
final void seal() {
// Transition user-visible state first
- final boolean success = SEALED_UPDATER.compareAndSet(this, 0, 1);
+ final boolean success = markSealed();
Preconditions.checkState(success, "Proxy %s was already sealed", getIdentifier());
- internalSeal();
+
+ if (!sealAndSend(Optional.absent())) {
+ sealSuccessor();
+ }
}
- final void ensureSealed() {
- if (SEALED_UPDATER.compareAndSet(this, 0, 1)) {
- internalSeal();
+ /**
+ * Internal seal propagation method, invoked when we have raced with reconnection thread. Note that there may have
+ * been multiple reconnects, so we have to make sure the action is propagate through all intermediate instances.
+ */
+ private void sealSuccessor() {
+ // Slow path: wait for the successor to complete
+ final AbstractProxyTransaction successor = awaitSuccessor();
+
+ // At this point the successor has completed transition and is possibly visible by the user thread, which is
+ // still stuck here. The successor has not seen final part of our state, nor the fact it is sealed.
+ // Propagate state and seal the successor.
+ flushState(successor);
+ successor.predecessorSealed();
+ }
+
+ private void predecessorSealed() {
+ if (markSealed() && !sealAndSend(Optional.absent())) {
+ sealSuccessor();
}
}
- private void internalSeal() {
- doSeal();
+ void sealOnly() {
parent.onTransactionSealed(this);
+ final boolean success = STATE_UPDATER.compareAndSet(this, OPEN, SEALED);
+ Verify.verify(success, "Attempted to replay seal on {}", this);
+ }
- // Now deal with state transfer, which can occur via successor or a follow-up canCommit() or directCommit().
- if (!STATE_UPDATER.compareAndSet(this, OPEN, SEALED)) {
- // Slow path: wait for the successor to complete
- final AbstractProxyTransaction successor = awaitSuccessor();
+ /**
+ * Seal this transaction and potentially send it out towards the backend. If this method reports false, the caller
+ * needs to deal with propagating the seal operation towards the successor.
+ *
+ * @param enqueuedTicks Enqueue ticks when this is invoked from replay path.
+ * @return True if seal operation was successful, false if this proxy has a successor.
+ */
+ boolean sealAndSend(final Optional<Long> enqueuedTicks) {
+ parent.onTransactionSealed(this);
- // At this point the successor has completed transition and is possibly visible by the user thread, which is
- // still stuck here. The successor has not seen final part of our state, nor the fact it is sealed.
- // Propagate state and seal the successor.
- flushState(successor);
- successor.ensureSealed();
- }
+ // Transition internal state to sealed and detect presence of a successor
+ return STATE_UPDATER.compareAndSet(this, OPEN, SEALED);
+ }
+
+ /**
+ * Mark this proxy as having been sealed.
+ *
+ * @return True if this call has transitioned to sealed state.
+ */
+ final boolean markSealed() {
+ return SEALED_UPDATER.compareAndSet(this, 0, 1);
}
private void checkNotSealed() {
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()));
}
final SuccessorState local = getSuccessorState();
final State prevState = local.getPrevState();
- final boolean isDone = DONE.equals(state)
- || state instanceof SuccessorState && ((SuccessorState) state).isDone();
final AbstractProxyTransaction successor = successorHistory.createTransactionProxy(getIdentifier(),
- isSnapshotOnly(), isDone);
+ isSnapshotOnly(), local.isDone());
LOG.debug("{} created successor {}", this, successor);
local.setSuccessor(successor);
if (SEALED.equals(prevState)) {
LOG.debug("Proxy {} reconnected while being sealed, propagating state to successor {}", this, successor);
flushState(successor);
- successor.ensureSealed();
+ if (successor.markSealed()) {
+ successor.sealAndSend(Optional.of(parent.currentTime()));
+ }
}
}
abstract CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(YangInstanceIdentifier path);
- abstract void doSeal();
-
@GuardedBy("this")
abstract void flushState(AbstractProxyTransaction successor);
Code Review / controller.git / blobdiff