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=e1919d1bb6c2a601486005425c5e5f6fe75423b4;hp=07b89e09230949da6c4849b3fb5dc03d4c3c36d8;hb=127042ea7e148d9dc0282acc3780b4754ca69e12;hpb=18ddbfdc55a1faddf7aeb2df6b25481d34c820ab

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 07b89e0923..e1919d1bb6 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
@@ -9,29 +9,29 @@ package org.opendaylight.controller.cluster.databroker.actors.dds;
 
 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;
+import com.google.common.util.concurrent.FluentFuture;
 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.Optional;
 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.eclipse.jdt.annotation.NonNull;
+import org.eclipse.jdt.annotation.Nullable;
 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;
@@ -45,7 +45,6 @@ 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;
-import org.opendaylight.mdsal.common.api.ReadFailedException;
 import org.opendaylight.yangtools.concepts.Identifiable;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
@@ -138,7 +137,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;
         }
@@ -305,12 +304,12 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         doWrite(path, data);
     }
 
-    final CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
+    final FluentFuture<Boolean> exists(final YangInstanceIdentifier path) {
         checkNotSealed();
         return doExists(path);
     }
 
-    final CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
+    final FluentFuture<Optional<NormalizedNode<?, ?>>> read(final YangInstanceIdentifier path) {
         checkNotSealed();
         return doRead(path);
     }
@@ -335,7 +334,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         final boolean success = markSealed();
         Preconditions.checkState(success, "Proxy %s was already sealed", getIdentifier());
 
-        if (!sealAndSend(Optional.absent())) {
+        if (!sealAndSend(Optional.empty())) {
             sealSuccessor();
         }
     }
@@ -351,20 +350,27 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         // 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);
+        final Optional<ModifyTransactionRequest> optState = flushState();
+        if (optState.isPresent()) {
+            forwardToSuccessor(successor, optState.get(), null);
+        }
         successor.predecessorSealed();
     }
 
     private void predecessorSealed() {
-        if (markSealed() && !sealAndSend(Optional.absent())) {
+        if (markSealed() && !sealAndSend(Optional.empty())) {
             sealSuccessor();
         }
     }
 
-    void sealOnly() {
-        parent.onTransactionSealed(this);
-        final boolean success = STATE_UPDATER.compareAndSet(this, OPEN, SEALED);
-        Verify.verify(success, "Attempted to replay seal on {}", this);
+    /**
+     * Seal this transaction. If this method reports false, the caller needs to deal with propagating the seal operation
+     * towards the successor.
+     *
+     * @return True if seal operation was successful, false if this proxy has a successor.
+     */
+    boolean sealOnly() {
+        return sealState();
     }
 
     /**
@@ -375,8 +381,11 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * @return True if seal operation was successful, false if this proxy has a successor.
      */
     boolean sealAndSend(final Optional<Long> enqueuedTicks) {
-        parent.onTransactionSealed(this);
+        return sealState();
+    }
 
+    private boolean sealState() {
+        parent.onTransactionSealed(this);
         // Transition internal state to sealed and detect presence of a successor
         return STATE_UPDATER.compareAndSet(this, OPEN, SEALED);
     }
@@ -410,7 +419,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
     }
 
-    final void recordSuccessfulRequest(final @Nonnull TransactionRequest<?> req) {
+    final void recordSuccessfulRequest(final @NonNull TransactionRequest<?> req) {
         successfulRequests.add(Verify.verifyNotNull(req));
     }
 
@@ -446,7 +455,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             } else if (t instanceof RequestFailure) {
                 ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
             } else {
-                ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
+                ret.voteNo(unhandledResponseException(t));
             }
 
             // This is a terminal request, hence we do not need to record it
@@ -504,7 +513,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                             ret.setException(cause);
                         }
                     } else {
-                        ret.setException(new IllegalStateException("Unhandled response " + t.getClass()));
+                        ret.setException(unhandledResponseException(t));
                     }
 
                     // This is a terminal request, hence we do not need to record it
@@ -535,7 +544,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                     } else if (t instanceof RequestFailure) {
                         ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
                     } else {
-                        ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
+                        ret.voteNo(unhandledResponseException(t));
                     }
 
                     recordSuccessfulRequest(req);
@@ -566,7 +575,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             } else if (t instanceof RequestFailure) {
                 ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
             } else {
-                ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
+                ret.voteNo(unhandledResponseException(t));
             }
 
             onPreCommitComplete(req);
@@ -599,7 +608,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             } else if (t instanceof RequestFailure) {
                 ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
             } else {
-                ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
+                ret.voteNo(unhandledResponseException(t));
             }
 
             LOG.debug("Transaction {} doCommit completed", this);
@@ -684,13 +693,13 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             for (Object obj : successfulRequests) {
                 if (obj instanceof TransactionRequest) {
                     LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
-                    successor.doReplayRequest((TransactionRequest<?>) obj, resp -> { }, now);
+                    successor.doReplayRequest((TransactionRequest<?>) obj, resp -> { /*NOOP*/ }, now);
                 } else {
                     Verify.verify(obj instanceof IncrementSequence);
                     final IncrementSequence increment = (IncrementSequence) obj;
                     successor.doReplayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
-                        increment.getSequence(), localActor(), isSnapshotOnly(), increment.getDelta()), resp -> { },
-                        now);
+                        increment.getSequence(), localActor(), isSnapshotOnly(),
+                        increment.getDelta()), resp -> { /*NOOP*/ }, now);
                     LOG.debug("Incrementing sequence {} to successor {}", obj, successor);
                 }
             }
@@ -719,9 +728,13 @@ 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);
+            final long enqueuedTicks = parent.currentTime();
+            final Optional<ModifyTransactionRequest> optState = flushState();
+            if (optState.isPresent()) {
+                successor.handleReplayedRemoteRequest(optState.get(), null, enqueuedTicks);
+            }
             if (successor.markSealed()) {
-                successor.sealAndSend(Optional.of(parent.currentTime()));
+                successor.sealAndSend(Optional.of(enqueuedTicks));
             }
         }
     }
@@ -790,12 +803,12 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
 
     abstract void doWrite(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
 
-    abstract CheckedFuture<Boolean, ReadFailedException> doExists(YangInstanceIdentifier path);
+    abstract FluentFuture<Boolean> doExists(YangInstanceIdentifier path);
 
-    abstract CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(YangInstanceIdentifier path);
+    abstract FluentFuture<Optional<NormalizedNode<?, ?>>> doRead(YangInstanceIdentifier path);
 
     @GuardedBy("this")
-    abstract void flushState(AbstractProxyTransaction successor);
+    abstract Optional<ModifyTransactionRequest> flushState();
 
     abstract TransactionRequest<?> abortRequest();
 
@@ -839,6 +852,10 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     abstract void handleReplayedRemoteRequest(TransactionRequest<?> request,
             @Nullable Consumer<Response<?, ?>> callback, long enqueuedTicks);
 
+    private static IllegalStateException unhandledResponseException(Response<?, ?> resp) {
+        return new IllegalStateException("Unhandled response " + resp.getClass());
+    }
+
     @Override
     public final String toString() {
         return MoreObjects.toStringHelper(this).add("identifier", getIdentifier()).add("state", state).toString();