X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-dom-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fmd%2Fsal%2Fdom%2Fbroker%2Fimpl%2FPingPongTransactionChain.java;h=b792826ec4887318736ca9e97329975cdd34af5e;hb=dceb9db7853dabfbd4abdfb3d886a79871097831;hp=c3a56ed454024b3cb316c1e5bec9d7184c897edc;hpb=a690f9b3faac1f44ca0d034bccb8dffe0e4c308a;p=controller.git 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 c3a56ed454..b792826ec4 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 @@ -57,6 +57,8 @@ public final class PingPongTransactionChain implements DOMTransactionChain { @GuardedBy("this") private boolean failed; + @GuardedBy("this") + private PingPongTransaction shutdownTx; /** * This updater is used to manipulate the "ready" transaction. We perform only atomic @@ -124,6 +126,8 @@ public final class PingPongTransactionChain implements DOMTransactionChain { } private synchronized PingPongTransaction slowAllocateTransaction() { + Preconditions.checkState(shutdownTx == null, "Transaction chain %s has been shut down", this); + final DOMDataReadWriteTransaction delegateTx = delegate.newReadWriteTransaction(); final PingPongTransaction newTx = new PingPongTransaction(delegateTx); @@ -158,13 +162,16 @@ public final class PingPongTransactionChain implements DOMTransactionChain { /* * This forces allocateTransaction() on a slow path, which has to happen after - * this method has completed executing. + * this method has completed executing. Also inflightTx may be updated outside + * the lock, hence we need to re-check. */ @GuardedBy("this") private void processIfReady() { - final PingPongTransaction tx = READY_UPDATER.getAndSet(this, null); - if (tx != null) { - processTransaction(tx); + if (inflightTx == null) { + final PingPongTransaction tx = READY_UPDATER.getAndSet(this, null); + if (tx != null) { + processTransaction(tx); + } } } @@ -175,7 +182,7 @@ public final class PingPongTransactionChain implements DOMTransactionChain { * @param tx Transaction which needs processing. */ @GuardedBy("this") - private void processTransaction(final @Nonnull PingPongTransaction tx) { + private void processTransaction(@Nonnull final PingPongTransaction tx) { if (failed) { LOG.debug("Cancelling transaction {}", tx); tx.getTransaction().cancel(); @@ -200,30 +207,56 @@ public final class PingPongTransactionChain implements DOMTransactionChain { }); } - private void transactionSuccessful(final PingPongTransaction tx, final Void result) { - LOG.debug("Transaction {} completed successfully", tx); - + /* + * We got invoked from the data store thread. We need to do two things: + * 1) release the in-flight transaction + * 2) process the potential next transaction + * + * We have to perform 2) under lock. We could perform 1) without locking, but that means the CAS result may + * not be accurate, as a user thread may submit the ready transaction before we acquire the lock -- and checking + * for next transaction is not enough, as that may have also be allocated (as a result of a quick + * submit/allocate/submit between 1) and 2)). Hence we'd end up doing the following: + * 1) CAS of inflightTx + * 2) take lock + * 3) volatile read of inflightTx + * + * Rather than doing that, we keep this method synchronized, hence performing only: + * 1) take lock + * 2) CAS of inflightTx + * + * Since the user thread is barred from submitting the transaction (in processIfReady), we can then proceed with + * the knowledge that inflightTx is null -- processTransaction() will still do a CAS, but that is only for + * correctness. + */ + private synchronized void processNextTransaction(final PingPongTransaction tx) { final boolean success = INFLIGHT_UPDATER.compareAndSet(this, tx, null); - Preconditions.checkState(success, "Successful transaction %s while %s was submitted", tx, inflightTx); + Preconditions.checkState(success, "Completed transaction %s while %s was submitted", tx, inflightTx); - synchronized (this) { - processIfReady(); + final PingPongTransaction nextTx = READY_UPDATER.getAndSet(this, null); + if (nextTx != null) { + processTransaction(nextTx); + } else if (shutdownTx != null) { + processTransaction(shutdownTx); + delegate.close(); + shutdownTx = null; } + } + + private void transactionSuccessful(final PingPongTransaction tx, final Void result) { + LOG.debug("Transaction {} completed successfully", tx); - // Can run unsynchronized tx.onSuccess(result); + processNextTransaction(tx); } private void transactionFailed(final PingPongTransaction tx, final Throwable t) { LOG.debug("Transaction {} failed", tx, t); - final boolean success = INFLIGHT_UPDATER.compareAndSet(this, tx, null); - Preconditions.checkState(success, "Failed transaction %s while %s was submitted", tx, inflightTx); - tx.onFailure(t); + processNextTransaction(tx); } - private void readyTransaction(final @Nonnull PingPongTransaction tx) { + private 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); @@ -235,7 +268,7 @@ public final class PingPongTransactionChain implements DOMTransactionChain { */ final boolean success = READY_UPDATER.compareAndSet(this, null, tx); Preconditions.checkState(success, "Transaction %s collided on ready state", tx, readyTx); - LOG.debug("Transaction {} readied"); + LOG.debug("Transaction {} readied", tx); /* * We do not see a transaction being in-flight, so we need to take care of dispatching @@ -251,12 +284,31 @@ public final class PingPongTransactionChain implements DOMTransactionChain { } @Override - public void close() { + public synchronized void close() { final PingPongTransaction notLocked = lockedTx; Preconditions.checkState(notLocked == null, "Attempted to close chain with outstanding transaction %s", notLocked); - synchronized (this) { - processIfReady(); + // This is not reliable, but if we observe it to be null and the process has already completed, + // the backend transaction chain will throw the appropriate error. + Preconditions.checkState(shutdownTx == null, "Attempted to close an already-closed chain"); + + // Force allocations on slow path, picking up a potentially-outstanding transaction + final PingPongTransaction tx = READY_UPDATER.getAndSet(this, null); + + if (tx != null) { + // We have one more transaction, which needs to be processed somewhere. If we do not + // a transaction in-flight, we need to push it down ourselves. + // If there is an in-flight transaction we will schedule this last one into a dedicated + // slot. Allocation slow path will check its presence and fail, the in-flight path will + // pick it up, submit and immediately close the chain. + if (inflightTx == null) { + processTransaction(tx); + delegate.close(); + } else { + shutdownTx = tx; + } + } else { + // Nothing outstanding, we can safely shutdown delegate.close(); } }