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.util.StringIdentifier;
39 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
40 import org.slf4j.Logger;
43 * Coordinates commits for a shard ensuring only one concurrent 3-phase commit.
45 * @author Thomas Pantelis
47 final class ShardCommitCoordinator {
49 // Interface hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts.
50 public interface CohortDecorator {
51 ShardDataTreeCohort decorate(Identifier transactionID, ShardDataTreeCohort actual);
54 private final Map<Identifier, CohortEntry> cohortCache = new HashMap<>();
56 private CohortEntry currentCohortEntry;
58 private final ShardDataTree dataTree;
60 private final DataTreeCohortActorRegistry cohortRegistry = new DataTreeCohortActorRegistry();
62 // We use a LinkedList here to avoid synchronization overhead with concurrent queue impls
63 // since this should only be accessed on the shard's dispatcher.
64 private final Queue<CohortEntry> queuedCohortEntries = new LinkedList<>();
66 private int queueCapacity;
68 private final Logger log;
70 private final String name;
72 private final long cacheExpiryTimeoutInMillis;
74 // This is a hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts.
75 private CohortDecorator cohortDecorator;
77 private ReadyTransactionReply readyTransactionReply;
79 private Runnable runOnPendingTransactionsComplete;
81 ShardCommitCoordinator(ShardDataTree dataTree, long cacheExpiryTimeoutInMillis, int queueCapacity, Logger log,
84 this.queueCapacity = queueCapacity;
87 this.dataTree = Preconditions.checkNotNull(dataTree);
88 this.cacheExpiryTimeoutInMillis = cacheExpiryTimeoutInMillis;
92 return queuedCohortEntries.size();
95 int getCohortCacheSize() {
96 return cohortCache.size();
99 void setQueueCapacity(int queueCapacity) {
100 this.queueCapacity = queueCapacity;
103 private ReadyTransactionReply readyTransactionReply(Shard shard) {
104 if(readyTransactionReply == null) {
105 readyTransactionReply = new ReadyTransactionReply(Serialization.serializedActorPath(shard.self()));
108 return readyTransactionReply;
111 private boolean queueCohortEntry(CohortEntry cohortEntry, ActorRef sender, Shard shard) {
112 if(queuedCohortEntries.size() < queueCapacity) {
113 queuedCohortEntries.offer(cohortEntry);
115 log.debug("{}: Enqueued transaction {}, queue size {}", name, cohortEntry.getTransactionID(),
116 queuedCohortEntries.size());
120 cohortCache.remove(cohortEntry.getTransactionID());
122 final RuntimeException ex = new RuntimeException(
123 String.format("%s: Could not enqueue transaction %s - the maximum commit queue"+
124 " capacity %d has been reached.",
125 name, cohortEntry.getTransactionID(), queueCapacity));
126 log.error(ex.getMessage());
127 sender.tell(new Failure(ex), shard.self());
133 * This method is called to ready a transaction that was prepared by ShardTransaction actor. It caches
134 * the prepared cohort entry for the given transactions ID in preparation for the subsequent 3-phase commit.
136 * @param ready the ForwardedReadyTransaction message to process
137 * @param sender the sender of the message
138 * @param shard the transaction's shard actor
141 void handleForwardedReadyTransaction(ForwardedReadyTransaction ready, ActorRef sender, Shard shard,
142 SchemaContext schema) {
143 log.debug("{}: Readying transaction {}, client version {}", name,
144 ready.getTransactionID(), ready.getTxnClientVersion());
146 final ShardDataTreeCohort cohort = ready.getTransaction().ready();
147 final CohortEntry cohortEntry = new CohortEntry(ready.getTransactionID(), cohort, cohortRegistry, schema, ready.getTxnClientVersion());
148 cohortCache.put(cohortEntry.getTransactionID(), cohortEntry);
150 if(!queueCohortEntry(cohortEntry, sender, shard)) {
154 if(ready.isDoImmediateCommit()) {
155 cohortEntry.setDoImmediateCommit(true);
156 cohortEntry.setReplySender(sender);
157 cohortEntry.setShard(shard);
158 handleCanCommit(cohortEntry);
160 // The caller does not want immediate commit - the 3-phase commit will be coordinated by the
161 // front-end so send back a ReadyTransactionReply with our actor path.
162 sender.tell(readyTransactionReply(shard), shard.self());
167 * This method handles a BatchedModifications message for a transaction being prepared directly on the
168 * Shard actor instead of via a ShardTransaction actor. If there's no currently cached
169 * DOMStoreWriteTransaction, one is created. The batched modifications are applied to the write Tx. If
170 * the BatchedModifications is ready to commit then a DOMStoreThreePhaseCommitCohort is created.
172 * @param batched the BatchedModifications message to process
173 * @param sender the sender of the message
174 * @param shard the transaction's shard actor
176 void handleBatchedModifications(BatchedModifications batched, ActorRef sender, Shard shard, SchemaContext schema) {
177 CohortEntry cohortEntry = cohortCache.get(new StringIdentifier(batched.getTransactionID()));
178 if(cohortEntry == null) {
179 cohortEntry = new CohortEntry(batched.getTransactionID(),
180 dataTree.newReadWriteTransaction(batched.getTransactionID(), batched.getTransactionChainID()),
181 cohortRegistry, schema, batched.getVersion());
182 cohortCache.put(cohortEntry.getTransactionID(), cohortEntry);
185 if(log.isDebugEnabled()) {
186 log.debug("{}: Applying {} batched modifications for Tx {}", name,
187 batched.getModifications().size(), batched.getTransactionID());
190 cohortEntry.applyModifications(batched.getModifications());
192 if(batched.isReady()) {
193 if(cohortEntry.getLastBatchedModificationsException() != null) {
194 cohortCache.remove(cohortEntry.getTransactionID());
195 throw cohortEntry.getLastBatchedModificationsException();
198 if(cohortEntry.getTotalBatchedModificationsReceived() != batched.getTotalMessagesSent()) {
199 cohortCache.remove(cohortEntry.getTransactionID());
200 throw new IllegalStateException(String.format(
201 "The total number of batched messages received %d does not match the number sent %d",
202 cohortEntry.getTotalBatchedModificationsReceived(), batched.getTotalMessagesSent()));
205 if(!queueCohortEntry(cohortEntry, sender, shard)) {
209 if(log.isDebugEnabled()) {
210 log.debug("{}: Readying Tx {}, client version {}", name,
211 batched.getTransactionID(), batched.getVersion());
214 cohortEntry.ready(cohortDecorator, batched.isDoCommitOnReady());
216 if(batched.isDoCommitOnReady()) {
217 cohortEntry.setReplySender(sender);
218 cohortEntry.setShard(shard);
219 handleCanCommit(cohortEntry);
221 sender.tell(readyTransactionReply(shard), shard.self());
224 sender.tell(new BatchedModificationsReply(batched.getModifications().size()), shard.self());
229 * This method handles {@link ReadyLocalTransaction} message. All transaction modifications have
230 * been prepared beforehand by the sender and we just need to drive them through into the
233 * @param message the ReadyLocalTransaction message to process
234 * @param sender the sender of the message
235 * @param shard the transaction's shard actor
237 void handleReadyLocalTransaction(ReadyLocalTransaction message, ActorRef sender, Shard shard,
238 SchemaContext schema) {
239 final ShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(dataTree, message.getModification(),
240 message.getTransactionID());
241 final CohortEntry cohortEntry = new CohortEntry(message.getTransactionID(), cohort, cohortRegistry, schema,
242 DataStoreVersions.CURRENT_VERSION);
243 cohortCache.put(cohortEntry.getTransactionID(), cohortEntry);
244 cohortEntry.setDoImmediateCommit(message.isDoCommitOnReady());
246 if(!queueCohortEntry(cohortEntry, sender, shard)) {
250 log.debug("{}: Applying local modifications for Tx {}", name, message.getTransactionID());
252 if (message.isDoCommitOnReady()) {
253 cohortEntry.setReplySender(sender);
254 cohortEntry.setShard(shard);
255 handleCanCommit(cohortEntry);
257 sender.tell(readyTransactionReply(shard), shard.self());
261 Collection<BatchedModifications> createForwardedBatchedModifications(final BatchedModifications from,
262 final int maxModificationsPerBatch) {
263 CohortEntry cohortEntry = getAndRemoveCohortEntry(new StringIdentifier(from.getTransactionID()));
264 if(cohortEntry == null || cohortEntry.getTransaction() == null) {
265 return Collections.singletonList(from);
268 cohortEntry.applyModifications(from.getModifications());
270 final LinkedList<BatchedModifications> newModifications = new LinkedList<>();
271 cohortEntry.getTransaction().getSnapshot().applyToCursor(new AbstractBatchedModificationsCursor() {
273 protected BatchedModifications getModifications() {
274 if(newModifications.isEmpty() ||
275 newModifications.getLast().getModifications().size() >= maxModificationsPerBatch) {
276 newModifications.add(new BatchedModifications(from.getTransactionID(),
277 from.getVersion(), from.getTransactionChainID()));
280 return newModifications.getLast();
284 BatchedModifications last = newModifications.getLast();
285 last.setDoCommitOnReady(from.isDoCommitOnReady());
286 last.setReady(from.isReady());
287 last.setTotalMessagesSent(newModifications.size());
288 return newModifications;
291 private void handleCanCommit(CohortEntry cohortEntry) {
292 cohortEntry.updateLastAccessTime();
294 if(currentCohortEntry != null) {
295 // There's already a Tx commit in progress so we can't process this entry yet - but it's in the
296 // queue and will get processed after all prior entries complete.
298 if(log.isDebugEnabled()) {
299 log.debug("{}: Commit for Tx {} already in progress - skipping canCommit for {} for now",
300 name, currentCohortEntry.getTransactionID(), cohortEntry.getTransactionID());
306 // No Tx commit currently in progress - check if this entry is the next one in the queue, If so make
307 // it the current entry and proceed with canCommit.
308 // Purposely checking reference equality here.
309 if(queuedCohortEntries.peek() == cohortEntry) {
310 currentCohortEntry = queuedCohortEntries.poll();
311 doCanCommit(currentCohortEntry);
313 if(log.isDebugEnabled()) {
314 log.debug("{}: Tx {} is the next pending canCommit - skipping {} for now", name,
315 queuedCohortEntries.peek() != null ? queuedCohortEntries.peek().getTransactionID() : "???",
316 cohortEntry.getTransactionID());
322 * This method handles the canCommit phase for a transaction.
324 * @param transactionID the ID of the transaction to canCommit
325 * @param sender the actor to which to send the response
326 * @param shard the transaction's shard actor
328 void handleCanCommit(Identifier transactionID, final ActorRef sender, final Shard shard) {
329 // Lookup the cohort entry that was cached previously (or should have been) by
330 // transactionReady (via the ForwardedReadyTransaction message).
331 final CohortEntry cohortEntry = cohortCache.get(transactionID);
332 if(cohortEntry == null) {
333 // Either canCommit was invoked before ready(shouldn't happen) or a long time passed
334 // between canCommit and ready and the entry was expired from the cache.
335 IllegalStateException ex = new IllegalStateException(
336 String.format("%s: No cohort entry found for transaction %s", name, transactionID));
337 log.error(ex.getMessage());
338 sender.tell(new Failure(ex), shard.self());
342 cohortEntry.setReplySender(sender);
343 cohortEntry.setShard(shard);
345 handleCanCommit(cohortEntry);
348 private void doCanCommit(final CohortEntry cohortEntry) {
349 boolean canCommit = false;
351 canCommit = cohortEntry.canCommit();
353 log.debug("{}: canCommit for {}: {}", name, cohortEntry.getTransactionID(), canCommit);
355 if(cohortEntry.isDoImmediateCommit()) {
357 doCommit(cohortEntry);
359 cohortEntry.getReplySender().tell(new Failure(new TransactionCommitFailedException(
360 "Can Commit failed, no detailed cause available.")), cohortEntry.getShard().self());
363 cohortEntry.getReplySender().tell(
364 canCommit ? CanCommitTransactionReply.yes(cohortEntry.getClientVersion()).toSerializable() :
365 CanCommitTransactionReply.no(cohortEntry.getClientVersion()).toSerializable(),
366 cohortEntry.getShard().self());
368 } catch (Exception e) {
369 log.debug("{}: An exception occurred during canCommit", name, e);
371 Throwable failure = e;
372 if(e instanceof ExecutionException) {
373 failure = e.getCause();
376 cohortEntry.getReplySender().tell(new Failure(failure), cohortEntry.getShard().self());
379 // Remove the entry from the cache now.
380 currentTransactionComplete(cohortEntry.getTransactionID(), true);
385 private boolean doCommit(CohortEntry cohortEntry) {
386 log.debug("{}: Committing transaction {}", name, cohortEntry.getTransactionID());
388 boolean success = false;
390 // We perform the preCommit phase here atomically with the commit phase. This is an
391 // optimization to eliminate the overhead of an extra preCommit message. We lose front-end
392 // coordination of preCommit across shards in case of failure but preCommit should not
393 // normally fail since we ensure only one concurrent 3-phase commit.
396 cohortEntry.preCommit();
398 cohortEntry.getShard().continueCommit(cohortEntry);
400 cohortEntry.updateLastAccessTime();
403 } catch (Exception e) {
404 log.error("{} An exception occurred while preCommitting transaction {}",
405 name, cohortEntry.getTransactionID(), e);
406 cohortEntry.getReplySender().tell(new Failure(e), cohortEntry.getShard().self());
408 currentTransactionComplete(cohortEntry.getTransactionID(), true);
415 * This method handles the preCommit and commit phases for a transaction.
417 * @param transactionID the ID of the transaction to commit
418 * @param sender the actor to which to send the response
419 * @param shard the transaction's shard actor
420 * @return true if the transaction was successfully prepared, false otherwise.
422 boolean handleCommit(final Identifier transactionID, final ActorRef sender, final Shard shard) {
423 // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to
425 final CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID);
426 if(cohortEntry == null) {
427 // We're not the current Tx - the Tx was likely expired b/c it took too long in
428 // between the canCommit and commit messages.
429 IllegalStateException ex = new IllegalStateException(
430 String.format("%s: Cannot commit transaction %s - it is not the current transaction",
431 name, transactionID));
432 log.error(ex.getMessage());
433 sender.tell(new Failure(ex), shard.self());
437 cohortEntry.setReplySender(sender);
438 return doCommit(cohortEntry);
441 void handleAbort(final Identifier transactionID, final ActorRef sender, final Shard shard) {
442 CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID);
443 if(cohortEntry != null) {
444 // We don't remove the cached cohort entry here (ie pass false) in case the Tx was
445 // aborted during replication in which case we may still commit locally if replication
447 currentTransactionComplete(transactionID, false);
449 cohortEntry = getAndRemoveCohortEntry(transactionID);
452 if(cohortEntry == null) {
456 log.debug("{}: Aborting transaction {}", name, transactionID);
458 final ActorRef self = shard.getSelf();
462 shard.getShardMBean().incrementAbortTransactionsCount();
465 sender.tell(AbortTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(), self);
467 } catch (Exception e) {
468 log.error("{}: An exception happened during abort", name, e);
471 sender.tell(new Failure(e), self);
476 void checkForExpiredTransactions(final long timeout, final Shard shard) {
477 CohortEntry cohortEntry = getCurrentCohortEntry();
478 if(cohortEntry != null) {
479 if(cohortEntry.isExpired(timeout)) {
480 log.warn("{}: Current transaction {} has timed out after {} ms - aborting",
481 name, cohortEntry.getTransactionID(), timeout);
483 handleAbort(cohortEntry.getTransactionID(), null, shard);
487 cleanupExpiredCohortEntries();
490 void abortPendingTransactions(final String reason, final Shard shard) {
491 if(currentCohortEntry == null && queuedCohortEntries.isEmpty()) {
495 List<CohortEntry> cohortEntries = getAndClearPendingCohortEntries();
497 log.debug("{}: Aborting {} pending queued transactions", name, cohortEntries.size());
499 for(CohortEntry cohortEntry: cohortEntries) {
500 if(cohortEntry.getReplySender() != null) {
501 cohortEntry.getReplySender().tell(new Failure(new RuntimeException(reason)), shard.self());
506 private List<CohortEntry> getAndClearPendingCohortEntries() {
507 List<CohortEntry> cohortEntries = new ArrayList<>();
509 if(currentCohortEntry != null) {
510 cohortEntries.add(currentCohortEntry);
511 cohortCache.remove(currentCohortEntry.getTransactionID());
512 currentCohortEntry = null;
515 for(CohortEntry cohortEntry: queuedCohortEntries) {
516 cohortEntries.add(cohortEntry);
517 cohortCache.remove(cohortEntry.getTransactionID());
520 queuedCohortEntries.clear();
521 return cohortEntries;
524 Collection<Object> convertPendingTransactionsToMessages(final int maxModificationsPerBatch) {
525 if(currentCohortEntry == null && queuedCohortEntries.isEmpty()) {
526 return Collections.emptyList();
529 Collection<Object> messages = new ArrayList<>();
530 List<CohortEntry> cohortEntries = getAndClearPendingCohortEntries();
531 for(CohortEntry cohortEntry: cohortEntries) {
532 if(cohortEntry.isExpired(cacheExpiryTimeoutInMillis) || cohortEntry.isAborted()) {
536 final LinkedList<BatchedModifications> newModifications = new LinkedList<>();
537 cohortEntry.getDataTreeModification().applyToCursor(new AbstractBatchedModificationsCursor() {
539 protected BatchedModifications getModifications() {
540 if(newModifications.isEmpty() ||
541 newModifications.getLast().getModifications().size() >= maxModificationsPerBatch) {
542 newModifications.add(new BatchedModifications(cohortEntry.getTransactionID().getString(),
543 cohortEntry.getClientVersion(), ""));
546 return newModifications.getLast();
550 if(!newModifications.isEmpty()) {
551 BatchedModifications last = newModifications.getLast();
552 last.setDoCommitOnReady(cohortEntry.isDoImmediateCommit());
554 last.setTotalMessagesSent(newModifications.size());
555 messages.addAll(newModifications);
557 if(!cohortEntry.isDoImmediateCommit() && cohortEntry.getState() == CohortEntry.State.CAN_COMMITTED) {
558 messages.add(new CanCommitTransaction(cohortEntry.getTransactionID().getString(),
559 cohortEntry.getClientVersion()));
562 if(!cohortEntry.isDoImmediateCommit() && cohortEntry.getState() == CohortEntry.State.PRE_COMMITTED) {
563 messages.add(new CommitTransaction(cohortEntry.getTransactionID().getString(),
564 cohortEntry.getClientVersion()));
573 * Returns the cohort entry for the Tx commit currently in progress if the given transaction ID
574 * matches the current entry.
576 * @param transactionID the ID of the transaction
577 * @return the current CohortEntry or null if the given transaction ID does not match the
580 CohortEntry getCohortEntryIfCurrent(Identifier transactionID) {
581 if(isCurrentTransaction(transactionID)) {
582 return currentCohortEntry;
588 CohortEntry getCurrentCohortEntry() {
589 return currentCohortEntry;
592 CohortEntry getAndRemoveCohortEntry(Identifier transactionID) {
593 return cohortCache.remove(transactionID);
596 boolean isCurrentTransaction(Identifier transactionID) {
597 return currentCohortEntry != null &&
598 currentCohortEntry.getTransactionID().equals(transactionID);
602 * This method is called when a transaction is complete, successful or not. If the given
603 * given transaction ID matches the current in-progress transaction, the next cohort entry,
604 * if any, is dequeued and processed.
606 * @param transactionID the ID of the completed transaction
607 * @param removeCohortEntry if true the CohortEntry for the transaction is also removed from
610 void currentTransactionComplete(Identifier transactionID, boolean removeCohortEntry) {
611 if(removeCohortEntry) {
612 cohortCache.remove(transactionID);
615 if(isCurrentTransaction(transactionID)) {
616 currentCohortEntry = null;
618 log.debug("{}: currentTransactionComplete: {}", name, transactionID);
620 maybeProcessNextCohortEntry();
624 private void maybeProcessNextCohortEntry() {
625 // Check if there's a next cohort entry waiting in the queue and if it is ready to commit. Also
626 // clean out expired entries.
627 final Iterator<CohortEntry> iter = queuedCohortEntries.iterator();
628 while(iter.hasNext()) {
629 final CohortEntry next = iter.next();
630 if(next.isReadyToCommit()) {
631 if(currentCohortEntry == null) {
632 if(log.isDebugEnabled()) {
633 log.debug("{}: Next entry to canCommit {}", name, next);
637 currentCohortEntry = next;
638 currentCohortEntry.updateLastAccessTime();
639 doCanCommit(currentCohortEntry);
643 } else if(next.isExpired(cacheExpiryTimeoutInMillis)) {
644 log.warn("{}: canCommit for transaction {} was not received within {} ms - entry removed from cache",
645 name, next.getTransactionID(), cacheExpiryTimeoutInMillis);
646 } else if(!next.isAborted()) {
651 cohortCache.remove(next.getTransactionID());
654 maybeRunOperationOnPendingTransactionsComplete();
657 void cleanupExpiredCohortEntries() {
658 maybeProcessNextCohortEntry();
661 void setRunOnPendingTransactionsComplete(Runnable operation) {
662 runOnPendingTransactionsComplete = operation;
663 maybeRunOperationOnPendingTransactionsComplete();
666 private void maybeRunOperationOnPendingTransactionsComplete() {
667 if(runOnPendingTransactionsComplete != null && currentCohortEntry == null && queuedCohortEntries.isEmpty()) {
668 log.debug("{}: Pending transactions complete - running operation {}", name, runOnPendingTransactionsComplete);
670 runOnPendingTransactionsComplete.run();
671 runOnPendingTransactionsComplete = null;
676 void setCohortDecorator(CohortDecorator cohortDecorator) {
677 this.cohortDecorator = cohortDecorator;
680 void processCohortRegistryCommand(ActorRef sender, CohortRegistryCommand message) {
681 cohortRegistry.process(sender, message);