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=d796ab35fa88522bd7f89d4802adaffe9c3d0a8e;hb=23a3dea5804c8198e61f550736911b5c992d6b6d;hp=540e2fe20ce52208ffd4a659b5e21c70eafe4b10;hpb=430114ac078864cd36fbe7d543440dde029a1dc7;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 540e2fe20c..d796ab35fa 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,34 +6,26 @@ */ package org.opendaylight.controller.md.sal.dom.broker.impl; -import java.util.Collections; -import java.util.List; +import com.google.common.base.Optional; +import com.google.common.base.Preconditions; +import com.google.common.base.Throwables; +import com.google.common.collect.Iterables; +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.Callable; import java.util.concurrent.ExecutionException; - +import java.util.concurrent.RejectedExecutionException; import javax.annotation.concurrent.GuardedBy; - -import org.opendaylight.controller.md.sal.common.api.TransactionStatus; import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException; import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction; -import org.opendaylight.controller.sal.common.util.Rpcs; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; -import org.opendaylight.yangtools.yang.common.RpcError; -import org.opendaylight.yangtools.yang.common.RpcResult; +import org.opendaylight.yangtools.util.DurationStatsTracker; +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 @@ -51,25 +43,7 @@ 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 DurationStatsTracker commitStatsTracker = new DurationStatsTracker(); private final ListeningExecutorService executor; /** @@ -83,19 +57,36 @@ public class DOMDataCommitCoordinatorImpl implements DOMDataCommitExecutor { this.executor = Preconditions.checkNotNull(executor, "executor must not be null."); } + public DurationStatsTracker getCommitStatsTracker() { + return commitStatsTracker; + } + @Override - public ListenableFuture> submit(final DOMDataWriteTransaction transaction, + public CheckedFuture submit(final DOMDataWriteTransaction transaction, final Iterable cohorts, final Optional listener) { 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 = executor.submit(new CommitCoordinationTask( - transaction, cohorts, listener)); + + ListenableFuture commitFuture = null; + try { + commitFuture = executor.submit(new CommitCoordinationTask(transaction, cohorts, + listener, commitStatsTracker)); + } catch(RejectedExecutionException e) { + LOG.error("The commit executor's queue is full - submit task was rejected. \n" + + executor, e); + return Futures.immediateFailedCheckedFuture( + new TransactionCommitFailedException( + "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 commitFuture; + + return MappingCheckedFuture.create(commitFuture, + TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER); } /** @@ -141,33 +132,44 @@ public class DOMDataCommitCoordinatorImpl implements DOMDataCommitExecutor { * support of cancelation. * */ - private static class CommitCoordinationTask implements Callable> { + private static class CommitCoordinationTask implements Callable { private final DOMDataWriteTransaction tx; private final Iterable cohorts; + private final DurationStatsTracker commitStatTracker; + private final int cohortSize; @GuardedBy("this") private CommitPhase currentPhase; public CommitCoordinationTask(final DOMDataWriteTransaction transaction, final Iterable cohorts, - final Optional listener) { + 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; + this.cohortSize = Iterables.size(cohorts); } @Override - public RpcResult call() throws TransactionCommitFailedException { + public Void call() throws TransactionCommitFailedException { + long startTime = System.nanoTime(); try { canCommitBlocking(); preCommitBlocking(); - return commitBlocking(); + 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); + } } } @@ -184,10 +186,39 @@ public class DOMDataCommitCoordinatorImpl implements DOMDataCommitExecutor { * */ private void canCommitBlocking() throws TransactionCommitFailedException { - final Boolean canCommitResult = canCommitAll().checkedGet(); - if (!canCommitResult) { - throw new TransactionCommitFailedException("Can Commit failed, no detailed cause available."); + for (ListenableFuture canCommit : canCommitAll()) { + try { + final Boolean result = (Boolean)canCommit.get(); + if (result == null || !result) { + throw new TransactionCommitFailedException("Can Commit failed, no detailed cause available."); + } + } catch (InterruptedException | ExecutionException e) { + throw TransactionCommitFailedExceptionMapper.CAN_COMMIT_ERROR_MAPPER.apply(e); + } + } + } + + /** + * + * 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 List of all cohorts futures from can commit phase. + * + */ + private ListenableFuture[] canCommitAll() { + changeStateFrom(CommitPhase.SUBMITTED, CommitPhase.CAN_COMMIT); + + final ListenableFuture[] ops = new ListenableFuture[cohortSize]; + int i = 0; + for (DOMStoreThreePhaseCommitCohort cohort : cohorts) { + ops[i++] = cohort.canCommit(); } + return ops; } /** @@ -204,7 +235,39 @@ public class DOMDataCommitCoordinatorImpl implements DOMDataCommitExecutor { * */ private void preCommitBlocking() throws TransactionCommitFailedException { - preCommitAll().checkedGet(); + final ListenableFuture[] preCommitFutures = preCommitAll(); + try { + for(ListenableFuture future : preCommitFutures) { + future.get(); + } + } catch (InterruptedException | ExecutionException e) { + throw TransactionCommitFailedExceptionMapper.PRE_COMMIT_MAPPER.apply(e); + } + } + + /** + * + * 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 List of all cohorts futures from can commit phase. + * + */ + private ListenableFuture[] preCommitAll() { + changeStateFrom(CommitPhase.CAN_COMMIT, CommitPhase.PRE_COMMIT); + + final ListenableFuture[] ops = new ListenableFuture[cohortSize]; + int i = 0; + for (DOMStoreThreePhaseCommitCohort cohort : cohorts) { + ops[i++] = cohort.preCommit(); + } + return ops; } /** @@ -219,9 +282,38 @@ public class DOMDataCommitCoordinatorImpl implements DOMDataCommitExecutor { * If one of cohorts failed preCommit * */ - private RpcResult commitBlocking() throws TransactionCommitFailedException { - commitAll().checkedGet(); - return Rpcs.getRpcResult(true, TransactionStatus.COMMITED, Collections. emptySet()); + private void commitBlocking() throws TransactionCommitFailedException { + final ListenableFuture[] commitFutures = commitAll(); + try { + for(ListenableFuture future : commitFutures) { + future.get(); + } + } catch (InterruptedException | ExecutionException e) { + throw TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER.apply(e); + } + } + + /** + * + * 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 List of all cohorts futures from can commit phase. + * + */ + private ListenableFuture[] commitAll() { + changeStateFrom(CommitPhase.PRE_COMMIT, CommitPhase.COMMIT); + + final ListenableFuture[] ops = new ListenableFuture[cohortSize]; + int i = 0; + for (DOMStoreThreePhaseCommitCohort cohort : cohorts) { + ops[i++] = cohort.commit(); + } + return ops; } /** @@ -259,98 +351,6 @@ public class DOMDataCommitCoordinatorImpl implements DOMDataCommitExecutor { 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 Futures.makeChecked(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 Futures.makeChecked(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 Futures - .makeChecked(allSuccessFuture, TransactionCommitFailedExceptionMapper.CAN_COMMIT_ERROR_MAPPER); - - } - /** * * Invokes abort on underlying cohorts and returns future which @@ -363,17 +363,20 @@ public class DOMDataCommitCoordinatorImpl implements DOMDataCommitExecutor { */ private ListenableFuture abortAsyncAll() { changeStateFrom(currentPhase, CommitPhase.ABORT); - Builder> ops = ImmutableList.builder(); + + final ListenableFuture[] ops = new ListenableFuture[cohortSize]; + int i = 0; for (DOMStoreThreePhaseCommitCohort cohort : cohorts) { - ops.add(cohort.abort()); + ops[i++] = 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()); + ListenableFuture compositeResult = (ListenableFuture) Futures.allAsList(ops); return compositeResult; }