X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatabroker%2Factors%2Fdds%2FAbstractProxyTransaction.java;h=51f528150d22a717034cc4281618591645cca92f;hb=7991491f2854dde2ec625ed6c08b44df7d258795;hp=549cab269fb7682b4575c141243dda12b028d23f;hpb=4ddfbaf460366ef4194e15c19649039a7d616399;p=controller.git

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java
index 549cab269f..51f528150d 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java
@@ -11,7 +11,6 @@ import akka.actor.ActorRef;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
-import com.google.common.base.Throwables;
 import com.google.common.base.Verify;
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.CheckedFuture;
@@ -30,7 +29,9 @@ import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.NotThreadSafe;
 import org.opendaylight.controller.cluster.access.client.ConnectionEntry;
 import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.ClosedTransactionException;
 import org.opendaylight.controller.cluster.access.commands.IncrementTransactionSequenceRequest;
+import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
 import org.opendaylight.controller.cluster.access.commands.TransactionCanCommitSuccess;
@@ -119,11 +120,14 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * 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");
         }
@@ -134,7 +138,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                 latch.await();
             } catch (InterruptedException e) {
                 LOG.warn("Interrupted while waiting for latch of {}", successor);
-                throw Throwables.propagate(e);
+                throw new RuntimeException(e);
             }
             return successor;
         }
@@ -144,13 +148,15 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
 
         State getPrevState() {
-            return prevState;
+            return Verify.verifyNotNull(prevState, "Attempted to access previous state, which was not set");
         }
 
         void setPrevState(final State prevState) {
             Verify.verify(this.prevState == null, "Attempted to set previous state to %s when we already have %s",
                     prevState, this.prevState);
             this.prevState = Preconditions.checkNotNull(prevState);
+            // We cannot have duplicate successor states, so this check is sufficient
+            this.done = DONE.equals(prevState);
         }
 
         // To be called from safe contexts, where successor is known to be completed
@@ -163,6 +169,14 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                     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);
@@ -237,11 +251,18 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * variable. It uses pre-allocated objects for fast paths (i.e. no successor present) and a per-transition object
      * for slow paths (when successor is injected/present).
      */
-    private volatile int sealed = 0;
-    private volatile State state = OPEN;
+    private volatile int sealed;
+    private volatile State state;
 
-    AbstractProxyTransaction(final ProxyHistory parent) {
+    AbstractProxyTransaction(final ProxyHistory parent, final boolean isDone) {
         this.parent = Preconditions.checkNotNull(parent);
+        if (isDone) {
+            state = DONE;
+            // DONE implies previous seal operation completed
+            sealed = 1;
+        } else {
+            state = OPEN;
+        }
     }
 
     final void executeInActor(final Runnable command) {
@@ -306,36 +327,70 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     }
 
     /**
-     * Seal this transaction before it is either committed or aborted.
+     * Seal this transaction before it is either committed or aborted. This method should only be invoked from
+     * application thread.
      */
     final void seal() {
         // Transition user-visible state first
-        final boolean success = SEALED_UPDATER.compareAndSet(this, 0, 1);
+        final boolean success = markSealed();
         Preconditions.checkState(success, "Proxy %s was already sealed", getIdentifier());
-        internalSeal();
+
+        if (!sealAndSend(Optional.absent())) {
+            sealSuccessor();
+        }
     }
 
-    final void ensureSealed() {
-        if (SEALED_UPDATER.compareAndSet(this, 0, 1)) {
-            internalSeal();
+    /**
+     * Internal seal propagation method, invoked when we have raced with reconnection thread. Note that there may have
+     * been multiple reconnects, so we have to make sure the action is propagate through all intermediate instances.
+     */
+    private void sealSuccessor() {
+        // Slow path: wait for the successor to complete
+        final AbstractProxyTransaction successor = awaitSuccessor();
+
+        // At this point the successor has completed transition and is possibly visible by the user thread, which is
+        // still stuck here. The successor has not seen final part of our state, nor the fact it is sealed.
+        // Propagate state and seal the successor.
+        final java.util.Optional<ModifyTransactionRequest> optState = flushState();
+        if (optState.isPresent()) {
+            forwardToSuccessor(successor, optState.get(), null);
         }
+        successor.predecessorSealed();
     }
 
-    private void internalSeal() {
-        doSeal();
+    private void predecessorSealed() {
+        if (markSealed() && !sealAndSend(Optional.absent())) {
+            sealSuccessor();
+        }
+    }
+
+    void sealOnly() {
         parent.onTransactionSealed(this);
+        final boolean success = STATE_UPDATER.compareAndSet(this, OPEN, SEALED);
+        Verify.verify(success, "Attempted to replay seal on %s", this);
+    }
 
-        // Now deal with state transfer, which can occur via successor or a follow-up canCommit() or directCommit().
-        if (!STATE_UPDATER.compareAndSet(this, OPEN, SEALED)) {
-            // Slow path: wait for the successor to complete
-            final AbstractProxyTransaction successor = awaitSuccessor();
+    /**
+     * Seal this transaction and potentially send it out towards the backend. If this method reports false, the caller
+     * needs to deal with propagating the seal operation towards the successor.
+     *
+     * @param enqueuedTicks Enqueue ticks when this is invoked from replay path.
+     * @return True if seal operation was successful, false if this proxy has a successor.
+     */
+    boolean sealAndSend(final Optional<Long> enqueuedTicks) {
+        parent.onTransactionSealed(this);
 
-            // At this point the successor has completed transition and is possibly visible by the user thread, which is
-            // still stuck here. The successor has not seen final part of our state, nor the fact it is sealed.
-            // Propagate state and seal the successor.
-            flushState(successor);
-            successor.ensureSealed();
-        }
+        // Transition internal state to sealed and detect presence of a successor
+        return STATE_UPDATER.compareAndSet(this, OPEN, SEALED);
+    }
+
+    /**
+     * Mark this proxy as having been sealed.
+     *
+     * @return True if this call has transitioned to sealed state.
+     */
+    final boolean markSealed() {
+        return SEALED_UPDATER.compareAndSet(this, 0, 1);
     }
 
     private void checkNotSealed() {
@@ -443,7 +498,14 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                     if (t instanceof TransactionCommitSuccess) {
                         ret.set(Boolean.TRUE);
                     } else if (t instanceof RequestFailure) {
-                        ret.setException(((RequestFailure<?, ?>) t).getCause().unwrap());
+                        final Throwable cause = ((RequestFailure<?, ?>) t).getCause().unwrap();
+                        if (cause instanceof ClosedTransactionException) {
+                            // This is okay, as it indicates the transaction has been completed. It can happen
+                            // when we lose connectivity with the backend after it has received the request.
+                            ret.set(Boolean.TRUE);
+                        } else {
+                            ret.setException(cause);
+                        }
                     } else {
                         ret.setException(new IllegalStateException("Unhandled response " + t.getClass()));
                     }
@@ -544,6 +606,10 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             }
 
             LOG.debug("Transaction {} doCommit completed", this);
+
+            // Needed for ProxyHistory$Local data tree rebase points.
+            parent.completeTransaction(this);
+
             enqueuePurge();
         });
     }
@@ -558,20 +624,30 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     }
 
     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
