* 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 final class SuccessorState extends State {
+ 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");
}
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);
}
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
final void completeTransaction(final AbstractProxyTransaction tx) {
lock.lock();
try {
- proxies.remove(tx.getIdentifier());
+ // Removal will be completed once purge completes
LOG.debug("Proxy {} completing transaction {}", this, tx);
onTransactionCompleted(tx);
} finally {
}
}
+ void purgeTransaction(final AbstractProxyTransaction tx) {
+ lock.lock();
+ try {
+ proxies.remove(tx.getIdentifier());
+ LOG.debug("Proxy {} purged transaction {}", this, tx);
+ } finally {
+ lock.unlock();
+ }
+ }
+
final void close() {
lock.lock();
try {