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%2Fdatastore%2FAbstractTransactionContextFactory.java;h=29b52f75579b57e2fbad4563585de9b10828d154;hb=546cd1fd100dbaa36908b22c2f422320dbd8c4b2;hp=69c69bb84b395dc5d89db35f74cdd44343987c95;hpb=925cb4a228d0fda99c7bfeb432eb25285a223887;p=controller.git

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContextFactory.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContextFactory.java
index 69c69bb84b..29b52f7557 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContextFactory.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContextFactory.java
@@ -7,24 +7,24 @@
  */
 package org.opendaylight.controller.cluster.datastore;
 
+import static java.util.Objects.requireNonNull;
+
 import akka.actor.ActorSelection;
-import akka.dispatch.OnComplete;
-import com.google.common.base.Preconditions;
 import java.util.Collection;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
-import javax.annotation.Nonnull;
+import org.eclipse.jdt.annotation.NonNull;
 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
 import org.opendaylight.controller.cluster.access.concepts.MemberName;
 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
-import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
-import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
-import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
+import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
+import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
+import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.ReadOnlyDataTree;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import scala.concurrent.Future;
@@ -42,20 +42,19 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
 
     private final ConcurrentMap<String, F> knownLocal = new ConcurrentHashMap<>();
     private final LocalHistoryIdentifier historyId;
-    private final ActorContext actorContext;
+    private final ActorUtils actorUtils;
 
     // Used via TX_COUNTER_UPDATER
     @SuppressWarnings("unused")
     private volatile long nextTx;
 
