X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatabroker%2Factors%2Fdds%2FAbstractProxyTransaction.java;h=ac91f2d0413ad8aca71391ba86921c92dc1b008b;hp=8e4c757d33767877bb2065a3de45ab201d1bfaf7;hb=9b4c07ca27b2c3c7ca5d5e790aa1f121ce4857f3;hpb=7f15e81c52f2efda779c670580f0697227557404

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 8e4c757d33..ac91f2d041 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
@@ -91,7 +91,10 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
     }
 
-    // Generic state base class. Direct instances are used for fast paths, sub-class is used for successor transitions
+    /**
+     * Base class for representing logical state of this proxy. See individual instantiations and {@link SuccessorState}
+     * for details.
+     */
     private static class State {
         private final String string;
 
@@ -105,14 +108,27 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
     }
 
-    // State class used when a successor has interfered. Contains coordinator latch, the successor and previous state
-    private static final class SuccessorState extends State {
+    /**
+     * State class used when a successor has interfered. Contains coordinator latch, the successor and previous state.
+     * This is a temporary state introduced during reconnection process and is necessary for correct state hand-off
+     * between the old connection (potentially being accessed by the user) and the new connection (being cleaned up
+     * by the actor.
+     *
+     * <p>
+     * When a user operation encounters this state, it synchronizes on the it and wait until reconnection completes,
+     * 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 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");
+            super("SUCCESSOR");
         }
 
         // Synchronize with succession process and return the successor
@@ -135,7 +151,8 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
 
         void setPrevState(final State prevState) {
-            Verify.verify(this.prevState == null);
+            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);
         }
 
@@ -145,9 +162,18 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
 
         void setSuccessor(final AbstractProxyTransaction successor) {
-            Verify.verify(this.successor == null);
+            Verify.verify(this.successor == null, "Attempted to set successor to %s when we already have %s",
+                    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);
@@ -155,9 +181,41 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             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");
+
+    /**
+     * Transaction has been open and is being actively worked on.
+     */
+    private static final State OPEN = new State("OPEN");
+
+    /**
+     * Transaction has been sealed by the user, but it has not completed flushing to the backed, yet. This is
+     * a transition state, as we are waiting for the user to initiate commit procedures.
+     *
+     * <p>
+     * Since the reconnect mechanics relies on state replay for transactions, this state needs to be flushed into the
+     * queue to re-create state in successor transaction (which may be based on different messages as locality may have
+     * changed). Hence the transition to {@link #FLUSHED} state needs to be handled in a thread-safe manner.
+     */
+    private static final State SEALED = new State("SEALED");
+
+    /**
+     * Transaction state has been flushed into the queue, i.e. it is visible by the successor and potentially
+     * the backend. At this point the transaction does not hold any state besides successful requests, all other state
+     * is held either in the connection's queue or the successor object.
+     *
+     * <p>
+     * Transition to this state indicates we have all input from the user we need to initiate the correct commit
+     * protocol.
+     */
+    private static final State FLUSHED = new State("FLUSHED");
+
+    /**
+     * Transaction state has been completely resolved, we have received confirmation of the transaction fate from
+     * the backend. The only remaining task left to do is finishing up the state cleanup, which is done via purge
+     * request. We need to hang on to the transaction until that is done, as we have to make sure backend completes
+     * purging its state -- otherwise we could have a leak on the backend.
+     */
+    private static final State DONE = new State("DONE");
 
     // Touched from client actor thread only
     private final Deque<Object> successfulRequests = new ArrayDeque<>();
@@ -197,6 +255,13 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         this.parent = Preconditions.checkNotNull(parent);
     }
 
+    final void executeInActor(final Runnable command) {
+        parent.context().executeInActor(behavior -> {
+            command.run();
+            return behavior;
+        });
+    }
+
     final ActorRef localActor() {
         return parent.localActor();
     }
