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=14ad54699161a60719ea846d353c46e73fb4adea;hb=HEAD;hp=e1919d1bb6c2a601486005425c5e5f6fe75423b4;hpb=127042ea7e148d9dc0282acc3780b4754ca69e12;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 e1919d1bb6..14ad546991 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
@@ -7,10 +7,13 @@
  */
 package org.opendaylight.controller.cluster.databroker.actors.dds;
 
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.base.Verify.verify;
+import static com.google.common.base.Verify.verifyNotNull;
+import static java.util.Objects.requireNonNull;
+
 import akka.actor.ActorRef;
 import com.google.common.base.MoreObjects;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Verify;
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.FluentFuture;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -19,12 +22,12 @@ import java.util.ArrayDeque;
 import java.util.Deque;
 import java.util.Iterator;
 import java.util.Optional;
+import java.util.OptionalLong;
 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.concurrent.GuardedBy;
-import javax.annotation.concurrent.NotThreadSafe;
+import org.checkerframework.checker.lock.qual.GuardedBy;
 import org.eclipse.jdt.annotation.NonNull;
 import org.eclipse.jdt.annotation.Nullable;
 import org.opendaylight.controller.cluster.access.client.ConnectionEntry;
@@ -46,6 +49,7 @@ 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.yangtools.concepts.Identifiable;
+import org.opendaylight.yangtools.yang.common.Empty;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 import org.slf4j.Logger;
@@ -61,15 +65,13 @@ import org.slf4j.LoggerFactory;
  * <p>
  * This class interacts with the queueing mechanism in ClientActorBehavior, hence once we arrive at a decision
  * to use either a local or remote implementation, we are stuck with it. We can re-evaluate on the next transaction.
- *
- * @author Robert Varga
  */
-abstract class AbstractProxyTransaction implements Identifiable<TransactionIdentifier> {
+abstract sealed class AbstractProxyTransaction implements Identifiable<TransactionIdentifier>
+        permits LocalProxyTransaction, RemoteProxyTransaction {
     /**
      * 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.
+     * and allows compressing multiple requests into a single entry. This class is not thread-safe.
      */
-    @NotThreadSafe
     private static final class IncrementSequence {
         private final long sequence;
         private long delta = 0;
@@ -99,7 +101,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         private final String string;
 
         State(final String string) {
-            this.string = Preconditions.checkNotNull(string);
+            this.string = requireNonNull(string);
         }
 
         @Override
@@ -137,7 +139,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                 latch.await();
             } catch (InterruptedException e) {
                 LOG.warn("Interrupted while waiting for latch of {}", successor);
-                throw new RuntimeException(e);
+                throw new IllegalStateException(e);
             }
             return successor;
         }
@@ -147,26 +149,26 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
 
         State getPrevState() {
-            return Verify.verifyNotNull(prevState, "Attempted to access previous state, which was not set");
+            return 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);
+            verify(this.prevState == null, "Attempted to set previous state to %s when we already have %s", prevState,
+                    this.prevState);
+            this.prevState = requireNonNull(prevState);
             // We cannot have duplicate successor states, so this check is sufficient
-            this.done = DONE.equals(prevState);
+            done = DONE.equals(prevState);
         }
 
         // To be called from safe contexts, where successor is known to be completed
         AbstractProxyTransaction getSuccessor() {
-            return Verify.verifyNotNull(successor);
+            return verifyNotNull(successor);
         }
 
         void setSuccessor(final AbstractProxyTransaction successor) {
-            Verify.verify(this.successor == null, "Attempted to set successor to %s when we already have %s",
-                    successor, this.successor);
-            this.successor = Preconditions.checkNotNull(successor);
+            verify(this.successor == null, "Attempted to set successor to %s when we already have %s", successor,
+                    this.successor);
+            this.successor = requireNonNull(successor);
         }
 
         boolean isDone() {
@@ -254,7 +256,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     private volatile State state;
 
     AbstractProxyTransaction(final ProxyHistory parent, final boolean isDone) {
-        this.parent = Preconditions.checkNotNull(parent);
+        this.parent = requireNonNull(parent);
         if (isDone) {
             state = DONE;
             // DONE implies previous seal operation completed
@@ -292,13 +294,13 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         doDelete(path);
     }
 
-    final void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
+    final void merge(final YangInstanceIdentifier path, final NormalizedNode data) {
         checkReadWrite();
         checkNotSealed();
         doMerge(path, data);
     }
 
-    final void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
+    final void write(final YangInstanceIdentifier path, final NormalizedNode data) {
         checkReadWrite();
         checkNotSealed();
         doWrite(path, data);
@@ -309,7 +311,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         return doExists(path);
     }
 
-    final FluentFuture<Optional<NormalizedNode<?, ?>>> read(final YangInstanceIdentifier path) {
+    final FluentFuture<Optional<NormalizedNode>> read(final YangInstanceIdentifier path) {
         checkNotSealed();
         return doRead(path);
     }
@@ -332,9 +334,9 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     final void seal() {
         // Transition user-visible state first
         final boolean success = markSealed();
-        Preconditions.checkState(success, "Proxy %s was already sealed", getIdentifier());
+        checkState(success, "Proxy %s was already sealed", getIdentifier());
 
-        if (!sealAndSend(Optional.empty())) {
+        if (!sealAndSend(OptionalLong.empty())) {
             sealSuccessor();
         }
     }
@@ -352,13 +354,13 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         // Propagate state and seal the successor.
         final Optional<ModifyTransactionRequest> optState = flushState();
         if (optState.isPresent()) {
-            forwardToSuccessor(successor, optState.get(), null);
+            forwardToSuccessor(successor, optState.orElseThrow(), null);
         }
         successor.predecessorSealed();
     }
 
     private void predecessorSealed() {
-        if (markSealed() && !sealAndSend(Optional.empty())) {
+        if (markSealed() && !sealAndSend(OptionalLong.empty())) {
             sealSuccessor();
         }
     }
@@ -380,7 +382,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * @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) {
+    boolean sealAndSend(final OptionalLong enqueuedTicks) {
         return sealState();
     }
 
@@ -400,16 +402,16 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     }
 
     private void checkNotSealed() {
-        Preconditions.checkState(sealed == 0, "Transaction %s has already been sealed", getIdentifier());
+        checkState(sealed == 0, "Transaction %s has already been sealed", getIdentifier());
     }
 
     private void checkSealed() {
-        Preconditions.checkState(sealed != 0, "Transaction %s has not been sealed yet", getIdentifier());
+        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);
+        verify(local instanceof SuccessorState, "State %s has unexpected class", local);
         return (SuccessorState) local;
     }
 
@@ -420,7 +422,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     }
 
     final void recordSuccessfulRequest(final @NonNull TransactionRequest<?> req) {
-        successfulRequests.add(Verify.verifyNotNull(req));
+        successfulRequests.add(verifyNotNull(req));
     }
 
     final void recordFinishedRequest(final Response<?, ?> response) {
@@ -446,7 +448,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         });
     }
 