-    protected AbstractTransactionContextFactory(final ActorContext actorContext,
-            final LocalHistoryIdentifier historyId) {
-        this.actorContext = Preconditions.checkNotNull(actorContext);
-        this.historyId = Preconditions.checkNotNull(historyId);
+    protected AbstractTransactionContextFactory(final ActorUtils actorUtils, final LocalHistoryIdentifier historyId) {
+        this.actorUtils = requireNonNull(actorUtils);
+        this.historyId = requireNonNull(historyId);
     }
 
-    final ActorContext getActorContext() {
-        return actorContext;
+    final ActorUtils getActorUtils() {
+        return actorUtils;
     }
 
     final LocalHistoryIdentifier getHistoryId() {
@@ -80,20 +79,48 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
         return null;
     }
 
-    private void onFindPrimaryShardSuccess(PrimaryShardInfo primaryShardInfo, TransactionProxy parent,
-            String shardName, TransactionContextWrapper transactionContextWrapper) {
+    private AbstractTransactionContextWrapper maybeCreateDirectTransactionContextWrapper(
+            final PrimaryShardInfo primaryShardInfo, final TransactionProxy parent,
+            final String shardName, final DelayedTransactionContextWrapper transactionContextWrapper) {
+        LOG.debug("Tx {}: Found primary {} for shard {}, trying to use DirectTransactionContextWrapper",
+                parent.getIdentifier(), primaryShardInfo.getPrimaryShardActor(), shardName);
+
+        updateShardInfo(shardName, primaryShardInfo);
+
+        final TransactionContext localContext = maybeCreateLocalTransactionContext(parent, shardName);
+        try {
+            if (localContext != null) {
+                LOG.debug("Tx {}: Local transaction context created successfully, using DirectTransactionWrapper",
+                        parent.getIdentifier());
+                return new DirectTransactionContextWrapper(parent.getIdentifier(), actorUtils, shardName,
+                        localContext);
+            } else {
+                LOG.debug("Tx {}: Local transaction context creation failed, using DelayedTransactionWrapper",
+                        parent.getIdentifier());
+                final RemoteTransactionContextSupport remote = new RemoteTransactionContextSupport(
+                        transactionContextWrapper, parent, shardName);
+                remote.setPrimaryShard(primaryShardInfo);
+                return transactionContextWrapper;
+            }
+        } finally {
+            onTransactionContextCreated(parent.getIdentifier());
+        }
+    }
+
+    private void onFindPrimaryShardSuccess(final PrimaryShardInfo primaryShardInfo, final TransactionProxy parent,
+            final String shardName, final DelayedTransactionContextWrapper transactionContextWrapper) {
         LOG.debug("Tx {}: Found primary {} for shard {}", parent.getIdentifier(),
                 primaryShardInfo.getPrimaryShardActor(), shardName);
 
         updateShardInfo(shardName, primaryShardInfo);
 
+        final TransactionContext localContext = maybeCreateLocalTransactionContext(parent, shardName);
         try {
-            TransactionContext localContext = maybeCreateLocalTransactionContext(parent, shardName);
             if (localContext != null) {
                 transactionContextWrapper.executePriorTransactionOperations(localContext);
             } else {
-                RemoteTransactionContextSupport remote = new RemoteTransactionContextSupport(transactionContextWrapper,
-                        parent, shardName);
+                final RemoteTransactionContextSupport remote = new RemoteTransactionContextSupport(
+                        transactionContextWrapper, parent, shardName);
                 remote.setPrimaryShard(primaryShardInfo);
             }
         } finally {
@@ -101,49 +128,47 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
         }
     }
 
-    private void onFindPrimaryShardFailure(Throwable failure, TransactionProxy parent,
-            String shardName, TransactionContextWrapper transactionContextWrapper) {
+    private void onFindPrimaryShardFailure(final Throwable failure, final TransactionProxy parent,
+            final String shardName, final DelayedTransactionContextWrapper transactionContextWrapper) {
         LOG.debug("Tx {}: Find primary for shard {} failed", parent.getIdentifier(), shardName, failure);
 
         try {
-            transactionContextWrapper.executePriorTransactionOperations(new NoOpTransactionContext(failure,
-                    parent.getIdentifier()));
+            transactionContextWrapper.executePriorTransactionOperations(
+                    new NoOpTransactionContext(failure, parent.getIdentifier()));
         } finally {
             onTransactionContextCreated(parent.getIdentifier());
         }
     }
 
-    final TransactionContextWrapper newTransactionContextWrapper(final TransactionProxy parent,
+    final AbstractTransactionContextWrapper newTransactionContextWrapper(final TransactionProxy parent,
             final String shardName) {
-        final TransactionContextWrapper transactionContextWrapper =
-                new TransactionContextWrapper(parent.getIdentifier(), actorContext);
-
-        Future<PrimaryShardInfo> findPrimaryFuture = findPrimaryShard(shardName, parent.getIdentifier());
+        final DelayedTransactionContextWrapper contextWrapper = new DelayedTransactionContextWrapper(
+                parent.getIdentifier(), actorUtils, shardName);
+        final Future<PrimaryShardInfo> findPrimaryFuture = findPrimaryShard(shardName, parent.getIdentifier());
         if (findPrimaryFuture.isCompleted()) {
-            Try<PrimaryShardInfo> maybe = findPrimaryFuture.value().get();
+            final Try<PrimaryShardInfo> maybe = findPrimaryFuture.value().get();
             if (maybe.isSuccess()) {
-                onFindPrimaryShardSuccess(maybe.get(), parent, shardName, transactionContextWrapper);
+                return maybeCreateDirectTransactionContextWrapper(maybe.get(), parent, shardName,
+                        contextWrapper);
             } else {
-                onFindPrimaryShardFailure(maybe.failed().get(), parent, shardName, transactionContextWrapper);
+                onFindPrimaryShardFailure(maybe.failed().get(), parent, shardName,
+                        contextWrapper);
             }
         } else {
-            findPrimaryFuture.onComplete(new OnComplete<PrimaryShardInfo>() {
-                @Override
-                public void onComplete(final Throwable failure, final PrimaryShardInfo primaryShardInfo) {
-                    if (failure == null) {
-                        onFindPrimaryShardSuccess(primaryShardInfo, parent, shardName, transactionContextWrapper);
-                    } else {
-                        onFindPrimaryShardFailure(failure, parent, shardName, transactionContextWrapper);
-                    }
+            findPrimaryFuture.onComplete(result -> {
+                if (result.isSuccess()) {
+                    onFindPrimaryShardSuccess(result.get(), parent, shardName, contextWrapper);
+                } else {
+                    onFindPrimaryShardFailure(result.failed().get(), parent, shardName, contextWrapper);
                 }
-            }, actorContext.getClientDispatcher());
+                return null;
+            }, actorUtils.getClientDispatcher());
         }
-
-        return transactionContextWrapper;
+        return contextWrapper;
     }
 
     private void updateShardInfo(final String shardName, final PrimaryShardInfo primaryShardInfo) {
-        final Optional<DataTree> maybeDataTree = primaryShardInfo.getLocalShardDataTree();
+        final Optional<ReadOnlyDataTree> maybeDataTree = primaryShardInfo.getLocalShardDataTree();
         if (maybeDataTree.isPresent()) {
             if (!knownLocal.containsKey(shardName)) {
                 LOG.debug("Shard {} resolved to local data tree - adding local factory", shardName);
@@ -177,8 +202,8 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
      * @param shardName Shard name
      * @return Future containing shard information.
      */
-    protected abstract Future<PrimaryShardInfo> findPrimaryShard(@Nonnull String shardName,
-            @Nonnull TransactionIdentifier txId);
+    protected abstract Future<PrimaryShardInfo> findPrimaryShard(@NonNull String shardName,
+            @NonNull TransactionIdentifier txId);
 
     /**
      * Create local transaction factory for specified shard, backed by specified shard leader
@@ -190,22 +215,22 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
      *                 read-only manner.
      * @return Transaction factory for local use.
      */
-    protected abstract F factoryForShard(String shardName, ActorSelection shardLeader, DataTree dataTree);
+    protected abstract F factoryForShard(String shardName, ActorSelection shardLeader, ReadOnlyDataTree dataTree);
 
     /**
      * Callback invoked from child transactions to push any futures, which need to
      * be waited for before the next transaction is allocated.
      * @param cohortFutures Collection of futures
      */
-    protected abstract <T> void onTransactionReady(@Nonnull TransactionIdentifier transaction,
-            @Nonnull Collection<Future<T>> cohortFutures);
+    protected abstract <T> void onTransactionReady(@NonNull TransactionIdentifier transaction,
+            @NonNull Collection<Future<T>> cohortFutures);
 
     /**
      * Callback invoked when the internal TransactionContext has been created for a transaction.
      *
      * @param transactionId the ID of the transaction.
      */
-    protected abstract void onTransactionContextCreated(@Nonnull TransactionIdentifier transactionId);
+    protected abstract void onTransactionContextCreated(@NonNull TransactionIdentifier transactionId);
 
     private static TransactionContext createLocalTransactionContext(final LocalTransactionFactory factory,
                                                                     final TransactionProxy parent) {
@@ -215,12 +240,12 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
                 final DOMStoreReadTransaction readOnly = factory.newReadOnlyTransaction(parent.getIdentifier());
                 return new LocalTransactionContext(readOnly, parent.getIdentifier(), factory) {
                     @Override
-                    protected DOMStoreWriteTransaction getWriteDelegate() {
+                    DOMStoreWriteTransaction getWriteDelegate() {
                         throw new UnsupportedOperationException();
                     }
 
                     @Override
-                    protected DOMStoreReadTransaction getReadDelegate() {
+                    DOMStoreReadTransaction getReadDelegate() {
                         return readOnly;
                     }
                 };
@@ -228,12 +253,12 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
                 final DOMStoreReadWriteTransaction readWrite = factory.newReadWriteTransaction(parent.getIdentifier());
                 return new LocalTransactionContext(readWrite, parent.getIdentifier(), factory) {
                     @Override
-                    protected DOMStoreWriteTransaction getWriteDelegate() {
+                    DOMStoreWriteTransaction getWriteDelegate() {
                         return readWrite;
                     }
 
                     @Override
-                    protected DOMStoreReadTransaction getReadDelegate() {
+                    DOMStoreReadTransaction getReadDelegate() {
                         return readWrite;
                     }
                 };
@@ -241,12 +266,12 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
                 final DOMStoreWriteTransaction writeOnly = factory.newWriteOnlyTransaction(parent.getIdentifier());
                 return new LocalTransactionContext(writeOnly, parent.getIdentifier(), factory) {
                     @Override
-                    protected DOMStoreWriteTransaction getWriteDelegate() {
+                    DOMStoreWriteTransaction getWriteDelegate() {
                         return writeOnly;
                     }
 
                     @Override
-                    protected DOMStoreReadTransaction getReadDelegate() {
+                    DOMStoreReadTransaction getReadDelegate() {
                         throw new UnsupportedOperationException();
                     }
                 };