BUG-6111: implement PingPongTransactionChain cancelation 58/43058/3
authorRobert Varga <rovarga@cisco.com>
Wed, 3 Aug 2016 15:15:57 +0000 (17:15 +0200)
committerTony Tkacik <ttkacik@cisco.com>
Mon, 8 Aug 2016 08:05:01 +0000 (08:05 +0000)
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 <rovarga@cisco.com>
opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/PingPongTransactionChain.java

index b792826..cbb5b0b 100644 (file)
@@ -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<PingPongTransaction, Throwable> 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<PingPongTransaction, Throwable> 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<Void, TransactionCommitFailedException> submit() {
                 readyTransaction(tx);
+                isOpen = false;
                 return tx.getSubmitFuture();
             }
 
+            @Deprecated
             @Override
             public ListenableFuture<RpcResult<TransactionStatus>> 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;
+                }
             }
         };