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%2FThreePhaseCommitCohortProxy.java;h=4445b14e2edc1b48b5c92012e8a659275357f433;hb=cad0f415169cae900315546bf2dae7801d2ee5e2;hp=6e5ba1c90c998102d8878d6b8e6a7c659a475ea4;hpb=139937c2e646894af6a9b2b8a8a1047c6ef82485;p=controller.git 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 6e5ba1c90c..4445b14e2e 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,10 +11,15 @@ 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.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,8 +32,6 @@ 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,6 +47,19 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho 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) { @@ -55,7 +71,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 @@ -67,7 +83,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho } return null; } - }, TransactionProxy.SAME_FAILURE_TRANSFORMER, actorContext.getActorSystem().dispatcher()); + }, TransactionProxy.SAME_FAILURE_TRANSFORMER, actorContext.getClientDispatcher()); } @Override @@ -95,7 +111,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho finishCanCommit(returnFuture); } } - }, actorContext.getActorSystem().dispatcher()); + }, actorContext.getClientDispatcher()); return returnFuture; } @@ -133,7 +149,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; } } @@ -142,7 +158,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho } returnFuture.set(Boolean.valueOf(result)); } - }, actorContext.getActorSystem().dispatcher()); + }, actorContext.getClientDispatcher()); } private Future> invokeCohorts(Object message) { @@ -151,11 +167,10 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho if(LOG.isDebugEnabled()) { LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message, cohort); } - - futureList.add(actorContext.executeOperationAsync(cohort, message)); + futureList.add(actorContext.executeOperationAsync(cohort, message, actorContext.getTransactionCommitOperationTimeout())); } - return Futures.sequence(futureList, actorContext.getActorSystem().dispatcher()); + return Futures.sequence(futureList, actorContext.getClientDispatcher()); } @Override @@ -179,12 +194,20 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho @Override public ListenableFuture commit() { - return voidOperation("commit", new CommitTransaction(transactionId).toSerializable(), - CommitTransactionReply.SERIALIZABLE_CLASS, true); + OperationCallback operationCallback = (cohortFutures.size() == 0) ? NO_OP_CALLBACK : + new CommitCallback(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) { + 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 OperationCallback callback) { if(LOG.isDebugEnabled()) { LOG.debug("Tx {} {}", transactionId, operationName); @@ -196,7 +219,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho if(cohorts != null) { finishVoidOperation(operationName, message, expectedResponseClass, propagateException, - returnFuture); + returnFuture, callback); } else { buildCohortList().onComplete(new OnComplete() { @Override @@ -213,21 +236,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>() { @@ -239,7 +265,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; } @@ -247,6 +273,7 @@ public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCoho } if(exceptionToPropagate != null) { + if(LOG.isDebugEnabled()) { LOG.debug("Tx {}: a {} cohort Future failed: {}", transactionId, operationName, exceptionToPropagate); @@ -265,18 +292,77 @@ 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> 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.get95thPercentile(); + + 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 + } + } + }