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();
}
*/
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
Verify.verify(obj instanceof IncrementSequence);
final IncrementSequence increment = (IncrementSequence) obj;
successor.replayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
- increment.getSequence(), localActor(), increment.getDelta()), resp -> { }, now);
+ increment.getSequence(), localActor(), isSnapshotOnly(), increment.getDelta()), resp -> { },
+ now);
LOG.debug("Incrementing sequence {} to successor {}", obj, successor);
}
}
* 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);
/**