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.Collection;
18 import java.util.Collections;
19 import java.util.HashMap;
20 import java.util.Iterator;
21 import java.util.LinkedList;
22 import java.util.List;
24 import java.util.Queue;
25 import java.util.concurrent.ExecutionException;
26 import java.util.concurrent.TimeUnit;
27 import org.opendaylight.controller.cluster.datastore.ShardCommitCoordinator.CohortEntry.State;
28 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
29 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
30 import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
31 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
32 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
33 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
34 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
35 import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
36 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
37 import org.opendaylight.controller.cluster.datastore.modification.Modification;
38 import org.opendaylight.controller.cluster.datastore.utils.AbstractBatchedModificationsCursor;
39 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
40 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
41 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
42 import org.slf4j.Logger;
45 * Coordinates commits for a shard ensuring only one concurrent 3-phase commit.
47 * @author Thomas Pantelis
49 class ShardCommitCoordinator {
51 // Interface hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts.
52 public interface CohortDecorator {
53 ShardDataTreeCohort decorate(String transactionID, ShardDataTreeCohort actual);
56 private final Map<String, CohortEntry> cohortCache = new HashMap<>();
58 private CohortEntry currentCohortEntry;
60 private final ShardDataTree dataTree;
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,
82 long cacheExpiryTimeoutInMillis, int queueCapacity, Logger log, String name) {
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 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
140 void handleForwardedReadyTransaction(ForwardedReadyTransaction ready, ActorRef sender, Shard shard) {
141 log.debug("{}: Readying transaction {}, client version {}", name,
142 ready.getTransactionID(), ready.getTxnClientVersion());
144 ShardDataTreeCohort cohort = ready.getTransaction().ready();
145 CohortEntry cohortEntry = new CohortEntry(ready.getTransactionID(), cohort, ready.getTxnClientVersion());
146 cohortCache.put(ready.getTransactionID(), cohortEntry);
148 if(!queueCohortEntry(cohortEntry, sender, shard)) {
152 if(ready.isDoImmediateCommit()) {
153 cohortEntry.setDoImmediateCommit(true);
154 cohortEntry.setReplySender(sender);
155 cohortEntry.setShard(shard);
156 handleCanCommit(cohortEntry);
158 // The caller does not want immediate commit - the 3-phase commit will be coordinated by the
159 // front-end so send back a ReadyTransactionReply with our actor path.
160 sender.tell(readyTransactionReply(shard), shard.self());
165 * This method handles a BatchedModifications message for a transaction being prepared directly on the
166 * Shard actor instead of via a ShardTransaction actor. If there's no currently cached
167 * DOMStoreWriteTransaction, one is created. The batched modifications are applied to the write Tx. If
168 * the BatchedModifications is ready to commit then a DOMStoreThreePhaseCommitCohort is created.
170 * @param batched the BatchedModifications message to process
171 * @param sender the sender of the message
172 * @param shard the transaction's shard actor
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 batched.getTransactionChainID()), batched.getVersion());
180 cohortCache.put(batched.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 dataTree.
230 * @param message the ReadyLocalTransaction message to process
231 * @param sender the sender of the message
232 * @param shard the transaction's shard actor
234 void handleReadyLocalTransaction(ReadyLocalTransaction message, ActorRef sender, Shard shard) {
235 final ShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(dataTree, message.getModification(),
236 message.getTransactionID());
237 final CohortEntry cohortEntry = new CohortEntry(message.getTransactionID(), cohort,
238 DataStoreVersions.CURRENT_VERSION);
239 cohortCache.put(message.getTransactionID(), cohortEntry);
240 cohortEntry.setDoImmediateCommit(message.isDoCommitOnReady());
242 if(!queueCohortEntry(cohortEntry, sender, shard)) {
246 log.debug("{}: Applying local modifications for Tx {}", name, message.getTransactionID());
248 if (message.isDoCommitOnReady()) {
249 cohortEntry.setReplySender(sender);
250 cohortEntry.setShard(shard);
251 handleCanCommit(cohortEntry);
253 sender.tell(readyTransactionReply(shard), shard.self());
257 Collection<BatchedModifications> createForwardedBatchedModifications(final BatchedModifications from,
258 final int maxModificationsPerBatch) {
259 CohortEntry cohortEntry = getAndRemoveCohortEntry(from.getTransactionID());
260 if(cohortEntry == null || cohortEntry.getTransaction() == null) {
261 return Collections.singletonList(from);
264 cohortEntry.applyModifications(from.getModifications());
266 final LinkedList<BatchedModifications> newModifications = new LinkedList<>();
267 cohortEntry.getTransaction().getSnapshot().applyToCursor(new AbstractBatchedModificationsCursor() {
269 protected BatchedModifications getModifications() {
270 if(newModifications.isEmpty() ||
271 newModifications.getLast().getModifications().size() >= maxModificationsPerBatch) {
272 newModifications.add(new BatchedModifications(from.getTransactionID(),
273 from.getVersion(), from.getTransactionChainID()));
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 String transactionID = cohortEntry.getTransactionID();
290 cohortEntry.updateLastAccessTime();
292 if(currentCohortEntry != null) {
293 // There's already a Tx commit in progress so we can't process this entry yet - but it's in the
294 // queue and will get processed after all prior entries complete.
296 if(log.isDebugEnabled()) {
297 log.debug("{}: Commit for Tx {} already in progress - skipping canCommit for {} for now",
298 name, currentCohortEntry.getTransactionID(), transactionID);
304 // No Tx commit currently in progress - check if this entry is the next one in the queue, If so make
305 // it the current entry and proceed with canCommit.
306 // Purposely checking reference equality here.
307 if(queuedCohortEntries.peek() == cohortEntry) {
308 currentCohortEntry = queuedCohortEntries.poll();
309 doCanCommit(currentCohortEntry);
311 if(log.isDebugEnabled()) {
312 log.debug("{}: Tx {} is the next pending canCommit - skipping {} for now", name,
313 queuedCohortEntries.peek() != null ? queuedCohortEntries.peek().getTransactionID() : "???",
320 * This method handles the canCommit phase for a transaction.
322 * @param transactionID the ID of the transaction to canCommit
323 * @param sender the actor to which to send the response
324 * @param shard the transaction's shard actor
326 void handleCanCommit(String transactionID, final ActorRef sender, final Shard shard) {
327 // Lookup the cohort entry that was cached previously (or should have been) by
328 // transactionReady (via the ForwardedReadyTransaction message).
329 final CohortEntry cohortEntry = cohortCache.get(transactionID);
330 if(cohortEntry == null) {
331 // Either canCommit was invoked before ready(shouldn't happen) or a long time passed
332 // between canCommit and ready and the entry was expired from the cache.
333 IllegalStateException ex = new IllegalStateException(
334 String.format("%s: No cohort entry found for transaction %s", name, transactionID));
335 log.error(ex.getMessage());
336 sender.tell(new Failure(ex), shard.self());
340 cohortEntry.setReplySender(sender);
341 cohortEntry.setShard(shard);
343 handleCanCommit(cohortEntry);
346 private void doCanCommit(final CohortEntry cohortEntry) {
347 boolean canCommit = false;
349 canCommit = cohortEntry.canCommit();
351 log.debug("{}: canCommit for {}: {}", name, cohortEntry.getTransactionID(), canCommit);
353 if(cohortEntry.isDoImmediateCommit()) {
355 doCommit(cohortEntry);
357 cohortEntry.getReplySender().tell(new Failure(new TransactionCommitFailedException(
358 "Can Commit failed, no detailed cause available.")), cohortEntry.getShard().self());
361 cohortEntry.getReplySender().tell(
362 canCommit ? CanCommitTransactionReply.yes(cohortEntry.getClientVersion()).toSerializable() :
363 CanCommitTransactionReply.no(cohortEntry.getClientVersion()).toSerializable(),
364 cohortEntry.getShard().self());
366 } catch (Exception e) {
367 log.debug("{}: An exception occurred during canCommit", name, e);
369 Throwable failure = e;
370 if(e instanceof ExecutionException) {
371 failure = e.getCause();
374 cohortEntry.getReplySender().tell(new Failure(failure), cohortEntry.getShard().self());
377 // Remove the entry from the cache now.
378 currentTransactionComplete(cohortEntry.getTransactionID(), true);
383 private boolean doCommit(CohortEntry cohortEntry) {
384 log.debug("{}: Committing transaction {}", name, cohortEntry.getTransactionID());
386 boolean success = false;
388 // We perform the preCommit phase here atomically with the commit phase. This is an
389 // optimization to eliminate the overhead of an extra preCommit message. We lose front-end
390 // coordination of preCommit across shards in case of failure but preCommit should not
391 // normally fail since we ensure only one concurrent 3-phase commit.
394 cohortEntry.preCommit();
396 cohortEntry.getShard().continueCommit(cohortEntry);
398 cohortEntry.updateLastAccessTime();
401 } catch (Exception e) {
402 log.error("{} An exception occurred while preCommitting transaction {}",
403 name, cohortEntry.getTransactionID(), e);
404 cohortEntry.getReplySender().tell(new Failure(e), cohortEntry.getShard().self());
406 currentTransactionComplete(cohortEntry.getTransactionID(), true);
413 * This method handles the preCommit and commit phases for a transaction.
415 * @param transactionID the ID of the transaction to commit
416 * @param sender the actor to which to send the response
417 * @param shard the transaction's shard actor
418 * @return true if the transaction was successfully prepared, false otherwise.
420 boolean handleCommit(final String transactionID, final ActorRef sender, final Shard shard) {
421 // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to
423 final CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID);
424 if(cohortEntry == null) {
425 // We're not the current Tx - the Tx was likely expired b/c it took too long in
426 // between the canCommit and commit messages.
427 IllegalStateException ex = new IllegalStateException(
428 String.format("%s: Cannot commit transaction %s - it is not the current transaction",
429 name, transactionID));
430 log.error(ex.getMessage());
431 sender.tell(new Failure(ex), shard.self());
435 cohortEntry.setReplySender(sender);
436 return doCommit(cohortEntry);
439 void handleAbort(final String transactionID, final ActorRef sender, final Shard shard) {
440 CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID);
441 if(cohortEntry != null) {
442 // We don't remove the cached cohort entry here (ie pass false) in case the Tx was
443 // aborted during replication in which case we may still commit locally if replication
445 currentTransactionComplete(transactionID, false);
447 cohortEntry = getAndRemoveCohortEntry(transactionID);
450 if(cohortEntry == null) {
454 log.debug("{}: Aborting transaction {}", name, transactionID);
456 final ActorRef self = shard.getSelf();
460 shard.getShardMBean().incrementAbortTransactionsCount();
463 sender.tell(AbortTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(), self);
465 } catch (Exception e) {
466 log.error("{}: An exception happened during abort", name, e);
469 sender.tell(new Failure(e), self);
474 void checkForExpiredTransactions(final long timeout, final Shard shard) {
475 CohortEntry cohortEntry = getCurrentCohortEntry();
476 if(cohortEntry != null) {
477 if(cohortEntry.isExpired(timeout)) {
478 log.warn("{}: Current transaction {} has timed out after {} ms - aborting",
479 name, cohortEntry.getTransactionID(), timeout);
481 handleAbort(cohortEntry.getTransactionID(), null, shard);
485 cleanupExpiredCohortEntries();
488 void abortPendingTransactions(final String reason, final Shard shard) {
489 if(currentCohortEntry == null && queuedCohortEntries.isEmpty()) {
493 List<CohortEntry> cohortEntries = getAndClearPendingCohortEntries();
495 log.debug("{}: Aborting {} pending queued transactions", name, cohortEntries.size());
497 for(CohortEntry cohortEntry: cohortEntries) {
498 if(cohortEntry.getReplySender() != null) {
499 cohortEntry.getReplySender().tell(new Failure(new RuntimeException(reason)), shard.self());
504 private List<CohortEntry> getAndClearPendingCohortEntries() {
505 List<CohortEntry> cohortEntries = new ArrayList<>();
506 if(currentCohortEntry != null) {
507 cohortEntries.add(currentCohortEntry);
508 cohortCache.remove(currentCohortEntry.getTransactionID());
509 currentCohortEntry = null;
512 for(CohortEntry cohortEntry: queuedCohortEntries) {
513 cohortEntries.add(cohortEntry);
514 cohortCache.remove(cohortEntry.getTransactionID());
517 queuedCohortEntries.clear();
518 return cohortEntries;
521 Collection<Object> convertPendingTransactionsToMessages(final int maxModificationsPerBatch) {
522 if(currentCohortEntry == null && queuedCohortEntries.isEmpty()) {
523 return Collections.emptyList();
526 Collection<Object> messages = new ArrayList<>();
527 List<CohortEntry> cohortEntries = getAndClearPendingCohortEntries();
528 for(CohortEntry cohortEntry: cohortEntries) {
529 if(cohortEntry.isExpired(cacheExpiryTimeoutInMillis) || cohortEntry.isAborted()) {
533 final LinkedList<BatchedModifications> newModifications = new LinkedList<>();
534 cohortEntry.getDataTreeModification().applyToCursor(new AbstractBatchedModificationsCursor() {
536 protected BatchedModifications getModifications() {
537 if(newModifications.isEmpty() ||
538 newModifications.getLast().getModifications().size() >= maxModificationsPerBatch) {
539 newModifications.add(new BatchedModifications(cohortEntry.getTransactionID(),
540 cohortEntry.getClientVersion(), ""));
543 return newModifications.getLast();
547 if(!newModifications.isEmpty()) {
548 BatchedModifications last = newModifications.getLast();
549 last.setDoCommitOnReady(cohortEntry.isDoImmediateCommit());
551 last.setTotalMessagesSent(newModifications.size());
552 messages.addAll(newModifications);
554 if(!cohortEntry.isDoImmediateCommit() && cohortEntry.getState() == State.CAN_COMMITTED) {
555 messages.add(new CanCommitTransaction(cohortEntry.getTransactionID(),
556 cohortEntry.getClientVersion()));
559 if(!cohortEntry.isDoImmediateCommit() && cohortEntry.getState() == State.PRE_COMMITTED) {
560 messages.add(new CommitTransaction(cohortEntry.getTransactionID(),
561 cohortEntry.getClientVersion()));
570 * Returns the cohort entry for the Tx commit currently in progress if the given transaction ID
571 * matches the current entry.
573 * @param transactionID the ID of the transaction
574 * @return the current CohortEntry or null if the given transaction ID does not match the
577 CohortEntry getCohortEntryIfCurrent(String transactionID) {
578 if(isCurrentTransaction(transactionID)) {
579 return currentCohortEntry;
585 CohortEntry getCurrentCohortEntry() {
586 return currentCohortEntry;
589 CohortEntry getAndRemoveCohortEntry(String transactionID) {
590 return cohortCache.remove(transactionID);
593 boolean isCurrentTransaction(String transactionID) {
594 return currentCohortEntry != null &&
595 currentCohortEntry.getTransactionID().equals(transactionID);
599 * This method is called when a transaction is complete, successful or not. If the given
600 * given transaction ID matches the current in-progress transaction, the next cohort entry,
601 * if any, is dequeued and processed.
603 * @param transactionID the ID of the completed transaction
604 * @param removeCohortEntry if true the CohortEntry for the transaction is also removed from
607 void currentTransactionComplete(String transactionID, boolean removeCohortEntry) {
608 if(removeCohortEntry) {
609 cohortCache.remove(transactionID);
612 if(isCurrentTransaction(transactionID)) {
613 currentCohortEntry = null;
615 log.debug("{}: currentTransactionComplete: {}", name, transactionID);
617 maybeProcessNextCohortEntry();
621 private void maybeProcessNextCohortEntry() {
622 // Check if there's a next cohort entry waiting in the queue and if it is ready to commit. Also
623 // clean out expired entries.
624 Iterator<CohortEntry> iter = queuedCohortEntries.iterator();
625 while(iter.hasNext()) {
626 CohortEntry next = iter.next();
627 if(next.isReadyToCommit()) {
628 if(currentCohortEntry == null) {
629 if(log.isDebugEnabled()) {
630 log.debug("{}: Next entry to canCommit {}", name, next);
634 currentCohortEntry = next;
635 currentCohortEntry.updateLastAccessTime();
636 doCanCommit(currentCohortEntry);
640 } else if(next.isExpired(cacheExpiryTimeoutInMillis)) {
641 log.warn("{}: canCommit for transaction {} was not received within {} ms - entry removed from cache",
642 name, next.getTransactionID(), cacheExpiryTimeoutInMillis);
643 } else if(!next.isAborted()) {
648 cohortCache.remove(next.getTransactionID());
651 maybeRunOperationOnPendingTransactionsComplete();
654 void cleanupExpiredCohortEntries() {
655 maybeProcessNextCohortEntry();
658 void setRunOnPendingTransactionsComplete(Runnable operation) {
659 runOnPendingTransactionsComplete = operation;
660 maybeRunOperationOnPendingTransactionsComplete();
663 private void maybeRunOperationOnPendingTransactionsComplete() {
664 if(runOnPendingTransactionsComplete != null && currentCohortEntry == null && queuedCohortEntries.isEmpty()) {
665 log.debug("{}: Pending transactions complete - running operation {}", name, runOnPendingTransactionsComplete);
667 runOnPendingTransactionsComplete.run();
668 runOnPendingTransactionsComplete = null;
673 void setCohortDecorator(CohortDecorator cohortDecorator) {
674 this.cohortDecorator = cohortDecorator;
677 static class CohortEntry {
686 private final String transactionID;
687 private ShardDataTreeCohort cohort;
688 private final ReadWriteShardDataTreeTransaction transaction;
689 private RuntimeException lastBatchedModificationsException;
690 private ActorRef replySender;
692 private boolean doImmediateCommit;
693 private final Stopwatch lastAccessTimer = Stopwatch.createStarted();
694 private int totalBatchedModificationsReceived;
695 private State state = State.PENDING;
696 private final short clientVersion;
698 CohortEntry(String transactionID, ReadWriteShardDataTreeTransaction transaction, short clientVersion) {
699 this.transaction = Preconditions.checkNotNull(transaction);
700 this.transactionID = transactionID;
701 this.clientVersion = clientVersion;
704 CohortEntry(String transactionID, ShardDataTreeCohort cohort, short clientVersion) {
705 this.transactionID = transactionID;
706 this.cohort = cohort;
707 this.transaction = null;
708 this.clientVersion = clientVersion;
711 void updateLastAccessTime() {
712 lastAccessTimer.reset();
713 lastAccessTimer.start();
716 String getTransactionID() {
717 return transactionID;
720 short getClientVersion() {
721 return clientVersion;
728 DataTreeCandidate getCandidate() {
729 return cohort.getCandidate();
732 DataTreeModification getDataTreeModification() {
733 return cohort.getDataTreeModification();
736 ReadWriteShardDataTreeTransaction getTransaction() {
740 int getTotalBatchedModificationsReceived() {
741 return totalBatchedModificationsReceived;
744 RuntimeException getLastBatchedModificationsException() {
745 return lastBatchedModificationsException;
748 void applyModifications(Iterable<Modification> modifications) {
749 totalBatchedModificationsReceived++;
750 if(lastBatchedModificationsException == null) {
751 for (Modification modification : modifications) {
753 modification.apply(transaction.getSnapshot());
754 } catch (RuntimeException e) {
755 lastBatchedModificationsException = e;
762 boolean canCommit() throws InterruptedException, ExecutionException {
763 state = State.CAN_COMMITTED;
765 // We block on the future here (and also preCommit(), commit(), abort()) so we don't have to worry
766 // about possibly accessing our state on a different thread outside of our dispatcher.
767 // TODO: the ShardDataTreeCohort returns immediate Futures anyway which begs the question - why
768 // bother even returning Futures from ShardDataTreeCohort if we have to treat them synchronously
769 // anyway?. The Futures are really a remnant from when we were using the InMemoryDataBroker.
770 return cohort.canCommit().get();
773 void preCommit() throws InterruptedException, ExecutionException {
774 state = State.PRE_COMMITTED;
775 cohort.preCommit().get();
778 void commit() throws InterruptedException, ExecutionException {
779 state = State.COMMITTED;
780 cohort.commit().get();
783 void abort() throws InterruptedException, ExecutionException {
784 state = State.ABORTED;
785 cohort.abort().get();
788 void ready(CohortDecorator cohortDecorator, boolean doImmediateCommit) {
789 Preconditions.checkState(cohort == null, "cohort was already set");
791 setDoImmediateCommit(doImmediateCommit);
793 cohort = transaction.ready();
795 if(cohortDecorator != null) {
796 // Call the hook for unit tests.
797 cohort = cohortDecorator.decorate(transactionID, cohort);
801 boolean isReadyToCommit() {
802 return replySender != null;
805 boolean isExpired(long expireTimeInMillis) {
806 return lastAccessTimer.elapsed(TimeUnit.MILLISECONDS) >= expireTimeInMillis;
809 boolean isDoImmediateCommit() {
810 return doImmediateCommit;
813 void setDoImmediateCommit(boolean doImmediateCommit) {
814 this.doImmediateCommit = doImmediateCommit;
817 ActorRef getReplySender() {
821 void setReplySender(ActorRef replySender) {
822 this.replySender = replySender;
829 void setShard(Shard shard) {
834 boolean isAborted() {
835 return state == State.ABORTED;
839 public String toString() {
840 StringBuilder builder = new StringBuilder();
841 builder.append("CohortEntry [transactionID=").append(transactionID).append(", doImmediateCommit=")
842 .append(doImmediateCommit).append("]");
843 return builder.toString();