Optimize inflight transitions

Change-Id: Iaab4941ab777eaef1068523fc055391d31ca3072
Signed-off-by: Robert Varga <rovarga@cisco.com>

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 b3c03b3185efa421b3913aba39362ef6f701bba0..abd348a9c73404b79f0866160f336bdee30aab59 100644 (file)
--- 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
@@ -55,8 +55,6 @@ public final class PingPongTransactionChain implements DOMTransactionChain {
     private static final Logger LOG = LoggerFactory.getLogger(PingPongTransactionChain.class);
     private final DOMTransactionChain delegate;
 
     private static final Logger LOG = LoggerFactory.getLogger(PingPongTransactionChain.class);
     private final DOMTransactionChain delegate;
 

-    @GuardedBy("this")
-    private PingPongTransaction inflightTransaction;

     @GuardedBy("this")
     private boolean failed;
 
     @GuardedBy("this")
     private boolean failed;
 


@@ -79,6 +77,14 @@ public final class PingPongTransactionChain implements DOMTransactionChain {
             AtomicReferenceFieldUpdater.newUpdater(PingPongTransactionChain.class, PingPongTransaction.class, "lockedTx");
     private volatile PingPongTransaction lockedTx;
 
             AtomicReferenceFieldUpdater.newUpdater(PingPongTransactionChain.class, PingPongTransaction.class, "lockedTx");
     private volatile PingPongTransaction lockedTx;
 

+    /**
+     * This updater is used to manipulate the "inflight" transaction. There can be at most
+     * one of these at any given time. We perform only compare-and-swap on these.
+     */
+    private static final AtomicReferenceFieldUpdater<PingPongTransactionChain, PingPongTransaction> INFLIGHT_UPDATER =
+            AtomicReferenceFieldUpdater.newUpdater(PingPongTransactionChain.class, PingPongTransaction.class, "inflightTx");
+    private volatile PingPongTransaction inflightTx;
+

     PingPongTransactionChain(final DOMDataBroker broker, final TransactionChainListener listener) {
         this.delegate = broker.createTransactionChain(new TransactionChainListener() {
             @Override
     PingPongTransactionChain(final DOMDataBroker broker, final TransactionChainListener listener) {
         this.delegate = broker.createTransactionChain(new TransactionChainListener() {
             @Override


@@ -86,14 +92,15 @@ public final class PingPongTransactionChain implements DOMTransactionChain {
                 LOG.debug("Delegate chain {} reported failure in {}", chain, transaction, cause);
 
                 final DOMDataReadWriteTransaction frontend;
                 LOG.debug("Delegate chain {} reported failure in {}", chain, transaction, cause);
 
                 final DOMDataReadWriteTransaction frontend;

-                if (inflightTransaction == null) {
+                final PingPongTransaction tx = inflightTx;
+                if (tx == null) {

                     LOG.warn("Transaction chain {} failed with no pending transactions", chain);
                     frontend = null;
                 } else {
                     LOG.warn("Transaction chain {} failed with no pending transactions", chain);
                     frontend = null;
                 } else {

-                    frontend = inflightTransaction.getFrontendTransaction();
+                    frontend = tx.getFrontendTransaction();

                 }
 
                 }
 

-                listener.onTransactionChainFailed(PingPongTransactionChain.this, frontend , cause);
+                listener.onTransactionChainFailed(PingPongTransactionChain.this, frontend, cause);

                 delegateFailed();
             }
 
                 delegateFailed();
             }
 


@@ -174,10 +181,11 @@ public final class PingPongTransactionChain implements DOMTransactionChain {
         }
 
         LOG.debug("Submitting transaction {}", tx);
         }
 
         LOG.debug("Submitting transaction {}", tx);

-        final CheckedFuture<Void, ?> f = tx.getTransaction().submit();
-        inflightTransaction = tx;
+        if (!INFLIGHT_UPDATER.compareAndSet(this, null, tx)) {
+            LOG.warn("Submitting transaction {} while {} is still running", tx, inflightTx);
+        }

 
 

-        Futures.addCallback(f, new FutureCallback<Void>() {
+        Futures.addCallback(tx.getTransaction().submit(), new FutureCallback<Void>() {

             @Override
             public void onSuccess(final Void result) {
                 transactionSuccessful(tx, result);
             @Override
             public void onSuccess(final Void result) {
                 transactionSuccessful(tx, result);


@@ -193,10 +201,10 @@ public final class PingPongTransactionChain implements DOMTransactionChain {
     private void transactionSuccessful(final PingPongTransaction tx, final Void result) {
         LOG.debug("Transaction {} completed successfully", tx);
 
     private void transactionSuccessful(final PingPongTransaction tx, final Void result) {
         LOG.debug("Transaction {} completed successfully", tx);
 

-        synchronized (this) {
-            Preconditions.checkState(inflightTransaction == tx, "Successful transaction %s while %s was submitted", tx, inflightTransaction);
+        final boolean success = INFLIGHT_UPDATER.compareAndSet(this, tx, null);
+        Preconditions.checkState(success, "Successful transaction %s while %s was submitted", tx, inflightTx);

 
 

-            inflightTransaction = null;
+        synchronized (this) {

             processIfReady();
         }
 
             processIfReady();
         }
 


@@ -207,10 +215,8 @@ public final class PingPongTransactionChain implements DOMTransactionChain {
     private void transactionFailed(final PingPongTransaction tx, final Throwable t) {
         LOG.debug("Transaction {} failed", tx, t);
 
     private void transactionFailed(final PingPongTransaction tx, final Throwable t) {
         LOG.debug("Transaction {} failed", tx, t);
 

-        synchronized (this) {
-            Preconditions.checkState(inflightTransaction == tx, "Failed transaction %s while %s was submitted", tx, inflightTransaction);
-            inflightTransaction = null;
-        }
+        final boolean success = INFLIGHT_UPDATER.compareAndSet(this, tx, null);
+        Preconditions.checkState(success, "Failed transaction %s while %s was submitted", tx, inflightTx);

 
         tx.onFailure(t);
     }
 
         tx.onFailure(t);
     }


@@ -221,8 +227,8 @@ public final class PingPongTransactionChain implements DOMTransactionChain {
 
         LOG.debug("Transaction {} unlocked", tx);
 
 
         LOG.debug("Transaction {} unlocked", tx);
 

-        synchronized (this) {
-            if (inflightTransaction == null) {
+        if (inflightTx == null) {
+            synchronized (this) {

                 processTransaction(tx);
             }
         }
                 processTransaction(tx);
             }
         }