X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Futils%2FActorContext.java;h=e7ab144a5a034dfd0b868d33f34bd81c9aeafbb5;hb=ae11ac10dfd3579b1b685455ea642bbb08de68f1;hp=5c514cf77502587fc15939df20e61187d0dfe297;hpb=d3a97264ecf47e8c60ea11a7caebce41b580e91d;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 5c514cf775..e7ab144a5a 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 @@ -14,10 +14,10 @@ import akka.actor.ActorRef; import akka.actor.ActorSelection; import akka.actor.ActorSystem; import akka.actor.Address; -import akka.actor.PoisonPill; import akka.dispatch.Mapper; import akka.dispatch.OnComplete; import akka.pattern.AskTimeoutException; +import akka.pattern.Patterns; import akka.util.Timeout; import com.codahale.metrics.MetricRegistry; import com.codahale.metrics.Timer; @@ -25,10 +25,12 @@ import com.google.common.base.Optional; import com.google.common.base.Preconditions; import com.google.common.base.Strings; import java.util.concurrent.TimeUnit; -import org.opendaylight.controller.cluster.common.actor.CommonConfig; +import java.util.function.Function; 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; @@ -43,6 +45,8 @@ import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound 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.raft.client.messages.Shutdown; 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; @@ -91,7 +95,6 @@ public class ActorContext { private Timeout operationTimeout; private final String selfAddressHostPort; private TransactionRateLimiter txRateLimiter; - private final int transactionOutstandingOperationLimit; private Timeout transactionCommitOperationTimeout; private Timeout shardInitializationTimeout; private final Dispatchers dispatchers; @@ -101,6 +104,7 @@ public class ActorContext { 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) { @@ -118,6 +122,7 @@ public class ActorContext { this.datastoreContext = datastoreContext; this.dispatchers = new Dispatchers(actorSystem.dispatchers()); this.primaryShardInfoCache = primaryShardInfoCache; + this.shardStrategyFactory = new ShardStrategyFactory(configuration); setCachedProperties(); @@ -128,13 +133,12 @@ public class ActorContext { selfAddressHostPort = null; } - transactionOutstandingOperationLimit = new CommonConfig(this.getActorSystem().settings().config()).getMailBoxCapacity(); } private void setCachedProperties() { 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( @@ -171,8 +175,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 @@ -185,7 +189,7 @@ public class ActorContext { updated = true; if(shardManager != null) { - shardManager.tell(context, ActorRef.noSender()); + shardManager.tell(contextFactory, ActorRef.noSender()); } } @@ -206,11 +210,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) { @@ -226,9 +232,10 @@ 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)); + PrimaryShardInfo info = new PrimaryShardInfo(actorSelection, primaryVersion, + Optional.fromNullable(localShardDataTree)); primaryShardInfoCache.putSuccessful(shardName, info); return info; } @@ -372,7 +379,12 @@ public class ActorContext { } public void shutdown() { - shardManager.tell(PoisonPill.getInstance(), ActorRef.noSender()); + FiniteDuration duration = datastoreContext.getShardRaftConfig().getElectionTimeOutInterval().$times(3); + try { + Await.ready(Patterns.gracefulStop(shardManager, duration, new Shutdown()), duration); + } catch(Exception e) { + LOG.warn("ShardManager for {} data store did not shutdown gracefully", getDataStoreName(), e); + } } public ClusterWrapper getClusterWrapper() { @@ -388,13 +400,14 @@ public class ActorContext { * * @param message */ - public void broadcast(final Object message){ + public void broadcast(final Function messageSupplier){ for(final String shardName : configuration.getAllShardNames()){ Future primaryFuture = findPrimaryShardAsync(shardName); primaryFuture.onComplete(new OnComplete() { @Override public void onComplete(Throwable failure, PrimaryShardInfo primaryShardInfo) { + Object message = messageSupplier.apply(primaryShardInfo.getPrimaryShardVersion()); if(failure != null) { LOG.warn("broadcast failed to send message {} to shard {}: {}", message.getClass().getSimpleName(), shardName, failure); @@ -410,6 +423,10 @@ public class ActorContext { return operationDuration; } + public Timeout getOperationTimeout() { + return operationTimeout; + } + public boolean isPathLocal(String path) { if (Strings.isNullOrEmpty(path)) { return false; @@ -437,42 +454,6 @@ public class ActorContext { } } - /** - * @deprecated This method is present only to support backward compatibility with Helium and should not be - * used any further - * - * - * @param primaryPath - * @param localPathOfRemoteActor - * @return - */ - @Deprecated - public String resolvePath(final String primaryPath, - final String localPathOfRemoteActor) { - StringBuilder builder = new StringBuilder(); - String[] primaryPathElements = primaryPath.split("/"); - builder.append(primaryPathElements[0]).append("//") - .append(primaryPathElements[1]).append(primaryPathElements[2]); - String[] remotePathElements = localPathOfRemoteActor.split("/"); - for (int i = 3; i < remotePathElements.length; i++) { - builder.append("/").append(remotePathElements[i]); - } - - 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. @@ -481,7 +462,7 @@ public class ActorContext { * @return */ public Timer getOperationTimer(String operationName){ - return getOperationTimer(datastoreContext.getDataStoreType(), operationName); + return getOperationTimer(datastoreContext.getDataStoreName(), operationName); } public Timer getOperationTimer(String dataStoreType, String operationName){ @@ -491,12 +472,12 @@ public class ActorContext { } /** - * Get the type of the data store to which this ActorContext belongs + * Get the name of the data store to which this ActorContext belongs * * @return */ - public String getDataStoreType() { - return datastoreContext.getDataStoreType(); + public String getDataStoreName() { + return datastoreContext.getDataStoreName(); } /** @@ -539,6 +520,10 @@ public class ActorContext { return configuration; } + public ShardStrategyFactory getShardStrategyFactory() { + return shardStrategyFactory; + } + protected Future doAsk(ActorRef actorRef, Object message, Timeout timeout){ return ask(actorRef, message, timeout); }