@@ -323,14 +388,18 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      */
     final void abort() {
         checkNotSealed();
-        doAbort();
         parent.abortTransaction(this);
+
+        sendRequest(abortRequest(), resp -> {
+            LOG.debug("Transaction {} abort completed with {}", getIdentifier(), resp);
+            enqueuePurge();
+        });
     }
 
     final void abort(final VotingFuture<Void> ret) {
         checkSealed();
 
-        sendAbort(t -> {
+        sendDoAbort(t -> {
             if (t instanceof TransactionAbortSuccess) {
                 ret.voteYes();
             } else if (t instanceof RequestFailure) {
@@ -341,16 +410,29 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
 
             // This is a terminal request, hence we do not need to record it
             LOG.debug("Transaction {} abort completed", this);
-            sendPurge();
+            enqueuePurge();
         });
     }
 
     final void enqueueAbort(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+        checkNotSealed();
+        parent.abortTransaction(this);
+
+        enqueueRequest(abortRequest(), resp -> {
+            LOG.debug("Transaction {} abort completed with {}", getIdentifier(), resp);
+            // Purge will be sent by the predecessor's callback
+            if (callback != null) {
+                callback.accept(resp);
+            }
+        }, enqueuedTicks);
+    }
+
+    final void enqueueDoAbort(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
         enqueueRequest(new TransactionAbortRequest(getIdentifier(), nextSequence(), localActor()), callback,
             enqueuedTicks);
     }
 
-    final void sendAbort(final Consumer<Response<?, ?>> callback) {
+    final void sendDoAbort(final Consumer<Response<?, ?>> callback) {
         sendRequest(new TransactionAbortRequest(getIdentifier(), nextSequence(), localActor()), callback);
     }
 
@@ -379,7 +461,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
 
                     // This is a terminal request, hence we do not need to record it
                     LOG.debug("Transaction {} directCommit completed", this);
-                    sendPurge();
+                    enqueuePurge();
                 });
 
                 return ret;
@@ -473,27 +555,45 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             }
 
             LOG.debug("Transaction {} doCommit completed", this);
-            sendPurge();
+
+            // Needed for ProxyHistory$Local data tree rebase points.
+            parent.completeTransaction(this);
+
+            enqueuePurge();
         });
     }
 
-    final void sendPurge() {
-        successfulRequests.clear();
+    private void enqueuePurge() {
+        enqueuePurge(null);
+    }
 
-        final TransactionRequest<?> req = new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor());
-        sendRequest(req, t -> {
-            LOG.debug("Transaction {} purge completed", this);
-            parent.completeTransaction(this);
-        });
+    final void enqueuePurge(final Consumer<Response<?, ?>> callback) {
+        // Purge request are dispatched internally, hence should not wait
+        enqueuePurge(callback, parent.currentTime());
     }
 
-    final void enqueuePurge(final long enqueuedTicks) {
+    final void enqueuePurge(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+        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();
 
-        final TransactionRequest<?> req = new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor());
-        enqueueRequest(req, t -> {
-            LOG.debug("Transaction {} purge completed", this);
-            parent.completeTransaction(this);
+        enqueueRequest(new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor()), resp -> {
+            LOG.debug("{}: purge completed", this);
+            parent.purgeTransaction(this);
+
+            if (callback != null) {
+                callback.accept(resp);
+            }
         }, enqueuedTicks);
     }
 
@@ -515,9 +615,13 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     }
 
     // Called with the connection locked
-    final void replayMessages(final AbstractProxyTransaction successor,
-            final Iterable<ConnectionEntry> enqueuedEntries) {
+    final void replayMessages(final ProxyHistory successorHistory, final Iterable<ConnectionEntry> enqueuedEntries) {
         final SuccessorState local = getSuccessorState();
+        final State prevState = local.getPrevState();
+
+        final AbstractProxyTransaction successor = successorHistory.createTransactionProxy(getIdentifier(),
+            isSnapshotOnly());
+        LOG.debug("{} created successor transaction proxy {}", this, successor);
         local.setSuccessor(successor);
 
         // Replay successful requests first
@@ -537,7 +641,8 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                     Verify.verify(obj instanceof IncrementSequence);
                     final IncrementSequence increment = (IncrementSequence) obj;
                     successor.replayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
-                        increment.getSequence(), localActor(), increment.getDelta()), resp -> { }, now);
+                        increment.getSequence(), localActor(), isSnapshotOnly(), increment.getDelta()), resp -> { },
+                        now);
                     LOG.debug("Incrementing sequence {} to successor {}", obj, successor);
                 }
             }
@@ -564,7 +669,6 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
          * 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);
@@ -637,11 +741,11 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
 
     abstract void doSeal();
 
-    abstract void doAbort();
-
     @GuardedBy("this")
     abstract void flushState(AbstractProxyTransaction successor);
 
+    abstract TransactionRequest<?> abortRequest();
+
     abstract TransactionRequest<?> commitRequest(boolean coordinated);
 
     /**