Make sure message handling executor references are appropriately managed when we…

… modify the service configuration Change-Id: I56866dd8c3359de0fcf827fc247024c65a63c5c2

Make sure message handling executor references are appropriately managed when we…
… modify the service configuration Change-Id: I56866dd8c3359de0fcf827fc247024c65a63c5c2
Madan Jampani · Gerrit Code Review
Commit cd48bfd694249c9852d94583f74a39078e12f4cd cd48bfd6 1 parent b5dcc51d
Showing 2 changed files with 70 additions and 54 deletions
core/store/dist/src/main/java/org/onosproject/store/flow/impl/DistributedFlowRuleStore.java
utils/netty/src/main/java/org/onlab/netty/NettyMessagingService.java
--- a/core/store/dist/src/main/java/org/onosproject/store/flow/impl/DistributedFlowRuleStore.java
View file @cd48bfd
+++ b/core/store/dist/src/main/java/org/onosproject/store/flow/impl/DistributedFlowRuleStore.java
View file @cd48bfd
@@ -180,6 +180,8 @@ public class DistributedFlowRuleStore
     private IdGenerator idGenerator;
+    private NodeId local;
+
     @Activate
     public void activate(ComponentContext context) {
         configService.registerProperties(getClass());
@@ -188,17 +190,76 @@ public class DistributedFlowRuleStore
         idGenerator = coreService.getIdGenerator(FlowRuleService.FLOW_OP_TOPIC);
+        local = clusterService.getLocalNode().id();
+
         // Cache to create SMap on demand
         smaps = CacheBuilder.newBuilder()
                 .softValues()
                 .build(new SMapLoader());
-        final NodeId local = clusterService.getLocalNode().id();
+        messageHandlingExecutor = Executors.newFixedThreadPool(
+                msgHandlerPoolSize, groupedThreads("onos/store/flow", "message-handlers"));
+
+        registerMessageHandlers(messageHandlingExecutor);
+
+        replicaInfoEventListener = new InternalReplicaInfoEventListener();
+
+        replicaInfoManager.addListener(replicaInfoEventListener);
+
+        logConfig("Started");
+    }
+    @Deactivate
+    public void deactivate(ComponentContext context) {
+        configService.unregisterProperties(getClass(), false);
+        unregisterMessageHandlers();
+        messageHandlingExecutor.shutdownNow();
+        replicaInfoManager.removeListener(replicaInfoEventListener);
+        log.info("Stopped");
+    }
+
+    @Modified
+    public void modified(ComponentContext context) {
+        if (context == null) {
+            backupEnabled = DEFAULT_BACKUP_ENABLED;
+            logConfig("Default config");
+            return;
+        }
+
+        Dictionary properties = context.getProperties();
+        int newPoolSize;
+        boolean newBackupEnabled;
+        try {
+            String s = get(properties, "msgHandlerPoolSize");
+            newPoolSize = isNullOrEmpty(s) ? msgHandlerPoolSize : Integer.parseInt(s.trim());
+
+            s = get(properties, "backupEnabled");
+            newBackupEnabled = isNullOrEmpty(s) ? backupEnabled : Boolean.parseBoolean(s.trim());
+
+        } catch (NumberFormatException | ClassCastException e) {
+            newPoolSize = MESSAGE_HANDLER_THREAD_POOL_SIZE;
+            newBackupEnabled = DEFAULT_BACKUP_ENABLED;
+        }
+
+        if (newBackupEnabled != backupEnabled) {
+            backupEnabled = newBackupEnabled;
+        }
+        if (newPoolSize != msgHandlerPoolSize) {
+            msgHandlerPoolSize = newPoolSize;
+            ExecutorService oldMsgHandler = messageHandlingExecutor;
             messageHandlingExecutor = Executors.newFixedThreadPool(
                     msgHandlerPoolSize, groupedThreads("onos/store/flow", "message-handlers"));
-        clusterCommunicator.addSubscriber(APPLY_BATCH_FLOWS, new OnStoreBatch(local), messageHandlingExecutor);
+            // replace previously registered handlers.
+            registerMessageHandlers(messageHandlingExecutor);
+            oldMsgHandler.shutdown();
+        }
+        logConfig("Reconfigured");
+    }
+
+    private void registerMessageHandlers(ExecutorService executor) {
+
+        clusterCommunicator.addSubscriber(APPLY_BATCH_FLOWS, new OnStoreBatch(local), executor);
         clusterCommunicator.addSubscriber(REMOTE_APPLY_COMPLETED, new ClusterMessageHandler() {
             @Override
@@ -207,7 +268,7 @@ public class DistributedFlowRuleStore
                 log.trace("received completed notification for {}", event);
                 notifyDelegate(event);
             }
-        }, messageHandlingExecutor);
+        }, executor);
         clusterCommunicator.addSubscriber(GET_FLOW_ENTRY, new ClusterMessageHandler() {
@@ -222,7 +283,7 @@ public class DistributedFlowRuleStore
                     log.error("Failed to respond back", e);
                 }
             }
