X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShardCommitCoordinator.java;h=0d63115754698688f5c64fba9d46d63b93336f87;hb=87c8362c7501408b281f5ddc9b78ed7440280fa1;hp=f3e1e33e347f7760820bce49897bf2aa3c20a5b6;hpb=103ceecd0195cca6c87fbd62a687d8addf128784;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardCommitCoordinator.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardCommitCoordinator.java index f3e1e33e34..0d63115754 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardCommitCoordinator.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardCommitCoordinator.java @@ -21,6 +21,7 @@ import java.util.Queue; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import org.opendaylight.controller.cluster.datastore.compat.BackwardsCompatibleThreePhaseCommitCohort; +import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply; import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply; @@ -28,8 +29,8 @@ import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTran import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction; import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply; import org.opendaylight.controller.cluster.datastore.modification.Modification; -import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification; import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate; import org.slf4j.Logger; /** @@ -37,7 +38,7 @@ import org.slf4j.Logger; * * @author Thomas Pantelis */ -public class ShardCommitCoordinator { +class ShardCommitCoordinator { // Interface hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts. public interface CohortDecorator { @@ -67,8 +68,8 @@ public class ShardCommitCoordinator { private ReadyTransactionReply readyTransactionReply; - public ShardCommitCoordinator(ShardDataTree dataTree, - long cacheExpiryTimeoutInMillis, int queueCapacity, ActorRef shardActor, Logger log, String name) { + ShardCommitCoordinator(ShardDataTree dataTree, + long cacheExpiryTimeoutInMillis, int queueCapacity, Logger log, String name) { this.queueCapacity = queueCapacity; this.log = log; @@ -77,7 +78,11 @@ public class ShardCommitCoordinator { this.cacheExpiryTimeoutInMillis = cacheExpiryTimeoutInMillis; } - public void setQueueCapacity(int queueCapacity) { + int getQueueSize() { + return queuedCohortEntries.size(); + } + + void setQueueCapacity(int queueCapacity) { this.queueCapacity = queueCapacity; } @@ -92,6 +97,10 @@ public class ShardCommitCoordinator { private boolean queueCohortEntry(CohortEntry cohortEntry, ActorRef sender, Shard shard) { if(queuedCohortEntries.size() < queueCapacity) { queuedCohortEntries.offer(cohortEntry); + + log.debug("{}: Enqueued transaction {}, queue size {}", name, cohortEntry.getTransactionID(), + queuedCohortEntries.size()); + return true; } else { cohortCache.remove(cohortEntry.getTransactionID()); @@ -109,13 +118,16 @@ public class ShardCommitCoordinator { /** * This method is called to ready a transaction that was prepared by ShardTransaction actor. It caches * the prepared cohort entry for the given transactions ID in preparation for the subsequent 3-phase commit. + * + * @param ready the ForwardedReadyTransaction message to process + * @param sender the sender of the message + * @param shard the transaction's shard actor */ - public void handleForwardedReadyTransaction(ForwardedReadyTransaction ready, ActorRef sender, Shard shard) { + void handleForwardedReadyTransaction(ForwardedReadyTransaction ready, ActorRef sender, Shard shard) { log.debug("{}: Readying transaction {}, client version {}", name, ready.getTransactionID(), ready.getTxnClientVersion()); - CohortEntry cohortEntry = new CohortEntry(ready.getTransactionID(), ready.getCohort(), - (MutableCompositeModification) ready.getModification()); + CohortEntry cohortEntry = new CohortEntry(ready.getTransactionID(), ready.getCohort()); cohortCache.put(ready.getTransactionID(), cohortEntry); if(!queueCohortEntry(cohortEntry, sender, shard)) { @@ -159,13 +171,11 @@ public class ShardCommitCoordinator { * DOMStoreWriteTransaction, one is created. The batched modifications are applied to the write Tx. If * the BatchedModifications is ready to commit then a DOMStoreThreePhaseCommitCohort is created. * - * @param batched the BatchedModifications - * @param shardActor the transaction's shard actor - * - * @throws ExecutionException if an error occurs loading the cache + * @param batched the BatchedModifications message to process + * @param sender the sender of the message + * @param shard the transaction's shard actor */ - void handleBatchedModifications(BatchedModifications batched, ActorRef sender, Shard shard) - throws ExecutionException { + void handleBatchedModifications(BatchedModifications batched, ActorRef sender, Shard shard) { CohortEntry cohortEntry = cohortCache.get(batched.getTransactionID()); if(cohortEntry == null) { cohortEntry = new CohortEntry(batched.getTransactionID(), @@ -182,6 +192,18 @@ public class ShardCommitCoordinator { cohortEntry.applyModifications(batched.getModifications()); if(batched.isReady()) { + if(cohortEntry.getLastBatchedModificationsException() != null) { + cohortCache.remove(cohortEntry.getTransactionID()); + throw cohortEntry.getLastBatchedModificationsException(); + } + + if(cohortEntry.getTotalBatchedModificationsReceived() != batched.getTotalMessagesSent()) { + cohortCache.remove(cohortEntry.getTransactionID()); + throw new IllegalStateException(String.format( + "The total number of batched messages received %d does not match the number sent %d", + cohortEntry.getTotalBatchedModificationsReceived(), batched.getTotalMessagesSent())); + } + if(!queueCohortEntry(cohortEntry, sender, shard)) { return; } @@ -209,12 +231,13 @@ public class ShardCommitCoordinator { * This method handles {@link ReadyLocalTransaction} message. All transaction modifications have * been prepared beforehand by the sender and we just need to drive them through into the dataTree. * - * @param message - * @param sender - * @param shard + * @param message the ReadyLocalTransaction message to process + * @param sender the sender of the message + * @param shard the transaction's shard actor */ void handleReadyLocalTransaction(ReadyLocalTransaction message, ActorRef sender, Shard shard) { - final ShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(dataTree, message.getModification()); + final ShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(dataTree, message.getModification(), + message.getTransactionID()); final CohortEntry cohortEntry = new CohortEntry(message.getTransactionID(), cohort); cohortCache.put(message.getTransactionID(), cohortEntry); cohortEntry.setDoImmediateCommit(message.isDoCommitOnReady()); @@ -268,11 +291,11 @@ public class ShardCommitCoordinator { /** * This method handles the canCommit phase for a transaction. * - * @param canCommit the CanCommitTransaction message - * @param sender the actor that sent the message + * @param transactionID the ID of the transaction to canCommit + * @param sender the actor to which to send the response * @param shard the transaction's shard actor */ - public void handleCanCommit(String transactionID, final ActorRef sender, final Shard shard) { + void handleCanCommit(String transactionID, final ActorRef sender, final Shard shard) { // Lookup the cohort entry that was cached previously (or should have been) by // transactionReady (via the ForwardedReadyTransaction message). final CohortEntry cohortEntry = cohortCache.get(transactionID); @@ -295,10 +318,7 @@ public class ShardCommitCoordinator { private void doCanCommit(final CohortEntry cohortEntry) { boolean canCommit = false; try { - // We block on the future here so we don't have to worry about possibly accessing our - // state on a different thread outside of our dispatcher. Also, the data store - // currently uses a same thread executor anyway. - canCommit = cohortEntry.getCohort().canCommit().get(); + canCommit = cohortEntry.canCommit(); log.debug("{}: canCommit for {}: {}", name, cohortEntry.getTransactionID(), canCommit); @@ -342,10 +362,7 @@ public class ShardCommitCoordinator { // normally fail since we ensure only one concurrent 3-phase commit. try { - // We block on the future here so we don't have to worry about possibly accessing our - // state on a different thread outside of our dispatcher. Also, the data store - // currently uses a same thread executor anyway. - cohortEntry.getCohort().preCommit().get(); + cohortEntry.preCommit(); cohortEntry.getShard().continueCommit(cohortEntry); @@ -363,6 +380,14 @@ public class ShardCommitCoordinator { return success; } + /** + * This method handles the preCommit and commit phases for a transaction. + * + * @param transactionID the ID of the transaction to commit + * @param sender the actor to which to send the response + * @param shard the transaction's shard actor + * @return true if the transaction was successfully prepared, false otherwise. + */ boolean handleCommit(final String transactionID, final ActorRef sender, final Shard shard) { // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to // this transaction. @@ -382,6 +407,41 @@ public class ShardCommitCoordinator { return doCommit(cohortEntry); } + void handleAbort(final String transactionID, final ActorRef sender, final Shard shard) { + CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID); + if(cohortEntry != null) { + // We don't remove the cached cohort entry here (ie pass false) in case the Tx was + // aborted during replication in which case we may still commit locally if replication + // succeeds. + currentTransactionComplete(transactionID, false); + } else { + cohortEntry = getAndRemoveCohortEntry(transactionID); + } + + if(cohortEntry == null) { + return; + } + + log.debug("{}: Aborting transaction {}", name, transactionID); + + final ActorRef self = shard.getSelf(); + try { + cohortEntry.abort(); + + shard.getShardMBean().incrementAbortTransactionsCount(); + + if(sender != null) { + sender.tell(new AbortTransactionReply().toSerializable(), self); + } + } catch (Exception e) { + log.error("{}: An exception happened during abort", name, e); + + if(sender != null) { + sender.tell(new akka.actor.Status.Failure(e), self); + } + } + } + /** * Returns the cohort entry for the Tx commit currently in progress if the given transaction ID * matches the current entry. @@ -456,12 +516,12 @@ public class ShardCommitCoordinator { } else if(next.isExpired(cacheExpiryTimeoutInMillis)) { log.warn("{}: canCommit for transaction {} was not received within {} ms - entry removed from cache", name, next.getTransactionID(), cacheExpiryTimeoutInMillis); - - iter.remove(); - cohortCache.remove(next.getTransactionID()); - } else { + } else if(!next.isAborted()) { break; } + + iter.remove(); + cohortCache.remove(next.getTransactionID()); } } @@ -478,23 +538,19 @@ public class ShardCommitCoordinator { private final String transactionID; private ShardDataTreeCohort cohort; private final ReadWriteShardDataTreeTransaction transaction; + private RuntimeException lastBatchedModificationsException; private ActorRef replySender; private Shard shard; private boolean doImmediateCommit; private final Stopwatch lastAccessTimer = Stopwatch.createStarted(); + private int totalBatchedModificationsReceived; + private boolean aborted; CohortEntry(String transactionID, ReadWriteShardDataTreeTransaction transaction) { this.transaction = Preconditions.checkNotNull(transaction); this.transactionID = transactionID; } - CohortEntry(String transactionID, ShardDataTreeCohort cohort, - MutableCompositeModification compositeModification) { - this.transactionID = transactionID; - this.cohort = cohort; - this.transaction = null; - } - CohortEntry(String transactionID, ShardDataTreeCohort cohort) { this.transactionID = transactionID; this.cohort = cohort; @@ -510,16 +566,54 @@ public class ShardCommitCoordinator { return transactionID; } - ShardDataTreeCohort getCohort() { - return cohort; + DataTreeCandidate getCandidate() { + return cohort.getCandidate(); + } + + int getTotalBatchedModificationsReceived() { + return totalBatchedModificationsReceived; + } + + RuntimeException getLastBatchedModificationsException() { + return lastBatchedModificationsException; } void applyModifications(Iterable modifications) { - for (Modification modification : modifications) { - modification.apply(transaction.getSnapshot()); + totalBatchedModificationsReceived++; + if(lastBatchedModificationsException == null) { + for (Modification modification : modifications) { + try { + modification.apply(transaction.getSnapshot()); + } catch (RuntimeException e) { + lastBatchedModificationsException = e; + throw e; + } + } } } + boolean canCommit() throws InterruptedException, ExecutionException { + // We block on the future here (and also preCommit(), commit(), abort()) so we don't have to worry + // about possibly accessing our state on a different thread outside of our dispatcher. + // TODO: the ShardDataTreeCohort returns immediate Futures anyway which begs the question - why + // bother even returning Futures from ShardDataTreeCohort if we have to treat them synchronously + // anyway?. The Futures are really a remnant from when we were using the InMemoryDataBroker. + return cohort.canCommit().get(); + } + + void preCommit() throws InterruptedException, ExecutionException { + cohort.preCommit().get(); + } + + void commit() throws InterruptedException, ExecutionException { + cohort.commit().get(); + } + + void abort() throws InterruptedException, ExecutionException { + aborted = true; + cohort.abort().get(); + } + void ready(CohortDecorator cohortDecorator, boolean doImmediateCommit) { Preconditions.checkState(cohort == null, "cohort was already set"); @@ -565,6 +659,11 @@ public class ShardCommitCoordinator { this.shard = shard; } + + boolean isAborted() { + return aborted; + } + @Override public String toString() { StringBuilder builder = new StringBuilder();