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%2Fdatastore%2FTransactionProxy.java;h=715f48c3492156d1b14005462da2c26aacb1768c;hp=6cf16b44268c6c16e26e0658632f61994ee33971;hb=531621aac4cff9d39cbd8668a53bdeba8a0e6d81;hpb=70c27e8bf6d323376a78aa5468faf4f27d081638 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 6cf16b4426..715f48c349 100644 --- 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,8 +156,8 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { if(remoteTransactionActorsMB.get()) { for(ActorSelection actor : remoteTransactionActors) { LOG.trace("Sending CloseTransaction to {}", actor); - actorContext.sendRemoteOperationAsync(actor, - new CloseTransaction().toSerializable()); + 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> cohortPathFutures = Lists.newArrayList(); + List> 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,7 +379,12 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { } try { - Object response = actorContext.executeShardOperation(shardName, + Optional 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)) { @@ -403,8 +408,12 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { remoteTransactionActorsMB.set(true); } + // TxActor is always created where the leader of the shard is. + // Check if TxActor is created in the same node + boolean isTxActorLocal = actorContext.isLocalPath(transactionPath); + transactionContext = new TransactionContextImpl(shardName, transactionPath, - transactionActor, identifier, actorContext, schemaContext); + transactionActor, identifier, actorContext, schemaContext, isTxActorLocal); remoteTransactionPaths.put(shardName, transactionContext); } else { @@ -433,7 +442,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { void closeTransaction(); - Future readyTransaction(); + Future readyTransaction(); void writeData(YangInstanceIdentifier path, NormalizedNode data); @@ -478,43 +487,42 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { private final SchemaContext schemaContext; private final String actorPath; private final ActorSelection actor; + private final boolean isTxActorLocal; private TransactionContextImpl(String shardName, String actorPath, ActorSelection actor, TransactionIdentifier identifier, ActorContext actorContext, - SchemaContext schemaContext) { + SchemaContext schemaContext, boolean isTxActorLocal) { super(shardName, identifier); this.actorPath = actorPath; this.actor = actor; this.actorContext = actorContext; this.schemaContext = schemaContext; + this.isTxActorLocal = isTxActorLocal; } private ActorSelection getActor() { 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 readyTransaction() { + public Future 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 replyFuture = actorContext.executeRemoteOperationAsync(getActor(), - new ReadyTransaction().toSerializable()); + ReadyTransaction readyTransaction = new ReadyTransaction(); + final Future replyFuture = actorContext.executeOperationAsync(getActor(), + isTxActorLocal ? readyTransaction : readyTransaction.toSerializable()); // Combine all the previously recorded put/merge/delete operation reply Futures and the // ReadyTransactionReply Future into one Future. If any one fails then the combined @@ -532,9 +540,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, ActorPath>() { + return combinedFutures.transform(new AbstractFunction1, ActorSelection>() { @Override - public ActorPath apply(Iterable notUsed) { + public ActorSelection apply(Iterable notUsed) { if(LOG.isDebugEnabled()) { LOG.debug("Tx {} readyTransaction: pending recorded operations succeeded", identifier); @@ -548,22 +556,15 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { // Note the Future get call here won't block as it's complete. Object serializedReadyReply = replyFuture.value().get().get(); - if(serializedReadyReply.getClass().equals( - ReadyTransactionReply.SERIALIZABLE_CLASS)) { - ReadyTransactionReply reply = ReadyTransactionReply.fromSerializable( - actorContext.getActorSystem(), serializedReadyReply); + if (serializedReadyReply instanceof ReadyTransactionReply) { + return actorContext.actorSelection(((ReadyTransactionReply)serializedReadyReply).getCohortPath()); - String resolvedCohortPath = getResolvedCohortPath( - reply.getCohortPath().toString()); + } else if(serializedReadyReply.getClass().equals(ReadyTransactionReply.SERIALIZABLE_CLASS)) { + ReadyTransactionReply reply = ReadyTransactionReply.fromSerializable(serializedReadyReply); + return actorContext.actorSelection(reply.getCohortPath()); - if(LOG.isDebugEnabled()) { - LOG.debug("Tx {} readyTransaction: resolved cohort path {}", - identifier, resolvedCohortPath); - } - return actorContext.actorFor(resolvedCohortPath); } else { // Throwing an exception here will fail the Future. - throw new IllegalArgumentException(String.format("Invalid reply type {}", serializedReadyReply.getClass())); } @@ -576,8 +577,10 @@ 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() )); + + DeleteData deleteData = new DeleteData(path); + recordedOperationFutures.add(actorContext.executeOperationAsync(getActor(), + isTxActorLocal ? deleteData : deleteData.toSerializable())); } @Override @@ -585,8 +588,10 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { if(LOG.isDebugEnabled()) { LOG.debug("Tx {} mergeData called path = {}", identifier, path); } - recordedOperationFutures.add(actorContext.executeRemoteOperationAsync(getActor(), - new MergeData(path, data, schemaContext).toSerializable())); + + MergeData mergeData = new MergeData(path, data, schemaContext); + recordedOperationFutures.add(actorContext.executeOperationAsync(getActor(), + isTxActorLocal ? mergeData : mergeData.toSerializable())); } @Override @@ -594,8 +599,10 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { if(LOG.isDebugEnabled()) { LOG.debug("Tx {} writeData called path = {}", identifier, path); } - recordedOperationFutures.add(actorContext.executeRemoteOperationAsync(getActor(), - new WriteData(path, data, schemaContext).toSerializable())); + + WriteData writeData = new WriteData(path, data, schemaContext); + recordedOperationFutures.add(actorContext.executeOperationAsync(getActor(), + isTxActorLocal ? writeData : writeData.toSerializable())); } @Override @@ -625,6 +632,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { Future> combinedFutures = akka.dispatch.Futures.sequence( Lists.newArrayList(recordedOperationFutures), actorContext.getActorSystem().dispatcher()); + OnComplete> onComplete = new OnComplete>() { @Override public void onComplete(Throwable failure, Iterable notUsed) @@ -669,25 +677,27 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { if(LOG.isDebugEnabled()) { LOG.debug("Tx {} read operation succeeded", identifier, failure); } - if (readResponse.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)) { - ReadDataReply reply = ReadDataReply.fromSerializable(schemaContext, - path, readResponse); - if (reply.getNormalizedNode() == null) { - returnFuture.set(Optional.>absent()); - } else { - returnFuture.set(Optional.>of( - reply.getNormalizedNode())); - } + + if (readResponse instanceof ReadDataReply) { + ReadDataReply reply = (ReadDataReply) readResponse; + returnFuture.set(Optional.>fromNullable(reply.getNormalizedNode())); + + } else if (readResponse.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)) { + ReadDataReply reply = ReadDataReply.fromSerializable(schemaContext, path, readResponse); + returnFuture.set(Optional.>fromNullable(reply.getNormalizedNode())); + } else { returnFuture.setException(new ReadFailedException( - "Invalid response reading data for path " + path)); + "Invalid response reading data for path " + path)); } } } }; - Future readFuture = actorContext.executeRemoteOperationAsync(getActor(), - new ReadData(path).toSerializable()); + ReadData readData = new ReadData(path); + Future readFuture = actorContext.executeOperationAsync(getActor(), + isTxActorLocal ? readData : readData.toSerializable()); + readFuture.onComplete(onComplete, actorContext.getActorSystem().dispatcher()); } @@ -762,9 +772,13 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { if(LOG.isDebugEnabled()) { LOG.debug("Tx {} dataExists operation succeeded", identifier, failure); } - if (response.getClass().equals(DataExistsReply.SERIALIZABLE_CLASS)) { - returnFuture.set(Boolean.valueOf(DataExistsReply. - fromSerializable(response).exists())); + + if (response instanceof DataExistsReply) { + returnFuture.set(Boolean.valueOf(((DataExistsReply) response).exists())); + + } else if (response.getClass().equals(DataExistsReply.SERIALIZABLE_CLASS)) { + returnFuture.set(Boolean.valueOf(DataExistsReply.fromSerializable(response).exists())); + } else { returnFuture.setException(new ReadFailedException( "Invalid response checking exists for path " + path)); @@ -773,8 +787,10 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { } }; - Future future = actorContext.executeRemoteOperationAsync(getActor(), - new DataExists(path).toSerializable()); + DataExists dataExists = new DataExists(path); + Future future = actorContext.executeOperationAsync(getActor(), + isTxActorLocal ? dataExists : dataExists.toSerializable()); + future.onComplete(onComplete, actorContext.getActorSystem().dispatcher()); } } @@ -799,7 +815,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { } @Override - public Future readyTransaction() { + public Future readyTransaction() { if(LOG.isDebugEnabled()) { LOG.debug("Tx {} readyTransaction called", identifier); }