Merge "Bug 2160: Added concurrent 3-phase commit coordinator"

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / TransactionProxy.java

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java
index 6cf16b44268c6c16e26e0658632f61994ee33971..ec198510d3586f88dee40d04f9294a3fb370a9ae 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java
@@ -8,7 +8,6 @@
 
 package org.opendaylight.controller.cluster.datastore;
 
-import akka.actor.ActorPath;
 import akka.actor.ActorSelection;
 import akka.dispatch.OnComplete;
 
@@ -22,6 +21,7 @@ import com.google.common.util.concurrent.CheckedFuture;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.SettableFuture;
 
+import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
 import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
@@ -156,7 +156,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
             if(remoteTransactionActorsMB.get()) {
                 for(ActorSelection actor : remoteTransactionActors) {
                     LOG.trace("Sending CloseTransaction to {}", actor);
-                    actorContext.sendRemoteOperationAsync(actor,
+                    actorContext.sendOperationAsync(actor,
                             new CloseTransaction().toSerializable());
                 }
             }
@@ -314,7 +314,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
             LOG.debug("Tx {} Trying to get {} transactions ready for commit", identifier,
                 remoteTransactionPaths.size());
         }
-        List<Future<ActorPath>> cohortPathFutures = Lists.newArrayList();
+        List<Future<ActorSelection>> cohortFutures = Lists.newArrayList();
 
         for(TransactionContext transactionContext : remoteTransactionPaths.values()) {
 
@@ -322,14 +322,14 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
                 LOG.debug("Tx {} Readying transaction for shard {}", identifier,
                     transactionContext.getShardName());
             }
-            cohortPathFutures.add(transactionContext.readyTransaction());
+            cohortFutures.add(transactionContext.readyTransaction());
         }
 
         if(transactionChainProxy != null){
-            transactionChainProxy.onTransactionReady(cohortPathFutures);
+            transactionChainProxy.onTransactionReady(cohortFutures);
         }
 
-        return new ThreePhaseCommitCohortProxy(actorContext, cohortPathFutures,
+        return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures,
                 identifier.toString());
     }
 
