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=1dc49dc88aca5468d075907033110af89368aed9;hb=refs%2Fchanges%2F52%2F31952%2F8;hp=b07cf28a7d7f03fe5de3abd7ca8c24c666e6c539;hpb=4379f102fa0c85abf58f60d81fec9c698582fb1a;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 b07cf28a7d..1dc49dc88a 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;
@@ -26,9 +26,10 @@ import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import java.util.concurrent.TimeUnit;
 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 +44,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;
@@ -100,6 +103,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) {
@@ -117,6 +121,7 @@ public class ActorContext {
         this.datastoreContext = datastoreContext;
         this.dispatchers = new Dispatchers(actorSystem.dispatchers());
         this.primaryShardInfoCache = primaryShardInfoCache;
+        this.shardStrategyFactory = new ShardStrategyFactory(configuration);
 
         setCachedProperties();
 
@@ -132,7 +137,7 @@ public class ActorContext {
     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(
@@ -169,8 +174,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
@@ -183,7 +188,7 @@ public class ActorContext {
         updated = true;
 
         if(shardManager != null) {
-            shardManager.tell(context, ActorRef.noSender());
+            shardManager.tell(contextFactory, ActorRef.noSender());
         }
     }
 
@@ -373,7 +378,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() {
@@ -411,6 +421,10 @@ public class ActorContext {
         return operationDuration;
     }
 
+    public Timeout getOperationTimeout() {
+        return operationTimeout;
+    }
+
     public boolean isPathLocal(String path) {
         if (Strings.isNullOrEmpty(path)) {
             return false;
@@ -470,7 +484,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){
@@ -479,13 +493,24 @@ public class ActorContext {
         return metricRegistry.timer(rate);
     }
 
+    /**
+     * Get the name of the data store to which this ActorContext belongs
+     *
+     * @return
+     */
+    public String getDataStoreName() {
+        return datastoreContext.getDataStoreName();
+    }
+
     /**
      * Get the type of the data store to which this ActorContext belongs
      *
      * @return
+     * @deprecated Use {@link #getDataStoreName()} instead.
      */
+    @Deprecated
     public String getDataStoreType() {
-        return datastoreContext.getDataStoreType();
+        return datastoreContext.getDataStoreName();
     }
 
     /**
@@ -528,6 +553,10 @@ public class ActorContext {
         return configuration;
     }
 
+    public ShardStrategyFactory getShardStrategyFactory() {
+        return shardStrategyFactory;
+    }
+
     protected Future<Object> doAsk(ActorRef actorRef, Object message, Timeout timeout){
         return ask(actorRef, message, timeout);
     }