Move event handling to background thread

Change-Id: I8ccd1631fac14b1f753da4fb4b4ed01e5a045edf
(cherry picked from commit 8f906bfa)

@@ -20,6 +20,7 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.LinkedListMultimap;
 import com.google.common.collect.LinkedListMultimap;
 import com.google.common.collect.Multimap;
 import com.google.common.collect.Multimap;
 import com.google.common.collect.Sets;
 import com.google.common.collect.Sets;
+
 import org.onosproject.core.ApplicationId;
 import org.onosproject.core.ApplicationId;
 import org.onosproject.mastership.MastershipService;
 import org.onosproject.mastership.MastershipService;
 import org.onosproject.net.Device;
 import org.onosproject.net.Device;
@@ -186,16 +187,19 @@ class FlowRuleDriverProvider extends AbstractProvider implements FlowRuleProvide
 
 
         @Override
         @Override
         public void event(DeviceEvent event) {
         public void event(DeviceEvent event) {
-            executor.schedule(() -> pollDeviceFlowEntries(event.subject()), 0, TimeUnit.SECONDS);
+            executor.execute(() -> handleEvent(event));
         }
         }
 
 
-        @Override
-        public boolean isRelevant(DeviceEvent event) {
+        private void handleEvent(DeviceEvent event) {
             Device device = event.subject();
             Device device = event.subject();
-            return mastershipService.isLocalMaster(device.id()) && device.is(FlowRuleProgrammable.class) &&
-                    (event.type() == DEVICE_ADDED ||
+            boolean isRelevant = mastershipService.isLocalMaster(device.id())
+                    && device.is(FlowRuleProgrammable.class)
+                    && (event.type() == DEVICE_ADDED ||
                         event.type() == DEVICE_UPDATED ||
                         event.type() == DEVICE_UPDATED ||
                         (event.type() == DEVICE_AVAILABILITY_CHANGED && deviceService.isAvailable(device.id())));
                         (event.type() == DEVICE_AVAILABILITY_CHANGED && deviceService.isAvailable(device.id())));
+            if (isRelevant) {
+                pollDeviceFlowEntries(event.subject());
+            }
         }
         }
     }
     }
 
 

@@ -163,6 +163,7 @@ public class DistributedFlowRuleStore
 
 
     private Map<Long, NodeId> pendingResponses = Maps.newConcurrentMap();
     private Map<Long, NodeId> pendingResponses = Maps.newConcurrentMap();
     private ExecutorService messageHandlingExecutor;
     private ExecutorService messageHandlingExecutor;
+    private ExecutorService eventHandler;
 
 
     private ScheduledFuture<?> backupTask;
     private ScheduledFuture<?> backupTask;
     private final ScheduledExecutorService backupSenderExecutor =
     private final ScheduledExecutorService backupSenderExecutor =
@@ -197,6 +198,8 @@ public class DistributedFlowRuleStore
 
 
         local = clusterService.getLocalNode().id();
         local = clusterService.getLocalNode().id();
 
 
+        eventHandler = Executors.newSingleThreadExecutor(
+                groupedThreads("onos/flow", "event-handler", log));
         messageHandlingExecutor = Executors.newFixedThreadPool(
         messageHandlingExecutor = Executors.newFixedThreadPool(
                 msgHandlerPoolSize, groupedThreads("onos/store/flow", "message-handlers", log));
                 msgHandlerPoolSize, groupedThreads("onos/store/flow", "message-handlers", log));
 
 
@@ -233,6 +236,7 @@ public class DistributedFlowRuleStore
         unregisterMessageHandlers();
         unregisterMessageHandlers();
         deviceTableStats.removeListener(tableStatsListener);
         deviceTableStats.removeListener(tableStatsListener);
         deviceTableStats.destroy();
         deviceTableStats.destroy();
+        eventHandler.shutdownNow();
         messageHandlingExecutor.shutdownNow();
         messageHandlingExecutor.shutdownNow();
         backupSenderExecutor.shutdownNow();
         backupSenderExecutor.shutdownNow();
         log.info("Stopped");
         log.info("Stopped");
@@ -663,6 +667,10 @@ public class DistributedFlowRuleStore
 
 
         @Override
         @Override
         public void event(ReplicaInfoEvent event) {
         public void event(ReplicaInfoEvent event) {
+            eventHandler.execute(() -> handleEvent(event));
+        }
+
+        private void handleEvent(ReplicaInfoEvent event) {
             if (!backupEnabled) {
             if (!backupEnabled) {
                 return;
                 return;
             }
             }

@@ -97,6 +97,7 @@ public class ConsistentDeviceMastershipStore
     private static final Pattern DEVICE_MASTERSHIP_TOPIC_PATTERN =
     private static final Pattern DEVICE_MASTERSHIP_TOPIC_PATTERN =
             Pattern.compile("device:(.*)");
             Pattern.compile("device:(.*)");
 
 
+    private ExecutorService eventHandler;
     private ExecutorService messageHandlingExecutor;
     private ExecutorService messageHandlingExecutor;
     private ScheduledExecutorService transferExecutor;
     private ScheduledExecutorService transferExecutor;
     private final LeadershipEventListener leadershipEventListener =
     private final LeadershipEventListener leadershipEventListener =
@@ -116,6 +117,10 @@ public class ConsistentDeviceMastershipStore
 
 
     @Activate
     @Activate
     public void activate() {
     public void activate() {
+
+        eventHandler = Executors.newSingleThreadExecutor(
+                        groupedThreads("onos/store/device/mastership", "event-handler", log));
+
         messageHandlingExecutor =
         messageHandlingExecutor =
                 Executors.newSingleThreadExecutor(
                 Executors.newSingleThreadExecutor(
                         groupedThreads("onos/store/device/mastership", "message-handler", log));
                         groupedThreads("onos/store/device/mastership", "message-handler", log));
@@ -136,10 +141,10 @@ public class ConsistentDeviceMastershipStore
     @Deactivate
     @Deactivate
     public void deactivate() {
     public void deactivate() {
         clusterCommunicator.removeSubscriber(ROLE_RELINQUISH_SUBJECT);
         clusterCommunicator.removeSubscriber(ROLE_RELINQUISH_SUBJECT);
+        leadershipService.removeListener(leadershipEventListener);
         messageHandlingExecutor.shutdown();
         messageHandlingExecutor.shutdown();
         transferExecutor.shutdown();
         transferExecutor.shutdown();
-        leadershipService.removeListener(leadershipEventListener);
-
+        eventHandler.shutdown();
         log.info("Stopped");
         log.info("Stopped");
     }
     }
 
 
@@ -308,6 +313,10 @@ public class ConsistentDeviceMastershipStore
 
 
         @Override
         @Override
         public void event(LeadershipEvent event) {
         public void event(LeadershipEvent event) {
+            eventHandler.execute(() -> handleEvent(event));
+        }
+
+        private void handleEvent(LeadershipEvent event) {
             Leadership leadership = event.subject();
             Leadership leadership = event.subject();
             DeviceId deviceId = extractDeviceIdFromTopic(leadership.topic());
             DeviceId deviceId = extractDeviceIdFromTopic(leadership.topic());
             RoleInfo roleInfo = getNodes(deviceId);
             RoleInfo roleInfo = getNodes(deviceId);