Bug 1534: Changed blocking calls to async in dist data store

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / DistributedDataStore.java

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DistributedDataStore.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DistributedDataStore.java
index 40e045f18e31bdfbcf2eb34c9fa0847a778f94f4..404a4e02033ea1c89f9fada4135cfe0e4b6a935e 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DistributedDataStore.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DistributedDataStore.java
@@ -10,9 +10,8 @@ package org.opendaylight.controller.cluster.datastore;
 
 import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
+
 import com.google.common.base.Preconditions;
-import com.google.common.util.concurrent.ListeningExecutorService;
-import com.google.common.util.concurrent.MoreExecutors;
 import org.opendaylight.controller.cluster.datastore.identifiers.ShardManagerIdentifier;
 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
@@ -21,14 +20,13 @@ import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategy
 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
+import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreConfigProperties;
 import org.opendaylight.controller.sal.core.spi.data.DOMStore;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
 import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.util.PropertyUtils;
-import org.opendaylight.yangtools.util.concurrent.SpecialExecutors;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
@@ -41,38 +39,14 @@ import org.slf4j.LoggerFactory;
  */
 public class DistributedDataStore implements DOMStore, SchemaContextListener, AutoCloseable {
 
-    private static final Logger
-        LOG = LoggerFactory.getLogger(DistributedDataStore.class);
-
-    private static final String EXECUTOR_MAX_POOL_SIZE_PROP =
-            "mdsal.dist-datastore-executor-pool.size";
-    private static final int DEFAULT_EXECUTOR_MAX_POOL_SIZE = 10;
-
-    private static final String EXECUTOR_MAX_QUEUE_SIZE_PROP =
-            "mdsal.dist-datastore-executor-queue.size";
-    private static final int DEFAULT_EXECUTOR_MAX_QUEUE_SIZE = 5000;
+    private static final Logger LOG = LoggerFactory.getLogger(DistributedDataStore.class);
 
     private final ActorContext actorContext;
 
     private SchemaContext schemaContext;
 
-    /**
-     * Executor used to run FutureTask's
-     *
-     * This is typically used when we need to make a request to an actor and
-     * wait for it's response and the consumer needs to be provided a Future.
-     */
-    private final ListeningExecutorService executor =
-            MoreExecutors.listeningDecorator(
-                    SpecialExecutors.newBlockingBoundedFastThreadPool(
-                            PropertyUtils.getIntSystemProperty(
-                                    EXECUTOR_MAX_POOL_SIZE_PROP,
-                                    DEFAULT_EXECUTOR_MAX_POOL_SIZE),
-                            PropertyUtils.getIntSystemProperty(
-                                    EXECUTOR_MAX_QUEUE_SIZE_PROP,
-                                    DEFAULT_EXECUTOR_MAX_QUEUE_SIZE), "DistDataStore"));
-
-    public DistributedDataStore(ActorSystem actorSystem, String type, ClusterWrapper cluster, Configuration configuration) {
+    public DistributedDataStore(ActorSystem actorSystem, String type, ClusterWrapper cluster,
+            Configuration configuration, InMemoryDOMDataStoreConfigProperties dataStoreProperties) {
         Preconditions.checkNotNull(actorSystem, "actorSystem should not be null");
         Preconditions.checkNotNull(type, "type should not be null");
         Preconditions.checkNotNull(cluster, "cluster should not be null");
@@ -84,7 +58,7 @@ public class DistributedDataStore implements DOMStore, SchemaContextListener, Au
         LOG.info("Creating ShardManager : {}", shardManagerId);
 
         this.actorContext = new ActorContext(actorSystem, actorSystem
-            .actorOf(ShardManager.props(type, cluster, configuration),
+            .actorOf(ShardManager.props(type, cluster, configuration, dataStoreProperties),
                 shardManagerId ), cluster, configuration);
     }
 
@@ -93,15 +67,16 @@ public class DistributedDataStore implements DOMStore, SchemaContextListener, Au
     }
 
 
+    @SuppressWarnings("unchecked")
     @Override
-    public <L extends AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>> ListenerRegistration<L> registerChangeListener(
+    public <L extends AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
+                                              ListenerRegistration<L> registerChangeListener(
         YangInstanceIdentifier path, L listener,
         AsyncDataBroker.DataChangeScope scope) {
 
         Preconditions.checkNotNull(path, "path should not be null");
         Preconditions.checkNotNull(listener, "listener should not be null");
 
-
         LOG.debug("Registering listener: {} for path: {} scope: {}", listener, path, scope);
 
         ActorRef dataChangeListenerActor = actorContext.getActorSystem().actorOf(
@@ -110,10 +85,8 @@ public class DistributedDataStore implements DOMStore, SchemaContextListener, Au
         String shardName = ShardStrategyFactory.getStrategy(path).findShard(path);
 
         Object result = actorContext.executeLocalShardOperation(shardName,
-            new RegisterChangeListener(path, dataChangeListenerActor.path(),
-                scope),
-            ActorContext.ASK_DURATION
-        );
+            new RegisterChangeListener(path, dataChangeListenerActor.path(), scope),
+            ActorContext.ASK_DURATION);
 
         if (result != null) {
             RegisterChangeListenerReply reply = (RegisterChangeListenerReply) result;
@@ -125,34 +98,31 @@ public class DistributedDataStore implements DOMStore, SchemaContextListener, Au
         LOG.debug(
             "No local shard for shardName {} was found so returning a noop registration",
             shardName);
+
         return new NoOpDataChangeListenerRegistration(listener);
     }
 
-
-
-
-
     @Override
     public DOMStoreTransactionChain createTransactionChain() {
-        return new TransactionChainProxy(actorContext, executor, schemaContext);
+        return new TransactionChainProxy(actorContext, schemaContext);
     }
 
     @Override
     public DOMStoreReadTransaction newReadOnlyTransaction() {
         return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_ONLY,
-            executor, schemaContext);
+            schemaContext);
     }
 
     @Override
     public DOMStoreWriteTransaction newWriteOnlyTransaction() {
         return new TransactionProxy(actorContext, TransactionProxy.TransactionType.WRITE_ONLY,
-            executor, schemaContext);
+            schemaContext);
     }
 
     @Override
     public DOMStoreReadWriteTransaction newReadWriteTransaction() {
         return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_WRITE,
-            executor, schemaContext);
+            schemaContext);
     }
 
     @Override public void onGlobalContextUpdated(SchemaContext schemaContext) {