From 72821ccd7b91e6400acbbd71f7007942e5fac872 Mon Sep 17 00:00:00 2001 From: Robert Varga Date: Wed, 3 Aug 2016 17:15:57 +0200 Subject: [PATCH] BUG-6111: implement PingPongTransactionChain cancelation This patch implements transaction cancelation in PingPongDataBroker, which has slightly different semantics -- if a transaction is canceled while being in a batch, proper isolation of the batch is maintained and after preceding batch completes, the transaction chain is aborted. Since there is no transaction isolation within a batch, this is the only course of action we can take. Change-Id: I0058503165dbfba8748a17a9ef9272265f4bc1c9 Signed-off-by: Robert Varga --- .../broker/impl/PingPongTransactionChain.java | 165 ++++++++++++++---- 1 file changed, 133 insertions(+), 32 deletions(-) diff --git a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/PingPongTransactionChain.java b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/PingPongTransactionChain.java index b792826ec4..cbb5b0bd5a 100644 --- a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/PingPongTransactionChain.java +++ b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/PingPongTransactionChain.java @@ -9,10 +9,14 @@ package org.opendaylight.controller.md.sal.dom.broker.impl; import com.google.common.base.Optional; import com.google.common.base.Preconditions; +import com.google.common.base.Verify; import com.google.common.util.concurrent.CheckedFuture; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; +import java.util.AbstractMap.SimpleImmutableEntry; +import java.util.Map.Entry; +import java.util.concurrent.CancellationException; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import javax.annotation.Nonnull; import javax.annotation.concurrent.GuardedBy; @@ -53,12 +57,15 @@ import org.slf4j.LoggerFactory; */ public final class PingPongTransactionChain implements DOMTransactionChain { private static final Logger LOG = LoggerFactory.getLogger(PingPongTransactionChain.class); + private final TransactionChainListener listener; private final DOMTransactionChain delegate; @GuardedBy("this") private boolean failed; @GuardedBy("this") private PingPongTransaction shutdownTx; + @GuardedBy("this") + private Entry deadTx; /** * This updater is used to manipulate the "ready" transaction. We perform only atomic @@ -87,47 +94,80 @@ public final class PingPongTransactionChain implements DOMTransactionChain { private volatile PingPongTransaction inflightTx; PingPongTransactionChain(final DOMDataBroker broker, final TransactionChainListener listener) { + this.listener = Preconditions.checkNotNull(listener); this.delegate = broker.createTransactionChain(new TransactionChainListener() { @Override - public void onTransactionChainFailed(final TransactionChain chain, final AsyncTransaction transaction, final Throwable cause) { - LOG.debug("Delegate chain {} reported failure in {}", chain, transaction, cause); - - final DOMDataReadWriteTransaction frontend; - final PingPongTransaction tx = inflightTx; - if (tx == null) { - LOG.warn("Transaction chain {} failed with no pending transactions", chain); - frontend = null; - } else { - frontend = tx.getFrontendTransaction(); - } - - listener.onTransactionChainFailed(PingPongTransactionChain.this, frontend, cause); - delegateFailed(); + public void onTransactionChainFailed(final TransactionChain chain, + final AsyncTransaction transaction, final Throwable cause) { + LOG.debug("Transaction chain {} reported failure in {}", chain, transaction, cause); + delegateFailed(chain, cause); } @Override public void onTransactionChainSuccessful(final TransactionChain chain) { - listener.onTransactionChainSuccessful(PingPongTransactionChain.this); + delegateSuccessful(chain); } }); } - private synchronized void delegateFailed() { - failed = true; + void delegateSuccessful(final TransactionChain chain) { + final Entry canceled; + synchronized (this) { + // This looks weird, but we need not hold the lock while invoking callbacks + canceled = deadTx; + } - /* - * If we do not have a locked transaction, we need to ensure that - * the backend transaction is cancelled. Otherwise we can defer - * until the user calls us. - */ - if (lockedTx == null) { - processIfReady(); + if (canceled == null) { + listener.onTransactionChainSuccessful(this); + return; + } + + // Backend shutdown successful, but we have a batch of transactions we have to report as dead due to the + // user calling cancel(). + final PingPongTransaction tx = canceled.getKey(); + final Throwable cause = canceled.getValue(); + LOG.debug("Transaction chain {} successful, failing cancelled transaction {}", chain, tx, cause); + + listener.onTransactionChainFailed(this, tx.getFrontendTransaction(), cause); + tx.onFailure(cause); + } + + void delegateFailed(final TransactionChain chain, final Throwable cause) { + + final DOMDataReadWriteTransaction frontend; + final PingPongTransaction tx = inflightTx; + if (tx == null) { + LOG.warn("Transaction chain {} failed with no pending transactions", chain); + frontend = null; + } else { + frontend = tx.getFrontendTransaction(); + } + + listener.onTransactionChainFailed(this, frontend, cause); + + synchronized (this) { + failed = true; + + /* + * If we do not have a locked transaction, we need to ensure that + * the backend transaction is cancelled. Otherwise we can defer + * until the user calls us. + */ + if (lockedTx == null) { + processIfReady(); + } } } private synchronized PingPongTransaction slowAllocateTransaction() { Preconditions.checkState(shutdownTx == null, "Transaction chain %s has been shut down", this); + if (deadTx != null) { + throw new IllegalStateException(String.format( + "Transaction chain %s has failed due to transaction %s being canceled", this, deadTx.getKey()), + deadTx.getValue()); + } + final DOMDataReadWriteTransaction delegateTx = delegate.newReadWriteTransaction(); final PingPongTransaction newTx = new PingPongTransaction(delegateTx); @@ -148,11 +188,9 @@ public final class PingPongTransactionChain implements DOMTransactionChain { return slowAllocateTransaction(); } - // Fast path: reuse current transaction. We will check - // failures and similar on submit(). + // Fast path: reuse current transaction. We will check failures and similar on submit(). if (!LOCKED_UPDATER.compareAndSet(this, null, oldTx)) { - // Ouch. Delegate chain has not detected a duplicate - // transaction allocation. This is the best we can do. + // Ouch. Delegate chain has not detected a duplicate transaction allocation. This is the best we can do. oldTx.getTransaction().cancel(); throw new IllegalStateException(String.format("Reusable transaction %s raced with transaction %s", oldTx, lockedTx)); } @@ -242,21 +280,21 @@ public final class PingPongTransactionChain implements DOMTransactionChain { } } - private void transactionSuccessful(final PingPongTransaction tx, final Void result) { + void transactionSuccessful(final PingPongTransaction tx, final Void result) { LOG.debug("Transaction {} completed successfully", tx); tx.onSuccess(result); processNextTransaction(tx); } - private void transactionFailed(final PingPongTransaction tx, final Throwable t) { + void transactionFailed(final PingPongTransaction tx, final Throwable t) { LOG.debug("Transaction {} failed", tx, t); tx.onFailure(t); processNextTransaction(tx); } - private void readyTransaction(@Nonnull final PingPongTransaction tx) { + void readyTransaction(@Nonnull final PingPongTransaction tx) { // First mark the transaction as not locked. final boolean lockedMatch = LOCKED_UPDATER.compareAndSet(this, tx, null); Preconditions.checkState(lockedMatch, "Attempted to submit transaction %s while we have %s", tx, lockedTx); @@ -283,6 +321,52 @@ public final class PingPongTransactionChain implements DOMTransactionChain { } } + /** + * Transaction cancellation is a heavyweight operation. We only support cancelation of a locked transaction + * and return false for everything else. Cancelling such a transaction will result in all transactions in the + * batch to be cancelled. + * + * @param tx Backend shared transaction + * @param frontendTx + * @param isOpen indicator whether the transaction was already closed + * @return True if cancellation succeeded, false otherwise + */ + synchronized void cancelTransaction(final PingPongTransaction tx, final DOMDataReadWriteTransaction frontendTx) { + // Attempt to unlock the operation. + final boolean lockedMatch = LOCKED_UPDATER.compareAndSet(this, tx, null); + Verify.verify(lockedMatch, "Cancelling transaction {} collided with locked transaction {}", tx, lockedTx); + + // Cancel the backend transaction, so we do not end up leaking it. + final boolean backendCancelled = tx.getTransaction().cancel(); + + if (failed) { + // The transaction has failed, this is probably the user just clearing up the transaction they had. We have + // already cancelled the transaction anyway, + return; + } else if (!backendCancelled) { + LOG.warn("Backend transaction cannot be cancelled during cancellation of {}, attempting to continue", tx); + } + + // We have dealt with canceling the backend transaction and have unlocked the transaction. Since we are still + // inside the synchronized block, any allocations are blocking on the slow path. Now we have to decide the fate + // of this transaction chain. + // + // If there are no other frontend transactions in this batch we are aligned with backend state and we can + // continue processing. + if (frontendTx.equals(tx.getFrontendTransaction())) { + LOG.debug("Cancelled transaction {} was head of the batch, resuming processing", tx); + return; + } + + // There are multiple frontend transactions in this batch. We have to report them as failed, which dooms this + // transaction chain, too. Since we just came off of a locked transaction, we do not have a ready transaction + // at the moment, but there may be some transaction in-flight. So we proceed to shutdown the backend chain + // and mark the fact that we should be turning its completion into a failure. + deadTx = new SimpleImmutableEntry<>(tx, + new CancellationException("Transaction " + frontendTx + " canceled").fillInStackTrace()); + delegate.close(); + } + @Override public synchronized void close() { final PingPongTransaction notLocked = lockedTx; @@ -292,6 +376,12 @@ public final class PingPongTransactionChain implements DOMTransactionChain { // the backend transaction chain will throw the appropriate error. Preconditions.checkState(shutdownTx == null, "Attempted to close an already-closed chain"); + // This may be a reaction to our failure callback, in that case the backend is already shutdown + if (deadTx != null) { + LOG.debug("Delegate {} is already closed due to failure {}", delegate, deadTx); + return; + } + // Force allocations on slow path, picking up a potentially-outstanding transaction final PingPongTransaction tx = READY_UPDATER.getAndSet(this, null); @@ -346,6 +436,8 @@ public final class PingPongTransactionChain implements DOMTransactionChain { public DOMDataReadWriteTransaction newReadWriteTransaction() { final PingPongTransaction tx = allocateTransaction(); final DOMDataReadWriteTransaction ret = new ForwardingDOMDataReadWriteTransaction() { + private boolean isOpen; + @Override protected DOMDataReadWriteTransaction delegate() { return tx.getTransaction(); @@ -354,18 +446,27 @@ public final class PingPongTransactionChain implements DOMTransactionChain { @Override public CheckedFuture submit() { readyTransaction(tx); + isOpen = false; return tx.getSubmitFuture(); } + @Deprecated @Override public ListenableFuture> commit() { readyTransaction(tx); + isOpen = false; return tx.getCommitFuture(); } @Override public boolean cancel() { - throw new UnsupportedOperationException("Transaction cancellation is not supported"); + if (isOpen) { + cancelTransaction(tx, this); + isOpen = false; + return true; + } else { + return false; + } } }; -- 2.36.6