Fixed issue with leaking various switch-related collectors, e.g. port stats, met…

…ers, table stats, flow stats.

Change-Id: If46102708fa88cf5f251a18cb9ce09393fb95752

@@ -180,6 +180,7 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
         controller.removeListener(listener);
         providerRegistry.unregister(this);
         collectors.values().forEach(PortStatsCollector::stop);
+        collectors.clear();
         providerService = null;
         LOG.info("Stopped");
     }
@@ -374,10 +375,9 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
             providerService.deviceConnected(did, description);
             providerService.updatePorts(did, buildPortDescriptions(sw));
 
-            PortStatsCollector psc =
-                    new PortStatsCollector(sw, portStatsPollFrequency);
+            PortStatsCollector psc = new PortStatsCollector(sw, portStatsPollFrequency);
+            stopCollectorIfNeeded(collectors.put(dpid, psc));
             psc.start();
-            collectors.put(dpid, psc);
 
             //figure out race condition for collectors.remove() and collectors.put()
             if (controller.getSwitch(dpid) == null) {
@@ -385,17 +385,19 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
             }
         }
 
+        private void stopCollectorIfNeeded(PortStatsCollector collector) {
+            if (collector != null) {
+                collector.stop();
+            }
+        }
+
         @Override
         public void switchRemoved(Dpid dpid) {
             if (providerService == null) {
                 return;
             }
             providerService.deviceDisconnected(deviceId(uri(dpid)));
-
-            PortStatsCollector collector = collectors.remove(dpid);
-            if (collector != null) {
-                collector.stop();
-            }
+            stopCollectorIfNeeded(collectors.remove(dpid));
         }
 
         @Override

@@ -31,7 +31,7 @@ import static org.slf4j.LoggerFactory.getLogger;
 /**
  * Collects flow statistics for the specified switch.
  */
