* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-
package org.opendaylight.controller.cluster.datastore;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
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.FutureCallback;
+import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
-import java.util.Collections;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
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.utils.ActorContext;
-import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
-import scala.runtime.AbstractFunction1;
/**
- * ThreePhaseCommitCohortProxy represents a set of remote cohort proxies
+ * ThreePhaseCommitCohortProxy represents a set of remote cohort proxies.
*/
-public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCohort{
+public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort<ActorSelection> {
private static final Logger LOG = LoggerFactory.getLogger(ThreePhaseCommitCohortProxy.class);
- private static final ListenableFuture<Void> IMMEDIATE_SUCCESS =
- com.google.common.util.concurrent.Futures.immediateFuture(null);
+ private static final MessageSupplier COMMIT_MESSAGE_SUPPLIER = new MessageSupplier() {
+ @Override
+ public Object newMessage(final TransactionIdentifier transactionId, final short version) {
+ return new CommitTransaction(transactionId, version).toSerializable();
+ }
- private final ActorContext actorContext;
- private final List<Future<ActorSelection>> cohortFutures;
- private volatile List<ActorSelection> cohorts;
- private final String transactionId;
- private static final OperationCallback NO_OP_CALLBACK = new OperationCallback() {
@Override
- public void run() {
+ public boolean isSerializedReplyType(final Object reply) {
+ return CommitTransactionReply.isSerializedType(reply);
}
+ };
+ private static final MessageSupplier ABORT_MESSAGE_SUPPLIER = new MessageSupplier() {
@Override
- public void success() {
+ public Object newMessage(final TransactionIdentifier transactionId, final short version) {
+ return new AbortTransaction(transactionId, version).toSerializable();
}
@Override
- public void failure() {
+ public boolean isSerializedReplyType(final Object reply) {
+ return AbortTransactionReply.isSerializedType(reply);
}
};
- public ThreePhaseCommitCohortProxy(ActorContext actorContext,
- List<Future<ActorSelection>> cohortFutures, String transactionId) {
- this.actorContext = actorContext;
- this.cohortFutures = cohortFutures;
- this.transactionId = transactionId;
- }
+ private final ActorUtils actorUtils;
+ private final List<CohortInfo> cohorts;
+ private final SettableFuture<Void> cohortsResolvedFuture = SettableFuture.create();
+ private final TransactionIdentifier transactionId;
+ private volatile OperationCallback commitOperationCallback;
- private Future<Void> buildCohortList() {
+ public ThreePhaseCommitCohortProxy(final ActorUtils actorUtils, final List<CohortInfo> cohorts,
+ final TransactionIdentifier transactionId) {
+ this.actorUtils = actorUtils;
+ this.cohorts = cohorts;
+ this.transactionId = requireNonNull(transactionId);
- Future<Iterable<ActorSelection>> combinedFutures = Futures.sequence(cohortFutures,
- actorContext.getClientDispatcher());
+ if (cohorts.isEmpty()) {
+ cohortsResolvedFuture.set(null);
+ }
+ }
- return combinedFutures.transform(new AbstractFunction1<Iterable<ActorSelection>, Void>() {
- @Override
- public Void apply(Iterable<ActorSelection> actorSelections) {
- cohorts = Lists.newArrayList(actorSelections);
- if(LOG.isDebugEnabled()) {
- LOG.debug("Tx {} successfully built cohort path list: {}",
- transactionId, cohorts);
+ private ListenableFuture<Void> resolveCohorts() {
+ if (cohortsResolvedFuture.isDone()) {
+ return cohortsResolvedFuture;
+ }
+
+ final AtomicInteger completed = new AtomicInteger(cohorts.size());
+ final Object lock = new Object();
+ for (final CohortInfo info: cohorts) {
+ info.getActorFuture().onComplete(new OnComplete<ActorSelection>() {
+ @Override
+ public void onComplete(final Throwable failure, final ActorSelection actor) {
+ synchronized (lock) {
+ boolean done = completed.decrementAndGet() == 0;
+ if (failure != null) {
+ LOG.debug("Tx {}: a cohort Future failed", transactionId, failure);
+ cohortsResolvedFuture.setException(failure);
+ } else if (!cohortsResolvedFuture.isDone()) {
+ LOG.debug("Tx {}: cohort actor {} resolved", transactionId, actor);
+
+ info.setResolvedActor(actor);
+ if (done) {
+ LOG.debug("Tx {}: successfully resolved all cohort actors", transactionId);
+ cohortsResolvedFuture.set(null);
+ }
+ }
+ }
}
- return null;
- }
- }, TransactionProxy.SAME_FAILURE_TRANSFORMER, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
+ }
+
+ return cohortsResolvedFuture;
}
@Override
public ListenableFuture<Boolean> canCommit() {
- if(LOG.isDebugEnabled()) {
- LOG.debug("Tx {} canCommit", transactionId);
- }
+ LOG.debug("Tx {} canCommit", transactionId);
+
final SettableFuture<Boolean> returnFuture = SettableFuture.create();
// The first phase of canCommit is to gather the list of cohort actor paths that will
// 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.
- buildCohortList().onComplete(new OnComplete<Void>() {
+ Futures.addCallback(resolveCohorts(), new FutureCallback<Void>() {
@Override
- public void onComplete(Throwable failure, Void notUsed) throws Throwable {
- if(failure != null) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("Tx {}: a cohort Future failed: {}", transactionId, failure);
- }
- returnFuture.setException(failure);
- } else {
- finishCanCommit(returnFuture);
- }
+ public void onSuccess(final Void notUsed) {
+ finishCanCommit(returnFuture);
+ }
+
+ @Override
+ public void onFailure(final Throwable failure) {
+ returnFuture.setException(failure);
}
- }, actorContext.getClientDispatcher());
+ }, MoreExecutors.directExecutor());
return returnFuture;
}
+ @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
+ justification = "https://github.com/spotbugs/spotbugs/issues/811")
private void finishCanCommit(final SettableFuture<Boolean> returnFuture) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("Tx {} finishCanCommit", transactionId);
+ LOG.debug("Tx {} finishCanCommit", transactionId);
+
+ // For empty transactions return immediately
+ if (cohorts.size() == 0) {
+ LOG.debug("Tx {}: canCommit returning result true", transactionId);
+ returnFuture.set(Boolean.TRUE);
+ return;
}
- // 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<Iterable<Object>> combinedFuture =
- invokeCohorts(new CanCommitTransaction(transactionId).toSerializable());
+ commitOperationCallback = new TransactionRateLimitingCallback(actorUtils);
+ commitOperationCallback.run();
- combinedFuture.onComplete(new OnComplete<Iterable<Object>>() {
+ final Iterator<CohortInfo> iterator = cohorts.iterator();
+
+ final OnComplete<Object> onComplete = new OnComplete<Object>() {
@Override
- public void onComplete(Throwable failure, Iterable<Object> responses) throws Throwable {
- if(failure != null) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("Tx {}: a canCommit cohort Future failed: {}", transactionId, failure);
- }
+ public void onComplete(final Throwable failure, final Object response) {
+ if (failure != null) {
+ LOG.debug("Tx {}: a canCommit cohort Future failed", transactionId, failure);
+
returnFuture.setException(failure);
+ commitOperationCallback.failure();
return;
}
+ // Only the first call to pause takes effect - subsequent calls before resume are no-ops. So
+ // this means we'll only time the first transaction canCommit which should be fine.
+ commitOperationCallback.pause();
+
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 (CanCommitTransactionReply.isSerializedType(response)) {
+ CanCommitTransactionReply reply = CanCommitTransactionReply.fromSerializable(response);
+
+ LOG.debug("Tx {}: received {}", transactionId, 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()) {
+
+ if (iterator.hasNext() && result) {
+ sendCanCommitTransaction(iterator.next(), this);
+ } else {
LOG.debug("Tx {}: canCommit returning result: {}", transactionId, result);
+ returnFuture.set(Boolean.valueOf(result));
}
- returnFuture.set(Boolean.valueOf(result));
+
}
- }, actorContext.getClientDispatcher());
+ };
+
+ sendCanCommitTransaction(iterator.next(), onComplete);
}
- private Future<Iterable<Object>> invokeCohorts(Object message) {
- List<Future<Object>> futureList = Lists.newArrayListWithCapacity(cohorts.size());
- for(ActorSelection cohort : cohorts) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message, cohort);
- }
- futureList.add(actorContext.executeOperationAsync(cohort, message, actorContext.getTransactionCommitOperationTimeout()));
+ private void sendCanCommitTransaction(final CohortInfo toCohortInfo, final OnComplete<Object> onComplete) {
+ CanCommitTransaction message = new CanCommitTransaction(transactionId, toCohortInfo.getActorVersion());
+
+ LOG.debug("Tx {}: sending {} to {}", transactionId, message, toCohortInfo.getResolvedActor());
+
+ Future<Object> future = actorUtils.executeOperationAsync(toCohortInfo.getResolvedActor(),
+ message.toSerializable(), actorUtils.getTransactionCommitOperationTimeout());
+ future.onComplete(onComplete, actorUtils.getClientDispatcher());
+ }
+
+ private Future<Iterable<Object>> invokeCohorts(final MessageSupplier messageSupplier) {
+ List<Future<Object>> futureList = new ArrayList<>(cohorts.size());
+ for (CohortInfo cohort : cohorts) {
+ Object message = messageSupplier.newMessage(transactionId, cohort.getActorVersion());
+
+ LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message , cohort.getResolvedActor());
+
+ futureList.add(actorUtils.executeOperationAsync(cohort.getResolvedActor(), message,
+ actorUtils.getTransactionCommitOperationTimeout()));
}
- return Futures.sequence(futureList, actorContext.getClientDispatcher());
+ return akka.dispatch.Futures.sequence(futureList, actorUtils.getClientDispatcher());
}
@Override
public ListenableFuture<Void> 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
// 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(transactionId).toSerializable(),
- AbortTransactionReply.SERIALIZABLE_CLASS, false);
+ return voidOperation("abort", ABORT_MESSAGE_SUPPLIER,
+ AbortTransactionReply.class, false, OperationCallback.NO_OP_CALLBACK);
}
@Override
public ListenableFuture<Void> commit() {
- OperationCallback operationCallback = (cohortFutures.size() == 0) ? NO_OP_CALLBACK :
- new CommitCallback(actorContext);
+ OperationCallback operationCallback = commitOperationCallback != null ? commitOperationCallback :
+ OperationCallback.NO_OP_CALLBACK;
- return voidOperation("commit", new CommitTransaction(transactionId).toSerializable(),
- CommitTransactionReply.SERIALIZABLE_CLASS, true, operationCallback);
+ return voidOperation("commit", COMMIT_MESSAGE_SUPPLIER,
+ CommitTransactionReply.class, true, operationCallback);
}
- private ListenableFuture<Void> voidOperation(final String operationName, final Object message,
- final Class<?> expectedResponseClass, final boolean propagateException) {
- return voidOperation(operationName, message, expectedResponseClass, propagateException, NO_OP_CALLBACK);
+ @SuppressWarnings("checkstyle:IllegalCatch")
+ private static boolean successfulFuture(final ListenableFuture<Void> future) {
+ if (!future.isDone()) {
+ return false;
+ }
+
+ try {
+ future.get();
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
}
- private ListenableFuture<Void> voidOperation(final String operationName, final Object message,
- final Class<?> expectedResponseClass, final boolean propagateException, final OperationCallback callback) {
+ private ListenableFuture<Void> voidOperation(final String operationName,
+ final MessageSupplier messageSupplier, final Class<?> expectedResponseClass,
+ final boolean propagateException, final OperationCallback callback) {
+ LOG.debug("Tx {} {}", transactionId, operationName);
- if(LOG.isDebugEnabled()) {
- LOG.debug("Tx {} {}", transactionId, operationName);
- }
final SettableFuture<Void> returnFuture = SettableFuture.create();
// 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(cohorts != null) {
- finishVoidOperation(operationName, message, expectedResponseClass, propagateException,
+ ListenableFuture<Void> future = resolveCohorts();
+ if (successfulFuture(future)) {
+ finishVoidOperation(operationName, messageSupplier, expectedResponseClass, propagateException,
returnFuture, callback);
} else {
- buildCohortList().onComplete(new OnComplete<Void>() {
+ Futures.addCallback(future, new FutureCallback<Void>() {
@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,
- operationName, failure);
- }
- if(propagateException) {
- returnFuture.setException(failure);
- } else {
- returnFuture.set(null);
- }
+ public void onSuccess(final Void notUsed) {
+ finishVoidOperation(operationName, messageSupplier, expectedResponseClass,
+ propagateException, returnFuture, callback);
+ }
+
+ @Override
+ public void onFailure(final Throwable failure) {
+ LOG.debug("Tx {}: a {} cohort path Future failed", transactionId, operationName, failure);
+
+ if (propagateException) {
+ returnFuture.setException(failure);
} else {
- finishVoidOperation(operationName, message, expectedResponseClass,
- propagateException, returnFuture, callback);
+ returnFuture.set(null);
}
}
- }, actorContext.getClientDispatcher());
+ }, MoreExecutors.directExecutor());
}
return returnFuture;
}
- private void finishVoidOperation(final String operationName, final Object message,
+ private void finishVoidOperation(final String operationName, final MessageSupplier messageSupplier,
final Class<?> expectedResponseClass, final boolean propagateException,
final SettableFuture<Void> returnFuture, final OperationCallback callback) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("Tx {} finish {}", transactionId, operationName);
- }
+ LOG.debug("Tx {} finish {}", transactionId, operationName);
- callback.run();
+ callback.resume();
- Future<Iterable<Object>> combinedFuture = invokeCohorts(message);
+ Future<Iterable<Object>> combinedFuture = invokeCohorts(messageSupplier);
combinedFuture.onComplete(new OnComplete<Iterable<Object>>() {
@Override
- public void onComplete(Throwable failure, Iterable<Object> responses) throws Throwable {
-
+ public void onComplete(final Throwable failure, final Iterable<Object> responses) {
Throwable exceptionToPropagate = failure;
- if(exceptionToPropagate == null) {
- for(Object response: responses) {
- if(!response.getClass().equals(expectedResponseClass)) {
+ if (exceptionToPropagate == null) {
+ for (Object response: responses) {
+ if (!response.getClass().equals(expectedResponseClass)) {
exceptionToPropagate = new IllegalArgumentException(
- String.format("Unexpected response type %s",
- response.getClass()));
+ String.format("Unexpected response type %s", response.getClass()));
break;
}
}
}
- if(exceptionToPropagate != null) {
-
- if(LOG.isDebugEnabled()) {
- LOG.debug("Tx {}: a {} cohort Future failed: {}", transactionId,
- operationName, exceptionToPropagate);
- }
- if(propagateException) {
+ if (exceptionToPropagate != null) {
+ LOG.debug("Tx {}: a {} cohort Future failed", transactionId, operationName, exceptionToPropagate);
+ if (propagateException) {
// We don't log the exception here to avoid redundant logging since we're
// propagating to the caller in MD-SAL core who will log it.
returnFuture.setException(exceptionToPropagate);
// Since the caller doesn't want us to propagate the exception we'll also
// not log it normally. But it's usually not good to totally silence
// exceptions so we'll log it to debug level.
- if(LOG.isDebugEnabled()) {
- LOG.debug(String.format("%s failed", message.getClass().getSimpleName()),
- exceptionToPropagate);
- }
returnFuture.set(null);
}
callback.failure();
} else {
+ LOG.debug("Tx {}: {} succeeded", transactionId, operationName);
- if(LOG.isDebugEnabled()) {
- LOG.debug("Tx {}: {} succeeded", transactionId, operationName);
- }
returnFuture.set(null);
callback.success();
}
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
}
- @VisibleForTesting
+ @Override
List<Future<ActorSelection>> getCohortFutures() {
- return Collections.unmodifiableList(cohortFutures);
- }
+ List<Future<ActorSelection>> cohortFutures = new ArrayList<>(cohorts.size());
+ for (CohortInfo info: cohorts) {
+ cohortFutures.add(info.getActorFuture());
+ }
- private static interface OperationCallback {
- void run();
- void success();
- void failure();
+ return cohortFutures;
}
- private static class CommitCallback implements OperationCallback{
-
- private static final Logger LOG = LoggerFactory.getLogger(CommitCallback.class);
- private static final String COMMIT = "commit";
+ static class CohortInfo {
+ private final Future<ActorSelection> actorFuture;
+ private final Supplier<Short> actorVersionSupplier;
- private final Timer commitTimer;
- private final ActorContext actorContext;
- private Timer.Context timerContext;
+ private volatile ActorSelection resolvedActor;
- CommitCallback(ActorContext actorContext){
- this.actorContext = actorContext;
- commitTimer = actorContext.getOperationTimer(COMMIT);
+ CohortInfo(final Future<ActorSelection> actorFuture, final Supplier<Short> actorVersionSupplier) {
+ this.actorFuture = actorFuture;
+ this.actorVersionSupplier = actorVersionSupplier;
}
- @Override
- public void run() {
- timerContext = commitTimer.time();
+ Future<ActorSelection> getActorFuture() {
+ return actorFuture;
}
- @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);
+ ActorSelection getResolvedActor() {
+ return resolvedActor;
+ }
- actorContext.setTxCreationLimit(newRateLimit);
+ void setResolvedActor(final ActorSelection resolvedActor) {
+ this.resolvedActor = resolvedActor;
}
- @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
+ short getActorVersion() {
+ checkState(resolvedActor != null, "getActorVersion cannot be called until the actor is resolved");
+ return actorVersionSupplier.get();
}
}
+ private interface MessageSupplier {
+ Object newMessage(TransactionIdentifier transactionId, short version);
+
+ boolean isSerializedReplyType(Object reply);
+ }
}