2 * Copyright (c) 2014 Brocade Communications Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.controller.cluster.datastore;
10 import akka.actor.ActorRef;
11 import akka.actor.Status.Failure;
12 import akka.serialization.Serialization;
13 import com.google.common.annotations.VisibleForTesting;
14 import com.google.common.base.Preconditions;
15 import com.google.common.base.Stopwatch;
16 import java.util.ArrayList;
17 import java.util.HashMap;
18 import java.util.Iterator;
19 import java.util.LinkedList;
20 import java.util.List;
22 import java.util.Queue;
23 import java.util.concurrent.ExecutionException;
24 import java.util.concurrent.TimeUnit;
25 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
26 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
27 import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
28 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
29 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
30 import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
31 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
32 import org.opendaylight.controller.cluster.datastore.modification.Modification;
33 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
34 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
35 import org.slf4j.Logger;
38 * Coordinates commits for a shard ensuring only one concurrent 3-phase commit.
40 * @author Thomas Pantelis
42 class ShardCommitCoordinator {
44 // Interface hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts.
45 public interface CohortDecorator {
46 ShardDataTreeCohort decorate(String transactionID, ShardDataTreeCohort actual);
49 private final Map<String, CohortEntry> cohortCache = new HashMap<>();
51 private CohortEntry currentCohortEntry;
53 private final ShardDataTree dataTree;
55 // We use a LinkedList here to avoid synchronization overhead with concurrent queue impls
56 // since this should only be accessed on the shard's dispatcher.
57 private final Queue<CohortEntry> queuedCohortEntries = new LinkedList<>();
59 private int queueCapacity;
61 private final Logger log;
63 private final String name;
65 private final long cacheExpiryTimeoutInMillis;
67 // This is a hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts.
68 private CohortDecorator cohortDecorator;
70 private ReadyTransactionReply readyTransactionReply;
72 private Runnable runOnPendingTransactionsComplete;
74 ShardCommitCoordinator(ShardDataTree dataTree,
75 long cacheExpiryTimeoutInMillis, int queueCapacity, Logger log, String name) {
77 this.queueCapacity = queueCapacity;
80 this.dataTree = Preconditions.checkNotNull(dataTree);
81 this.cacheExpiryTimeoutInMillis = cacheExpiryTimeoutInMillis;
85 return queuedCohortEntries.size();
88 int getCohortCacheSize() {
89 return cohortCache.size();
92 void setQueueCapacity(int queueCapacity) {
93 this.queueCapacity = queueCapacity;
96 private ReadyTransactionReply readyTransactionReply(Shard shard) {
97 if(readyTransactionReply == null) {
98 readyTransactionReply = new ReadyTransactionReply(Serialization.serializedActorPath(shard.self()));
101 return readyTransactionReply;
104 private boolean queueCohortEntry(CohortEntry cohortEntry, ActorRef sender, Shard shard) {
105 if(queuedCohortEntries.size() < queueCapacity) {
106 queuedCohortEntries.offer(cohortEntry);
108 log.debug("{}: Enqueued transaction {}, queue size {}", name, cohortEntry.getTransactionID(),
109 queuedCohortEntries.size());
113 cohortCache.remove(cohortEntry.getTransactionID());
115 RuntimeException ex = new RuntimeException(
116 String.format("%s: Could not enqueue transaction %s - the maximum commit queue"+
117 " capacity %d has been reached.",
118 name, cohortEntry.getTransactionID(), queueCapacity));
119 log.error(ex.getMessage());
120 sender.tell(new Failure(ex), shard.self());
126 * This method is called to ready a transaction that was prepared by ShardTransaction actor. It caches
127 * the prepared cohort entry for the given transactions ID in preparation for the subsequent 3-phase commit.
129 * @param ready the ForwardedReadyTransaction message to process
130 * @param sender the sender of the message
131 * @param shard the transaction's shard actor
133 void handleForwardedReadyTransaction(ForwardedReadyTransaction ready, ActorRef sender, Shard shard) {
134 log.debug("{}: Readying transaction {}, client version {}", name,
135 ready.getTransactionID(), ready.getTxnClientVersion());
137 ShardDataTreeCohort cohort = ready.getTransaction().ready();
138 CohortEntry cohortEntry = new CohortEntry(ready.getTransactionID(), cohort, ready.getTxnClientVersion());
139 cohortCache.put(ready.getTransactionID(), cohortEntry);
141 if(!queueCohortEntry(cohortEntry, sender, shard)) {
145 if(ready.isDoImmediateCommit()) {
146 cohortEntry.setDoImmediateCommit(true);
147 cohortEntry.setReplySender(sender);
148 cohortEntry.setShard(shard);
149 handleCanCommit(cohortEntry);
151 // The caller does not want immediate commit - the 3-phase commit will be coordinated by the
152 // front-end so send back a ReadyTransactionReply with our actor path.
153 sender.tell(readyTransactionReply(shard), shard.self());
158 * This method handles a BatchedModifications message for a transaction being prepared directly on the
159 * Shard actor instead of via a ShardTransaction actor. If there's no currently cached
160 * DOMStoreWriteTransaction, one is created. The batched modifications are applied to the write Tx. If
161 * the BatchedModifications is ready to commit then a DOMStoreThreePhaseCommitCohort is created.
163 * @param batched the BatchedModifications message to process
164 * @param sender the sender of the message
165 * @param shard the transaction's shard actor
167 void handleBatchedModifications(BatchedModifications batched, ActorRef sender, Shard shard) {
168 CohortEntry cohortEntry = cohortCache.get(batched.getTransactionID());
169 if(cohortEntry == null) {
170 cohortEntry = new CohortEntry(batched.getTransactionID(),
171 dataTree.newReadWriteTransaction(batched.getTransactionID(),
172 batched.getTransactionChainID()), batched.getVersion());
173 cohortCache.put(batched.getTransactionID(), cohortEntry);
176 if(log.isDebugEnabled()) {
177 log.debug("{}: Applying {} batched modifications for Tx {}", name,
178 batched.getModifications().size(), batched.getTransactionID());
181 cohortEntry.applyModifications(batched.getModifications());
183 if(batched.isReady()) {
184 if(cohortEntry.getLastBatchedModificationsException() != null) {
185 cohortCache.remove(cohortEntry.getTransactionID());
186 throw cohortEntry.getLastBatchedModificationsException();
189 if(cohortEntry.getTotalBatchedModificationsReceived() != batched.getTotalMessagesSent()) {
190 cohortCache.remove(cohortEntry.getTransactionID());
191 throw new IllegalStateException(String.format(
192 "The total number of batched messages received %d does not match the number sent %d",
193 cohortEntry.getTotalBatchedModificationsReceived(), batched.getTotalMessagesSent()));
196 if(!queueCohortEntry(cohortEntry, sender, shard)) {
200 if(log.isDebugEnabled()) {
201 log.debug("{}: Readying Tx {}, client version {}", name,
202 batched.getTransactionID(), batched.getVersion());
205 cohortEntry.ready(cohortDecorator, batched.isDoCommitOnReady());
207 if(batched.isDoCommitOnReady()) {
208 cohortEntry.setReplySender(sender);
209 cohortEntry.setShard(shard);
210 handleCanCommit(cohortEntry);
212 sender.tell(readyTransactionReply(shard), shard.self());
215 sender.tell(new BatchedModificationsReply(batched.getModifications().size()), shard.self());
220 * This method handles {@link ReadyLocalTransaction} message. All transaction modifications have
221 * been prepared beforehand by the sender and we just need to drive them through into the dataTree.
223 * @param message the ReadyLocalTransaction message to process
224 * @param sender the sender of the message
225 * @param shard the transaction's shard actor
227 void handleReadyLocalTransaction(ReadyLocalTransaction message, ActorRef sender, Shard shard) {
228 final ShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(dataTree, message.getModification(),
229 message.getTransactionID());
230 final CohortEntry cohortEntry = new CohortEntry(message.getTransactionID(), cohort,
231 DataStoreVersions.CURRENT_VERSION);
232 cohortCache.put(message.getTransactionID(), cohortEntry);
233 cohortEntry.setDoImmediateCommit(message.isDoCommitOnReady());
235 if(!queueCohortEntry(cohortEntry, sender, shard)) {
239 log.debug("{}: Applying local modifications for Tx {}", name, message.getTransactionID());
241 if (message.isDoCommitOnReady()) {
242 cohortEntry.setReplySender(sender);
243 cohortEntry.setShard(shard);
244 handleCanCommit(cohortEntry);
246 sender.tell(readyTransactionReply(shard), shard.self());
250 private void handleCanCommit(CohortEntry cohortEntry) {
251 String transactionID = cohortEntry.getTransactionID();
253 cohortEntry.updateLastAccessTime();
255 if(currentCohortEntry != null) {
256 // There's already a Tx commit in progress so we can't process this entry yet - but it's in the
257 // queue and will get processed after all prior entries complete.
259 if(log.isDebugEnabled()) {
260 log.debug("{}: Commit for Tx {} already in progress - skipping canCommit for {} for now",
261 name, currentCohortEntry.getTransactionID(), transactionID);
267 // No Tx commit currently in progress - check if this entry is the next one in the queue, If so make
268 // it the current entry and proceed with canCommit.
269 // Purposely checking reference equality here.
270 if(queuedCohortEntries.peek() == cohortEntry) {
271 currentCohortEntry = queuedCohortEntries.poll();
272 doCanCommit(currentCohortEntry);
274 if(log.isDebugEnabled()) {
275 log.debug("{}: Tx {} is the next pending canCommit - skipping {} for now",
276 name, queuedCohortEntries.peek().getTransactionID(), transactionID);
282 * This method handles the canCommit phase for a transaction.
284 * @param transactionID the ID of the transaction to canCommit
285 * @param sender the actor to which to send the response
286 * @param shard the transaction's shard actor
288 void handleCanCommit(String transactionID, final ActorRef sender, final Shard shard) {
289 // Lookup the cohort entry that was cached previously (or should have been) by
290 // transactionReady (via the ForwardedReadyTransaction message).
291 final CohortEntry cohortEntry = cohortCache.get(transactionID);
292 if(cohortEntry == null) {
293 // Either canCommit was invoked before ready(shouldn't happen) or a long time passed
294 // between canCommit and ready and the entry was expired from the cache.
295 IllegalStateException ex = new IllegalStateException(
296 String.format("%s: No cohort entry found for transaction %s", name, transactionID));
297 log.error(ex.getMessage());
298 sender.tell(new Failure(ex), shard.self());
302 cohortEntry.setReplySender(sender);
303 cohortEntry.setShard(shard);
305 handleCanCommit(cohortEntry);
308 private void doCanCommit(final CohortEntry cohortEntry) {
309 boolean canCommit = false;
311 canCommit = cohortEntry.canCommit();
313 log.debug("{}: canCommit for {}: {}", name, cohortEntry.getTransactionID(), canCommit);
315 if(cohortEntry.isDoImmediateCommit()) {
317 doCommit(cohortEntry);
319 cohortEntry.getReplySender().tell(new Failure(new TransactionCommitFailedException(
320 "Can Commit failed, no detailed cause available.")), cohortEntry.getShard().self());
323 // FIXME - use caller's version
324 cohortEntry.getReplySender().tell(
325 canCommit ? CanCommitTransactionReply.yes(cohortEntry.getClientVersion()).toSerializable() :
326 CanCommitTransactionReply.no(cohortEntry.getClientVersion()).toSerializable(),
327 cohortEntry.getShard().self());
329 } catch (Exception e) {
330 log.debug("{}: An exception occurred during canCommit", name, e);
332 Throwable failure = e;
333 if(e instanceof ExecutionException) {
334 failure = e.getCause();
337 cohortEntry.getReplySender().tell(new Failure(failure), cohortEntry.getShard().self());
340 // Remove the entry from the cache now.
341 currentTransactionComplete(cohortEntry.getTransactionID(), true);
346 private boolean doCommit(CohortEntry cohortEntry) {
347 log.debug("{}: Committing transaction {}", name, cohortEntry.getTransactionID());
349 boolean success = false;
351 // We perform the preCommit phase here atomically with the commit phase. This is an
352 // optimization to eliminate the overhead of an extra preCommit message. We lose front-end
353 // coordination of preCommit across shards in case of failure but preCommit should not
354 // normally fail since we ensure only one concurrent 3-phase commit.
357 cohortEntry.preCommit();
359 cohortEntry.getShard().continueCommit(cohortEntry);
361 cohortEntry.updateLastAccessTime();
364 } catch (Exception e) {
365 log.error("{} An exception occurred while preCommitting transaction {}",
366 name, cohortEntry.getTransactionID(), e);
367 cohortEntry.getReplySender().tell(new Failure(e), cohortEntry.getShard().self());
369 currentTransactionComplete(cohortEntry.getTransactionID(), true);
376 * This method handles the preCommit and commit phases for a transaction.
378 * @param transactionID the ID of the transaction to commit
379 * @param sender the actor to which to send the response
380 * @param shard the transaction's shard actor
381 * @return true if the transaction was successfully prepared, false otherwise.
383 boolean handleCommit(final String transactionID, final ActorRef sender, final Shard shard) {
384 // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to
386 final CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID);
387 if(cohortEntry == null) {
388 // We're not the current Tx - the Tx was likely expired b/c it took too long in
389 // between the canCommit and commit messages.
390 IllegalStateException ex = new IllegalStateException(
391 String.format("%s: Cannot commit transaction %s - it is not the current transaction",
392 name, transactionID));
393 log.error(ex.getMessage());
394 sender.tell(new Failure(ex), shard.self());
398 cohortEntry.setReplySender(sender);
399 return doCommit(cohortEntry);
402 void handleAbort(final String transactionID, final ActorRef sender, final Shard shard) {
403 CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID);
404 if(cohortEntry != null) {
405 // We don't remove the cached cohort entry here (ie pass false) in case the Tx was
406 // aborted during replication in which case we may still commit locally if replication
408 currentTransactionComplete(transactionID, false);
410 cohortEntry = getAndRemoveCohortEntry(transactionID);
413 if(cohortEntry == null) {
417 log.debug("{}: Aborting transaction {}", name, transactionID);
419 final ActorRef self = shard.getSelf();
423 shard.getShardMBean().incrementAbortTransactionsCount();
426 sender.tell(AbortTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(), self);
428 } catch (Exception e) {
429 log.error("{}: An exception happened during abort", name, e);
432 sender.tell(new Failure(e), self);
437 void checkForExpiredTransactions(final long timeout, final Shard shard) {
438 CohortEntry cohortEntry = getCurrentCohortEntry();
439 if(cohortEntry != null) {
440 if(cohortEntry.isExpired(timeout)) {
441 log.warn("{}: Current transaction {} has timed out after {} ms - aborting",
442 name, cohortEntry.getTransactionID(), timeout);
444 handleAbort(cohortEntry.getTransactionID(), null, shard);
448 cleanupExpiredCohortEntries();
451 void abortPendingTransactions(final String reason, final Shard shard) {
452 if(currentCohortEntry == null && queuedCohortEntries.isEmpty()) {
456 List<CohortEntry> cohortEntries = new ArrayList<>();
458 if(currentCohortEntry != null) {
459 cohortEntries.add(currentCohortEntry);
460 currentCohortEntry = null;
463 cohortEntries.addAll(queuedCohortEntries);
464 queuedCohortEntries.clear();
466 for(CohortEntry cohortEntry: cohortEntries) {
467 if(cohortEntry.getReplySender() != null) {
468 cohortEntry.getReplySender().tell(new Failure(new RuntimeException(reason)), shard.self());
474 * Returns the cohort entry for the Tx commit currently in progress if the given transaction ID
475 * matches the current entry.
477 * @param transactionID the ID of the transaction
478 * @return the current CohortEntry or null if the given transaction ID does not match the
481 CohortEntry getCohortEntryIfCurrent(String transactionID) {
482 if(isCurrentTransaction(transactionID)) {
483 return currentCohortEntry;
489 CohortEntry getCurrentCohortEntry() {
490 return currentCohortEntry;
493 CohortEntry getAndRemoveCohortEntry(String transactionID) {
494 return cohortCache.remove(transactionID);
497 boolean isCurrentTransaction(String transactionID) {
498 return currentCohortEntry != null &&
499 currentCohortEntry.getTransactionID().equals(transactionID);
503 * This method is called when a transaction is complete, successful or not. If the given
504 * given transaction ID matches the current in-progress transaction, the next cohort entry,
505 * if any, is dequeued and processed.
507 * @param transactionID the ID of the completed transaction
508 * @param removeCohortEntry if true the CohortEntry for the transaction is also removed from
511 void currentTransactionComplete(String transactionID, boolean removeCohortEntry) {
512 if(removeCohortEntry) {
513 cohortCache.remove(transactionID);
516 if(isCurrentTransaction(transactionID)) {
517 currentCohortEntry = null;
519 log.debug("{}: currentTransactionComplete: {}", name, transactionID);
521 maybeProcessNextCohortEntry();
525 private void maybeProcessNextCohortEntry() {
526 // Check if there's a next cohort entry waiting in the queue and if it is ready to commit. Also
527 // clean out expired entries.
528 Iterator<CohortEntry> iter = queuedCohortEntries.iterator();
529 while(iter.hasNext()) {
530 CohortEntry next = iter.next();
531 if(next.isReadyToCommit()) {
532 if(currentCohortEntry == null) {
533 if(log.isDebugEnabled()) {
534 log.debug("{}: Next entry to canCommit {}", name, next);
538 currentCohortEntry = next;
539 currentCohortEntry.updateLastAccessTime();
540 doCanCommit(currentCohortEntry);
544 } else if(next.isExpired(cacheExpiryTimeoutInMillis)) {
545 log.warn("{}: canCommit for transaction {} was not received within {} ms - entry removed from cache",
546 name, next.getTransactionID(), cacheExpiryTimeoutInMillis);
547 } else if(!next.isAborted()) {
552 cohortCache.remove(next.getTransactionID());
555 maybeRunOperationOnPendingTransactionsComplete();
558 void cleanupExpiredCohortEntries() {
559 maybeProcessNextCohortEntry();
562 void setRunOnPendingTransactionsComplete(Runnable operation) {
563 runOnPendingTransactionsComplete = operation;
564 maybeRunOperationOnPendingTransactionsComplete();
567 private void maybeRunOperationOnPendingTransactionsComplete() {
568 if(runOnPendingTransactionsComplete != null && currentCohortEntry == null && queuedCohortEntries.isEmpty()) {
569 log.debug("{}: Pending transactions complete - running operation {}", name, runOnPendingTransactionsComplete);
571 runOnPendingTransactionsComplete.run();
572 runOnPendingTransactionsComplete = null;
577 void setCohortDecorator(CohortDecorator cohortDecorator) {
578 this.cohortDecorator = cohortDecorator;
581 static class CohortEntry {
582 private final String transactionID;
583 private ShardDataTreeCohort cohort;
584 private final ReadWriteShardDataTreeTransaction transaction;
585 private RuntimeException lastBatchedModificationsException;
586 private ActorRef replySender;
588 private boolean doImmediateCommit;
589 private final Stopwatch lastAccessTimer = Stopwatch.createStarted();
590 private int totalBatchedModificationsReceived;
591 private boolean aborted;
592 private final short clientVersion;
594 CohortEntry(String transactionID, ReadWriteShardDataTreeTransaction transaction, short clientVersion) {
595 this.transaction = Preconditions.checkNotNull(transaction);
596 this.transactionID = transactionID;
597 this.clientVersion = clientVersion;
600 CohortEntry(String transactionID, ShardDataTreeCohort cohort, short clientVersion) {
601 this.transactionID = transactionID;
602 this.cohort = cohort;
603 this.transaction = null;
604 this.clientVersion = clientVersion;
607 void updateLastAccessTime() {
608 lastAccessTimer.reset();
609 lastAccessTimer.start();
612 String getTransactionID() {
613 return transactionID;
616 short getClientVersion() {
617 return clientVersion;
620 DataTreeCandidate getCandidate() {
621 return cohort.getCandidate();
624 int getTotalBatchedModificationsReceived() {
625 return totalBatchedModificationsReceived;
628 RuntimeException getLastBatchedModificationsException() {
629 return lastBatchedModificationsException;
632 void applyModifications(Iterable<Modification> modifications) {
633 totalBatchedModificationsReceived++;
634 if(lastBatchedModificationsException == null) {
635 for (Modification modification : modifications) {
637 modification.apply(transaction.getSnapshot());
638 } catch (RuntimeException e) {
639 lastBatchedModificationsException = e;
646 boolean canCommit() throws InterruptedException, ExecutionException {
647 // We block on the future here (and also preCommit(), commit(), abort()) so we don't have to worry
648 // about possibly accessing our state on a different thread outside of our dispatcher.
649 // TODO: the ShardDataTreeCohort returns immediate Futures anyway which begs the question - why
650 // bother even returning Futures from ShardDataTreeCohort if we have to treat them synchronously
651 // anyway?. The Futures are really a remnant from when we were using the InMemoryDataBroker.
652 return cohort.canCommit().get();
655 void preCommit() throws InterruptedException, ExecutionException {
656 cohort.preCommit().get();
659 void commit() throws InterruptedException, ExecutionException {
660 cohort.commit().get();
663 void abort() throws InterruptedException, ExecutionException {
665 cohort.abort().get();
668 void ready(CohortDecorator cohortDecorator, boolean doImmediateCommit) {
669 Preconditions.checkState(cohort == null, "cohort was already set");
671 setDoImmediateCommit(doImmediateCommit);
673 cohort = transaction.ready();
675 if(cohortDecorator != null) {
676 // Call the hook for unit tests.
677 cohort = cohortDecorator.decorate(transactionID, cohort);
681 boolean isReadyToCommit() {
682 return replySender != null;
685 boolean isExpired(long expireTimeInMillis) {
686 return lastAccessTimer.elapsed(TimeUnit.MILLISECONDS) >= expireTimeInMillis;
689 boolean isDoImmediateCommit() {
690 return doImmediateCommit;
693 void setDoImmediateCommit(boolean doImmediateCommit) {
694 this.doImmediateCommit = doImmediateCommit;
697 ActorRef getReplySender() {
701 void setReplySender(ActorRef replySender) {
702 this.replySender = replySender;
709 void setShard(Shard shard) {
714 boolean isAborted() {
719 public String toString() {
720 StringBuilder builder = new StringBuilder();
721 builder.append("CohortEntry [transactionID=").append(transactionID).append(", doImmediateCommit=")
722 .append(doImmediateCommit).append("]");
723 return builder.toString();