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 java.util.ArrayList;
16 import java.util.Collection;
17 import java.util.Collections;
18 import java.util.HashMap;
19 import java.util.Iterator;
20 import java.util.LinkedList;
21 import java.util.List;
23 import java.util.Queue;
24 import java.util.concurrent.ExecutionException;
25 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActorRegistry.CohortRegistryCommand;
26 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
27 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
28 import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
29 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
30 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
31 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
32 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
33 import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
34 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
35 import org.opendaylight.controller.cluster.datastore.utils.AbstractBatchedModificationsCursor;
36 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
37 import org.opendaylight.yangtools.concepts.Identifier;
38 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
39 import org.slf4j.Logger;
42 * Coordinates commits for a shard ensuring only one concurrent 3-phase commit.
44 * @author Thomas Pantelis
46 final class ShardCommitCoordinator {
48 // Interface hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts.
49 public interface CohortDecorator {
50 ShardDataTreeCohort decorate(Identifier transactionID, ShardDataTreeCohort actual);
53 private final Map<Identifier, CohortEntry> cohortCache = new HashMap<>();
55 private CohortEntry currentCohortEntry;
57 private final ShardDataTree dataTree;
59 private final DataTreeCohortActorRegistry cohortRegistry = new DataTreeCohortActorRegistry();
61 // We use a LinkedList here to avoid synchronization overhead with concurrent queue impls
62 // since this should only be accessed on the shard's dispatcher.
63 private final Queue<CohortEntry> queuedCohortEntries = new LinkedList<>();
65 private int queueCapacity;
67 private final Logger log;
69 private final String name;
71 private final long cacheExpiryTimeoutInMillis;
73 // This is a hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts.
74 private CohortDecorator cohortDecorator;
76 private ReadyTransactionReply readyTransactionReply;
78 private Runnable runOnPendingTransactionsComplete;
80 ShardCommitCoordinator(ShardDataTree dataTree, long cacheExpiryTimeoutInMillis, int queueCapacity, Logger log,
83 this.queueCapacity = queueCapacity;
86 this.dataTree = Preconditions.checkNotNull(dataTree);
87 this.cacheExpiryTimeoutInMillis = cacheExpiryTimeoutInMillis;
91 return queuedCohortEntries.size();
94 int getCohortCacheSize() {
95 return cohortCache.size();
98 void setQueueCapacity(int queueCapacity) {
99 this.queueCapacity = queueCapacity;
102 private ReadyTransactionReply readyTransactionReply(Shard shard) {
103 if(readyTransactionReply == null) {
104 readyTransactionReply = new ReadyTransactionReply(Serialization.serializedActorPath(shard.self()));
107 return readyTransactionReply;
110 private boolean queueCohortEntry(CohortEntry cohortEntry, ActorRef sender, Shard shard) {
111 if(queuedCohortEntries.size() < queueCapacity) {
112 queuedCohortEntries.offer(cohortEntry);
114 log.debug("{}: Enqueued transaction {}, queue size {}", name, cohortEntry.getTransactionID(),
115 queuedCohortEntries.size());
119 cohortCache.remove(cohortEntry.getTransactionID());
121 final RuntimeException ex = new RuntimeException(
122 String.format("%s: Could not enqueue transaction %s - the maximum commit queue"+
123 " capacity %d has been reached.",
124 name, cohortEntry.getTransactionID(), queueCapacity));
125 log.error(ex.getMessage());
126 sender.tell(new Failure(ex), shard.self());
132 * This method is called to ready a transaction that was prepared by ShardTransaction actor. It caches
133 * the prepared cohort entry for the given transactions ID in preparation for the subsequent 3-phase commit.
135 * @param ready the ForwardedReadyTransaction message to process
136 * @param sender the sender of the message
137 * @param shard the transaction's shard actor
140 void handleForwardedReadyTransaction(ForwardedReadyTransaction ready, ActorRef sender, Shard shard,
141 SchemaContext schema) {
142 log.debug("{}: Readying transaction {}, client version {}", name,
143 ready.getTransactionID(), ready.getTxnClientVersion());
145 final ShardDataTreeCohort cohort = ready.getTransaction().ready();
146 final CohortEntry cohortEntry = new CohortEntry(ready.getTransactionID(), cohort, cohortRegistry, schema, ready.getTxnClientVersion());
147 cohortCache.put(cohortEntry.getTransactionID(), cohortEntry);
149 if(!queueCohortEntry(cohortEntry, sender, shard)) {
153 if(ready.isDoImmediateCommit()) {
154 cohortEntry.setDoImmediateCommit(true);
155 cohortEntry.setReplySender(sender);
156 cohortEntry.setShard(shard);
157 handleCanCommit(cohortEntry);
159 // The caller does not want immediate commit - the 3-phase commit will be coordinated by the
160 // front-end so send back a ReadyTransactionReply with our actor path.
161 sender.tell(readyTransactionReply(shard), shard.self());
166 * This method handles a BatchedModifications message for a transaction being prepared directly on the
167 * Shard actor instead of via a ShardTransaction actor. If there's no currently cached
168 * DOMStoreWriteTransaction, one is created. The batched modifications are applied to the write Tx. If
169 * the BatchedModifications is ready to commit then a DOMStoreThreePhaseCommitCohort is created.
171 * @param batched the BatchedModifications message to process
172 * @param sender the sender of the message
174 void handleBatchedModifications(BatchedModifications batched, ActorRef sender, Shard shard) {
175 CohortEntry cohortEntry = cohortCache.get(batched.getTransactionID());
176 if(cohortEntry == null) {
177 cohortEntry = new CohortEntry(batched.getTransactionID(),
178 dataTree.newReadWriteTransaction(batched.getTransactionID()),
179 cohortRegistry, dataTree.getSchemaContext(), batched.getVersion());
180 cohortCache.put(cohortEntry.getTransactionID(), cohortEntry);
183 if(log.isDebugEnabled()) {
184 log.debug("{}: Applying {} batched modifications for Tx {}", name,
185 batched.getModifications().size(), batched.getTransactionID());
188 cohortEntry.applyModifications(batched.getModifications());
190 if(batched.isReady()) {
191 if(cohortEntry.getLastBatchedModificationsException() != null) {
192 cohortCache.remove(cohortEntry.getTransactionID());
193 throw cohortEntry.getLastBatchedModificationsException();
196 if(cohortEntry.getTotalBatchedModificationsReceived() != batched.getTotalMessagesSent()) {
197 cohortCache.remove(cohortEntry.getTransactionID());
198 throw new IllegalStateException(String.format(
199 "The total number of batched messages received %d does not match the number sent %d",
200 cohortEntry.getTotalBatchedModificationsReceived(), batched.getTotalMessagesSent()));
203 if(!queueCohortEntry(cohortEntry, sender, shard)) {
207 if(log.isDebugEnabled()) {
208 log.debug("{}: Readying Tx {}, client version {}", name,
209 batched.getTransactionID(), batched.getVersion());
212 cohortEntry.ready(cohortDecorator, batched.isDoCommitOnReady());
214 if(batched.isDoCommitOnReady()) {
215 cohortEntry.setReplySender(sender);
216 cohortEntry.setShard(shard);
217 handleCanCommit(cohortEntry);
219 sender.tell(readyTransactionReply(shard), shard.self());
222 sender.tell(new BatchedModificationsReply(batched.getModifications().size()), shard.self());
227 * This method handles {@link ReadyLocalTransaction} message. All transaction modifications have
228 * been prepared beforehand by the sender and we just need to drive them through into the
231 * @param message the ReadyLocalTransaction message to process
232 * @param sender the sender of the message
233 * @param shard the transaction's shard actor
235 void handleReadyLocalTransaction(ReadyLocalTransaction message, ActorRef sender, Shard shard) {
236 final ShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(dataTree, message.getModification(),
237 message.getTransactionID());
238 final CohortEntry cohortEntry = new CohortEntry(message.getTransactionID(), cohort, cohortRegistry,
239 dataTree.getSchemaContext(), DataStoreVersions.CURRENT_VERSION);
240 cohortCache.put(cohortEntry.getTransactionID(), cohortEntry);
241 cohortEntry.setDoImmediateCommit(message.isDoCommitOnReady());
243 if(!queueCohortEntry(cohortEntry, sender, shard)) {
247 log.debug("{}: Applying local modifications for Tx {}", name, message.getTransactionID());
249 if (message.isDoCommitOnReady()) {
250 cohortEntry.setReplySender(sender);
251 cohortEntry.setShard(shard);
252 handleCanCommit(cohortEntry);
254 sender.tell(readyTransactionReply(shard), shard.self());
258 Collection<BatchedModifications> createForwardedBatchedModifications(final BatchedModifications from,
259 final int maxModificationsPerBatch) {
260 CohortEntry cohortEntry = getAndRemoveCohortEntry(from.getTransactionID());
261 if(cohortEntry == null || cohortEntry.getTransaction() == null) {
262 return Collections.singletonList(from);
265 cohortEntry.applyModifications(from.getModifications());
267 final LinkedList<BatchedModifications> newModifications = new LinkedList<>();
268 cohortEntry.getTransaction().getSnapshot().applyToCursor(new AbstractBatchedModificationsCursor() {
270 protected BatchedModifications getModifications() {
271 if(newModifications.isEmpty() ||
272 newModifications.getLast().getModifications().size() >= maxModificationsPerBatch) {
273 newModifications.add(new BatchedModifications(from.getTransactionID(), from.getVersion()));
276 return newModifications.getLast();
280 BatchedModifications last = newModifications.getLast();
281 last.setDoCommitOnReady(from.isDoCommitOnReady());
282 last.setReady(from.isReady());
283 last.setTotalMessagesSent(newModifications.size());
284 return newModifications;
287 private void handleCanCommit(CohortEntry cohortEntry) {
288 cohortEntry.updateLastAccessTime();
290 if(currentCohortEntry != null) {
291 // There's already a Tx commit in progress so we can't process this entry yet - but it's in the
292 // queue and will get processed after all prior entries complete.
294 if(log.isDebugEnabled()) {
295 log.debug("{}: Commit for Tx {} already in progress - skipping canCommit for {} for now",
296 name, currentCohortEntry.getTransactionID(), cohortEntry.getTransactionID());
302 // No Tx commit currently in progress - check if this entry is the next one in the queue, If so make
303 // it the current entry and proceed with canCommit.
304 // Purposely checking reference equality here.
305 if(queuedCohortEntries.peek() == cohortEntry) {
306 currentCohortEntry = queuedCohortEntries.poll();
307 doCanCommit(currentCohortEntry);
309 if(log.isDebugEnabled()) {
310 log.debug("{}: Tx {} is the next pending canCommit - skipping {} for now", name,
311 queuedCohortEntries.peek() != null ? queuedCohortEntries.peek().getTransactionID() : "???",
312 cohortEntry.getTransactionID());
318 * This method handles the canCommit phase for a transaction.
320 * @param transactionID the ID of the transaction to canCommit
321 * @param sender the actor to which to send the response
322 * @param shard the transaction's shard actor
324 void handleCanCommit(Identifier transactionID, final ActorRef sender, final Shard shard) {
325 // Lookup the cohort entry that was cached previously (or should have been) by
326 // transactionReady (via the ForwardedReadyTransaction message).
327 final CohortEntry cohortEntry = cohortCache.get(transactionID);
328 if(cohortEntry == null) {
329 // Either canCommit was invoked before ready(shouldn't happen) or a long time passed
330 // between canCommit and ready and the entry was expired from the cache.
331 IllegalStateException ex = new IllegalStateException(
332 String.format("%s: No cohort entry found for transaction %s", name, transactionID));
333 log.error(ex.getMessage());
334 sender.tell(new Failure(ex), shard.self());
338 cohortEntry.setReplySender(sender);
339 cohortEntry.setShard(shard);
341 handleCanCommit(cohortEntry);
344 private void doCanCommit(final CohortEntry cohortEntry) {
345 boolean canCommit = false;
347 canCommit = cohortEntry.canCommit();
349 log.debug("{}: canCommit for {}: {}", name, cohortEntry.getTransactionID(), canCommit);
351 if(cohortEntry.isDoImmediateCommit()) {
353 doCommit(cohortEntry);
355 cohortEntry.getReplySender().tell(new Failure(new TransactionCommitFailedException(
356 "Can Commit failed, no detailed cause available.")), cohortEntry.getShard().self());
359 cohortEntry.getReplySender().tell(
360 canCommit ? CanCommitTransactionReply.yes(cohortEntry.getClientVersion()).toSerializable() :
361 CanCommitTransactionReply.no(cohortEntry.getClientVersion()).toSerializable(),
362 cohortEntry.getShard().self());
364 } catch (Exception e) {
365 log.debug("{}: An exception occurred during canCommit", name, e);
367 Throwable failure = e;
368 if(e instanceof ExecutionException) {
369 failure = e.getCause();
372 cohortEntry.getReplySender().tell(new Failure(failure), cohortEntry.getShard().self());
375 // Remove the entry from the cache now.
376 currentTransactionComplete(cohortEntry.getTransactionID(), true);
381 private boolean doCommit(CohortEntry cohortEntry) {
382 log.debug("{}: Committing transaction {}", name, cohortEntry.getTransactionID());
384 boolean success = false;
386 // We perform the preCommit phase here atomically with the commit phase. This is an
387 // optimization to eliminate the overhead of an extra preCommit message. We lose front-end
388 // coordination of preCommit across shards in case of failure but preCommit should not
389 // normally fail since we ensure only one concurrent 3-phase commit.
392 cohortEntry.preCommit();
394 cohortEntry.getShard().continueCommit(cohortEntry);
396 cohortEntry.updateLastAccessTime();
399 } catch (Exception e) {
400 log.error("{} An exception occurred while preCommitting transaction {}",
401 name, cohortEntry.getTransactionID(), e);
402 cohortEntry.getReplySender().tell(new Failure(e), cohortEntry.getShard().self());
404 currentTransactionComplete(cohortEntry.getTransactionID(), true);
411 * This method handles the preCommit and commit phases for a transaction.
413 * @param transactionID the ID of the transaction to commit
414 * @param sender the actor to which to send the response
415 * @param shard the transaction's shard actor
416 * @return true if the transaction was successfully prepared, false otherwise.
418 boolean handleCommit(final Identifier transactionID, final ActorRef sender, final Shard shard) {
419 // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to
421 final CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID);
422 if(cohortEntry == null) {
423 // We're not the current Tx - the Tx was likely expired b/c it took too long in
424 // between the canCommit and commit messages.
425 IllegalStateException ex = new IllegalStateException(
426 String.format("%s: Cannot commit transaction %s - it is not the current transaction",
427 name, transactionID));
428 log.error(ex.getMessage());
429 sender.tell(new Failure(ex), shard.self());
433 cohortEntry.setReplySender(sender);
434 return doCommit(cohortEntry);
437 void handleAbort(final Identifier transactionID, final ActorRef sender, final Shard shard) {
438 CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID);
439 if(cohortEntry != null) {
440 // We don't remove the cached cohort entry here (ie pass false) in case the Tx was
441 // aborted during replication in which case we may still commit locally if replication
443 currentTransactionComplete(transactionID, false);
445 cohortEntry = getAndRemoveCohortEntry(transactionID);
448 if(cohortEntry == null) {
452 log.debug("{}: Aborting transaction {}", name, transactionID);
454 final ActorRef self = shard.getSelf();
458 shard.getShardMBean().incrementAbortTransactionsCount();
461 sender.tell(AbortTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(), self);
463 } catch (Exception e) {
464 log.error("{}: An exception happened during abort", name, e);
467 sender.tell(new Failure(e), self);
472 void checkForExpiredTransactions(final long timeout, final Shard shard) {
473 CohortEntry cohortEntry = getCurrentCohortEntry();
474 if(cohortEntry != null) {
475 if(cohortEntry.isExpired(timeout)) {
476 log.warn("{}: Current transaction {} has timed out after {} ms - aborting",
477 name, cohortEntry.getTransactionID(), timeout);
479 handleAbort(cohortEntry.getTransactionID(), null, shard);
483 cleanupExpiredCohortEntries();
486 void abortPendingTransactions(final String reason, final Shard shard) {
487 if(currentCohortEntry == null && queuedCohortEntries.isEmpty()) {
491 List<CohortEntry> cohortEntries = getAndClearPendingCohortEntries();
493 log.debug("{}: Aborting {} pending queued transactions", name, cohortEntries.size());
495 for(CohortEntry cohortEntry: cohortEntries) {
496 if(cohortEntry.getReplySender() != null) {
497 cohortEntry.getReplySender().tell(new Failure(new RuntimeException(reason)), shard.self());
502 private List<CohortEntry> getAndClearPendingCohortEntries() {
503 List<CohortEntry> cohortEntries = new ArrayList<>();
505 if(currentCohortEntry != null) {
506 cohortEntries.add(currentCohortEntry);
507 cohortCache.remove(currentCohortEntry.getTransactionID());
508 currentCohortEntry = null;
511 for(CohortEntry cohortEntry: queuedCohortEntries) {
512 cohortEntries.add(cohortEntry);
513 cohortCache.remove(cohortEntry.getTransactionID());
516 queuedCohortEntries.clear();
517 return cohortEntries;
520 Collection<Object> convertPendingTransactionsToMessages(final int maxModificationsPerBatch) {
521 if(currentCohortEntry == null && queuedCohortEntries.isEmpty()) {
522 return Collections.emptyList();
525 Collection<Object> messages = new ArrayList<>();
526 List<CohortEntry> cohortEntries = getAndClearPendingCohortEntries();
527 for(CohortEntry cohortEntry: cohortEntries) {
528 if(cohortEntry.isExpired(cacheExpiryTimeoutInMillis) || cohortEntry.isAborted()) {
532 final LinkedList<BatchedModifications> newModifications = new LinkedList<>();
533 cohortEntry.getDataTreeModification().applyToCursor(new AbstractBatchedModificationsCursor() {
535 protected BatchedModifications getModifications() {
536 if(newModifications.isEmpty() ||
537 newModifications.getLast().getModifications().size() >= maxModificationsPerBatch) {
538 newModifications.add(new BatchedModifications(cohortEntry.getTransactionID(),
539 cohortEntry.getClientVersion()));
542 return newModifications.getLast();
546 if(!newModifications.isEmpty()) {
547 BatchedModifications last = newModifications.getLast();
548 last.setDoCommitOnReady(cohortEntry.isDoImmediateCommit());
550 last.setTotalMessagesSent(newModifications.size());
551 messages.addAll(newModifications);
553 if(!cohortEntry.isDoImmediateCommit() && cohortEntry.getState() == CohortEntry.State.CAN_COMMITTED) {
554 messages.add(new CanCommitTransaction(cohortEntry.getTransactionID(),
555 cohortEntry.getClientVersion()));
558 if(!cohortEntry.isDoImmediateCommit() && cohortEntry.getState() == CohortEntry.State.PRE_COMMITTED) {
559 messages.add(new CommitTransaction(cohortEntry.getTransactionID(),
560 cohortEntry.getClientVersion()));
569 * Returns the cohort entry for the Tx commit currently in progress if the given transaction ID
570 * matches the current entry.
572 * @param transactionID the ID of the transaction
573 * @return the current CohortEntry or null if the given transaction ID does not match the
576 CohortEntry getCohortEntryIfCurrent(Identifier transactionID) {
577 if(isCurrentTransaction(transactionID)) {
578 return currentCohortEntry;
584 CohortEntry getCurrentCohortEntry() {
585 return currentCohortEntry;
588 CohortEntry getAndRemoveCohortEntry(Identifier transactionID) {
589 return cohortCache.remove(transactionID);
592 boolean isCurrentTransaction(Identifier transactionID) {
593 return currentCohortEntry != null &&
594 currentCohortEntry.getTransactionID().equals(transactionID);
598 * This method is called when a transaction is complete, successful or not. If the given
599 * given transaction ID matches the current in-progress transaction, the next cohort entry,
600 * if any, is dequeued and processed.
602 * @param transactionID the ID of the completed transaction
603 * @param removeCohortEntry if true the CohortEntry for the transaction is also removed from
606 void currentTransactionComplete(Identifier transactionID, boolean removeCohortEntry) {
607 if(removeCohortEntry) {
608 cohortCache.remove(transactionID);
611 if(isCurrentTransaction(transactionID)) {
612 currentCohortEntry = null;
614 log.debug("{}: currentTransactionComplete: {}", name, transactionID);
616 maybeProcessNextCohortEntry();
620 private void maybeProcessNextCohortEntry() {
621 // Check if there's a next cohort entry waiting in the queue and if it is ready to commit. Also
622 // clean out expired entries.
623 final Iterator<CohortEntry> iter = queuedCohortEntries.iterator();
624 while(iter.hasNext()) {
625 final CohortEntry next = iter.next();
626 if(next.isReadyToCommit()) {
627 if(currentCohortEntry == null) {
628 if(log.isDebugEnabled()) {
629 log.debug("{}: Next entry to canCommit {}", name, next);
633 currentCohortEntry = next;
634 currentCohortEntry.updateLastAccessTime();
635 doCanCommit(currentCohortEntry);
639 } else if(next.isExpired(cacheExpiryTimeoutInMillis)) {
640 log.warn("{}: canCommit for transaction {} was not received within {} ms - entry removed from cache",
641 name, next.getTransactionID(), cacheExpiryTimeoutInMillis);
642 } else if(!next.isAborted()) {
647 cohortCache.remove(next.getTransactionID());
650 maybeRunOperationOnPendingTransactionsComplete();
653 void cleanupExpiredCohortEntries() {
654 maybeProcessNextCohortEntry();
657 void setRunOnPendingTransactionsComplete(Runnable operation) {
658 runOnPendingTransactionsComplete = operation;
659 maybeRunOperationOnPendingTransactionsComplete();
662 private void maybeRunOperationOnPendingTransactionsComplete() {
663 if(runOnPendingTransactionsComplete != null && currentCohortEntry == null && queuedCohortEntries.isEmpty()) {
664 log.debug("{}: Pending transactions complete - running operation {}", name, runOnPendingTransactionsComplete);
666 runOnPendingTransactionsComplete.run();
667 runOnPendingTransactionsComplete = null;
672 void setCohortDecorator(CohortDecorator cohortDecorator) {
673 this.cohortDecorator = cohortDecorator;
676 void processCohortRegistryCommand(ActorRef sender, CohortRegistryCommand message) {
677 cohortRegistry.process(sender, message);