-        }, messageHandlingExecutor);
+        }, executor);
         clusterCommunicator.addSubscriber(GET_DEVICE_FLOW_ENTRIES, new ClusterMessageHandler() {
@@ -237,7 +298,7 @@ public class DistributedFlowRuleStore
                     log.error("Failed to respond to peer's getFlowEntries request", e);
                 }
             }
-        }, messageHandlingExecutor);
+        }, executor);
         clusterCommunicator.addSubscriber(REMOVE_FLOW_ENTRY, new ClusterMessageHandler() {
@@ -252,60 +313,15 @@ public class DistributedFlowRuleStore
                     log.error("Failed to respond back", e);
                 }
             }
-        }, messageHandlingExecutor);
+        }, executor);
-
-        replicaInfoEventListener = new InternalReplicaInfoEventListener();
-
-        replicaInfoManager.addListener(replicaInfoEventListener);
-
-        logConfig("Started");
     }
-    @Deactivate
+    private void unregisterMessageHandlers() {
-    public void deactivate(ComponentContext context) {
-        configService.unregisterProperties(getClass(), false);
         clusterCommunicator.removeSubscriber(REMOVE_FLOW_ENTRY);
         clusterCommunicator.removeSubscriber(GET_DEVICE_FLOW_ENTRIES);
         clusterCommunicator.removeSubscriber(GET_FLOW_ENTRY);
         clusterCommunicator.removeSubscriber(APPLY_BATCH_FLOWS);
         clusterCommunicator.removeSubscriber(REMOTE_APPLY_COMPLETED);
-        messageHandlingExecutor.shutdown();
-        replicaInfoManager.removeListener(replicaInfoEventListener);
-        log.info("Stopped");
-    }
-
-    @Modified
-    public void modified(ComponentContext context) {
-        if (context == null) {
-            backupEnabled = DEFAULT_BACKUP_ENABLED;
-            logConfig("Default config");
-            return;
-        }
-
-        Dictionary properties = context.getProperties();
-        int newPoolSize;
-        boolean newBackupEnabled;
-        try {
-            String s = get(properties, "msgHandlerPoolSize");
-            newPoolSize = isNullOrEmpty(s) ? msgHandlerPoolSize : Integer.parseInt(s.trim());
-
-            s = get(properties, "backupEnabled");
-            newBackupEnabled = isNullOrEmpty(s) ? backupEnabled : Boolean.parseBoolean(s.trim());
-
-        } catch (NumberFormatException | ClassCastException e) {
-            newPoolSize = MESSAGE_HANDLER_THREAD_POOL_SIZE;
-            newBackupEnabled = DEFAULT_BACKUP_ENABLED;
-        }
-
-        if (newPoolSize != msgHandlerPoolSize || newBackupEnabled != backupEnabled) {
-            msgHandlerPoolSize = newPoolSize;
-            backupEnabled = newBackupEnabled;
-            ExecutorService oldMsgHandler = messageHandlingExecutor;
-            messageHandlingExecutor = Executors.newFixedThreadPool(
-                    msgHandlerPoolSize, groupedThreads("onos/store/flow", "message-handlers"));
-            oldMsgHandler.shutdown();
-            logConfig("Reconfigured");
-        }
     }
     private void logConfig(String prefix) {
--- a/utils/netty/src/main/java/org/onlab/netty/NettyMessagingService.java
View file @cd48bfd
+++ b/utils/netty/src/main/java/org/onlab/netty/NettyMessagingService.java
View file @cd48bfd
@@ -200,12 +200,12 @@ public class NettyMessagingService implements MessagingService {
     @Override
     public void registerHandler(String type, MessageHandler handler) {
-        handlers.putIfAbsent(type, handler);
+        handlers.put(type, handler);
     }
     @Override
     public void registerHandler(String type, MessageHandler handler, ExecutorService executor) {
-        handlers.putIfAbsent(type, new MessageHandler() {
+        handlers.put(type, new MessageHandler() {
             @Override
             public void handle(Message message) throws IOException {
                 executor.submit(() -> {