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=9b9b86f11b84b04bda361e1b56ad837a60c393db;hb=ac9b62ec80da4d68e5d867bde0ceaa73062de444;hp=51f528150d22a717034cc4281618591645cca92f;hpb=7991491f2854dde2ec625ed6c08b44df7d258795;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 51f528150d..9b9b86f11b 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,26 +7,29 @@
  */
 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.Optional;
-import com.google.common.base.Preconditions;
-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.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.Nonnull;
-import javax.annotation.Nullable;
-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;
 import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.ClosedTransactionException;
@@ -45,8 +48,8 @@ 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.common.Empty;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 import org.slf4j.Logger;
@@ -65,12 +68,12 @@ import org.slf4j.LoggerFactory;
  *
  * @author Robert Varga
  */
+// FIXME: sealed when we have JDK17+
 abstract class AbstractProxyTransaction implements Identifiable<TransactionIdentifier> {
     /**
      * 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;
@@ -100,7 +103,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
@@ -138,7 +141,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;
         }
@@ -148,26 +151,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() {
@@ -255,7 +258,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
@@ -293,24 +296,24 @@ 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);
     }
 
-    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);
     }
@@ -333,9 +336,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.absent())) {
+        if (!sealAndSend(OptionalLong.empty())) {
             sealSuccessor();
         }
     }
@@ -351,7 +354,7 @@ 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.
-        final java.util.Optional<ModifyTransactionRequest> optState = flushState();
+        final Optional<ModifyTransactionRequest> optState = flushState();
         if (optState.isPresent()) {
             forwardToSuccessor(successor, optState.get(), null);
         }
@@ -359,15 +362,19 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     }
 
     private void predecessorSealed() {
-        if (markSealed() && !sealAndSend(Optional.absent())) {
+        if (markSealed() && !sealAndSend(OptionalLong.empty())) {
             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);
+    /**
+     * 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();
     }
 
     /**
@@ -377,9 +384,12 @@ 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) {
-        parent.onTransactionSealed(this);
+    boolean sealAndSend(final OptionalLong enqueuedTicks) {
+        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);
     }
@@ -394,16 +404,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;
     }
 
@@ -413,8 +423,8 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
     }
 
-    final void recordSuccessfulRequest(final @Nonnull TransactionRequest<?> req) {
-        successfulRequests.add(Verify.verifyNotNull(req));
+    final void recordSuccessfulRequest(final @NonNull TransactionRequest<?> req) {
+        successfulRequests.add(verifyNotNull(req));
     }
 
     final void recordFinishedRequest(final Response<?, ?> response) {
@@ -440,7 +450,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         });
     }
 
-    final void abort(final VotingFuture<Void> ret) {
+    final void abort(final VotingFuture<Empty> ret) {
         checkSealed();
 
         sendDoAbort(t -> {
@@ -449,7 +459,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
@@ -494,7 +504,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) {
@@ -507,7 +517,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
@@ -530,7 +540,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) {
@@ -538,7 +548,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);
@@ -569,7 +579,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);
@@ -602,7 +612,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);
@@ -656,7 +666,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
@@ -687,13 +697,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);
+                    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);
                 }
             }
@@ -708,7 +718,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();
@@ -723,12 +733,12 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         if (SEALED.equals(prevState)) {
             LOG.debug("Proxy {} reconnected while being sealed, propagating state to successor {}", this, successor);
             final long enqueuedTicks = parent.currentTime();
-            final java.util.Optional<ModifyTransactionRequest> optState = flushState();
+            final Optional<ModifyTransactionRequest> optState = flushState();
             if (optState.isPresent()) {
                 successor.handleReplayedRemoteRequest(optState.get(), null, enqueuedTicks);
             }
             if (successor.markSealed()) {
-                successor.sealAndSend(Optional.of(enqueuedTicks));
+                successor.sealAndSend(OptionalLong.of(enqueuedTicks));
             }
         }
     }
@@ -793,16 +803,16 @@ 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 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 java.util.Optional<ModifyTransactionRequest> flushState();
+    abstract Optional<ModifyTransactionRequest> flushState();
 
     abstract TransactionRequest<?> abortRequest();
 
@@ -846,6 +856,14 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     abstract void handleReplayedRemoteRequest(TransactionRequest<?> request,
             @Nullable Consumer<Response<?, ?>> callback, long enqueuedTicks);
 
+    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());
+    }
+
     @Override
     public final String toString() {
         return MoreObjects.toStringHelper(this).add("identifier", getIdentifier()).add("state", state).toString();