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=8f6271605727abba421dafd8bb9dee3a070bdd0c;hp=09b6568e1ade016accf0ab80788065950d469ec1;hb=de64c6bbf2d5aeb51f4036f9dd606a9bf6f71afb;hpb=06691686cff16d5766a1431ecb43d22ca410b71c 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 09b6568e1a..8f62716057 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 @@ -9,14 +9,19 @@ package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorSelection; -import akka.dispatch.Futures; import akka.dispatch.OnComplete; +import com.google.common.base.Preconditions; +import com.google.common.base.Supplier; 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.SettableFuture; -import java.util.Collections; +import java.util.ArrayList; import java.util.Iterator; import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +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; @@ -27,51 +32,92 @@ import org.opendaylight.controller.cluster.datastore.utils.ActorContext; 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 extends AbstractThreePhaseCommitCohort { private static final Logger LOG = LoggerFactory.getLogger(ThreePhaseCommitCohortProxy.class); + private static final MessageSupplier COMMIT_MESSAGE_SUPPLIER = new MessageSupplier() { + @Override + public Object newMessage(TransactionIdentifier transactionId, short version) { + return new CommitTransaction(transactionId, version).toSerializable(); + } + + @Override + public boolean isSerializedReplyType(Object reply) { + return CommitTransactionReply.isSerializedType(reply); + } + }; + + private static final MessageSupplier ABORT_MESSAGE_SUPPLIER = new MessageSupplier() { + @Override + public Object newMessage(TransactionIdentifier transactionId, short version) { + return new AbortTransaction(transactionId, version).toSerializable(); + } + + @Override + public boolean isSerializedReplyType(Object reply) { + return AbortTransactionReply.isSerializedType(reply); + } + }; + private final ActorContext actorContext; - private final List> cohortFutures; - private volatile List cohorts; - private final String transactionId; + private final List cohorts; + private final SettableFuture cohortsResolvedFuture = SettableFuture.create(); + private final TransactionIdentifier transactionId; private volatile OperationCallback commitOperationCallback; - public ThreePhaseCommitCohortProxy(ActorContext actorContext, - List> cohortFutures, String transactionId) { + public ThreePhaseCommitCohortProxy(ActorContext actorContext, List cohorts, + TransactionIdentifier transactionId) { this.actorContext = actorContext; - this.cohortFutures = cohortFutures; - this.transactionId = transactionId; - } + this.cohorts = cohorts; + this.transactionId = Preconditions.checkNotNull(transactionId); - private Future buildCohortList() { + if (cohorts.isEmpty()) { + cohortsResolvedFuture.set(null); + } + } - Future> combinedFutures = Futures.sequence(cohortFutures, - actorContext.getClientDispatcher()); + private ListenableFuture resolveCohorts() { + if (cohortsResolvedFuture.isDone()) { + return cohortsResolvedFuture; + } - return combinedFutures.transform(new AbstractFunction1, Void>() { - @Override - public Void apply(Iterable actorSelections) { - cohorts = Lists.newArrayList(actorSelections); - if(LOG.isDebugEnabled()) { - LOG.debug("Tx {} successfully built cohort path list: {}", - transactionId, cohorts); + final AtomicInteger completed = new AtomicInteger(cohorts.size()); + final Object lock = new Object(); + for (final CohortInfo info: cohorts) { + info.getActorFuture().onComplete(new OnComplete() { + @Override + public void onComplete(Throwable failure, 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; - } - }, TransactionReadyReplyMapper.SAME_FAILURE_TRANSFORMER, actorContext.getClientDispatcher()); + }, actorContext.getClientDispatcher()); + } + + return cohortsResolvedFuture; } @Override public ListenableFuture canCommit() { - if(LOG.isDebugEnabled()) { - LOG.debug("Tx {} canCommit", transactionId); - } + LOG.debug("Tx {} canCommit", transactionId); + final SettableFuture returnFuture = SettableFuture.create(); // The first phase of canCommit is to gather the list of cohort actor paths that will @@ -80,53 +126,42 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort< // 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() { + Futures.addCallback(resolveCohorts(), new FutureCallback() { @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(Void notUsed) { + finishCanCommit(returnFuture); } - }, actorContext.getClientDispatcher()); + + @Override + public void onFailure(Throwable failure) { + returnFuture.setException(failure); + } + }); return returnFuture; } private void finishCanCommit(final SettableFuture returnFuture) { - if(LOG.isDebugEnabled()) { - LOG.debug("Tx {} finishCanCommit", transactionId); - } + LOG.debug("Tx {} finishCanCommit", transactionId); // For empty transactions return immediately - if(cohorts.size() == 0){ - if(LOG.isDebugEnabled()) { - LOG.debug("Tx {}: canCommit returning result: {}", transactionId, true); - } + if (cohorts.size() == 0) { + LOG.debug("Tx {}: canCommit returning result true", transactionId); returnFuture.set(Boolean.TRUE); return; } - commitOperationCallback = cohortFutures.isEmpty() ? OperationCallback.NO_OP_CALLBACK : - new TransactionRateLimitingCallback(actorContext); - + commitOperationCallback = new TransactionRateLimitingCallback(actorContext); commitOperationCallback.run(); - final Object message = new CanCommitTransaction(transactionId).toSerializable(); - - final Iterator iterator = cohorts.iterator(); + final Iterator iterator = cohorts.iterator(); final OnComplete onComplete = new OnComplete() { @Override - public void onComplete(Throwable failure, Object response) throws Throwable { + public void onComplete(Throwable failure, Object response) { if (failure != null) { - if (LOG.isDebugEnabled()) { - LOG.debug("Tx {}: a canCommit cohort Future failed: {}", transactionId, failure); - } + LOG.debug("Tx {}: a canCommit cohort Future failed", transactionId, failure); + returnFuture.setException(failure); commitOperationCallback.failure(); return; @@ -137,9 +172,11 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort< commitOperationCallback.pause(); boolean result = true; - if (response.getClass().equals(CanCommitTransactionReply.SERIALIZABLE_CLASS)) { - CanCommitTransactionReply reply = - CanCommitTransactionReply.fromSerializable(response); + if (CanCommitTransactionReply.isSerializedType(response)) { + CanCommitTransactionReply reply = CanCommitTransactionReply.fromSerializable(response); + + LOG.debug("Tx {}: received {}", transactionId, response); + if (!reply.getCanCommit()) { result = false; } @@ -150,35 +187,41 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort< return; } - if(iterator.hasNext() && result){ - Future future = actorContext.executeOperationAsync(iterator.next(), message, - actorContext.getTransactionCommitOperationTimeout()); - future.onComplete(this, actorContext.getClientDispatcher()); + if (iterator.hasNext() && result) { + sendCanCommitTransaction(iterator.next(), this); } else { - if(LOG.isDebugEnabled()) { - LOG.debug("Tx {}: canCommit returning result: {}", transactionId, result); - } + LOG.debug("Tx {}: canCommit returning result: {}", transactionId, result); returnFuture.set(Boolean.valueOf(result)); } } }; - Future future = actorContext.executeOperationAsync(iterator.next(), message, - actorContext.getTransactionCommitOperationTimeout()); + sendCanCommitTransaction(iterator.next(), onComplete); + } + + private void sendCanCommitTransaction(CohortInfo toCohortInfo, OnComplete onComplete) { + CanCommitTransaction message = new CanCommitTransaction(transactionId, toCohortInfo.getActorVersion()); + + LOG.debug("Tx {}: sending {} to {}", transactionId, message, toCohortInfo.getResolvedActor()); + + Future future = actorContext.executeOperationAsync(toCohortInfo.getResolvedActor(), + message.toSerializable(), actorContext.getTransactionCommitOperationTimeout()); future.onComplete(onComplete, actorContext.getClientDispatcher()); } - private Future> invokeCohorts(Object message) { + private Future> invokeCohorts(MessageSupplier messageSupplier) { List> 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())); + for (CohortInfo cohort : cohorts) { + Object message = messageSupplier.newMessage(transactionId, cohort.getActorVersion()); + + LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message , cohort); + + futureList.add(actorContext.executeOperationAsync(cohort.getResolvedActor(), message, + actorContext.getTransactionCommitOperationTimeout())); } - return Futures.sequence(futureList, actorContext.getClientDispatcher()); + return akka.dispatch.Futures.sequence(futureList, actorContext.getClientDispatcher()); } @Override @@ -196,8 +239,8 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort< // 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 @@ -205,89 +248,88 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort< 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 voidOperation(final String operationName, final Object message, - final Class expectedResponseClass, final boolean propagateException) { - return voidOperation(operationName, message, expectedResponseClass, propagateException, - OperationCallback.NO_OP_CALLBACK); + @SuppressWarnings("checkstyle:IllegalCatch") + private static boolean successfulFuture(ListenableFuture future) { + if (!future.isDone()) { + return false; + } + + try { + future.get(); + return true; + } catch (Exception e) { + return false; + } } - private ListenableFuture voidOperation(final String operationName, final Object message, - final Class expectedResponseClass, final boolean propagateException, final OperationCallback callback) { + private ListenableFuture 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 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 future = resolveCohorts(); + if (successfulFuture(future)) { + finishVoidOperation(operationName, messageSupplier, expectedResponseClass, propagateException, returnFuture, callback); } else { - buildCohortList().onComplete(new OnComplete() { + Futures.addCallback(future, new FutureCallback() { @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(Void notUsed) { + finishVoidOperation(operationName, messageSupplier, expectedResponseClass, + propagateException, returnFuture, callback); + } + + @Override + public void onFailure(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()); + }); } return returnFuture; } - private void finishVoidOperation(final String operationName, final Object message, + private void finishVoidOperation(final String operationName, MessageSupplier messageSupplier, final Class expectedResponseClass, final boolean propagateException, final SettableFuture returnFuture, final OperationCallback callback) { - if(LOG.isDebugEnabled()) { - LOG.debug("Tx {} finish {}", transactionId, operationName); - } + LOG.debug("Tx {} finish {}", transactionId, operationName); callback.resume(); - Future> combinedFuture = invokeCohorts(message); + Future> combinedFuture = invokeCohorts(messageSupplier); combinedFuture.onComplete(new OnComplete>() { @Override public void onComplete(Throwable failure, Iterable responses) throws Throwable { - 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); @@ -295,19 +337,13 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort< // 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(); @@ -318,6 +354,46 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort< @Override List> getCohortFutures() { - return Collections.unmodifiableList(cohortFutures); + List> cohortFutures = new ArrayList<>(cohorts.size()); + for (CohortInfo info: cohorts) { + cohortFutures.add(info.getActorFuture()); + } + + return cohortFutures; + } + + static class CohortInfo { + private final Future actorFuture; + private volatile ActorSelection resolvedActor; + private final Supplier actorVersionSupplier; + + CohortInfo(Future actorFuture, Supplier actorVersionSupplier) { + this.actorFuture = actorFuture; + this.actorVersionSupplier = actorVersionSupplier; + } + + Future getActorFuture() { + return actorFuture; + } + + ActorSelection getResolvedActor() { + return resolvedActor; + } + + void setResolvedActor(ActorSelection resolvedActor) { + this.resolvedActor = resolvedActor; + } + + short getActorVersion() { + Preconditions.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); } }