-    final void abort(final VotingFuture<Void> ret) {
+    final void abort(final VotingFuture<Empty> ret) {
         checkSealed();
 
         sendDoAbort(t -> {
@@ -500,7 +502,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         synchronized (this) {
             if (STATE_UPDATER.compareAndSet(this, SEALED, FLUSHED)) {
                 final SettableFuture<Boolean> ret = SettableFuture.create();
-                sendRequest(Verify.verifyNotNull(commitRequest(false)), t -> {
+                sendRequest(verifyNotNull(commitRequest(false)), t -> {
                     if (t instanceof TransactionCommitSuccess) {
                         ret.set(Boolean.TRUE);
                     } else if (t instanceof RequestFailure) {
@@ -536,7 +538,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         // Precludes startReconnect() from interfering with the fast path
         synchronized (this) {
             if (STATE_UPDATER.compareAndSet(this, SEALED, FLUSHED)) {
-                final TransactionRequest<?> req = Verify.verifyNotNull(commitRequest(true));
+                final TransactionRequest<?> req = verifyNotNull(commitRequest(true));
 
                 sendRequest(req, t -> {
                     if (t instanceof TransactionCanCommitSuccess) {
@@ -662,7 +664,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         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,
+        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
@@ -695,7 +697,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                     LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
                     successor.doReplayRequest((TransactionRequest<?>) obj, resp -> { /*NOOP*/ }, now);
                 } else {
-                    Verify.verify(obj instanceof IncrementSequence);
+                    verify(obj instanceof IncrementSequence);
                     final IncrementSequence increment = (IncrementSequence) obj;
                     successor.doReplayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
                         increment.getSequence(), localActor(), isSnapshotOnly(),
@@ -714,7 +716,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             final Request<?, ?> req = e.getRequest();
 
             if (getIdentifier().equals(req.getTarget())) {
-                Verify.verify(req instanceof TransactionRequest, "Unhandled request %s", req);
+                verify(req instanceof TransactionRequest, "Unhandled request %s", req);
                 LOG.debug("Replaying queued request {} to successor {}", req, successor);
                 successor.doReplayRequest((TransactionRequest<?>) req, e.getCallback(), e.getEnqueuedTicks());
                 it.remove();
@@ -731,10 +733,10 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             final long enqueuedTicks = parent.currentTime();
             final Optional<ModifyTransactionRequest> optState = flushState();
             if (optState.isPresent()) {
-                successor.handleReplayedRemoteRequest(optState.get(), null, enqueuedTicks);
+                successor.handleReplayedRemoteRequest(optState.orElseThrow(), null, enqueuedTicks);
             }
             if (successor.markSealed()) {
-                successor.sealAndSend(Optional.of(enqueuedTicks));
+                successor.sealAndSend(OptionalLong.of(enqueuedTicks));
             }
         }
     }
@@ -799,13 +801,13 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
 
     abstract void doDelete(YangInstanceIdentifier path);
 
-    abstract void doMerge(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
+    abstract void doMerge(YangInstanceIdentifier path, NormalizedNode data);
 
-    abstract void doWrite(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
+    abstract void doWrite(YangInstanceIdentifier path, NormalizedNode data);
 
     abstract FluentFuture<Boolean> doExists(YangInstanceIdentifier path);
 
-    abstract FluentFuture<Optional<NormalizedNode<?, ?>>> doRead(YangInstanceIdentifier path);
+    abstract FluentFuture<Optional<NormalizedNode>> doRead(YangInstanceIdentifier path);
 
     @GuardedBy("this")
     abstract Optional<ModifyTransactionRequest> flushState();
@@ -852,7 +854,11 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     abstract void handleReplayedRemoteRequest(TransactionRequest<?> request,
             @Nullable Consumer<Response<?, ?>> callback, long enqueuedTicks);
 
-    private static IllegalStateException unhandledResponseException(Response<?, ?> resp) {
+    static final @NonNull IllegalArgumentException unhandledRequest(final TransactionRequest<?> request) {
+        return new IllegalArgumentException("Unhandled request " + request);
+    }
+
+    private static @NonNull IllegalStateException unhandledResponseException(final Response<?, ?> resp) {
         return new IllegalStateException("Unhandled response " + resp.getClass());
     }