opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxy.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.datastore;
  10
  11 import akka.actor.ActorPath;
  12 import akka.actor.ActorSelection;
  13 import akka.dispatch.Futures;
  14 import akka.dispatch.OnComplete;
  15
  16 import com.google.common.annotations.VisibleForTesting;
  17 import com.google.common.collect.Lists;
  18 import com.google.common.util.concurrent.ListenableFuture;
  19 import com.google.common.util.concurrent.SettableFuture;
  20
  21 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
  22 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
  23 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
  24 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
  25 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
  26 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  27 import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransaction;
  28 import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransactionReply;
  29 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  30 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  31 import org.slf4j.Logger;
  32 import org.slf4j.LoggerFactory;
  33
  34 import scala.concurrent.Future;
  35 import scala.runtime.AbstractFunction1;
  36
  37 import java.util.Collections;
  38 import java.util.List;
  39
  40 /**
  41  * ThreePhaseCommitCohortProxy represents a set of remote cohort proxies
  42  */
  43 public class ThreePhaseCommitCohortProxy implements DOMStoreThreePhaseCommitCohort{
  44
  45     private static final Logger LOG = LoggerFactory.getLogger(ThreePhaseCommitCohortProxy.class);
  46
  47     private final ActorContext actorContext;
  48     private final List<Future<ActorPath>> cohortPathFutures;
  49     private volatile List<ActorPath> cohortPaths;
  50     private final String transactionId;
  51
  52     public ThreePhaseCommitCohortProxy(ActorContext actorContext,
  53             List<Future<ActorPath>> cohortPathFutures, String transactionId) {
  54         this.actorContext = actorContext;
  55         this.cohortPathFutures = cohortPathFutures;
  56         this.transactionId = transactionId;
  57     }
  58
  59     private Future<Void> buildCohortPathsList() {
  60
  61         Future<Iterable<ActorPath>> combinedFutures = Futures.sequence(cohortPathFutures,
  62                 actorContext.getActorSystem().dispatcher());
  63
  64         return combinedFutures.transform(new AbstractFunction1<Iterable<ActorPath>, Void>() {
  65             @Override
  66             public Void apply(Iterable<ActorPath> paths) {
  67                 cohortPaths = Lists.newArrayList(paths);
  68
  69                 LOG.debug("Tx {} successfully built cohort path list: {}",
  70                         transactionId, cohortPaths);
  71                 return null;
  72             }
  73         }, TransactionProxy.SAME_FAILURE_TRANSFORMER, actorContext.getActorSystem().dispatcher());
  74     }
  75
  76     @Override
  77     public ListenableFuture<Boolean> canCommit() {
  78         LOG.debug("Tx {} canCommit", transactionId);
  79
  80         final SettableFuture<Boolean> returnFuture = SettableFuture.create();
  81
  82         // The first phase of canCommit is to gather the list of cohort actor paths that will
  83         // participate in the commit. buildCohortPathsList combines the cohort path Futures into
  84         // one Future which we wait on asynchronously here. The cohort actor paths are
  85         // extracted from ReadyTransactionReply messages by the Futures that were obtained earlier
  86         // and passed to us from upstream processing. If any one fails then  we'll fail canCommit.
  87
  88         buildCohortPathsList().onComplete(new OnComplete<Void>() {
  89             @Override
  90             public void onComplete(Throwable failure, Void notUsed) throws Throwable {
  91                 if(failure != null) {
  92                     LOG.debug("Tx {}: a cohort path Future failed: {}", transactionId, failure);
  93                     returnFuture.setException(failure);
  94                 } else {
  95                     finishCanCommit(returnFuture);
  96                 }
  97             }
  98         }, actorContext.getActorSystem().dispatcher());
  99
 100         return returnFuture;
 101     }
 102
 103     private void finishCanCommit(final SettableFuture<Boolean> returnFuture) {
 104
 105         LOG.debug("Tx {} finishCanCommit", transactionId);
 106
 107         // The last phase of canCommit is to invoke all the cohort actors asynchronously to perform
 108         // their canCommit processing. If any one fails then we'll fail canCommit.
 109
 110         Future<Iterable<Object>> combinedFuture =
 111                 invokeCohorts(new CanCommitTransaction().toSerializable());
 112
 113         combinedFuture.onComplete(new OnComplete<Iterable<Object>>() {
 114             @Override
 115             public void onComplete(Throwable failure, Iterable<Object> responses) throws Throwable {
 116                 if(failure != null) {
 117                     LOG.debug("Tx {}: a canCommit cohort Future failed: {}", transactionId, failure);
 118                     returnFuture.setException(failure);
 119                     return;
 120                 }
 121
 122                 boolean result = true;
 123                 for(Object response: responses) {
 124                     if (response.getClass().equals(CanCommitTransactionReply.SERIALIZABLE_CLASS)) {
 125                         CanCommitTransactionReply reply =
 126                                 CanCommitTransactionReply.fromSerializable(response);
 127                         if (!reply.getCanCommit()) {
 128                             result = false;
 129                             break;
 130                         }
 131                     } else {
 132                         LOG.error("Unexpected response type {}", response.getClass());
 133                         returnFuture.setException(new IllegalArgumentException(
 134                                 String.format("Unexpected response type {}", response.getClass())));
 135                         return;
 136                     }
 137                 }
 138
 139                 LOG.debug("Tx {}: canCommit returning result: {}", transactionId, result);
 140
 141                 returnFuture.set(Boolean.valueOf(result));
 142             }
 143         }, actorContext.getActorSystem().dispatcher());
 144     }
 145
 146     private Future<Iterable<Object>> invokeCohorts(Object message) {
 147         List<Future<Object>> futureList = Lists.newArrayListWithCapacity(cohortPaths.size());
 148         for(ActorPath actorPath : cohortPaths) {
 149
 150             LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message, actorPath);
 151
 152             ActorSelection cohort = actorContext.actorSelection(actorPath);
 153
 154             futureList.add(actorContext.executeRemoteOperationAsync(cohort, message,
 155                     ActorContext.ASK_DURATION));
 156         }
 157
 158         return Futures.sequence(futureList, actorContext.getActorSystem().dispatcher());
 159     }
 160
 161     @Override
 162     public ListenableFuture<Void> preCommit() {
 163         return voidOperation("preCommit",  new PreCommitTransaction().toSerializable(),
 164                 PreCommitTransactionReply.SERIALIZABLE_CLASS, true);
 165     }
 166
 167     @Override
 168     public ListenableFuture<Void> abort() {
 169         // Note - we pass false for propagateException. In the front-end data broker, this method
 170         // is called when one of the 3 phases fails with an exception. We'd rather have that
 171         // original exception propagated to the client. If our abort fails and we propagate the
 172         // exception then that exception will supersede and suppress the original exception. But
 173         // it's the original exception that is the root cause and of more interest to the client.
 174
 175         return voidOperation("abort", new AbortTransaction().toSerializable(),
 176                 AbortTransactionReply.SERIALIZABLE_CLASS, false);
 177     }
 178
 179     @Override
 180     public ListenableFuture<Void> commit() {
 181         return voidOperation("commit",  new CommitTransaction().toSerializable(),
 182                 CommitTransactionReply.SERIALIZABLE_CLASS, true);
 183     }
 184
 185     private ListenableFuture<Void> voidOperation(final String operationName, final Object message,
 186             final Class<?> expectedResponseClass, final boolean propagateException) {
 187
 188         LOG.debug("Tx {} {}", transactionId, operationName);
 189
 190         final SettableFuture<Void> returnFuture = SettableFuture.create();
 191
 192         // The cohort actor list should already be built at this point by the canCommit phase but,
 193         // if not for some reason, we'll try to build it here.
 194
 195         if(cohortPaths != null) {
 196             finishVoidOperation(operationName, message, expectedResponseClass, propagateException,
 197                     returnFuture);
 198         } else {
 199             buildCohortPathsList().onComplete(new OnComplete<Void>() {
 200                 @Override
 201                 public void onComplete(Throwable failure, Void notUsed) throws Throwable {
 202                     if(failure != null) {
 203                         LOG.debug("Tx {}: a {} cohort path Future failed: {}", transactionId,
 204                                 operationName, failure);
 205
 206                         if(propagateException) {
 207                             returnFuture.setException(failure);
 208                         } else {
 209                             returnFuture.set(null);
 210                         }
 211                     } else {
 212                         finishVoidOperation(operationName, message, expectedResponseClass,
 213                                 propagateException, returnFuture);
 214                     }
 215                 }
 216             }, actorContext.getActorSystem().dispatcher());
 217         }
 218
 219         return returnFuture;
 220     }
 221
 222     private void finishVoidOperation(final String operationName, final Object message,
 223             final Class<?> expectedResponseClass, final boolean propagateException,
 224             final SettableFuture<Void> returnFuture) {
 225
 226         LOG.debug("Tx {} finish {}", transactionId, operationName);
 227
 228         Future<Iterable<Object>> combinedFuture = invokeCohorts(message);
 229
 230         combinedFuture.onComplete(new OnComplete<Iterable<Object>>() {
 231             @Override
 232             public void onComplete(Throwable failure, Iterable<Object> responses) throws Throwable {
 233
 234                 Throwable exceptionToPropagate = failure;
 235                 if(exceptionToPropagate == null) {
 236                     for(Object response: responses) {
 237                         if(!response.getClass().equals(expectedResponseClass)) {
 238                             exceptionToPropagate = new IllegalArgumentException(
 239                                     String.format("Unexpected response type {}",
 240                                             response.getClass()));
 241                             break;
 242                         }
 243                     }
 244                 }
 245
 246                 if(exceptionToPropagate != null) {
 247                     LOG.debug("Tx {}: a {} cohort Future failed: {}", transactionId,
 248                             operationName, exceptionToPropagate);
 249
 250                     if(propagateException) {
 251                         // We don't log the exception here to avoid redundant logging since we're
 252                         // propagating to the caller in MD-SAL core who will log it.
 253                         returnFuture.setException(exceptionToPropagate);
 254                     } else {
 255                         // Since the caller doesn't want us to propagate the exception we'll also
 256                         // not log it normally. But it's usually not good to totally silence
 257                         // exceptions so we'll log it to debug level.
 258                         LOG.debug(String.format("%s failed",  message.getClass().getSimpleName()),
 259                                 exceptionToPropagate);
 260                         returnFuture.set(null);
 261                     }
 262                 } else {
 263                     LOG.debug("Tx {}: {} succeeded", transactionId, operationName);
 264                     returnFuture.set(null);
 265                 }
 266             }
 267         }, actorContext.getActorSystem().dispatcher());
 268     }
 269
 270     @VisibleForTesting
 271     List<Future<ActorPath>> getCohortPathFutures() {
 272         return Collections.unmodifiableList(cohortPathFutures);
 273     }
 274 }