X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Futils%2FActorContext.java;h=792064cd6700cc2d0cf6a09423e8964ee980eab1;hp=6f9bb7fc9feb4ede3a00d06b07d689d615f51458;hb=c2d1b9207fe82d36db83501e1baaffe7bc7da9ae;hpb=6dcee56392712348b1abdcdc0d1d5f94dfcf505c 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 6f9bb7fc9f..792064cd67 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,40 +15,38 @@ 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.JmxReporter; 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.DatastoreContextFactory; +import org.opendaylight.controller.cluster.datastore.config.Configuration; import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException; import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException; import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException; import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException; import org.opendaylight.controller.cluster.datastore.exceptions.UnknownMessageException; -import org.opendaylight.controller.cluster.datastore.messages.ActorNotInitialized; import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard; import org.opendaylight.controller.cluster.datastore.messages.FindPrimary; +import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardFound; import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound; import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound; -import org.opendaylight.controller.cluster.datastore.messages.PrimaryFound; -import org.opendaylight.controller.cluster.datastore.messages.PrimaryNotFound; +import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo; +import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; +import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory; +import org.opendaylight.controller.cluster.reporting.MetricsReporter; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree; import org.opendaylight.yangtools.yang.model.api.SchemaContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -66,10 +64,8 @@ import scala.concurrent.duration.FiniteDuration; */ public class ActorContext { private static final Logger LOG = LoggerFactory.getLogger(ActorContext.class); - private static final String UNKNOWN_DATA_STORE_TYPE = "unknown"; private static final String DISTRIBUTED_DATA_STORE_METRIC_REGISTRY = "distributed-data-store"; private static final String METRIC_RATE = "rate"; - private static final String DOMAIN = "org.opendaylight.controller.cluster.datastore"; private static final Mapper FIND_PRIMARY_FAILURE_TRANSFORMER = new Mapper() { @Override @@ -86,6 +82,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; @@ -95,33 +92,35 @@ public class ActorContext { private FiniteDuration operationDuration; private Timeout operationTimeout; private final String selfAddressHostPort; - private RateLimiter txRateLimiter; - private final MetricRegistry metricRegistry = new MetricRegistry(); - private final JmxReporter jmxReporter = JmxReporter.forRegistry(metricRegistry).inDomain(DOMAIN).build(); - private final int transactionOutstandingOperationLimit; + private TransactionRateLimiter txRateLimiter; private Timeout transactionCommitOperationTimeout; private Timeout shardInitializationTimeout; private final Dispatchers dispatchers; - private Cache> primaryShardActorSelectionCache; private volatile SchemaContext schemaContext; private volatile boolean updated; + private final MetricRegistry metricRegistry = MetricsReporter.getInstance(DatastoreContext.METRICS_DOMAIN).getMetricsRegistry(); + + private final PrimaryShardInfoFutureCache primaryShardInfoCache; + private final ShardStrategyFactory shardStrategyFactory; 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; + this.shardStrategyFactory = new ShardStrategyFactory(configuration); setCachedProperties(); @@ -132,25 +131,18 @@ public class ActorContext { selfAddressHostPort = null; } - transactionOutstandingOperationLimit = new CommonConfig(this.getActorSystem().settings().config()).getMailBoxCapacity(); - jmxReporter.start(); - } 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)); - - primaryShardActorSelectionCache = CacheBuilder.newBuilder() - .expireAfterWrite(datastoreContext.getShardLeaderElectionTimeout().duration().toMillis(), TimeUnit.MILLISECONDS) - .build(); } public DatastoreContext getDatastoreContext() { @@ -181,8 +173,8 @@ public class ActorContext { } } - public void setDatastoreContext(DatastoreContext context) { - this.datastoreContext = context; + public void setDatastoreContext(DatastoreContextFactory contextFactory) { + this.datastoreContext = contextFactory.getBaseDatastoreContext(); setCachedProperties(); // We write the 'updated' volatile to trigger a write memory barrier so that the writes above @@ -195,7 +187,7 @@ public class ActorContext { updated = true; if(shardManager != null) { - shardManager.tell(context, ActorRef.noSender()); + shardManager.tell(contextFactory, ActorRef.noSender()); } } @@ -203,31 +195,30 @@ public class ActorContext { return schemaContext; } - public Future findPrimaryShardAsync(final String shardName) { - Future ret = primaryShardActorSelectionCache.getIfPresent(shardName); + public Future findPrimaryShardAsync(final String shardName) { + Future ret = primaryShardInfoCache.getIfPresent(shardName); if(ret != null){ return ret; } Future future = executeOperationAsync(shardManager, - new FindPrimary(shardName, true).toSerializable(), shardInitializationTimeout); + new FindPrimary(shardName, true), shardInitializationTimeout); - return future.transform(new Mapper() { + return future.transform(new Mapper() { @Override - public ActorSelection checkedApply(Object response) throws Exception { - if(PrimaryFound.SERIALIZABLE_CLASS.isInstance(response)) { - PrimaryFound found = PrimaryFound.fromSerializable(response); - - LOG.debug("Primary found {}", found.getPrimaryPath()); - ActorSelection actorSelection = actorSystem.actorSelection(found.getPrimaryPath()); - primaryShardActorSelectionCache.put(shardName, Futures.successful(actorSelection)); - return actorSelection; - } else if(response instanceof ActorNotInitialized) { - throw new NotInitializedException( - String.format("Found primary shard %s but it's not initialized yet. " + - "Please try again later", shardName)); - } else if(response instanceof PrimaryNotFound) { - throw new PrimaryNotFoundException( - String.format("No primary shard found for %S.", shardName)); + public PrimaryShardInfo checkedApply(Object response) throws Exception { + if(response instanceof RemotePrimaryShardFound) { + LOG.debug("findPrimaryShardAsync received: {}", response); + 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(), DataStoreVersions.CURRENT_VERSION, + found.getLocalShardDataTree()); + } else if(response instanceof NotInitializedException) { + throw (NotInitializedException)response; + } else if(response instanceof PrimaryNotFoundException) { + throw (PrimaryNotFoundException)response; } else if(response instanceof NoShardLeaderException) { throw (NoShardLeaderException)response; } @@ -238,6 +229,15 @@ public class ActorContext { }, FIND_PRIMARY_FAILURE_TRANSFORMER, getClientDispatcher()); } + private PrimaryShardInfo onPrimaryShardFound(String shardName, String primaryActorPath, + short primaryVersion, DataTree localShardDataTree) { + ActorSelection actorSelection = actorSystem.actorSelection(primaryActorPath); + PrimaryShardInfo info = new PrimaryShardInfo(actorSelection, primaryVersion, + Optional.fromNullable(localShardDataTree)); + primaryShardInfoCache.putSuccessful(shardName, info); + return info; + } + /** * Finds a local shard given its shard name and return it's ActorRef * @@ -274,10 +274,8 @@ public class ActorContext { LocalShardFound found = (LocalShardFound)response; LOG.debug("Local shard found {}", found.getPath()); return found.getPath(); - } else if(response instanceof ActorNotInitialized) { - throw new NotInitializedException( - String.format("Found local shard for %s but it's not initialized yet.", - shardName)); + } else if(response instanceof NotInitializedException) { + throw (NotInitializedException)response; } else if(response instanceof LocalShardNotFound) { throw new LocalShardNotFoundException( String.format("Local shard for %s does not exist.", shardName)); @@ -379,8 +377,7 @@ public class ActorContext { } public void shutdown() { - shardManager.tell(PoisonPill.getInstance(), null); - actorSystem.shutdown(); + shardManager.tell(PoisonPill.getInstance(), ActorRef.noSender()); } public ClusterWrapper getClusterWrapper() { @@ -399,15 +396,15 @@ public class ActorContext { public void broadcast(final Object message){ for(final String shardName : configuration.getAllShardNames()){ - Future primaryFuture = findPrimaryShardAsync(shardName); - primaryFuture.onComplete(new OnComplete() { + Future primaryFuture = findPrimaryShardAsync(shardName); + primaryFuture.onComplete(new OnComplete() { @Override - public void onComplete(Throwable failure, ActorSelection primaryShard) { + public void onComplete(Throwable failure, PrimaryShardInfo primaryShardInfo) { if(failure != null) { LOG.warn("broadcast failed to send message {} to shard {}: {}", message.getClass().getSimpleName(), shardName, failure); } else { - primaryShard.tell(message, ActorRef.noSender()); + primaryShardInfo.getPrimaryShardActor().tell(message, ActorRef.noSender()); } } }, getClientDispatcher()); @@ -418,6 +415,10 @@ public class ActorContext { return operationDuration; } + public Timeout getOperationTimeout() { + return operationTimeout; + } + public boolean isPathLocal(String path) { if (Strings.isNullOrEmpty(path)) { return false; @@ -469,18 +470,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. @@ -489,7 +478,12 @@ public class ActorContext { * @return */ public Timer getOperationTimer(String operationName){ - final String rate = MetricRegistry.name(DISTRIBUTED_DATA_STORE_METRIC_REGISTRY, datastoreContext.getDataStoreType(), operationName, METRIC_RATE); + return getOperationTimer(datastoreContext.getDataStoreType(), operationName); + } + + public Timer getOperationTimer(String dataStoreType, String operationName){ + final String rate = MetricRegistry.name(DISTRIBUTED_DATA_STORE_METRIC_REGISTRY, dataStoreType, + operationName, METRIC_RATE); return metricRegistry.timer(rate); } @@ -502,21 +496,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(); } /** @@ -547,6 +532,14 @@ public class ActorContext { return this.dispatchers.getDispatcherPath(Dispatchers.DispatcherType.Notification); } + public Configuration getConfiguration() { + return configuration; + } + + public ShardStrategyFactory getShardStrategyFactory() { + return shardStrategyFactory; + } + protected Future doAsk(ActorRef actorRef, Object message, Timeout timeout){ return ask(actorRef, message, timeout); } @@ -555,8 +548,7 @@ public class ActorContext { return ask(actorRef, message, timeout); } - @VisibleForTesting - Cache> getPrimaryShardActorSelectionCache() { - return primaryShardActorSelectionCache; + public PrimaryShardInfoFutureCache getPrimaryShardInfoCache() { + return primaryShardInfoCache; } }