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.IncrementTransactionSequenceRequest;
import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
import org.opendaylight.controller.cluster.access.commands.TransactionCanCommitSuccess;
*/
@NotThreadSafe
private static final class IncrementSequence {
- private long delta = 1;
+ private final long sequence;
+ private long delta = 0;
+
+ IncrementSequence(final long sequence) {
+ this.sequence = sequence;
+ }
long getDelta() {
return delta;
}
+ long getSequence() {
+ return sequence;
+ }
+
void incrementDelta() {
delta++;
}
this.parent = Preconditions.checkNotNull(parent);
}
+ final void executeInActor(final Runnable command) {
+ parent.context().executeInActor(behavior -> {
+ command.run();
+ return behavior;
+ });
+ }
+
final ActorRef localActor() {
return parent.localActor();
}
- private void incrementSequence(final long delta) {
+ final void incrementSequence(final long delta) {
sequence += delta;
LOG.debug("Transaction {} incremented sequence to {}", this, sequence);
}
successfulRequests.add(Verify.verifyNotNull(req));
}
- final void recordFinishedRequest() {
+ final void recordFinishedRequest(final Response<?, ?> response) {
final Object last = successfulRequests.peekLast();
if (last instanceof IncrementSequence) {
((IncrementSequence) last).incrementDelta();
} else {
- successfulRequests.addLast(new IncrementSequence());
+ successfulRequests.addLast(new IncrementSequence(response.getSequence()));
}
}
*/
final void abort() {
checkNotSealed();
- doAbort();
parent.abortTransaction(this);
+
+ sendRequest(abortRequest(), resp -> {
+ LOG.debug("Transaction {} abort completed with {}", getIdentifier(), resp);
+ enqueuePurge();
+ });
}
final void abort(final VotingFuture<Void> ret) {
checkSealed();
- sendAbort(t -> {
+ sendDoAbort(t -> {
if (t instanceof TransactionAbortSuccess) {
ret.voteYes();
} else if (t instanceof RequestFailure) {
// This is a terminal request, hence we do not need to record it
LOG.debug("Transaction {} abort completed", this);
- sendPurge();
+ enqueuePurge();
});
}
final void enqueueAbort(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+ checkNotSealed();
+ parent.abortTransaction(this);
+
+ enqueueRequest(abortRequest(), resp -> {
+ LOG.debug("Transaction {} abort completed with {}", getIdentifier(), resp);
+ // Purge will be sent by the predecessor's callback
+ if (callback != null) {
+ callback.accept(resp);
+ }
+ }, enqueuedTicks);
+ }
+
+ final void enqueueDoAbort(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
enqueueRequest(new TransactionAbortRequest(getIdentifier(), nextSequence(), localActor()), callback,
enqueuedTicks);
}
- final void sendAbort(final Consumer<Response<?, ?>> callback) {
+ final void sendDoAbort(final Consumer<Response<?, ?>> callback) {
sendRequest(new TransactionAbortRequest(getIdentifier(), nextSequence(), localActor()), callback);
}
// This is a terminal request, hence we do not need to record it
LOG.debug("Transaction {} directCommit completed", this);
- sendPurge();
+ enqueuePurge();
});
return ret;
}
LOG.debug("Transaction {} doCommit completed", this);
- sendPurge();
+ enqueuePurge();
});
}
- final void sendPurge() {
- successfulRequests.clear();
-
- final TransactionRequest<?> req = new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor());
- sendRequest(req, t -> {
- LOG.debug("Transaction {} purge completed", this);
- parent.completeTransaction(this);
- });
+ private void enqueuePurge() {
+ enqueuePurge(null);
}
- final void enqueuePurge(final long enqueuedTicks) {
- successfulRequests.clear();
+ final void enqueuePurge(final Consumer<Response<?, ?>> callback) {
+ // Purge request are dispatched internally, hence should not wait
+ enqueuePurge(callback, parent.currentTime());
+ }
- final TransactionRequest<?> req = new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor());
- enqueueRequest(req, t -> {
+ final void enqueuePurge(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+ enqueueRequest(purgeRequest(), resp -> {
LOG.debug("Transaction {} purge completed", this);
parent.completeTransaction(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());
+ LOG.debug("{} created successor transaction proxy {}", this, successor);
local.setSuccessor(successor);
// Replay successful requests first
successor.replayRequest((TransactionRequest<?>) obj, resp -> { }, now);
} else {
Verify.verify(obj instanceof IncrementSequence);
- successor.incrementSequence(((IncrementSequence) obj).getDelta());
+ final IncrementSequence increment = (IncrementSequence) obj;
+ successor.replayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
+ increment.getSequence(), localActor(), isSnapshotOnly(), increment.getDelta()), resp -> { },
+ now);
+ LOG.debug("Incrementing sequence {} to successor {}", obj, successor);
}
}
LOG.debug("{} replayed {} successful requests", getIdentifier(), successfulRequests.size());
* 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);
abstract void doSeal();
- abstract void doAbort();
-
@GuardedBy("this")
abstract void flushState(AbstractProxyTransaction successor);
+ abstract TransactionRequest<?> abortRequest();
+
abstract TransactionRequest<?> commitRequest(boolean coordinated);
/**