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=57749a1a736bc935da74222ca984d834083d6206;hp=4f472266c1f56acbd8fc531ae7189f1ae91951b4;hb=2727bea09c83646b6cbd2ef9672d0b7f6cf3b22f;hpb=c68b9f38d84bd45c85b8133c2054ecdd4c413a8f 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 4f472266c1..57749a1a73 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 @@ -11,15 +11,12 @@ package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorSelection; import akka.dispatch.Futures; import akka.dispatch.OnComplete; -import com.codahale.metrics.Snapshot; -import com.codahale.metrics.Timer; -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.Iterator; import java.util.List; -import java.util.concurrent.TimeUnit; import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction; import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction; @@ -27,7 +24,6 @@ import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransacti import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction; import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply; 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; @@ -36,30 +32,14 @@ import scala.runtime.AbstractFunction1; /** * ThreePhaseCommitCohortProxy represents a set of remote cohort proxies */ -public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCohort{ +public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort { 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> 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> cohortFutures, String transactionId) { @@ -71,7 +51,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho private Future buildCohortList() { Future> combinedFutures = Futures.sequence(cohortFutures, - actorContext.getActorSystem().dispatcher()); + actorContext.getClientDispatcher()); return combinedFutures.transform(new AbstractFunction1, Void>() { @Override @@ -83,7 +63,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho } return null; } - }, TransactionProxy.SAME_FAILURE_TRANSFORMER, actorContext.getActorSystem().dispatcher()); + }, TransactionProxy.SAME_FAILURE_TRANSFORMER, actorContext.getClientDispatcher()); } @Override @@ -111,7 +91,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho finishCanCommit(returnFuture); } } - }, actorContext.getActorSystem().dispatcher()); + }, actorContext.getClientDispatcher()); return returnFuture; } @@ -120,17 +100,25 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho if(LOG.isDebugEnabled()) { LOG.debug("Tx {} finishCanCommit", transactionId); } - // The last phase of canCommit is to invoke all the cohort actors asynchronously to perform - // their canCommit processing. If any one fails then we'll fail canCommit. - Future> combinedFuture = - invokeCohorts(new CanCommitTransaction(transactionId).toSerializable()); + // For empty transactions return immediately + if(cohorts.size() == 0){ + if(LOG.isDebugEnabled()) { + LOG.debug("Tx {}: canCommit returning result: {}", transactionId, true); + } + returnFuture.set(Boolean.TRUE); + return; + } - combinedFuture.onComplete(new OnComplete>() { + final Object message = new CanCommitTransaction(transactionId).toSerializable(); + + final Iterator iterator = cohorts.iterator(); + + final OnComplete onComplete = new OnComplete() { @Override - public void onComplete(Throwable failure, Iterable responses) throws Throwable { - if(failure != null) { - if(LOG.isDebugEnabled()) { + public void onComplete(Throwable failure, Object response) throws Throwable { + if (failure != null) { + if (LOG.isDebugEnabled()) { LOG.debug("Tx {}: a canCommit cohort Future failed: {}", transactionId, failure); } returnFuture.setException(failure); @@ -138,27 +126,36 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho } boolean result = true; - for(Object response: responses) { - if (response.getClass().equals(CanCommitTransactionReply.SERIALIZABLE_CLASS)) { - CanCommitTransactionReply reply = - CanCommitTransactionReply.fromSerializable(response); - if (!reply.getCanCommit()) { - result = false; - break; - } - } else { - LOG.error("Unexpected response type {}", response.getClass()); - returnFuture.setException(new IllegalArgumentException( - String.format("Unexpected response type %s", response.getClass()))); - return; + if (response.getClass().equals(CanCommitTransactionReply.SERIALIZABLE_CLASS)) { + CanCommitTransactionReply reply = + CanCommitTransactionReply.fromSerializable(response); + if (!reply.getCanCommit()) { + result = false; } + } else { + LOG.error("Unexpected response type {}", response.getClass()); + returnFuture.setException(new IllegalArgumentException( + String.format("Unexpected response type %s", response.getClass()))); + return; } - if(LOG.isDebugEnabled()) { - LOG.debug("Tx {}: canCommit returning result: {}", transactionId, result); + + if(iterator.hasNext() && result){ + Future future = actorContext.executeOperationAsync(iterator.next(), message, + actorContext.getTransactionCommitOperationTimeout()); + future.onComplete(this, actorContext.getClientDispatcher()); + } else { + if(LOG.isDebugEnabled()) { + LOG.debug("Tx {}: canCommit returning result: {}", transactionId, result); + } + returnFuture.set(Boolean.valueOf(result)); } - returnFuture.set(Boolean.valueOf(result)); + } - }, actorContext.getActorSystem().dispatcher()); + }; + + Future future = actorContext.executeOperationAsync(iterator.next(), message, + actorContext.getTransactionCommitOperationTimeout()); + future.onComplete(onComplete, actorContext.getClientDispatcher()); } private Future> invokeCohorts(Object message) { @@ -170,14 +167,14 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho 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() { // We don't need to do anything here - preCommit is done atomically with the commit phase // by the shard. - return IMMEDIATE_SUCCESS; + return IMMEDIATE_VOID_SUCCESS; } @Override @@ -194,8 +191,8 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho @Override public ListenableFuture commit() { - OperationCallback operationCallback = (cohortFutures.size() == 0) ? NO_OP_CALLBACK : - new CommitCallback(actorContext); + OperationCallback operationCallback = cohortFutures.isEmpty() ? OperationCallback.NO_OP_CALLBACK : + new TransactionRateLimitingCallback(actorContext); return voidOperation("commit", new CommitTransaction(transactionId).toSerializable(), CommitTransactionReply.SERIALIZABLE_CLASS, true, operationCallback); @@ -203,7 +200,8 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho private ListenableFuture voidOperation(final String operationName, final Object message, final Class expectedResponseClass, final boolean propagateException) { - return voidOperation(operationName, message, expectedResponseClass, propagateException, NO_OP_CALLBACK); + return voidOperation(operationName, message, expectedResponseClass, propagateException, + OperationCallback.NO_OP_CALLBACK); } private ListenableFuture voidOperation(final String operationName, final Object message, @@ -239,7 +237,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho propagateException, returnFuture, callback); } } - }, actorContext.getActorSystem().dispatcher()); + }, actorContext.getClientDispatcher()); } return returnFuture; @@ -304,65 +302,11 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho callback.success(); } } - }, actorContext.getActorSystem().dispatcher()); + }, actorContext.getClientDispatcher()); } - @VisibleForTesting + @Override List> getCohortFutures() { return Collections.unmodifiableList(cohortFutures); } - - private static interface OperationCallback { - void run(); - void success(); - void failure(); - } - - private static class CommitCallback implements OperationCallback{ - - private static final Logger LOG = LoggerFactory.getLogger(CommitCallback.class); - private static final String COMMIT = "commit"; - - private final Timer commitTimer; - private final ActorContext actorContext; - private Timer.Context timerContext; - - CommitCallback(ActorContext actorContext){ - this.actorContext = actorContext; - commitTimer = actorContext.getOperationTimer(COMMIT); - } - - @Override - public void run() { - timerContext = commitTimer.time(); - } - - @Override - public void success() { - timerContext.stop(); - - Snapshot timerSnapshot = commitTimer.getSnapshot(); - double allowedLatencyInNanos = timerSnapshot.get98thPercentile(); - - long commitTimeoutInSeconds = actorContext.getDatastoreContext() - .getShardTransactionCommitTimeoutInSeconds(); - long commitTimeoutInNanos = TimeUnit.SECONDS.toNanos(commitTimeoutInSeconds); - - // Here we are trying to find out how many transactions per second are allowed - double newRateLimit = ((double) commitTimeoutInNanos / allowedLatencyInNanos) / commitTimeoutInSeconds; - - LOG.debug("Data Store {} commit rateLimit adjusted to {} allowedLatencyInNanos = {}", - actorContext.getDataStoreType(), newRateLimit, allowedLatencyInNanos); - - actorContext.setTxCreationLimit(newRateLimit); - } - - @Override - public void failure() { - // This would mean we couldn't get a transaction completed in 30 seconds which is - // the default transaction commit timeout. Using the timeout information to figure out the rate limit is - // not going to be useful - so we leave it as it is - } - } - }