Use two thread pools for handling the OpenFlow STATS and BARRIER messages.

This fixes a problem where a large number of incoming STATS messages
is practically using all available threds from the pool (16), and
there are no available threads to handle the BARRIER messages.

Change-Id: I1130eb8f3b5a17d5d3a7825f32da68eacb99569a

fixing other threadpool issues, ie. not using cachedThreadPool

Change-Id: I40ef10e1f704aef779b2a23c0497dfb7992520eb

@@ -103,7 +103,8 @@ public class FlowRuleManager
 
 
     @Activate
     @Activate
     public void activate() {
     public void activate() {
-        futureService = Executors.newCachedThreadPool(namedThreads("provider-future-listeners-%d"));
+        futureService =
+                Executors.newFixedThreadPool(32, namedThreads("provider-future-listeners-%d"));
         store.setDelegate(delegate);
         store.setDelegate(delegate);
         eventDispatcher.addSink(FlowRuleEvent.class, listenerRegistry);
         eventDispatcher.addSink(FlowRuleEvent.class, listenerRegistry);
         log.info("Started");
         log.info("Started");

@@ -67,8 +67,13 @@ public class OpenFlowControllerImpl implements OpenFlowController {
     private static final Logger log =
     private static final Logger log =
             LoggerFactory.getLogger(OpenFlowControllerImpl.class);
             LoggerFactory.getLogger(OpenFlowControllerImpl.class);
 
 
-    private final ExecutorService executor = Executors.newFixedThreadPool(16,
-            namedThreads("of-event-%d"));
+    private final ExecutorService executorMsgs =
+        Executors.newFixedThreadPool(32,
+                                     namedThreads("of-event-stats-%d"));
+
+    private final ExecutorService executorBarrier =
+        Executors.newFixedThreadPool(4,
+                                     namedThreads("of-event-barrier-%d"));
 
 
     protected ConcurrentHashMap<Dpid, OpenFlowSwitch> connectedSwitches =
     protected ConcurrentHashMap<Dpid, OpenFlowSwitch> connectedSwitches =
             new ConcurrentHashMap<Dpid, OpenFlowSwitch>();
             new ConcurrentHashMap<Dpid, OpenFlowSwitch>();
@@ -189,6 +194,12 @@ public class OpenFlowControllerImpl implements OpenFlowController {
                 p.handlePacket(pktCtx);
                 p.handlePacket(pktCtx);
             }
             }
             break;
             break;
+        // TODO: Consider using separate threadpool for sensitive messages.
+        //    ie. Back to back error could cause us to starve.
+        case FLOW_REMOVED:
+        case ERROR:
+            executorMsgs.submit(new OFMessageHandler(dpid, msg));
+            break;
         case STATS_REPLY:
         case STATS_REPLY:
             OFStatsReply reply = (OFStatsReply) msg;
             OFStatsReply reply = (OFStatsReply) msg;
             if (reply.getStatsType().equals(OFStatsType.PORT_DESC)) {
             if (reply.getStatsType().equals(OFStatsType.PORT_DESC)) {
@@ -201,14 +212,11 @@ public class OpenFlowControllerImpl implements OpenFlowController {
                 OFFlowStatsReply.Builder rep =
                 OFFlowStatsReply.Builder rep =
                         OFFactories.getFactory(msg.getVersion()).buildFlowStatsReply();
                         OFFactories.getFactory(msg.getVersion()).buildFlowStatsReply();
                 rep.setEntries(Lists.newLinkedList(stats));
                 rep.setEntries(Lists.newLinkedList(stats));
-                executor.submit(new OFMessageHandler(dpid, rep.build()));
+                executorMsgs.submit(new OFMessageHandler(dpid, rep.build()));
             }
             }
             break;
             break;
-
-        case FLOW_REMOVED:
-        case ERROR:
         case BARRIER_REPLY:
         case BARRIER_REPLY:
-            executor.submit(new OFMessageHandler(dpid, msg));
+            executorBarrier.submit(new OFMessageHandler(dpid, msg));
             break;
             break;
         case EXPERIMENTER:
         case EXPERIMENTER:
             // Handle optical port stats
             // Handle optical port stats

@@ -329,7 +329,7 @@ public class OpenFlowRuleProvider extends AbstractProvider implements FlowRulePr
             // Do nothing here for now.
             // Do nothing here for now.
         }
         }
 
 
-        private synchronized void pushFlowMetrics(Dpid dpid, OFStatsReply stats) {
+        private void pushFlowMetrics(Dpid dpid, OFStatsReply stats) {
 
 
             DeviceId did = DeviceId.deviceId(Dpid.uri(dpid));
             DeviceId did = DeviceId.deviceId(Dpid.uri(dpid));
             final OFFlowStatsReply replies = (OFFlowStatsReply) stats;
             final OFFlowStatsReply replies = (OFFlowStatsReply) stats;