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         }
 156
 157         return Futures.sequence(futureList, actorContext.getActorSystem().dispatcher());
 158     }
 159
 160     @Override
 161     public ListenableFuture<Void> preCommit() {
 162         return voidOperation("preCommit",  new PreCommitTransaction().toSerializable(),
 163                 PreCommitTransactionReply.SERIALIZABLE_CLASS, true);
 164     }
 165
 166     @Override
 167     public ListenableFuture<Void> abort() {
 168         // Note - we pass false for propagateException. In the front-end data broker, this method
 169         // is called when one of the 3 phases fails with an exception. We'd rather have that
 170         // original exception propagated to the client. If our abort fails and we propagate the
 171         // exception then that exception will supersede and suppress the original exception. But
 172         // it's the original exception that is the root cause and of more interest to the client.
 173
 174         return voidOperation("abort", new AbortTransaction().toSerializable(),
 175                 AbortTransactionReply.SERIALIZABLE_CLASS, false);
 176     }
 177
 178     @Override
 179     public ListenableFuture<Void> commit() {
 180         return voidOperation("commit",  new CommitTransaction().toSerializable(),
 181                 CommitTransactionReply.SERIALIZABLE_CLASS, true);
 182     }
 183
 184     private ListenableFuture<Void> voidOperation(final String operationName, final Object message,
 185             final Class<?> expectedResponseClass, final boolean propagateException) {
 186
 187         LOG.debug("Tx {} {}", transactionId, operationName);
 188
 189         final SettableFuture<Void> returnFuture = SettableFuture.create();
 190
 191         // The cohort actor list should already be built at this point by the canCommit phase but,
 192         // if not for some reason, we'll try to build it here.
 193
 194         if(cohortPaths != null) {
 195             finishVoidOperation(operationName, message, expectedResponseClass, propagateException,
 196                     returnFuture);
 197         } else {
 198             buildCohortPathsList().onComplete(new OnComplete<Void>() {
 199                 @Override
 200                 public void onComplete(Throwable failure, Void notUsed) throws Throwable {
 201                     if(failure != null) {
 202                         LOG.debug("Tx {}: a {} cohort path Future failed: {}", transactionId,
 203                                 operationName, failure);
 204
 205                         if(propagateException) {
 206                             returnFuture.setException(failure);
 207                         } else {
 208                             returnFuture.set(null);
 209                         }
 210                     } else {
 211                         finishVoidOperation(operationName, message, expectedResponseClass,
 212                                 propagateException, returnFuture);
 213                     }
 214                 }
 215             }, actorContext.getActorSystem().dispatcher());
 216         }
 217
 218         return returnFuture;
 219     }
 220
 221     private void finishVoidOperation(final String operationName, final Object message,
 222             final Class<?> expectedResponseClass, final boolean propagateException,
 223             final SettableFuture<Void> returnFuture) {
 224
 225         LOG.debug("Tx {} finish {}", transactionId, operationName);
 226
 227         Future<Iterable<Object>> combinedFuture = invokeCohorts(message);
 228
 229         combinedFuture.onComplete(new OnComplete<Iterable<Object>>() {
 230             @Override
 231             public void onComplete(Throwable failure, Iterable<Object> responses) throws Throwable {
 232
 233                 Throwable exceptionToPropagate = failure;
 234                 if(exceptionToPropagate == null) {
 235                     for(Object response: responses) {
 236                         if(!response.getClass().equals(expectedResponseClass)) {
 237                             exceptionToPropagate = new IllegalArgumentException(
 238                                     String.format("Unexpected response type {}",
 239                                             response.getClass()));
 240                             break;
 241                         }
 242                     }
 243                 }
 244
 245                 if(exceptionToPropagate != null) {
 246                     LOG.debug("Tx {}: a {} cohort Future failed: {}", transactionId,
 247                             operationName, exceptionToPropagate);
 248
 249                     if(propagateException) {
 250                         // We don't log the exception here to avoid redundant logging since we're
 251                         // propagating to the caller in MD-SAL core who will log it.
 252                         returnFuture.setException(exceptionToPropagate);
 253                     } else {
 254                         // Since the caller doesn't want us to propagate the exception we'll also
 255                         // not log it normally. But it's usually not good to totally silence
 256                         // exceptions so we'll log it to debug level.
 257                         LOG.debug(String.format("%s failed",  message.getClass().getSimpleName()),
 258                                 exceptionToPropagate);
 259                         returnFuture.set(null);
 260                     }
 261                 } else {
 262                     LOG.debug("Tx {}: {} succeeded", transactionId, operationName);
 263                     returnFuture.set(null);
 264                 }
 265             }
 266         }, actorContext.getActorSystem().dispatcher());
 267     }
 268
 269     @VisibleForTesting
 270     List<Future<ActorPath>> getCohortPathFutures() {
 271         return Collections.unmodifiableList(cohortPathFutures);
 272     }
 273 }