X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatabroker%2Factors%2Fdds%2FAbstractProxyTransaction.java;h=0ba660234a15cee42c84f1c00b61d2ad2ef244e6;hb=refs%2Fchanges%2F50%2F49250%2F5;hp=adfc0df876cb6180961ee535519dd44aa1e946ee;hpb=320a4e5cd2d9d80468a3f82798744f2035488218;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 adfc0df876..0ba660234a 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
@@ -10,12 +10,23 @@ package org.opendaylight.controller.cluster.databroker.actors.dds;
 import akka.actor.ActorRef;
 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.util.concurrent.CheckedFuture;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.function.Consumer;
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+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.TransactionAbortRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
 import org.opendaylight.controller.cluster.access.commands.TransactionCanCommitSuccess;
@@ -24,7 +35,7 @@ import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRe
 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitSuccess;
 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
-import org.opendaylight.controller.cluster.access.concepts.RequestException;
+import org.opendaylight.controller.cluster.access.concepts.Request;
 import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
 import org.opendaylight.controller.cluster.access.concepts.Response;
 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
@@ -49,13 +60,124 @@ import org.slf4j.LoggerFactory;
  * @author Robert Varga
  */
 abstract class AbstractProxyTransaction implements Identifiable<TransactionIdentifier> {
-    private static final Logger LOG = LoggerFactory.getLogger(AbstractProxyTransaction.class);
+    /**
+     * Marker object used instead of read-type of requests, which are satisfied only once. This has a lower footprint
+     * and allows compressing multiple requests into a single entry.
+     */
+    @NotThreadSafe
+    private static final class IncrementSequence {
+        private long delta = 1;
+
+        long getDelta() {
+            return delta;
+        }
+
+        void incrementDelta() {
+            delta++;
+        }
+    }
+
+    // Generic state base class. Direct instances are used for fast paths, sub-class is used for successor transitions
+    private static class State {
+        private final String string;
+
+        State(final String string) {
+            this.string = Preconditions.checkNotNull(string);
+        }
+
+        @Override
+        public final String toString() {
+            return string;
+        }
+    }
+
+    // State class used when a successor has interfered. Contains coordinator latch, the successor and previous state
+    private static final class SuccessorState extends State {
+        private final CountDownLatch latch = new CountDownLatch(1);
+        private AbstractProxyTransaction successor;
+        private State prevState;
+
+        SuccessorState() {
+            super("successor");
+        }
+
+        // Synchronize with succession process and return the successor
+        AbstractProxyTransaction await() {
+            try {
+                latch.await();
+            } catch (InterruptedException e) {
+                LOG.warn("Interrupted while waiting for latch of {}", successor);
+                throw Throwables.propagate(e);
+            }
+            return successor;
+        }
+
+        void finish() {
+            latch.countDown();
+        }
+
+        State getPrevState() {
+            return prevState;
+        }
 
+        void setPrevState(final State prevState) {
+            Verify.verify(this.prevState == null);
+            this.prevState = Preconditions.checkNotNull(prevState);
+        }
+
+        // To be called from safe contexts, where successor is known to be completed
+        AbstractProxyTransaction getSuccessor() {
+            return Verify.verifyNotNull(successor);
+        }
+
+        void setSuccessor(final AbstractProxyTransaction successor) {
+            Verify.verify(this.successor == null);
+            this.successor = Preconditions.checkNotNull(successor);
+        }
+    }
+
+    private static final Logger LOG = LoggerFactory.getLogger(AbstractProxyTransaction.class);
+    private static final AtomicIntegerFieldUpdater<AbstractProxyTransaction> SEALED_UPDATER =
+            AtomicIntegerFieldUpdater.newUpdater(AbstractProxyTransaction.class, "sealed");
+    private static final AtomicReferenceFieldUpdater<AbstractProxyTransaction, State> STATE_UPDATER =
+            AtomicReferenceFieldUpdater.newUpdater(AbstractProxyTransaction.class, State.class, "state");
+    private static final State OPEN = new State("open");
+    private static final State SEALED = new State("sealed");
+    private static final State FLUSHED = new State("flushed");
+
+    // Touched from client actor thread only
+    private final Deque<Object> successfulRequests = new ArrayDeque<>();
     private final ProxyHistory parent;
 
-    private AbstractProxyTransaction successor;
+    // Accessed from user thread only, which may not access this object concurrently
     private long sequence;
-    private boolean sealed;
+
+    /*
+     * Atomic state-keeping is required to synchronize the process of propagating completed transaction state towards
+     * the backend -- which may include a successor.
+     *
+     * Successor, unlike {@link AbstractProxyTransaction#seal()} is triggered from the client actor thread, which means
+     * the successor placement needs to be atomic with regard to the application thread.
+     *
+     * In the common case, the application thread performs performs the seal operations and then "immediately" sends
+     * the corresponding message. The uncommon case is when the seal and send operations race with a connect completion
+     * or timeout, when a successor is injected.
+     *
+     * This leaves the problem of needing to completely transferring state just after all queued messages are replayed
+     * after a successor was injected, so that it can be properly sealed if we are racing. Further complication comes
+     * from lock ordering, where the successor injection works with a locked queue and locks proxy objects -- leading
+     * to a potential AB-BA deadlock in case of a naive implementation.
+     *
+     * For tracking user-visible state we use a single volatile int, which is flipped atomically from 0 to 1 exactly
+     * once in {@link AbstractProxyTransaction#seal()}. That keeps common operations fast, as they need to perform
+     * only a single volatile read to assert state correctness.
+     *
+     * For synchronizing client actor (successor-injecting) and user (commit-driving) thread, we keep a separate state
+     * 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;
 
     AbstractProxyTransaction(final ProxyHistory parent) {
         this.parent = Preconditions.checkNotNull(parent);
@@ -65,21 +187,31 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         return parent.localActor();
     }
 
+    private void incrementSequence(final long delta) {
+        sequence += delta;
+        LOG.debug("Transaction {} incremented sequence to {}", this, sequence);
+    }
+
     final long nextSequence() {
-        return sequence++;
+        final long ret = sequence++;
+        LOG.debug("Transaction {} allocated sequence {}", this, ret);
+        return ret;
     }
 
     final void delete(final YangInstanceIdentifier path) {
+        checkReadWrite();
         checkNotSealed();
         doDelete(path);
     }
 
     final void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
+        checkReadWrite();
         checkNotSealed();
         doMerge(path, data);
     }
 
     final void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
+        checkReadWrite();
         checkNotSealed();
         doWrite(path, data);
     }
@@ -103,18 +235,66 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * Seal this transaction before it is either committed or aborted.
      */
     final void seal() {
-        checkNotSealed();
+        // Transition user-visible state first
+        final boolean success = SEALED_UPDATER.compareAndSet(this, 0, 1);
+        Preconditions.checkState(success, "Proxy %s was already sealed", getIdentifier());
+        internalSeal();
+    }
+
+    final void ensureSealed() {
+        if (SEALED_UPDATER.compareAndSet(this, 0, 1)) {
+            internalSeal();
+        }
+    }
+
+    private void internalSeal() {
         doSeal();
-        sealed = true;
         parent.onTransactionSealed(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();
+
+            // 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();
+        }
     }
 
     private void checkNotSealed() {
-        Preconditions.checkState(!sealed, "Transaction %s has already been sealed", getIdentifier());
+        Preconditions.checkState(sealed == 0, "Transaction %s has already been sealed", getIdentifier());
     }
 
     private void checkSealed() {
-        Preconditions.checkState(sealed, "Transaction %s has not been sealed yet", getIdentifier());
+        Preconditions.checkState(sealed != 0, "Transaction %s has not been sealed yet", getIdentifier());
+    }
+
+    private SuccessorState getSuccessorState() {
+        final State local = state;
+        Verify.verify(local instanceof SuccessorState, "State %s has unexpected class", local);
+        return (SuccessorState) local;
+    }
+
+    private void checkReadWrite() {
+        if (isSnapshotOnly()) {
+            throw new UnsupportedOperationException("Transaction " + getIdentifier() + " is a read-only snapshot");
+        }
+    }
+
+    final void recordSuccessfulRequest(final @Nonnull TransactionRequest<?> req) {
+        successfulRequests.add(Verify.verifyNotNull(req));
+    }
+
+    final void recordFinishedRequest() {
+        final Object last = successfulRequests.peekLast();
+        if (last instanceof IncrementSequence) {
+            ((IncrementSequence) last).incrementDelta();
+        } else {
+            successfulRequests.addLast(new IncrementSequence());
+        }
     }
 
     /**
@@ -139,6 +319,8 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
             }
 
+            // This is a terminal request, hence we do not need to record it
+            LOG.debug("Transaction {} abort completed", this);
             parent.completeTransaction(this);
         });
     }
@@ -154,42 +336,76 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * @return Future completion
      */
     final ListenableFuture<Boolean> directCommit() {
+        checkReadWrite();
         checkSealed();
 
-        final SettableFuture<Boolean> ret = SettableFuture.create();
-        sendRequest(Verify.verifyNotNull(commitRequest(false)), t -> {
-            if (t instanceof TransactionCommitSuccess) {
-                ret.set(Boolean.TRUE);
-            } else if (t instanceof RequestFailure) {
-                ret.setException(((RequestFailure<?, ?>) t).getCause());
-            } else {
-                ret.setException(new IllegalStateException("Unhandled response " + t.getClass()));
+        // Precludes startReconnect() from interfering with the fast path
+        synchronized (this) {
+            if (STATE_UPDATER.compareAndSet(this, SEALED, FLUSHED)) {
+                final SettableFuture<Boolean> ret = SettableFuture.create();
+                sendRequest(Verify.verifyNotNull(commitRequest(false)), t -> {
+                    if (t instanceof TransactionCommitSuccess) {
+                        ret.set(Boolean.TRUE);
+                    } else if (t instanceof RequestFailure) {
+                        ret.setException(((RequestFailure<?, ?>) t).getCause());
+                    } else {
+                        ret.setException(new IllegalStateException("Unhandled response " + t.getClass()));
+                    }
+
+                    // This is a terminal request, hence we do not need to record it
+                    LOG.debug("Transaction {} directCommit completed", this);
+                    parent.completeTransaction(this);
+                });
+
+                return ret;
             }
+        }
 
-            parent.completeTransaction(this);
-        });
-        return ret;
+        // We have had some interference with successor injection, wait for it to complete and defer to the successor.
+        return awaitSuccessor().directCommit();
     }
 
-
-    void canCommit(final VotingFuture<?> ret) {
+    final void canCommit(final VotingFuture<?> ret) {
+        checkReadWrite();
         checkSealed();
 
-        sendRequest(Verify.verifyNotNull(commitRequest(true)), t -> {
-            if (t instanceof TransactionCanCommitSuccess) {
-                ret.voteYes();
-            } else if (t instanceof RequestFailure) {
-                ret.voteNo(((RequestFailure<?, ?>) t).getCause());
-            } else {
-                ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
+        // Precludes startReconnect() from interfering with the fast path
+        synchronized (this) {
+            if (STATE_UPDATER.compareAndSet(this, SEALED, FLUSHED)) {
+                final TransactionRequest<?> req = Verify.verifyNotNull(commitRequest(true));
+
+                sendRequest(req, t -> {
+                    if (t instanceof TransactionCanCommitSuccess) {
+                        ret.voteYes();
+                    } else if (t instanceof RequestFailure) {
+                        ret.voteNo(((RequestFailure<?, ?>) t).getCause());
+                    } else {
+                        ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
+                    }
+
+                    recordSuccessfulRequest(req);
+                    LOG.debug("Transaction {} canCommit completed", this);
+                });
+
+                return;
             }
-        });
+        }
+
+        // We have had some interference with successor injection, wait for it to complete and defer to the successor.
+        awaitSuccessor().canCommit(ret);
     }
 
-    void preCommit(final VotingFuture<?> ret) {
+    private AbstractProxyTransaction awaitSuccessor() {
+        return getSuccessorState().await();
+    }
+
+    final void preCommit(final VotingFuture<?> ret) {
+        checkReadWrite();
         checkSealed();
 
-        sendRequest(new TransactionPreCommitRequest(getIdentifier(), nextSequence(), localActor()), t -> {
+        final TransactionRequest<?> req = new TransactionPreCommitRequest(getIdentifier(), nextSequence(),
+            localActor());
+        sendRequest(req, t -> {
             if (t instanceof TransactionPreCommitSuccess) {
                 ret.voteYes();
             } else if (t instanceof RequestFailure) {
@@ -197,10 +413,14 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             } else {
                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
             }
+
+            recordSuccessfulRequest(req);
+            LOG.debug("Transaction {} preCommit completed", this);
         });
     }
 
-    void doCommit(final VotingFuture<?> ret) {
+    final void doCommit(final VotingFuture<?> ret) {
+        checkReadWrite();
         checkSealed();
 
         sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), t -> {
@@ -212,12 +432,81 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
             }
 
+            LOG.debug("Transaction {} doCommit completed", this);
             parent.completeTransaction(this);
         });
     }
 
-    void replaySuccessfulRequests(final AbstractProxyTransaction successor) {
-        this.successor = Preconditions.checkNotNull(successor);
+    // 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
+        // state. This method is called with the queue still unlocked.
+        final SuccessorState nextState = new SuccessorState();
+        final State prevState = STATE_UPDATER.getAndSet(this, nextState);
+
+        LOG.debug("Start reconnect of proxy {} previous state {}", this, prevState);
+        Verify.verify(!(prevState instanceof SuccessorState), "Proxy %s duplicate reconnect attempt after %s", this,
+            prevState);
+
+        // We have asserted a slow-path state, seal(), canCommit(), directCommit() are forced to slow paths, which will
+        // wait until we unblock nextState's latch before accessing state. Now we record prevState for later use and we
+        // are done.
+        nextState.setPrevState(prevState);
+    }
+
+    // Called with the connection locked
+    final void replayMessages(final AbstractProxyTransaction successor,
+            final Iterable<ConnectionEntry> enqueuedEntries) {
+        final SuccessorState local = getSuccessorState();
+        local.setSuccessor(successor);
+
+        // Replay successful requests first
+        for (Object obj : successfulRequests) {
+            if (obj instanceof TransactionRequest) {
+                LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
+                successor.handleForwardedRemoteRequest((TransactionRequest<?>) obj, null);
+            } else {
+                Verify.verify(obj instanceof IncrementSequence);
+                successor.incrementSequence(((IncrementSequence) obj).getDelta());
+            }
+        }
+        LOG.debug("{} replayed {} successful requests", getIdentifier(), successfulRequests.size());
+        successfulRequests.clear();
+
+        // Now replay whatever is in the connection
+        final Iterator<ConnectionEntry> it = enqueuedEntries.iterator();
+        while (it.hasNext()) {
+            final ConnectionEntry e = it.next();
+            final Request<?, ?> req = e.getRequest();
+
+            if (getIdentifier().equals(req.getTarget())) {
+                Verify.verify(req instanceof TransactionRequest, "Unhandled request %s", req);
+                LOG.debug("Forwarding queued request{} to successor {}", req, successor);
+                successor.handleForwardedRemoteRequest((TransactionRequest<?>) req, e.getCallback());
+                it.remove();
+            }
+        }
+
+        /*
+         * Check the state at which we have started the reconnect attempt. State transitions triggered while we were
+         * reconnecting have been forced to slow paths, which will be unlocked once we unblock the state latch
+         * at the end of this method.
+         */
+        final State prevState = local.getPrevState();
+        if (SEALED.equals(prevState)) {
+            LOG.debug("Proxy {} reconnected while being sealed, propagating state to successor {}", this, successor);
+            flushState(successor);
+            successor.ensureSealed();
+        }
+    }
+
+    // Called with the connection locked
+    final void finishReconnect() {
+        final SuccessorState local = getSuccessorState();
+        LOG.debug("Finishing reconnect of proxy {}", this);
+
+        // All done, release the latch, unblocking seal() and canCommit() slow paths
+        local.finish();
     }
 
     /**
@@ -226,11 +515,9 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      *
      * @param request Request to be forwarded
      * @param callback Original callback
-     * @throws RequestException when the request is unhandled by the successor
      */
-    final void replayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback)
-            throws RequestException {
-        Preconditions.checkState(successor != null, "%s does not have a successor set", this);
+    final void replayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
+        final AbstractProxyTransaction successor = getSuccessorState().getSuccessor();
 
         if (successor instanceof LocalProxyTransaction) {
             forwardToLocal((LocalProxyTransaction)successor, request, callback);
@@ -241,6 +528,8 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
     }
 
+    abstract boolean isSnapshotOnly();
+
     abstract void doDelete(final YangInstanceIdentifier path);
 
     abstract void doMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data);
@@ -256,6 +545,9 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
 
     abstract void doAbort();
 
+    @GuardedBy("this")
+    abstract void flushState(AbstractProxyTransaction successor);
+
     abstract TransactionRequest<?> commitRequest(boolean coordinated);
 
     /**
@@ -276,11 +568,11 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * Replay a request originating in this proxy to a successor remote proxy.
      */
     abstract void forwardToRemote(RemoteProxyTransaction successor, TransactionRequest<?> request,
-            Consumer<Response<?, ?>> callback) throws RequestException;
+            Consumer<Response<?, ?>> callback);
 
     /**
      * Replay a request originating in this proxy to a successor local proxy.
      */
     abstract void forwardToLocal(LocalProxyTransaction successor, TransactionRequest<?> request,
-            Consumer<Response<?, ?>> callback) throws RequestException;
+            Consumer<Response<?, ?>> callback);
 }