X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-dom-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fmd%2Fsal%2Fdom%2Fbroker%2Fimpl%2FDOMDataCommitCoordinatorImpl.java;h=7b5350023113fee05df7e56aa6df6b54204400a1;hb=5764146d24df9d9450ebfbb0aec10cbbcdcc655c;hp=3fde8d360f8af6df8cb0bcd705a9e3289d9fd35e;hpb=1d639169d7afd590f4f756242031768e97d95b61;p=controller.git diff --git a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMDataCommitCoordinatorImpl.java b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMDataCommitCoordinatorImpl.java index 3fde8d360f..7b53500231 100644 --- a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMDataCommitCoordinatorImpl.java +++ b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMDataCommitCoordinatorImpl.java @@ -6,32 +6,20 @@ */ package org.opendaylight.controller.md.sal.dom.broker.impl; -import java.util.List; -import java.util.concurrent.Callable; -import java.util.concurrent.ExecutionException; +import com.google.common.base.Preconditions; +import com.google.common.util.concurrent.CheckedFuture; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.ListeningExecutorService; import java.util.concurrent.RejectedExecutionException; - -import javax.annotation.concurrent.GuardedBy; - import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException; import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; -import org.opendaylight.yangtools.util.DurationStatsTracker; +import org.opendaylight.yangtools.util.DurationStatisticsTracker; import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.base.Function; -import com.google.common.base.Optional; -import com.google.common.base.Preconditions; -import com.google.common.base.Throwables; -import com.google.common.collect.ImmutableList; -import com.google.common.collect.ImmutableList.Builder; -import com.google.common.util.concurrent.CheckedFuture; -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.ListeningExecutorService; - /** * * Implementation of blocking three phase commit coordinator, which which @@ -49,29 +37,9 @@ import com.google.common.util.concurrent.ListeningExecutorService; public class DOMDataCommitCoordinatorImpl implements DOMDataCommitExecutor { private static final Logger LOG = LoggerFactory.getLogger(DOMDataCommitCoordinatorImpl.class); - - /** - * Runs AND binary operation between all booleans in supplied iteration of booleans. - * - * This method will stop evaluating iterables if first found is false. - */ - private static final Function, Boolean> AND_FUNCTION = new Function, Boolean>() { - - @Override - public Boolean apply(final Iterable input) { - for(boolean value : input) { - if(!value) { - return Boolean.FALSE; - } - } - return Boolean.TRUE; - } - }; - + private final DurationStatisticsTracker commitStatsTracker = DurationStatisticsTracker.createConcurrent(); private final ListeningExecutorService executor; - private final DurationStatsTracker commitStatsTracker = new DurationStatsTracker(); - /** * * Construct DOMDataCommitCoordinator which uses supplied executor to @@ -83,22 +51,21 @@ public class DOMDataCommitCoordinatorImpl implements DOMDataCommitExecutor { this.executor = Preconditions.checkNotNull(executor, "executor must not be null."); } - public DurationStatsTracker getCommitStatsTracker() { + public DurationStatisticsTracker getCommitStatsTracker() { return commitStatsTracker; } @Override public CheckedFuture submit(final DOMDataWriteTransaction transaction, - final Iterable cohorts, final Optional listener) { + final Iterable cohorts) { Preconditions.checkArgument(transaction != null, "Transaction must not be null."); Preconditions.checkArgument(cohorts != null, "Cohorts must not be null."); - Preconditions.checkArgument(listener != null, "Listener must not be null"); LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier()); ListenableFuture commitFuture = null; try { commitFuture = executor.submit(new CommitCoordinationTask(transaction, cohorts, - listener, commitStatsTracker)); + commitStatsTracker)); } catch(RejectedExecutionException e) { LOG.error("The commit executor's queue is full - submit task was rejected. \n" + executor, e); @@ -107,330 +74,7 @@ public class DOMDataCommitCoordinatorImpl implements DOMDataCommitExecutor { "Could not submit the commit task - the commit queue capacity has been exceeded.", e)); } - if (listener.isPresent()) { - Futures.addCallback(commitFuture, new DOMDataCommitErrorInvoker(transaction, listener.get())); - } - return MappingCheckedFuture.create(commitFuture, TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER); } - - /** - * - * Phase of 3PC commit - * - * Represents phase of 3PC Commit - * - * - */ - private static enum CommitPhase { - /** - * - * Commit Coordination Task is submitted for executing - * - */ - SUBMITTED, - /** - * Commit Coordination Task is in can commit phase of 3PC - * - */ - CAN_COMMIT, - /** - * Commit Coordination Task is in pre-commit phase of 3PC - * - */ - PRE_COMMIT, - /** - * Commit Coordination Task is in commit phase of 3PC - * - */ - COMMIT, - /** - * Commit Coordination Task is in abort phase of 3PC - * - */ - ABORT - } - - /** - * - * Implementation of blocking three-phase commit-coordination tasks without - * support of cancelation. - * - */ - private static class CommitCoordinationTask implements Callable { - - private final DOMDataWriteTransaction tx; - private final Iterable cohorts; - private final DurationStatsTracker commitStatTracker; - - @GuardedBy("this") - private CommitPhase currentPhase; - - public CommitCoordinationTask(final DOMDataWriteTransaction transaction, - final Iterable cohorts, - final Optional listener, - final DurationStatsTracker commitStatTracker) { - this.tx = Preconditions.checkNotNull(transaction, "transaction must not be null"); - this.cohorts = Preconditions.checkNotNull(cohorts, "cohorts must not be null"); - this.currentPhase = CommitPhase.SUBMITTED; - this.commitStatTracker = commitStatTracker; - } - - @Override - public Void call() throws TransactionCommitFailedException { - - long startTime = System.nanoTime(); - try { - canCommitBlocking(); - preCommitBlocking(); - commitBlocking(); - return null; - } catch (TransactionCommitFailedException e) { - LOG.warn("Tx: {} Error during phase {}, starting Abort", tx.getIdentifier(), currentPhase, e); - abortBlocking(e); - throw e; - } finally { - if(commitStatTracker != null) { - commitStatTracker.addDuration(System.nanoTime() - startTime); - } - } - } - - /** - * - * Invokes canCommit on underlying cohorts and blocks till - * all results are returned. - * - * Valid state transition is from SUBMITTED to CAN_COMMIT, - * if currentPhase is not SUBMITTED throws IllegalStateException. - * - * @throws TransactionCommitFailedException - * If one of cohorts failed can Commit - * - */ - private void canCommitBlocking() throws TransactionCommitFailedException { - final Boolean canCommitResult = canCommitAll().checkedGet(); - if (!canCommitResult) { - throw new TransactionCommitFailedException("Can Commit failed, no detailed cause available."); - } - } - - /** - * - * Invokes preCommit on underlying cohorts and blocks till - * all results are returned. - * - * Valid state transition is from CAN_COMMIT to PRE_COMMIT, if current - * state is not CAN_COMMIT - * throws IllegalStateException. - * - * @throws TransactionCommitFailedException - * If one of cohorts failed preCommit - * - */ - private void preCommitBlocking() throws TransactionCommitFailedException { - preCommitAll().checkedGet(); - } - - /** - * - * Invokes commit on underlying cohorts and blocks till - * all results are returned. - * - * Valid state transition is from PRE_COMMIT to COMMIT, if not throws - * IllegalStateException. - * - * @throws TransactionCommitFailedException - * If one of cohorts failed preCommit - * - */ - private void commitBlocking() throws TransactionCommitFailedException { - commitAll().checkedGet(); - } - - /** - * Aborts transaction. - * - * Invokes {@link DOMStoreThreePhaseCommitCohort#abort()} on all - * cohorts, blocks - * for all results. If any of the abort failed throws - * IllegalStateException, - * which will contains originalCause as suppressed Exception. - * - * If aborts we're successful throws supplied exception - * - * @param originalCause - * Exception which should be used to fail transaction for - * consumers of transaction - * future and listeners of transaction failure. - * @throws TransactionCommitFailedException - * on invocation of this method. - * originalCa - * @throws IllegalStateException - * if abort failed. - */ - private void abortBlocking(final TransactionCommitFailedException originalCause) - throws TransactionCommitFailedException { - LOG.warn("Tx: {} Error during phase {}, starting Abort", tx.getIdentifier(), currentPhase, originalCause); - Exception cause = originalCause; - try { - abortAsyncAll().get(); - } catch (InterruptedException | ExecutionException e) { - LOG.error("Tx: {} Error during Abort.", tx.getIdentifier(), e); - cause = new IllegalStateException("Abort failed.", e); - cause.addSuppressed(e); - } - Throwables.propagateIfPossible(cause, TransactionCommitFailedException.class); - } - - /** - * - * Invokes preCommit on underlying cohorts and returns future - * which will complete once all preCommit on cohorts completed or - * failed. - * - * - * Valid state transition is from CAN_COMMIT to PRE_COMMIT, if current - * state is not CAN_COMMIT - * throws IllegalStateException. - * - * @return Future which will complete once all cohorts completed - * preCommit. - * Future throws TransactionCommitFailedException - * If any of cohorts failed preCommit - * - */ - private CheckedFuture preCommitAll() { - changeStateFrom(CommitPhase.CAN_COMMIT, CommitPhase.PRE_COMMIT); - Builder> ops = ImmutableList.builder(); - for (DOMStoreThreePhaseCommitCohort cohort : cohorts) { - ops.add(cohort.preCommit()); - } - /* - * We are returing all futures as list, not only succeeded ones in - * order to fail composite future if any of them failed. - * See Futures.allAsList for this description. - */ - @SuppressWarnings({ "unchecked", "rawtypes" }) - ListenableFuture compositeResult = (ListenableFuture) Futures.allAsList(ops.build()); - return MappingCheckedFuture.create(compositeResult, - TransactionCommitFailedExceptionMapper.PRE_COMMIT_MAPPER); - } - - /** - * - * Invokes commit on underlying cohorts and returns future which - * completes - * once all commits on cohorts are completed. - * - * Valid state transition is from PRE_COMMIT to COMMIT, if not throws - * IllegalStateException - * - * @return Future which will complete once all cohorts completed - * commit. - * Future throws TransactionCommitFailedException - * If any of cohorts failed preCommit - * - */ - private CheckedFuture commitAll() { - changeStateFrom(CommitPhase.PRE_COMMIT, CommitPhase.COMMIT); - Builder> ops = ImmutableList.builder(); - for (DOMStoreThreePhaseCommitCohort cohort : cohorts) { - ops.add(cohort.commit()); - } - /* - * We are returing all futures as list, not only succeeded ones in - * order to fail composite future if any of them failed. - * See Futures.allAsList for this description. - */ - @SuppressWarnings({ "unchecked", "rawtypes" }) - ListenableFuture compositeResult = (ListenableFuture) Futures.allAsList(ops.build()); - return MappingCheckedFuture.create(compositeResult, - TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER); - } - - /** - * - * Invokes canCommit on underlying cohorts and returns composite future - * which will contains {@link Boolean#TRUE} only and only if - * all cohorts returned true. - * - * Valid state transition is from SUBMITTED to CAN_COMMIT, - * if currentPhase is not SUBMITTED throws IllegalStateException. - * - * @return Future which will complete once all cohorts completed - * preCommit. - * Future throws TransactionCommitFailedException - * If any of cohorts failed preCommit - * - */ - private CheckedFuture canCommitAll() { - changeStateFrom(CommitPhase.SUBMITTED, CommitPhase.CAN_COMMIT); - Builder> canCommitOperations = ImmutableList.builder(); - for (DOMStoreThreePhaseCommitCohort cohort : cohorts) { - canCommitOperations.add(cohort.canCommit()); - } - ListenableFuture> allCanCommits = Futures.allAsList(canCommitOperations.build()); - ListenableFuture allSuccessFuture = Futures.transform(allCanCommits, AND_FUNCTION); - return MappingCheckedFuture.create(allSuccessFuture, - TransactionCommitFailedExceptionMapper.CAN_COMMIT_ERROR_MAPPER); - - } - - /** - * - * Invokes abort on underlying cohorts and returns future which - * completes - * once all abort on cohorts are completed. - * - * @return Future which will complete once all cohorts completed - * abort. - * - */ - private ListenableFuture abortAsyncAll() { - changeStateFrom(currentPhase, CommitPhase.ABORT); - Builder> ops = ImmutableList.builder(); - for (DOMStoreThreePhaseCommitCohort cohort : cohorts) { - ops.add(cohort.abort()); - } - /* - * We are returing all futures as list, not only succeeded ones in - * order to fail composite future if any of them failed. - * See Futures.allAsList for this description. - */ - @SuppressWarnings({ "unchecked", "rawtypes" }) - ListenableFuture compositeResult = (ListenableFuture) Futures.allAsList(ops.build()); - return compositeResult; - } - - /** - * Change phase / state of transaction from expected value to new value - * - * This method checks state and updates state to new state of - * of this task if current state equals expected state. - * If expected state and current state are different raises - * IllegalStateException - * which means there is probably bug in implementation of commit - * coordination. - * - * If transition is successful, it logs transition on DEBUG level. - * - * @param currentExpected - * Required phase for change of state - * @param newState - * New Phase which will be entered by transaction. - * @throws IllegalStateException - * If currentState of task does not match expected state - */ - private synchronized void changeStateFrom(final CommitPhase currentExpected, final CommitPhase newState) { - Preconditions.checkState(currentPhase.equals(currentExpected), - "Invalid state transition: Tx: %s current state: %s new state: %s", tx.getIdentifier(), - currentPhase, newState); - LOG.debug("Transaction {}: Phase {} Started ", tx.getIdentifier(), newState); - currentPhase = newState; - }; - - } - }