opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMDataCommitCoordinatorImpl.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
   3  * This program and the accompanying materials are made available under the
   4  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   5  * and is available at http://www.eclipse.org/legal/epl-v10.html
   6  */
   7 package org.opendaylight.controller.md.sal.dom.broker.impl;
   8
   9 import java.util.List;
  10 import java.util.concurrent.Callable;
  11 import java.util.concurrent.ExecutionException;
  12
  13 import javax.annotation.concurrent.GuardedBy;
  14
  15 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  16 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
  17 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  18 import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture;
  19 import org.slf4j.Logger;
  20 import org.slf4j.LoggerFactory;
  21
  22 import com.google.common.base.Function;
  23 import com.google.common.base.Optional;
  24 import com.google.common.base.Preconditions;
  25 import com.google.common.base.Throwables;
  26 import com.google.common.collect.ImmutableList;
  27 import com.google.common.collect.ImmutableList.Builder;
  28 import com.google.common.util.concurrent.CheckedFuture;
  29 import com.google.common.util.concurrent.Futures;
  30 import com.google.common.util.concurrent.ListenableFuture;
  31 import com.google.common.util.concurrent.ListeningExecutorService;
  32
  33 /**
  34  *
  35  * Implementation of blocking three phase commit coordinator, which which
  36  * supports coordination on multiple {@link DOMStoreThreePhaseCommitCohort}.
  37  *
  38  * This implementation does not support cancelation of commit,
  39  *
  40  * In order to advance to next phase of three phase commit all subtasks of
  41  * previous step must be finish.
  42  *
  43  * This executor does not have an upper bound on subtask timeout.
  44  *
  45  *
  46  */
  47 public class DOMDataCommitCoordinatorImpl implements DOMDataCommitExecutor {
  48
  49     private static final Logger LOG = LoggerFactory.getLogger(DOMDataCommitCoordinatorImpl.class);
  50
  51     /**
  52      * Runs AND binary operation between all booleans in supplied iteration of booleans.
  53      *
  54      * This method will stop evaluating iterables if first found is false.
  55      */
  56     private static final Function<Iterable<Boolean>, Boolean> AND_FUNCTION = new Function<Iterable<Boolean>, Boolean>() {
  57
  58         @Override
  59         public Boolean apply(final Iterable<Boolean> input) {
  60             for(boolean value : input) {
  61                if(!value) {
  62                    return Boolean.FALSE;
  63                }
  64             }
  65             return Boolean.TRUE;
  66         }
  67     };
  68
  69     private final ListeningExecutorService executor;
  70
  71     /**
  72      *
  73      * Construct DOMDataCommitCoordinator which uses supplied executor to
  74      * process commit coordinations.
  75      *
  76      * @param executor
  77      */
  78     public DOMDataCommitCoordinatorImpl(final ListeningExecutorService executor) {
  79         this.executor = Preconditions.checkNotNull(executor, "executor must not be null.");
  80     }
  81
  82     @Override
  83     public CheckedFuture<Void,TransactionCommitFailedException> submit(final DOMDataWriteTransaction transaction,
  84             final Iterable<DOMStoreThreePhaseCommitCohort> cohorts, final Optional<DOMDataCommitErrorListener> listener) {
  85         Preconditions.checkArgument(transaction != null, "Transaction must not be null.");
  86         Preconditions.checkArgument(cohorts != null, "Cohorts must not be null.");
  87         Preconditions.checkArgument(listener != null, "Listener must not be null");
  88         LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier());
  89         ListenableFuture<Void> commitFuture = executor.submit(new CommitCoordinationTask(
  90                 transaction, cohorts, listener));
  91         if (listener.isPresent()) {
  92             Futures.addCallback(commitFuture, new DOMDataCommitErrorInvoker(transaction, listener.get()));
  93         }
  94
  95         return MappingCheckedFuture.create(commitFuture,
  96                 TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER);
  97     }
  98
  99     /**
 100      *
 101      * Phase of 3PC commit
 102      *
 103      * Represents phase of 3PC Commit
 104      *
 105      *
 106      */
 107     private static enum CommitPhase {
 108         /**
 109          *
 110          * Commit Coordination Task is submitted for executing
 111          *
 112          */
 113         SUBMITTED,
 114         /**
 115          * Commit Coordination Task is in can commit phase of 3PC
 116          *
 117          */
 118         CAN_COMMIT,
 119         /**
 120          * Commit Coordination Task is in pre-commit phase of 3PC
 121          *
 122          */
 123         PRE_COMMIT,
 124         /**
 125          * Commit Coordination Task is in commit phase of 3PC
 126          *
 127          */
 128         COMMIT,
 129         /**
 130          * Commit Coordination Task is in abort phase of 3PC
 131          *
 132          */
 133         ABORT
 134     }
 135
 136     /**
 137      *
 138      * Implementation of blocking three-phase commit-coordination tasks without
 139      * support of cancelation.
 140      *
 141      */
 142     private static class CommitCoordinationTask implements Callable<Void> {
 143
 144         private final DOMDataWriteTransaction tx;
 145         private final Iterable<DOMStoreThreePhaseCommitCohort> cohorts;
 146
 147         @GuardedBy("this")
 148         private CommitPhase currentPhase;
 149
 150         public CommitCoordinationTask(final DOMDataWriteTransaction transaction,
 151                 final Iterable<DOMStoreThreePhaseCommitCohort> cohorts,
 152                 final Optional<DOMDataCommitErrorListener> listener) {
 153             this.tx = Preconditions.checkNotNull(transaction, "transaction must not be null");
 154             this.cohorts = Preconditions.checkNotNull(cohorts, "cohorts must not be null");
 155             this.currentPhase = CommitPhase.SUBMITTED;
 156         }
 157
 158         @Override
 159         public Void call() throws TransactionCommitFailedException {
 160
 161             try {
 162                 canCommitBlocking();
 163                 preCommitBlocking();
 164                 commitBlocking();
 165                 return null;
 166             } catch (TransactionCommitFailedException e) {
 167                 LOG.warn("Tx: {} Error during phase {}, starting Abort", tx.getIdentifier(), currentPhase, e);
 168                 abortBlocking(e);
 169                 throw e;
 170             }
 171         }
 172
 173         /**
 174          *
 175          * Invokes canCommit on underlying cohorts and blocks till
 176          * all results are returned.
 177          *
 178          * Valid state transition is from SUBMITTED to CAN_COMMIT,
 179          * if currentPhase is not SUBMITTED throws IllegalStateException.
 180          *
 181          * @throws TransactionCommitFailedException
 182          *             If one of cohorts failed can Commit
 183          *
 184          */
 185         private void canCommitBlocking() throws TransactionCommitFailedException {
 186             final Boolean canCommitResult = canCommitAll().checkedGet();
 187             if (!canCommitResult) {
 188                 throw new TransactionCommitFailedException("Can Commit failed, no detailed cause available.");
 189             }
 190         }
 191
 192         /**
 193          *
 194          * Invokes preCommit on underlying cohorts and blocks till
 195          * all results are returned.
 196          *
 197          * Valid state transition is from CAN_COMMIT to PRE_COMMIT, if current
 198          * state is not CAN_COMMIT
 199          * throws IllegalStateException.
 200          *
 201          * @throws TransactionCommitFailedException
 202          *             If one of cohorts failed preCommit
 203          *
 204          */
 205         private void preCommitBlocking() throws TransactionCommitFailedException {
 206             preCommitAll().checkedGet();
 207         }
 208
 209         /**
 210          *
 211          * Invokes commit on underlying cohorts and blocks till
 212          * all results are returned.
 213          *
 214          * Valid state transition is from PRE_COMMIT to COMMIT, if not throws
 215          * IllegalStateException.
 216          *
 217          * @throws TransactionCommitFailedException
 218          *             If one of cohorts failed preCommit
 219          *
 220          */
 221         private void commitBlocking() throws TransactionCommitFailedException {
 222             commitAll().checkedGet();
 223         }
 224
 225         /**
 226          * Aborts transaction.
 227          *
 228          * Invokes {@link DOMStoreThreePhaseCommitCohort#abort()} on all
 229          * cohorts, blocks
 230          * for all results. If any of the abort failed throws
 231          * IllegalStateException,
 232          * which will contains originalCause as suppressed Exception.
 233          *
 234          * If aborts we're successful throws supplied exception
 235          *
 236          * @param originalCause
 237          *            Exception which should be used to fail transaction for
 238          *            consumers of transaction
 239          *            future and listeners of transaction failure.
 240          * @throws TransactionCommitFailedException
 241          *             on invocation of this method.
 242          *             originalCa
 243          * @throws IllegalStateException
 244          *             if abort failed.
 245          */
 246         private void abortBlocking(final TransactionCommitFailedException originalCause)
 247                 throws TransactionCommitFailedException {
 248             LOG.warn("Tx: {} Error during phase {}, starting Abort", tx.getIdentifier(), currentPhase, originalCause);
 249             Exception cause = originalCause;
 250             try {
 251                 abortAsyncAll().get();
 252             } catch (InterruptedException | ExecutionException e) {
 253                 LOG.error("Tx: {} Error during Abort.", tx.getIdentifier(), e);
 254                 cause = new IllegalStateException("Abort failed.", e);
 255                 cause.addSuppressed(e);
 256             }
 257             Throwables.propagateIfPossible(cause, TransactionCommitFailedException.class);
 258         }
 259
 260         /**
 261          *
 262          * Invokes preCommit on underlying cohorts and returns future
 263          * which will complete once all preCommit on cohorts completed or
 264          * failed.
 265          *
 266          *
 267          * Valid state transition is from CAN_COMMIT to PRE_COMMIT, if current
 268          * state is not CAN_COMMIT
 269          * throws IllegalStateException.
 270          *
 271          * @return Future which will complete once all cohorts completed
 272          *         preCommit.
 273          *         Future throws TransactionCommitFailedException
 274          *         If any of cohorts failed preCommit
 275          *
 276          */
 277         private CheckedFuture<Void, TransactionCommitFailedException> preCommitAll() {
 278             changeStateFrom(CommitPhase.CAN_COMMIT, CommitPhase.PRE_COMMIT);
 279             Builder<ListenableFuture<Void>> ops = ImmutableList.builder();
 280             for (DOMStoreThreePhaseCommitCohort cohort : cohorts) {
 281                 ops.add(cohort.preCommit());
 282             }
 283             /*
 284              * We are returing all futures as list, not only succeeded ones in
 285              * order to fail composite future if any of them failed.
 286              * See Futures.allAsList for this description.
 287              */
 288             @SuppressWarnings({ "unchecked", "rawtypes" })
 289             ListenableFuture<Void> compositeResult = (ListenableFuture) Futures.allAsList(ops.build());
 290             return MappingCheckedFuture.create(compositeResult,
 291                                          TransactionCommitFailedExceptionMapper.PRE_COMMIT_MAPPER);
 292         }
 293
 294         /**
 295          *
 296          * Invokes commit on underlying cohorts and returns future which
 297          * completes
 298          * once all commits on cohorts are completed.
 299          *
 300          * Valid state transition is from PRE_COMMIT to COMMIT, if not throws
 301          * IllegalStateException
 302          *
 303          * @return Future which will complete once all cohorts completed
 304          *         commit.
 305          *         Future throws TransactionCommitFailedException
 306          *         If any of cohorts failed preCommit
 307          *
 308          */
 309         private CheckedFuture<Void, TransactionCommitFailedException> commitAll() {
 310             changeStateFrom(CommitPhase.PRE_COMMIT, CommitPhase.COMMIT);
 311             Builder<ListenableFuture<Void>> ops = ImmutableList.builder();
 312             for (DOMStoreThreePhaseCommitCohort cohort : cohorts) {
 313                 ops.add(cohort.commit());
 314             }
 315             /*
 316              * We are returing all futures as list, not only succeeded ones in
 317              * order to fail composite future if any of them failed.
 318              * See Futures.allAsList for this description.
 319              */
 320             @SuppressWarnings({ "unchecked", "rawtypes" })
 321             ListenableFuture<Void> compositeResult = (ListenableFuture) Futures.allAsList(ops.build());
 322             return MappingCheckedFuture.create(compositeResult,
 323                                      TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER);
 324         }
 325
 326         /**
 327          *
 328          * Invokes canCommit on underlying cohorts and returns composite future
 329          * which will contains {@link Boolean#TRUE} only and only if
 330          * all cohorts returned true.
 331          *
 332          * Valid state transition is from SUBMITTED to CAN_COMMIT,
 333          * if currentPhase is not SUBMITTED throws IllegalStateException.
 334          *
 335          * @return Future which will complete once all cohorts completed
 336          *         preCommit.
 337          *         Future throws TransactionCommitFailedException
 338          *         If any of cohorts failed preCommit
 339          *
 340          */
 341         private CheckedFuture<Boolean, TransactionCommitFailedException> canCommitAll() {
 342             changeStateFrom(CommitPhase.SUBMITTED, CommitPhase.CAN_COMMIT);
 343             Builder<ListenableFuture<Boolean>> canCommitOperations = ImmutableList.builder();
 344             for (DOMStoreThreePhaseCommitCohort cohort : cohorts) {
 345                 canCommitOperations.add(cohort.canCommit());
 346             }
 347             ListenableFuture<List<Boolean>> allCanCommits = Futures.allAsList(canCommitOperations.build());
 348             ListenableFuture<Boolean> allSuccessFuture = Futures.transform(allCanCommits, AND_FUNCTION);
 349             return MappingCheckedFuture.create(allSuccessFuture,
 350                                        TransactionCommitFailedExceptionMapper.CAN_COMMIT_ERROR_MAPPER);
 351
 352         }
 353
 354         /**
 355          *
 356          * Invokes abort on underlying cohorts and returns future which
 357          * completes
 358          * once all abort on cohorts are completed.
 359          *
 360          * @return Future which will complete once all cohorts completed
 361          *         abort.
 362          *
 363          */
 364         private ListenableFuture<Void> abortAsyncAll() {
 365             changeStateFrom(currentPhase, CommitPhase.ABORT);
 366             Builder<ListenableFuture<Void>> ops = ImmutableList.builder();
 367             for (DOMStoreThreePhaseCommitCohort cohort : cohorts) {
 368                 ops.add(cohort.abort());
 369             }
 370             /*
 371              * We are returing all futures as list, not only succeeded ones in
 372              * order to fail composite future if any of them failed.
 373              * See Futures.allAsList for this description.
 374              */
 375             @SuppressWarnings({ "unchecked", "rawtypes" })
 376             ListenableFuture<Void> compositeResult = (ListenableFuture) Futures.allAsList(ops.build());
 377             return compositeResult;
 378         }
 379
 380         /**
 381          * Change phase / state of transaction from expected value to new value
 382          *
 383          * This method checks state and updates state to new state of
 384          * of this task if current state equals expected state.
 385          * If expected state and current state are different raises
 386          * IllegalStateException
 387          * which means there is probably bug in implementation of commit
 388          * coordination.
 389          *
 390          * If transition is successful, it logs transition on DEBUG level.
 391          *
 392          * @param currentExpected
 393          *            Required phase for change of state
 394          * @param newState
 395          *            New Phase which will be entered by transaction.
 396          * @throws IllegalStateException
 397          *             If currentState of task does not match expected state
 398          */
 399         private synchronized void changeStateFrom(final CommitPhase currentExpected, final CommitPhase newState) {
 400             Preconditions.checkState(currentPhase.equals(currentExpected),
 401                     "Invalid state transition: Tx: %s current state: %s new state: %s", tx.getIdentifier(),
 402                     currentPhase, newState);
 403             LOG.debug("Transaction {}: Phase {} Started ", tx.getIdentifier(), newState);
 404             currentPhase = newState;
 405         };
 406
 407     }
 408
 409 }