@@ -595,8 +671,8 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         final State prevState = local.getPrevState();
 
         final AbstractProxyTransaction successor = successorHistory.createTransactionProxy(getIdentifier(),
-            isSnapshotOnly());
-        LOG.debug("{} created successor transaction proxy {}", this, successor);
+            isSnapshotOnly(), local.isDone());
+        LOG.debug("{} created successor {}", this, successor);
         local.setSuccessor(successor);
 
         // Replay successful requests first
@@ -611,11 +687,11 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             for (Object obj : successfulRequests) {
                 if (obj instanceof TransactionRequest) {
                     LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
-                    successor.replayRequest((TransactionRequest<?>) obj, resp -> { }, now);
+                    successor.doReplayRequest((TransactionRequest<?>) obj, resp -> { }, now);
                 } else {
                     Verify.verify(obj instanceof IncrementSequence);
                     final IncrementSequence increment = (IncrementSequence) obj;
-                    successor.replayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
+                    successor.doReplayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
                         increment.getSequence(), localActor(), isSnapshotOnly(), increment.getDelta()), resp -> { },
                         now);
                     LOG.debug("Incrementing sequence {} to successor {}", obj, successor);
@@ -634,7 +710,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             if (getIdentifier().equals(req.getTarget())) {
                 Verify.verify(req instanceof TransactionRequest, "Unhandled request %s", req);
                 LOG.debug("Replaying queued request {} to successor {}", req, successor);
-                successor.replayRequest((TransactionRequest<?>) req, e.getCallback(), e.getEnqueuedTicks());
+                successor.doReplayRequest((TransactionRequest<?>) req, e.getCallback(), e.getEnqueuedTicks());
                 it.remove();
             }
         }
@@ -646,8 +722,14 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
          */
         if (SEALED.equals(prevState)) {
             LOG.debug("Proxy {} reconnected while being sealed, propagating state to successor {}", this, successor);
-            flushState(successor);
-            successor.ensureSealed();
+            final long enqueuedTicks = parent.currentTime();
+            final java.util.Optional<ModifyTransactionRequest> optState = flushState();
+            if (optState.isPresent()) {
+                successor.handleReplayedRemoteRequest(optState.get(), null, enqueuedTicks);
+            }
+            if (successor.markSealed()) {
+                successor.sealAndSend(Optional.of(enqueuedTicks));
+            }
         }
     }
 
@@ -662,7 +744,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * @param callback Callback to be invoked once the request completes
      * @param enqueuedTicks ticker-based time stamp when the request was enqueued
      */
-    private void replayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
+    private void doReplayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
             final long enqueuedTicks) {
         if (request instanceof AbstractLocalTransactionRequest) {
             handleReplayedLocalRequest((AbstractLocalTransactionRequest<?>) request, callback, enqueuedTicks);
@@ -702,6 +784,11 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
     }
 
+    final void replayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
+            final long enqueuedTicks) {
+        getSuccessorState().getSuccessor().doReplayRequest(request, callback, enqueuedTicks);
+    }
+
     abstract boolean isSnapshotOnly();
 
     abstract void doDelete(YangInstanceIdentifier path);
@@ -714,10 +801,8 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
 
     abstract CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(YangInstanceIdentifier path);
 
-    abstract void doSeal();
-
     @GuardedBy("this")
-    abstract void flushState(AbstractProxyTransaction successor);
+    abstract java.util.Optional<ModifyTransactionRequest> flushState();
 
     abstract TransactionRequest<?> abortRequest();