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=ad05a1ca71001285a8aed1f5afe16b1f511f3e96;hp=6f9bb7fc9feb4ede3a00d06b07d689d615f51458;hb=daaef05cbf70e6cbec9af181258faead6d9620a6;hpb=4349b034606957d3e876b82b14a292e6739a986a 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..ad05a1ca71 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 @@ -20,7 +20,6 @@ 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; @@ -29,8 +28,13 @@ 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.cache.RemovalListener; +import com.google.common.cache.RemovalNotification; import com.google.common.util.concurrent.RateLimiter; +import java.util.ArrayList; +import java.util.Collection; import java.util.concurrent.TimeUnit; +import javax.annotation.concurrent.GuardedBy; import org.opendaylight.controller.cluster.common.actor.CommonConfig; import org.opendaylight.controller.cluster.datastore.ClusterWrapper; import org.opendaylight.controller.cluster.datastore.Configuration; @@ -41,14 +45,16 @@ import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedEx 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.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; @@ -64,12 +70,10 @@ import scala.concurrent.duration.FiniteDuration; * easily. An ActorContext can be freely passed around to local object instances * but should not be passed to actors especially remote actors */ -public class ActorContext { +public class ActorContext implements RemovalListener> { 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 @@ -96,16 +100,17 @@ public class ActorContext { 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 Timeout transactionCommitOperationTimeout; private Timeout shardInitializationTimeout; private final Dispatchers dispatchers; - private Cache> primaryShardActorSelectionCache; + private Cache> primaryShardInfoCache; private volatile SchemaContext schemaContext; private volatile boolean updated; + private final MetricRegistry metricRegistry = MetricsReporter.getInstance(DatastoreContext.METRICS_DOMAIN).getMetricsRegistry(); + @GuardedBy("shardInfoListeners") + private final Collection> shardInfoListeners = new ArrayList<>(); public ActorContext(ActorSystem actorSystem, ActorRef shardManager, ClusterWrapper clusterWrapper, Configuration configuration) { @@ -133,8 +138,6 @@ public class ActorContext { } transactionOutstandingOperationLimit = new CommonConfig(this.getActorSystem().settings().config()).getMailBoxCapacity(); - jmxReporter.start(); - } private void setCachedProperties() { @@ -148,8 +151,9 @@ public class ActorContext { shardInitializationTimeout = new Timeout(datastoreContext.getShardInitializationTimeout().duration().$times(2)); - primaryShardActorSelectionCache = CacheBuilder.newBuilder() + primaryShardInfoCache = CacheBuilder.newBuilder() .expireAfterWrite(datastoreContext.getShardLeaderElectionTimeout().duration().toMillis(), TimeUnit.MILLISECONDS) + .removalListener(this) .build(); } @@ -203,31 +207,28 @@ 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); + return onPrimaryShardFound(shardName, ((RemotePrimaryShardFound)response).getPrimaryPath(), null); + } else if(response instanceof LocalPrimaryShardFound) { + LOG.debug("findPrimaryShardAsync received: {}", response); + LocalPrimaryShardFound found = (LocalPrimaryShardFound)response; + return onPrimaryShardFound(shardName, found.getPrimaryPath(), 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 +239,20 @@ public class ActorContext { }, FIND_PRIMARY_FAILURE_TRANSFORMER, getClientDispatcher()); } + private PrimaryShardInfo onPrimaryShardFound(String shardName, String primaryActorPath, + DataTree localShardDataTree) { + ActorSelection actorSelection = actorSystem.actorSelection(primaryActorPath); + PrimaryShardInfo info = new PrimaryShardInfo(actorSelection, Optional.fromNullable(localShardDataTree)); + primaryShardInfoCache.put(shardName, Futures.successful(info)); + + synchronized (shardInfoListeners) { + for (ShardInfoListenerRegistration reg : shardInfoListeners) { + reg.getInstance().onShardInfoUpdated(shardName, info); + } + } + return info; + } + /** * Finds a local shard given its shard name and return it's ActorRef * @@ -274,10 +289,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 +392,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 +411,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()); @@ -489,7 +501,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); } @@ -547,6 +564,10 @@ public class ActorContext { return this.dispatchers.getDispatcherPath(Dispatchers.DispatcherType.Notification); } + public Configuration getConfiguration() { + return configuration; + } + protected Future doAsk(ActorRef actorRef, Object message, Timeout timeout){ return ask(actorRef, message, timeout); } @@ -556,7 +577,31 @@ public class ActorContext { } @VisibleForTesting - Cache> getPrimaryShardActorSelectionCache() { - return primaryShardActorSelectionCache; + Cache> getPrimaryShardInfoCache() { + return primaryShardInfoCache; + } + + public ShardInfoListenerRegistration registerShardInfoListener(final T listener) { + final ShardInfoListenerRegistration reg = new ShardInfoListenerRegistration(listener, this); + + synchronized (shardInfoListeners) { + shardInfoListeners.add(reg); + } + return reg; + } + + protected void removeShardInfoListener(final ShardInfoListenerRegistration registration) { + synchronized (shardInfoListeners) { + shardInfoListeners.remove(registration); + } + } + + @Override + public void onRemoval(final RemovalNotification> notification) { + synchronized (shardInfoListeners) { + for (ShardInfoListenerRegistration reg : shardInfoListeners) { + reg.getInstance().onShardInfoUpdated(notification.getKey(), null); + } + } } }