+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.md.sal.dom.broker.impl;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+
+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.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
+ * supports coordination on multiple {@link DOMStoreThreePhaseCommitCohort}.
+ *
+ * This implementation does not support cancelation of commit,
+ *
+ * In order to advance to next phase of three phase commit all subtasks of
+ * previous step must be finish.
+ *
+ * This executor does not have an upper bound on subtask timeout.
+ *
+ *
+ */
+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<Iterable<Boolean>, Boolean> AND_FUNCTION = new Function<Iterable<Boolean>, Boolean>() {
+
+ @Override
+ public Boolean apply(final Iterable<Boolean> input) {
+ for(boolean value : input) {
+ if(!value) {
+ return Boolean.FALSE;
+ }
+ }
+ return Boolean.TRUE;
+ }
+ };
+
+ private final ListeningExecutorService executor;
+
+ /**
+ *
+ * Construct DOMDataCommitCoordinator which uses supplied executor to
+ * process commit coordinations.
+ *
+ * @param executor
+ */
+ public DOMDataCommitCoordinatorImpl(final ListeningExecutorService executor) {
+ this.executor = Preconditions.checkNotNull(executor, "executor must not be null.");
+ }
+
+ @Override
+ public ListenableFuture<RpcResult<TransactionStatus>> submit(final DOMDataWriteTransaction transaction,
+ final Iterable<DOMStoreThreePhaseCommitCohort> cohorts, final Optional<DOMDataCommitErrorListener> 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<RpcResult<TransactionStatus>> commitFuture = executor.submit(new CommitCoordinationTask(
+ transaction, cohorts, listener));
+ if (listener.isPresent()) {
+ Futures.addCallback(commitFuture, new DOMDataCommitErrorInvoker(transaction, listener.get()));
+ }
+ return commitFuture;
+ }
+
+ /**
+ *
+ * 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<RpcResult<TransactionStatus>> {
+
+ private final DOMDataWriteTransaction tx;
+ private final Iterable<DOMStoreThreePhaseCommitCohort> cohorts;
+
+ @GuardedBy("this")
+ private CommitPhase currentPhase;
+
+ public CommitCoordinationTask(final DOMDataWriteTransaction transaction,
+ final Iterable<DOMStoreThreePhaseCommitCohort> cohorts,
+ final Optional<DOMDataCommitErrorListener> listener) {
+ this.tx = Preconditions.checkNotNull(transaction, "transaction must not be null");
+ this.cohorts = Preconditions.checkNotNull(cohorts, "cohorts must not be null");
+ this.currentPhase = CommitPhase.SUBMITTED;
+ }
+
+ @Override
+ public RpcResult<TransactionStatus> call() throws TransactionCommitFailedException {
+
+ try {
+ canCommitBlocking();
+ preCommitBlocking();
+ return commitBlocking();
+ } catch (TransactionCommitFailedException e) {
+ LOG.warn("Tx: {} Error during phase {}, starting Abort", tx.getIdentifier(), currentPhase, e);
+ abortBlocking(e);
+ throw e;
+ }
+ }
+
+ /**
+ *
+ * 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 RpcResult<TransactionStatus> commitBlocking() throws TransactionCommitFailedException {
+ commitAll().checkedGet();
+ return Rpcs.getRpcResult(true, TransactionStatus.COMMITED, Collections.<RpcError> emptySet());
+ }
+
+ /**
+ * 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<Void, TransactionCommitFailedException> preCommitAll() {
+ changeStateFrom(CommitPhase.CAN_COMMIT, CommitPhase.PRE_COMMIT);
+ Builder<ListenableFuture<Void>> 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<Void> 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<Void, TransactionCommitFailedException> commitAll() {
+ changeStateFrom(CommitPhase.PRE_COMMIT, CommitPhase.COMMIT);
+ Builder<ListenableFuture<Void>> 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<Void> 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<Boolean, TransactionCommitFailedException> canCommitAll() {
+ changeStateFrom(CommitPhase.SUBMITTED, CommitPhase.CAN_COMMIT);
+ Builder<ListenableFuture<Boolean>> canCommitOperations = ImmutableList.builder();
+ for (DOMStoreThreePhaseCommitCohort cohort : cohorts) {
+ canCommitOperations.add(cohort.canCommit());
+ }
+ ListenableFuture<List<Boolean>> allCanCommits = Futures.allAsList(canCommitOperations.build());
+ ListenableFuture<Boolean> allSuccessFuture = Futures.transform(allCanCommits, AND_FUNCTION);
+ return Futures
+ .makeChecked(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<Void> abortAsyncAll() {
+ changeStateFrom(currentPhase, CommitPhase.ABORT);
+ Builder<ListenableFuture<Void>> 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<Void> 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;
+ };
+
+ }
+
+}
Code Review / controller.git / commitdiff