@@ -379,9 +379,14 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
         }
 
         try {
-            Object response = actorContext.executeShardOperation(shardName,
-                new CreateTransaction(identifier.toString(), this.transactionType.ordinal(),
-                    getTransactionChainId()).toSerializable());
+            Optional<ActorSelection> primaryShard = actorContext.findPrimaryShard(shardName);
+            if (!primaryShard.isPresent()) {
+                throw new PrimaryNotFoundException("Primary could not be found for shard " + shardName);
+            }
+
+            Object response = actorContext.executeOperation(primaryShard.get(),
+                    new CreateTransaction(identifier.toString(), this.transactionType.ordinal(),
+                            getTransactionChainId()).toSerializable());
             if (response.getClass().equals(CreateTransactionReply.SERIALIZABLE_CLASS)) {
                 CreateTransactionReply reply =
                     CreateTransactionReply.fromSerializable(response);
@@ -433,7 +438,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
 
         void closeTransaction();
 
-        Future<ActorPath> readyTransaction();
+        Future<ActorSelection> readyTransaction();
 
         void writeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
 
@@ -493,27 +498,23 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
             return actor;
         }
 
-        private String getResolvedCohortPath(String cohortPath) {
-            return actorContext.resolvePath(actorPath, cohortPath);
-        }
-
         @Override
         public void closeTransaction() {
             if(LOG.isDebugEnabled()) {
                 LOG.debug("Tx {} closeTransaction called", identifier);
             }
-            actorContext.sendRemoteOperationAsync(getActor(), new CloseTransaction().toSerializable());
+            actorContext.sendOperationAsync(getActor(), new CloseTransaction().toSerializable());
         }
 
         @Override
-        public Future<ActorPath> readyTransaction() {
+        public Future<ActorSelection> readyTransaction() {
             if(LOG.isDebugEnabled()) {
                 LOG.debug("Tx {} readyTransaction called with {} previous recorded operations pending",
                     identifier, recordedOperationFutures.size());
             }
             // Send the ReadyTransaction message to the Tx actor.
 
-            final Future<Object> replyFuture = actorContext.executeRemoteOperationAsync(getActor(),
+            final Future<Object> replyFuture = actorContext.executeOperationAsync(getActor(),
                     new ReadyTransaction().toSerializable());
 
             // Combine all the previously recorded put/merge/delete operation reply Futures and the
@@ -532,9 +533,9 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
             // Transform the combined Future into a Future that returns the cohort actor path from
             // the ReadyTransactionReply. That's the end result of the ready operation.
 
-            return combinedFutures.transform(new AbstractFunction1<Iterable<Object>, ActorPath>() {
+            return combinedFutures.transform(new AbstractFunction1<Iterable<Object>, ActorSelection>() {
                 @Override
-                public ActorPath apply(Iterable<Object> notUsed) {
+                public ActorSelection apply(Iterable<Object> notUsed) {
                     if(LOG.isDebugEnabled()) {
                         LOG.debug("Tx {} readyTransaction: pending recorded operations succeeded",
                             identifier);
@@ -551,16 +552,9 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
                     if(serializedReadyReply.getClass().equals(
                                                      ReadyTransactionReply.SERIALIZABLE_CLASS)) {
                         ReadyTransactionReply reply = ReadyTransactionReply.fromSerializable(
-                                actorContext.getActorSystem(), serializedReadyReply);
+                               serializedReadyReply);
 
-                        String resolvedCohortPath = getResolvedCohortPath(
-                                reply.getCohortPath().toString());
-
-                        if(LOG.isDebugEnabled()) {
-                            LOG.debug("Tx {} readyTransaction: resolved cohort path {}",
-                                identifier, resolvedCohortPath);
-                        }
-                        return actorContext.actorFor(resolvedCohortPath);
+                        return actorContext.actorSelection(reply.getCohortPath());
                     } else {
                         // Throwing an exception here will fail the Future.
 
@@ -576,8 +570,8 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
             if(LOG.isDebugEnabled()) {
                 LOG.debug("Tx {} deleteData called path = {}", identifier, path);
             }
-            recordedOperationFutures.add(actorContext.executeRemoteOperationAsync(getActor(),
-                    new DeleteData(path).toSerializable() ));
+            recordedOperationFutures.add(actorContext.executeOperationAsync(getActor(),
+                    new DeleteData(path).toSerializable()));
         }
 
         @Override
@@ -585,7 +579,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
             if(LOG.isDebugEnabled()) {
                 LOG.debug("Tx {} mergeData called path = {}", identifier, path);
             }
-            recordedOperationFutures.add(actorContext.executeRemoteOperationAsync(getActor(),
+            recordedOperationFutures.add(actorContext.executeOperationAsync(getActor(),
                     new MergeData(path, data, schemaContext).toSerializable()));
         }
 
@@ -594,7 +588,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
             if(LOG.isDebugEnabled()) {
                 LOG.debug("Tx {} writeData called path = {}", identifier, path);
             }
-            recordedOperationFutures.add(actorContext.executeRemoteOperationAsync(getActor(),
+            recordedOperationFutures.add(actorContext.executeOperationAsync(getActor(),
                     new WriteData(path, data, schemaContext).toSerializable()));
         }
 
@@ -686,7 +680,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
                 }
             };
 
-            Future<Object> readFuture = actorContext.executeRemoteOperationAsync(getActor(),
+            Future<Object> readFuture = actorContext.executeOperationAsync(getActor(),
                     new ReadData(path).toSerializable());
             readFuture.onComplete(onComplete, actorContext.getActorSystem().dispatcher());
         }
@@ -773,7 +767,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
                 }
             };
 
-            Future<Object> future = actorContext.executeRemoteOperationAsync(getActor(),
+            Future<Object> future = actorContext.executeOperationAsync(getActor(),
                     new DataExists(path).toSerializable());
             future.onComplete(onComplete, actorContext.getActorSystem().dispatcher());
         }
@@ -799,7 +793,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
         }
 
         @Override
-        public Future<ActorPath> readyTransaction() {
+        public Future<ActorSelection> readyTransaction() {
             if(LOG.isDebugEnabled()) {
                 LOG.debug("Tx {} readyTransaction called", identifier);
             }