-class FlowStatsCollector {
+class FlowStatsCollector implements SwitchDataCollector {
 
     private final Logger log = getLogger(getClass());
 

@@ -55,7 +55,7 @@ import static org.slf4j.LoggerFactory.getLogger;
 /**
  * Efficiently and adaptively collects flow statistics for the specified switch.
  */
-public class NewAdaptiveFlowStatsCollector {
+public class NewAdaptiveFlowStatsCollector implements SwitchDataCollector {
     private final Logger log = getLogger(getClass());
 
     private static final String CHECK_AND_MOVE_LOG =

@@ -131,7 +131,6 @@ public class OpenFlowRuleProvider extends AbstractProvider
 
     // NewAdaptiveFlowStatsCollector Set
     private final Map<Dpid, NewAdaptiveFlowStatsCollector> afsCollectors = Maps.newHashMap();
-    private final Map<Dpid, FlowStatsCollector> collectors = Maps.newHashMap();
     private final Map<Dpid, TableStatisticsCollector> tableStatsCollectors = Maps.newHashMap();
 
     /**
@@ -142,7 +141,7 @@ public class OpenFlowRuleProvider extends AbstractProvider
     }
 
     @Activate
-    public void activate(ComponentContext context) {
+    protected void activate(ComponentContext context) {
         cfgService.registerProperties(getClass());
         providerService = providerRegistry.register(this);
         controller.addListener(listener);
@@ -159,7 +158,7 @@ public class OpenFlowRuleProvider extends AbstractProvider
     }
 
     @Deactivate
-    public void deactivate(ComponentContext context) {
+    protected void deactivate(ComponentContext context) {
         cfgService.unregisterProperties(getClass(), false);
         stopCollectors();
         providerRegistry.unregister(this);
@@ -169,7 +168,7 @@ public class OpenFlowRuleProvider extends AbstractProvider
     }
 
     @Modified
-    public void modified(ComponentContext context) {
+    protected void modified(ComponentContext context) {
         Dictionary<?, ?> properties = context.getProperties();
         int newFlowPollFrequency;
         try {
@@ -223,15 +222,21 @@ public class OpenFlowRuleProvider extends AbstractProvider
             NewAdaptiveFlowStatsCollector fsc =
                     new NewAdaptiveFlowStatsCollector(driverService, sw, flowPollFrequency);
             fsc.start();
-            afsCollectors.put(new Dpid(sw.getId()), fsc);
+            stopCollectorIfNeeded(afsCollectors.put(new Dpid(sw.getId()), fsc));
         } else {
             FlowStatsCollector fsc = new FlowStatsCollector(timer, sw, flowPollFrequency);
             fsc.start();
-            simpleCollectors.put(new Dpid(sw.getId()), fsc);
+            stopCollectorIfNeeded(simpleCollectors.put(new Dpid(sw.getId()), fsc));
         }
         TableStatisticsCollector tsc = new TableStatisticsCollector(timer, sw, flowPollFrequency);
         tsc.start();
-        tableStatsCollectors.put(new Dpid(sw.getId()), tsc);
+        stopCollectorIfNeeded(tableStatsCollectors.put(new Dpid(sw.getId()), tsc));
+    }
+
+    private void stopCollectorIfNeeded(SwitchDataCollector collector) {
+        if (collector != null) {
+            collector.stop();
+        }
     }
 
     private void stopCollectors() {
@@ -398,29 +403,17 @@ public class OpenFlowRuleProvider extends AbstractProvider
 
         @Override
         public void switchAdded(Dpid dpid) {
-
-            OpenFlowSwitch sw = controller.getSwitch(dpid);
-
             createCollector(controller.getSwitch(dpid));
         }
 
         @Override
         public void switchRemoved(Dpid dpid) {
             if (adaptiveFlowSampling) {
-                NewAdaptiveFlowStatsCollector collector = afsCollectors.remove(dpid);
-                if (collector != null) {
-                    collector.stop();
-                }
+                stopCollectorIfNeeded(afsCollectors.remove(dpid));
             } else {
-                FlowStatsCollector collector = simpleCollectors.remove(dpid);
-                if (collector != null) {
-                    collector.stop();
-                }
-            }
-            TableStatisticsCollector tsc = tableStatsCollectors.remove(dpid);
-            if (tsc != null) {
-                tsc.stop();
+                stopCollectorIfNeeded(simpleCollectors.remove(dpid));
             }
+            stopCollectorIfNeeded(tableStatsCollectors.remove(dpid));
         }
 
         @Override

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.provider.of.flow.impl;
+
+/**
+ * Auxiliary abstraction.
+ */
+interface SwitchDataCollector {
+
+    /**
+     * Starts the collector.
+     */
+    void start();
+
+    /**
+     * Stops the collector.
+     */
+    void stop();
+}
\ No newline at end of file

@@ -29,7 +29,7 @@ import static org.slf4j.LoggerFactory.getLogger;
 /**
  * Collects Table statistics for the specified switch.
  */
-class TableStatisticsCollector {
+class TableStatisticsCollector implements SwitchDataCollector {
 
     private final Logger log = getLogger(getClass());
 

@@ -363,10 +363,7 @@ public class OpenFlowGroupProvider extends AbstractProvider implements GroupProv
             if (isGroupSupported(sw)) {
                 GroupStatsCollector gsc = new GroupStatsCollector(sw, POLL_INTERVAL);
                 gsc.start();
-                GroupStatsCollector prevGsc = collectors.put(dpid, gsc);
-                if (prevGsc != null) {
-                    prevGsc.stop();
-                }
+                stopCollectorIfNeeded(collectors.put(dpid, gsc));
             }
 
             //figure out race condition
@@ -377,7 +374,10 @@ public class OpenFlowGroupProvider extends AbstractProvider implements GroupProv
 
         @Override
         public void switchRemoved(Dpid dpid) {
-            GroupStatsCollector collector = collectors.remove(dpid);
+            stopCollectorIfNeeded(collectors.remove(dpid));
+        }
+
+        private void stopCollectorIfNeeded(GroupStatsCollector collector) {
             if (collector != null) {
                 collector.stop();
             }

@@ -207,7 +207,13 @@ public class OpenFlowMeterProvider extends AbstractProvider implements MeterProv
         if (isMeterSupported(sw)) {
             MeterStatsCollector msc = new MeterStatsCollector(sw, POLL_INTERVAL);
             msc.start();
-            collectors.put(new Dpid(sw.getId()), msc);
+            stopCollectorIfNeeded(collectors.put(new Dpid(sw.getId()), msc));
+        }
+    }
+
+    private void stopCollectorIfNeeded(MeterStatsCollector collector) {
+        if (collector != null) {
+            collector.stop();
         }
     }
 
@@ -370,10 +376,7 @@ public class OpenFlowMeterProvider extends AbstractProvider implements MeterProv
 
         @Override
         public void switchRemoved(Dpid dpid) {
-            MeterStatsCollector msc = collectors.remove(dpid);
-            if (msc != null) {
-                msc.stop();
-            }
+            stopCollectorIfNeeded(collectors.remove(dpid));
         }
 
         @Override