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%2FThreePhaseCommitCohortProxy.java;h=c479da73127760977d4c27b3ff9873d10c295c57;hp=a7a5b31b174e4e0d03db192aa367a43ebe67ad62;hb=61722bffcf885c04d11f684f2f03b09fa96f2002;hpb=5a15471e74536f8fe6d62747b7b822655a17dd4e diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxy.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxy.java index a7a5b31b17..c479da7312 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxy.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxy.java @@ -8,35 +8,28 @@ package org.opendaylight.controller.cluster.datastore; -import akka.actor.ActorPath; import akka.actor.ActorSelection; import akka.dispatch.Futures; import akka.dispatch.OnComplete; - import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Lists; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.SettableFuture; - +import java.util.Collections; +import java.util.List; import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction; import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction; import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction; import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply; -import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransaction; -import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransactionReply; import org.opendaylight.controller.cluster.datastore.utils.ActorContext; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import scala.concurrent.Future; import scala.runtime.AbstractFunction1; -import java.util.Collections; -import java.util.List; - /** * ThreePhaseCommitCohortProxy represents a set of remote cohort proxies */ @@ -44,34 +37,50 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho private static final Logger LOG = LoggerFactory.getLogger(ThreePhaseCommitCohortProxy.class); + private static final ListenableFuture IMMEDIATE_SUCCESS = + com.google.common.util.concurrent.Futures.immediateFuture(null); + private final ActorContext actorContext; - private final List> cohortPathFutures; - private volatile List cohortPaths; + private final List> cohortFutures; + private volatile List cohorts; private final String transactionId; + private static final OperationCallback NO_OP_CALLBACK = new OperationCallback() { + @Override + public void run() { + } + + @Override + public void success() { + } + + @Override + public void failure() { + } + }; public ThreePhaseCommitCohortProxy(ActorContext actorContext, - List> cohortPathFutures, String transactionId) { + List> cohortFutures, String transactionId) { this.actorContext = actorContext; - this.cohortPathFutures = cohortPathFutures; + this.cohortFutures = cohortFutures; this.transactionId = transactionId; } - private Future buildCohortPathsList() { + private Future buildCohortList() { - Future> combinedFutures = Futures.sequence(cohortPathFutures, - actorContext.getActorSystem().dispatcher()); + Future> combinedFutures = Futures.sequence(cohortFutures, + actorContext.getClientDispatcher()); - return combinedFutures.transform(new AbstractFunction1, Void>() { + return combinedFutures.transform(new AbstractFunction1, Void>() { @Override - public Void apply(Iterable paths) { - cohortPaths = Lists.newArrayList(paths); + public Void apply(Iterable actorSelections) { + cohorts = Lists.newArrayList(actorSelections); if(LOG.isDebugEnabled()) { LOG.debug("Tx {} successfully built cohort path list: {}", - transactionId, cohortPaths); + transactionId, cohorts); } return null; } - }, TransactionProxy.SAME_FAILURE_TRANSFORMER, actorContext.getActorSystem().dispatcher()); + }, TransactionProxy.SAME_FAILURE_TRANSFORMER, actorContext.getClientDispatcher()); } @Override @@ -87,19 +96,19 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho // extracted from ReadyTransactionReply messages by the Futures that were obtained earlier // and passed to us from upstream processing. If any one fails then we'll fail canCommit. - buildCohortPathsList().onComplete(new OnComplete() { + buildCohortList().onComplete(new OnComplete() { @Override public void onComplete(Throwable failure, Void notUsed) throws Throwable { if(failure != null) { if(LOG.isDebugEnabled()) { - LOG.debug("Tx {}: a cohort path Future failed: {}", transactionId, failure); + LOG.debug("Tx {}: a cohort Future failed: {}", transactionId, failure); } returnFuture.setException(failure); } else { finishCanCommit(returnFuture); } } - }, actorContext.getActorSystem().dispatcher()); + }, actorContext.getClientDispatcher()); return returnFuture; } @@ -112,7 +121,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho // their canCommit processing. If any one fails then we'll fail canCommit. Future> combinedFuture = - invokeCohorts(new CanCommitTransaction().toSerializable()); + invokeCohorts(new CanCommitTransaction(transactionId).toSerializable()); combinedFuture.onComplete(new OnComplete>() { @Override @@ -137,7 +146,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho } else { LOG.error("Unexpected response type {}", response.getClass()); returnFuture.setException(new IllegalArgumentException( - String.format("Unexpected response type {}", response.getClass()))); + String.format("Unexpected response type %s", response.getClass()))); return; } } @@ -146,27 +155,26 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho } returnFuture.set(Boolean.valueOf(result)); } - }, actorContext.getActorSystem().dispatcher()); + }, actorContext.getClientDispatcher()); } private Future> invokeCohorts(Object message) { - List> futureList = Lists.newArrayListWithCapacity(cohortPaths.size()); - for(ActorPath actorPath : cohortPaths) { + List> futureList = Lists.newArrayListWithCapacity(cohorts.size()); + for(ActorSelection cohort : cohorts) { if(LOG.isDebugEnabled()) { - LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message, actorPath); + LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message, cohort); } - ActorSelection cohort = actorContext.actorSelection(actorPath); - - futureList.add(actorContext.executeRemoteOperationAsync(cohort, message)); + futureList.add(actorContext.executeOperationAsync(cohort, message, actorContext.getTransactionCommitOperationTimeout())); } - return Futures.sequence(futureList, actorContext.getActorSystem().dispatcher()); + return Futures.sequence(futureList, actorContext.getClientDispatcher()); } @Override public ListenableFuture preCommit() { - return voidOperation("preCommit", new PreCommitTransaction().toSerializable(), - PreCommitTransactionReply.SERIALIZABLE_CLASS, true); + // We don't need to do anything here - preCommit is done atomically with the commit phase + // by the shard. + return IMMEDIATE_SUCCESS; } @Override @@ -177,18 +185,26 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho // exception then that exception will supersede and suppress the original exception. But // it's the original exception that is the root cause and of more interest to the client. - return voidOperation("abort", new AbortTransaction().toSerializable(), + return voidOperation("abort", new AbortTransaction(transactionId).toSerializable(), AbortTransactionReply.SERIALIZABLE_CLASS, false); } @Override public ListenableFuture commit() { - return voidOperation("commit", new CommitTransaction().toSerializable(), - CommitTransactionReply.SERIALIZABLE_CLASS, true); + OperationCallback operationCallback = (cohortFutures.size() == 0) ? NO_OP_CALLBACK : + new TransactionRateLimitingCallback(actorContext); + + return voidOperation("commit", new CommitTransaction(transactionId).toSerializable(), + CommitTransactionReply.SERIALIZABLE_CLASS, true, operationCallback); + } + + private ListenableFuture voidOperation(final String operationName, final Object message, + final Class expectedResponseClass, final boolean propagateException) { + return voidOperation(operationName, message, expectedResponseClass, propagateException, NO_OP_CALLBACK); } private ListenableFuture voidOperation(final String operationName, final Object message, - final Class expectedResponseClass, final boolean propagateException) { + final Class expectedResponseClass, final boolean propagateException, final OperationCallback callback) { if(LOG.isDebugEnabled()) { LOG.debug("Tx {} {}", transactionId, operationName); @@ -198,11 +214,11 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho // The cohort actor list should already be built at this point by the canCommit phase but, // if not for some reason, we'll try to build it here. - if(cohortPaths != null) { + if(cohorts != null) { finishVoidOperation(operationName, message, expectedResponseClass, propagateException, - returnFuture); + returnFuture, callback); } else { - buildCohortPathsList().onComplete(new OnComplete() { + buildCohortList().onComplete(new OnComplete() { @Override public void onComplete(Throwable failure, Void notUsed) throws Throwable { if(failure != null) { @@ -217,21 +233,24 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho } } else { finishVoidOperation(operationName, message, expectedResponseClass, - propagateException, returnFuture); + propagateException, returnFuture, callback); } } - }, actorContext.getActorSystem().dispatcher()); + }, actorContext.getClientDispatcher()); } return returnFuture; } private void finishVoidOperation(final String operationName, final Object message, - final Class expectedResponseClass, final boolean propagateException, - final SettableFuture returnFuture) { + final Class expectedResponseClass, final boolean propagateException, + final SettableFuture returnFuture, final OperationCallback callback) { if(LOG.isDebugEnabled()) { LOG.debug("Tx {} finish {}", transactionId, operationName); } + + callback.run(); + Future> combinedFuture = invokeCohorts(message); combinedFuture.onComplete(new OnComplete>() { @@ -243,7 +262,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho for(Object response: responses) { if(!response.getClass().equals(expectedResponseClass)) { exceptionToPropagate = new IllegalArgumentException( - String.format("Unexpected response type {}", + String.format("Unexpected response type %s", response.getClass())); break; } @@ -251,6 +270,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho } if(exceptionToPropagate != null) { + if(LOG.isDebugEnabled()) { LOG.debug("Tx {}: a {} cohort Future failed: {}", transactionId, operationName, exceptionToPropagate); @@ -269,18 +289,23 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho } returnFuture.set(null); } + + callback.failure(); } else { + if(LOG.isDebugEnabled()) { LOG.debug("Tx {}: {} succeeded", transactionId, operationName); } returnFuture.set(null); + + callback.success(); } } - }, actorContext.getActorSystem().dispatcher()); + }, actorContext.getClientDispatcher()); } @VisibleForTesting - List> getCohortPathFutures() { - return Collections.unmodifiableList(cohortPathFutures); + List> getCohortFutures() { + return Collections.unmodifiableList(cohortFutures); } }