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=5b4f54daf8ea9b56f43e1afa1eb3f38dc9b5c1fd;hb=3b0499cd187bcdeda057465350d381c8bc28847c;hp=17d988005fadf32c8209837671d43d0aa2981b60;hpb=9f1061c46af5220ad95d8d0b94411ba2fd832a50;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 17d988005f..5b4f54daf8 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 @@ -28,8 +28,12 @@ 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 com.google.common.cache.RemovalListener; +import com.google.common.cache.RemovalNotification; +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; @@ -42,11 +46,14 @@ import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException import org.opendaylight.controller.cluster.datastore.exceptions.UnknownMessageException; 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.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; @@ -62,7 +69,7 @@ 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 DISTRIBUTED_DATA_STORE_METRIC_REGISTRY = "distributed-data-store"; private static final String METRIC_RATE = "rate"; @@ -82,6 +89,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; @@ -91,16 +99,18 @@ public class ActorContext { private FiniteDuration operationDuration; private Timeout operationTimeout; private final String selfAddressHostPort; - private RateLimiter txRateLimiter; + private TransactionRateLimiter txRateLimiter; 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) { @@ -131,7 +141,7 @@ public class ActorContext { } private void setCachedProperties() { - txRateLimiter = RateLimiter.create(datastoreContext.getTransactionCreationInitialRateLimit()); + txRateLimiter = new TransactionRateLimiter(this); operationDuration = Duration.create(datastoreContext.getOperationTimeoutInSeconds(), TimeUnit.SECONDS); operationTimeout = new Timeout(operationDuration); @@ -141,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(); } @@ -196,24 +207,24 @@ 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), shardInitializationTimeout); - return future.transform(new Mapper() { + return future.transform(new Mapper() { @Override - public ActorSelection checkedApply(Object response) throws Exception { - if(response instanceof PrimaryFound) { - PrimaryFound found = (PrimaryFound)response; - - LOG.debug("Primary found {}", found.getPrimaryPath()); - ActorSelection actorSelection = actorSystem.actorSelection(found.getPrimaryPath()); - primaryShardActorSelectionCache.put(shardName, Futures.successful(actorSelection)); - return actorSelection; + 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) { @@ -228,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 * @@ -367,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() { @@ -387,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()); @@ -495,21 +519,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(); } /** @@ -553,7 +568,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); + } + } } }