X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Futils%2FActorContext.java;h=42a58878a8918a66c650d1c636b67bd506dc5400;hb=5105751d47439e5d71d3a3b8035e4afd262c1890;hp=73f1a8f328d2a671c5699d770b782a860afcc7f7;hpb=2727bea09c83646b6cbd2ef9672d0b7f6cf3b22f;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java index 73f1a8f328..42a58878a8 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java @@ -15,24 +15,19 @@ import akka.actor.ActorSelection; import akka.actor.ActorSystem; import akka.actor.Address; import akka.actor.PoisonPill; -import akka.dispatch.Futures; import akka.dispatch.Mapper; import akka.dispatch.OnComplete; import akka.pattern.AskTimeoutException; import akka.util.Timeout; import com.codahale.metrics.MetricRegistry; import com.codahale.metrics.Timer; -import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Optional; import com.google.common.base.Preconditions; import com.google.common.base.Strings; -import com.google.common.cache.Cache; -import com.google.common.cache.CacheBuilder; -import com.google.common.util.concurrent.RateLimiter; import java.util.concurrent.TimeUnit; -import org.opendaylight.controller.cluster.common.actor.CommonConfig; import org.opendaylight.controller.cluster.datastore.ClusterWrapper; import org.opendaylight.controller.cluster.datastore.Configuration; +import org.opendaylight.controller.cluster.datastore.DataStoreVersions; import org.opendaylight.controller.cluster.datastore.DatastoreContext; import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException; import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; @@ -85,6 +80,7 @@ public class ActorContext { } }; public static final String MAILBOX = "bounded-mailbox"; + public static final String COMMIT = "commit"; private final ActorSystem actorSystem; private final ActorRef shardManager; @@ -94,32 +90,33 @@ public class ActorContext { private FiniteDuration operationDuration; private Timeout operationTimeout; private final String selfAddressHostPort; - private RateLimiter txRateLimiter; - private final int transactionOutstandingOperationLimit; + private TransactionRateLimiter txRateLimiter; private Timeout transactionCommitOperationTimeout; private Timeout shardInitializationTimeout; private final Dispatchers dispatchers; - private Cache> primaryShardInfoCache; private volatile SchemaContext schemaContext; private volatile boolean updated; private final MetricRegistry metricRegistry = MetricsReporter.getInstance(DatastoreContext.METRICS_DOMAIN).getMetricsRegistry(); + private final PrimaryShardInfoFutureCache primaryShardInfoCache; + public ActorContext(ActorSystem actorSystem, ActorRef shardManager, ClusterWrapper clusterWrapper, Configuration configuration) { this(actorSystem, shardManager, clusterWrapper, configuration, - DatastoreContext.newBuilder().build()); + DatastoreContext.newBuilder().build(), new PrimaryShardInfoFutureCache()); } public ActorContext(ActorSystem actorSystem, ActorRef shardManager, ClusterWrapper clusterWrapper, Configuration configuration, - DatastoreContext datastoreContext) { + DatastoreContext datastoreContext, PrimaryShardInfoFutureCache primaryShardInfoCache) { this.actorSystem = actorSystem; this.shardManager = shardManager; this.clusterWrapper = clusterWrapper; this.configuration = configuration; this.datastoreContext = datastoreContext; this.dispatchers = new Dispatchers(actorSystem.dispatchers()); + this.primaryShardInfoCache = primaryShardInfoCache; setCachedProperties(); @@ -130,23 +127,18 @@ public class ActorContext { selfAddressHostPort = null; } - transactionOutstandingOperationLimit = new CommonConfig(this.getActorSystem().settings().config()).getMailBoxCapacity(); } private void setCachedProperties() { - txRateLimiter = RateLimiter.create(datastoreContext.getTransactionCreationInitialRateLimit()); + txRateLimiter = new TransactionRateLimiter(this); - operationDuration = Duration.create(datastoreContext.getOperationTimeoutInSeconds(), TimeUnit.SECONDS); + operationDuration = Duration.create(datastoreContext.getOperationTimeoutInMillis(), TimeUnit.MILLISECONDS); operationTimeout = new Timeout(operationDuration); transactionCommitOperationTimeout = new Timeout(Duration.create( datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS)); shardInitializationTimeout = new Timeout(datastoreContext.getShardInitializationTimeout().duration().$times(2)); - - primaryShardInfoCache = CacheBuilder.newBuilder() - .expireAfterWrite(datastoreContext.getShardLeaderElectionTimeout().duration().toMillis(), TimeUnit.MILLISECONDS) - .build(); } public DatastoreContext getDatastoreContext() { @@ -212,11 +204,13 @@ public class ActorContext { public PrimaryShardInfo checkedApply(Object response) throws Exception { if(response instanceof RemotePrimaryShardFound) { LOG.debug("findPrimaryShardAsync received: {}", response); - return onPrimaryShardFound(shardName, ((RemotePrimaryShardFound)response).getPrimaryPath(), null); + RemotePrimaryShardFound found = (RemotePrimaryShardFound)response; + return onPrimaryShardFound(shardName, found.getPrimaryPath(), found.getPrimaryVersion(), null); } else if(response instanceof LocalPrimaryShardFound) { LOG.debug("findPrimaryShardAsync received: {}", response); LocalPrimaryShardFound found = (LocalPrimaryShardFound)response; - return onPrimaryShardFound(shardName, found.getPrimaryPath(), found.getLocalShardDataTree()); + return onPrimaryShardFound(shardName, found.getPrimaryPath(), DataStoreVersions.CURRENT_VERSION, + found.getLocalShardDataTree()); } else if(response instanceof NotInitializedException) { throw (NotInitializedException)response; } else if(response instanceof PrimaryNotFoundException) { @@ -232,10 +226,11 @@ public class ActorContext { } private PrimaryShardInfo onPrimaryShardFound(String shardName, String primaryActorPath, - DataTree localShardDataTree) { + short primaryVersion, DataTree localShardDataTree) { ActorSelection actorSelection = actorSystem.actorSelection(primaryActorPath); - PrimaryShardInfo info = new PrimaryShardInfo(actorSelection, Optional.fromNullable(localShardDataTree)); - primaryShardInfoCache.put(shardName, Futures.successful(info)); + PrimaryShardInfo info = new PrimaryShardInfo(actorSelection, primaryVersion, + Optional.fromNullable(localShardDataTree)); + primaryShardInfoCache.putSuccessful(shardName, info); return info; } @@ -378,8 +373,7 @@ public class ActorContext { } public void shutdown() { - shardManager.tell(PoisonPill.getInstance(), null); - actorSystem.shutdown(); + shardManager.tell(PoisonPill.getInstance(), ActorRef.noSender()); } public ClusterWrapper getClusterWrapper() { @@ -417,6 +411,10 @@ public class ActorContext { return operationDuration; } + public Timeout getOperationTimeout() { + return operationTimeout; + } + public boolean isPathLocal(String path) { if (Strings.isNullOrEmpty(path)) { return false; @@ -468,18 +466,6 @@ public class ActorContext { return builder.toString(); } - /** - * Get the maximum number of operations that are to be permitted within a transaction before the transaction - * should begin throttling the operations - * - * Parking reading this configuration here because we need to get to the actor system settings - * - * @return - */ - public int getTransactionOutstandingOperationLimit(){ - return transactionOutstandingOperationLimit; - } - /** * This is a utility method that lets us get a Timer object for any operation. This is a little open-ended to allow * us to create a timer for pretty much anything. @@ -506,21 +492,12 @@ public class ActorContext { return datastoreContext.getDataStoreType(); } - /** - * Set the number of transaction creation permits that are to be allowed - * - * @param permitsPerSecond - */ - public void setTxCreationLimit(double permitsPerSecond){ - txRateLimiter.setRate(permitsPerSecond); - } - /** * Get the current transaction creation rate limit * @return */ public double getTxCreationLimit(){ - return txRateLimiter.getRate(); + return txRateLimiter.getTxCreationLimit(); } /** @@ -563,8 +540,7 @@ public class ActorContext { return ask(actorRef, message, timeout); } - @VisibleForTesting - Cache> getPrimaryShardInfoCache() { + public PrimaryShardInfoFutureCache getPrimaryShardInfoCache() { return primaryShardInfoCache; } }