Improves host location provider introducing a pool of worker threads

Change-Id: Icc6d5a436cdaee10dde4f751bff944ccda69d8aa

Author: pierventre

providers/host/src/main/java/org/onosproject/provider/host/impl/HostLocationProvider.java

@@ -42,6 +42,7 @@
 import org.onlab.packet.ndp.NeighborSolicitation;
 import org.onlab.packet.ndp.RouterAdvertisement;
 import org.onlab.packet.ndp.RouterSolicitation;
+import org.onlab.util.PredictableExecutor;
 import org.onlab.util.Tools;
 import org.onosproject.cfg.ComponentConfigService;
 import org.onosproject.core.ApplicationId;
@@ -185,7 +186,9 @@ public class HostLocationProvider extends AbstractProvider implements HostProvid
 
     ExecutorService deviceEventHandler;
     private ExecutorService probeEventHandler;
-    private ExecutorService packetHandler;
+    // Packet workers - 0 will leverage available processors
+    private static final int DEFAULT_THREADS = 0;
+    private PredictableExecutor packetWorkers;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
     protected NetworkConfigService netcfgService;
@@ -215,8 +218,8 @@ public void activate(ComponentContext context) {
                 "device-event-handler", log));
         probeEventHandler = newSingleThreadScheduledExecutor(groupedThreads("onos/host-loc-provider",
                 "probe-event-handler", log));
-        packetHandler = newSingleThreadScheduledExecutor(groupedThreads("onos/host-loc-provider",
-                "packet-handler", log));
+        packetWorkers = new PredictableExecutor(DEFAULT_THREADS, groupedThreads("onos/host-loc-provider",
+                                                                                 "packet-worker-%d", log));
         providerService = providerRegistry.register(this);
         packetService.addProcessor(processor, PacketProcessor.advisor(1));
         deviceService.addListener(deviceListener);
@@ -238,7 +241,7 @@ public void deactivate() {
         deviceService.removeListener(deviceListener);
         deviceEventHandler.shutdown();
         probeEventHandler.shutdown();
-        packetHandler.shutdown();
+        packetWorkers.shutdown();
         providerService = null;
         registry.unregisterConfigFactory(hostLearningConfig);
         netcfgService.removeListener(cfgListener);
@@ -525,18 +528,22 @@ private void updateHostIp(HostId hid, IpAddress ip) {
 
         @Override
         public void process(PacketContext context) {
-            packetHandler.execute(() -> processPacketInternal(context));
-        }
-
-        private void processPacketInternal(PacketContext context) {
+            // Verify valid context
             if (context == null) {
                 return;
             }
-
+            // Verify valid Ethernet packet
             Ethernet eth = context.inPacket().parsed();
             if (eth == null) {
                 return;
             }
+            // Dispatch to a worker thread
+            HostId hostId = HostId.hostId(eth.getSourceMAC(), VlanId.vlanId(eth.getVlanID()));
+            packetWorkers.execute(() -> processPacketInternal(context), hostId.hashCode());
+        }
+
+        private void processPacketInternal(PacketContext context) {
+            Ethernet eth = context.inPacket().parsed();
 
             MacAddress srcMac = eth.getSourceMAC();
             if (srcMac.isBroadcast() || srcMac.isMulticast()) {