*/
package org.opendaylight.controller.cluster.datastore;
+import akka.actor.ActorRef;
+import akka.actor.Status;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.modification.CompositeModification;
import org.opendaylight.controller.cluster.datastore.modification.Modification;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import akka.actor.ActorRef;
-import akka.actor.Status;
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
/**
* Coordinates commits for a shard ensuring only one concurrent 3-phase commit.
*/
public class ShardCommitCoordinator {
- private static final Logger LOG = LoggerFactory.getLogger(ShardCommitCoordinator.class);
-
- private static final Object CAN_COMMIT_REPLY_TRUE =
- new CanCommitTransactionReply(Boolean.TRUE).toSerializable();
-
- private static final Object CAN_COMMIT_REPLY_FALSE =
- new CanCommitTransactionReply(Boolean.FALSE).toSerializable();
-
private final Cache<String, CohortEntry> cohortCache;
private CohortEntry currentCohortEntry;
private final Queue<CohortEntry> queuedCohortEntries;
- private final int queueCapacity;
+ private int queueCapacity;
+
+ private final Logger log;
+
+ private final String name;
- public ShardCommitCoordinator(long cacheExpiryTimeoutInSec, int queueCapacity) {
+ public ShardCommitCoordinator(long cacheExpiryTimeoutInSec, int queueCapacity, Logger log,
+ String name) {
cohortCache = CacheBuilder.newBuilder().expireAfterAccess(
cacheExpiryTimeoutInSec, TimeUnit.SECONDS).build();
this.queueCapacity = queueCapacity;
+ this.log = log;
+ this.name = name;
// We use a LinkedList here to avoid synchronization overhead with concurrent queue impls
// since this should only be accessed on the shard's dispatcher.
queuedCohortEntries = new LinkedList<>();
}
+ public void setQueueCapacity(int queueCapacity) {
+ this.queueCapacity = queueCapacity;
+ }
+
/**
* This method caches a cohort entry for the given transactions ID in preparation for the
* subsequent 3-phase commit.
public void handleCanCommit(CanCommitTransaction canCommit, final ActorRef sender,
final ActorRef shard) {
String transactionID = canCommit.getTransactionID();
- if(LOG.isDebugEnabled()) {
- LOG.debug("Processing canCommit for transaction {} for shard {}",
- transactionID, shard.path());
+ if(log.isDebugEnabled()) {
+ log.debug("{}: Processing canCommit for transaction {} for shard {}",
+ name, transactionID, shard.path());
}
// Lookup the cohort entry that was cached previously (or should have been) by
// Either canCommit was invoked before ready(shouldn't happen) or a long time passed
// between canCommit and ready and the entry was expired from the cache.
IllegalStateException ex = new IllegalStateException(
- String.format("No cohort entry found for transaction %s", transactionID));
- LOG.error(ex.getMessage());
+ String.format("%s: No cohort entry found for transaction %s", name, transactionID));
+ log.error(ex.getMessage());
sender.tell(new Status.Failure(ex), shard);
return;
}
if(currentCohortEntry != null) {
// There's already a Tx commit in progress - attempt to queue this entry to be
// committed after the current Tx completes.
- LOG.debug("Transaction {} is already in progress - queueing transaction {}",
- currentCohortEntry.getTransactionID(), transactionID);
+ log.debug("{}: Transaction {} is already in progress - queueing transaction {}",
+ name, currentCohortEntry.getTransactionID(), transactionID);
if(queuedCohortEntries.size() < queueCapacity) {
queuedCohortEntries.offer(cohortEntry);
removeCohortEntry(transactionID);
RuntimeException ex = new RuntimeException(
- String.format("Could not enqueue transaction %s - the maximum commit queue"+
+ String.format("%s: Could not enqueue transaction %s - the maximum commit queue"+
" capacity %d has been reached.",
- transactionID, queueCapacity));
- LOG.error(ex.getMessage());
+ name, transactionID, queueCapacity));
+ log.error(ex.getMessage());
sender.tell(new Status.Failure(ex), shard);
}
} else {
Boolean canCommit = cohortEntry.getCohort().canCommit().get();
cohortEntry.getCanCommitSender().tell(
- canCommit ? CAN_COMMIT_REPLY_TRUE : CAN_COMMIT_REPLY_FALSE, cohortEntry.getShard());
+ canCommit ? CanCommitTransactionReply.YES.toSerializable() :
+ CanCommitTransactionReply.NO.toSerializable(), cohortEntry.getShard());
if(!canCommit) {
// Remove the entry from the cache now since the Tx will be aborted.
removeCohortEntry(cohortEntry.getTransactionID());
}
} catch (InterruptedException | ExecutionException e) {
- LOG.debug("An exception occurred during canCommit", e);
+ log.debug("{}: An exception occurred during canCommit: {}", name, e);
// Remove the entry from the cache now since the Tx will be aborted.
removeCohortEntry(cohortEntry.getTransactionID());
// Dequeue the next cohort entry waiting in the queue.
currentCohortEntry = queuedCohortEntries.poll();
if(currentCohortEntry != null) {
+ currentCohortEntry.updateLastAccessTime();
doCanCommit(currentCohortEntry);
}
}
void setShard(ActorRef shard) {
this.shard = shard;
}
+
+ boolean hasModifications(){
+ if(modification instanceof CompositeModification){
+ return ((CompositeModification) modification).getModifications().size() > 0;
+ }
+ return true;
+ }
}
}
Code Review / controller.git / blobdiff