commit 2af244a9770f85e802bca4a22982fe244c90604e
author Madan Jampani <madan@onlab.us> Sun Feb 22 13:12:01 2015 -0800
committer Madan Jampani <madan@onlab.us> Mon Feb 23 20:37:25 2015 +0000
tree 7ebd3d30b89ba6d53b9d194c6efc99de8e4f0841
parent a853709a6b1e9a671ff484b10a23570865d182b6

Never process incoming messages on the netty event loop thread pool.
Currently in a lot of places we are deserializing incoming messages on this threadpool and that could be significantly limiting throughput.

Change-Id: I83eb7e91004cea4addb28bc28f27e50de10028fe

core/store/dist/src/main/java/org/onosproject/store/flowext/impl/DefaultFlowRuleExtRouter.java

diff --git a/core/store/dist/src/main/java/org/onosproject/store/flowext/impl/DefaultFlowRuleExtRouter.java b/core/store/dist/src/main/java/org/onosproject/store/flowext/impl/DefaultFlowRuleExtRouter.java
index 6effe1b..708faf6 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/flowext/impl/DefaultFlowRuleExtRouter.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/flowext/impl/DefaultFlowRuleExtRouter.java
@@ -78,6 +78,9 @@
 
     private final Logger log = getLogger(getClass());
 
+    // TODO: Make configurable.
+    private static final int MESSAGE_HANDLER_THREAD_POOL_SIZE = 4;
+
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected ReplicaInfoService replicaInfoManager;
 
@@ -102,6 +105,8 @@
     private final ExecutorService futureListeners = Executors
             .newCachedThreadPool(groupedThreads("onos/flow", "store-peer-responders"));
 
+    private ExecutorService messageHandlingExecutor;
+
     protected static final StoreSerializer SERIALIZER = new KryoSerializer() {
         @Override
         protected void setupKryoPool() {
@@ -120,6 +125,11 @@
 
     @Activate
     public void activate() {
+
+        messageHandlingExecutor = Executors.newFixedThreadPool(
+                MESSAGE_HANDLER_THREAD_POOL_SIZE,
+                groupedThreads("onos/flow", "message-handlers"));
+
         clusterCommunicator.addSubscriber(APPLY_EXTEND_FLOWS,
             new ClusterMessageHandler() {
 
@@ -141,7 +151,7 @@
                       }
                   }, futureListeners);
               }
-            });
+            }, messageHandlingExecutor);
 
         replicaInfoManager.addListener(replicaInfoEventListener);
 
@@ -151,6 +161,7 @@
     @Deactivate
     public void deactivate() {
         clusterCommunicator.removeSubscriber(APPLY_EXTEND_FLOWS);
+        messageHandlingExecutor.shutdown();
         replicaInfoManager.removeListener(replicaInfoEventListener);
         log.info("Stopped");
     }