Thomas Vachuska
Committed by Gerrit Code Review

Fixed a slew of shutdown exceptions that arose due to improper or out-of-order r…

…esource clean-up, e.g. listeners, timers, executors.

Change-Id: I37c351c4202b32e92c076d9d566b96d7ff8d313a
...@@ -114,11 +114,8 @@ public class ComponentConfigManager implements ComponentConfigService { ...@@ -114,11 +114,8 @@ public class ComponentConfigManager implements ComponentConfigService {
114 String componentName = componentClass.getName(); 114 String componentName = componentClass.getName();
115 checkNotNull(componentName, COMPONENT_NULL); 115 checkNotNull(componentName, COMPONENT_NULL);
116 Map<String, ConfigProperty> cps = properties.remove(componentName); 116 Map<String, ConfigProperty> cps = properties.remove(componentName);
117 - if (cps != null) { 117 + if (clear && cps != null) {
118 cps.keySet().forEach(name -> store.unsetProperty(componentName, name)); 118 cps.keySet().forEach(name -> store.unsetProperty(componentName, name));
119 - }
120 -
121 - if (clear) {
122 clearExistingValues(componentName); 119 clearExistingValues(componentName);
123 } 120 }
124 } 121 }
......
...@@ -95,11 +95,11 @@ public class ProxyArpManager implements ProxyArpService { ...@@ -95,11 +95,11 @@ public class ProxyArpManager implements ProxyArpService {
95 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) 95 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
96 protected DeviceService deviceService; 96 protected DeviceService deviceService;
97 97
98 - private final Multimap<Device, PortNumber> internalPorts = 98 + private final Multimap<Device, PortNumber> internalPorts = HashMultimap.create();
99 - HashMultimap.<Device, PortNumber>create(); 99 + private final Multimap<Device, PortNumber> externalPorts = HashMultimap.create();
100 100
101 - private final Multimap<Device, PortNumber> externalPorts = 101 + private final DeviceListener deviceListener = new InternalDeviceListener();
102 - HashMultimap.<Device, PortNumber>create(); 102 + private final InternalLinkListener linkListener = new InternalLinkListener();
103 103
104 /** 104 /**
105 * Listens to both device service and link service to determine 105 * Listens to both device service and link service to determine
...@@ -107,16 +107,17 @@ public class ProxyArpManager implements ProxyArpService { ...@@ -107,16 +107,17 @@ public class ProxyArpManager implements ProxyArpService {
107 */ 107 */
108 @Activate 108 @Activate
109 public void activate() { 109 public void activate() {
110 - deviceService.addListener(new InternalDeviceListener()); 110 + deviceService.addListener(deviceListener);
111 - linkService.addListener(new InternalLinkListener()); 111 + linkService.addListener(linkListener);
112 determinePortLocations(); 112 determinePortLocations();
113 -
114 log.info("Started"); 113 log.info("Started");
115 } 114 }
116 115
117 116
118 @Deactivate 117 @Deactivate
119 public void deactivate() { 118 public void deactivate() {
119 + deviceService.removeListener(deviceListener);
120 + linkService.removeListener(linkListener);
120 log.info("Stopped"); 121 log.info("Stopped");
121 } 122 }
122 123
......
...@@ -96,14 +96,14 @@ public abstract class AbstractOpenFlowSwitch extends AbstractHandlerBehaviour ...@@ -96,14 +96,14 @@ public abstract class AbstractOpenFlowSwitch extends AbstractHandlerBehaviour
96 96
97 @Override 97 @Override
98 public final void sendMsg(OFMessage m) { 98 public final void sendMsg(OFMessage m) {
99 - if (role == RoleState.MASTER) { 99 + if (role == RoleState.MASTER && channel.isWritable()) {
100 channel.write(Collections.singletonList(m)); 100 channel.write(Collections.singletonList(m));
101 } 101 }
102 } 102 }
103 103
104 @Override 104 @Override
105 public final void sendMsg(List<OFMessage> msgs) { 105 public final void sendMsg(List<OFMessage> msgs) {
106 - if (role == RoleState.MASTER) { 106 + if (role == RoleState.MASTER && channel.isWritable()) {
107 channel.write(msgs); 107 channel.write(msgs);
108 } 108 }
109 } 109 }
......
...@@ -73,6 +73,8 @@ import static org.slf4j.LoggerFactory.getLogger; ...@@ -73,6 +73,8 @@ import static org.slf4j.LoggerFactory.getLogger;
73 @Component(immediate = true) 73 @Component(immediate = true)
74 public class LLDPLinkProvider extends AbstractProvider implements LinkProvider { 74 public class LLDPLinkProvider extends AbstractProvider implements LinkProvider {
75 75
76 + private static final String PROVIDER_NAME = "org.onosproject.provider.lldp";
77 +
76 private static final String PROP_USE_BDDP = "useBDDP"; 78 private static final String PROP_USE_BDDP = "useBDDP";
77 private static final String PROP_DISABLE_LD = "disableLinkDiscovery"; 79 private static final String PROP_DISABLE_LD = "disableLinkDiscovery";
78 private static final String PROP_LLDP_SUPPRESSION = "lldpSuppression"; 80 private static final String PROP_LLDP_SUPPRESSION = "lldpSuppression";
...@@ -132,13 +134,13 @@ public class LLDPLinkProvider extends AbstractProvider implements LinkProvider { ...@@ -132,13 +134,13 @@ public class LLDPLinkProvider extends AbstractProvider implements LinkProvider {
132 * Creates an OpenFlow link provider. 134 * Creates an OpenFlow link provider.
133 */ 135 */
134 public LLDPLinkProvider() { 136 public LLDPLinkProvider() {
135 - super(new ProviderId("lldp", "org.onosproject.provider.lldp")); 137 + super(new ProviderId("lldp", PROVIDER_NAME));
136 } 138 }
137 139
138 @Activate 140 @Activate
139 public void activate(ComponentContext context) { 141 public void activate(ComponentContext context) {
140 cfgService.registerProperties(getClass()); 142 cfgService.registerProperties(getClass());
141 - appId = coreService.registerApplication("org.onosproject.provider.lldp"); 143 + appId = coreService.registerApplication(PROVIDER_NAME);
142 144
143 // to load configuration at startup 145 // to load configuration at startup
144 modified(context); 146 modified(context);
...@@ -188,14 +190,14 @@ public class LLDPLinkProvider extends AbstractProvider implements LinkProvider { ...@@ -188,14 +190,14 @@ public class LLDPLinkProvider extends AbstractProvider implements LinkProvider {
188 if (disableLinkDiscovery) { 190 if (disableLinkDiscovery) {
189 return; 191 return;
190 } 192 }
191 - executor.shutdownNow();
192 - for (LinkDiscovery ld : discoverers.values()) {
193 - ld.stop();
194 - }
195 providerRegistry.unregister(this); 193 providerRegistry.unregister(this);
196 deviceService.removeListener(listener); 194 deviceService.removeListener(listener);
197 packetService.removeProcessor(listener); 195 packetService.removeProcessor(listener);
198 masterService.removeListener(roleListener); 196 masterService.removeListener(roleListener);
197 +
198 + executor.shutdownNow();
199 + discoverers.values().forEach(LinkDiscovery::stop);
200 + discoverers.clear();
199 providerService = null; 201 providerService = null;
200 202
201 log.info("Stopped"); 203 log.info("Stopped");
......
...@@ -104,8 +104,8 @@ public class LinkDiscovery implements TimerTask { ...@@ -104,8 +104,8 @@ public class LinkDiscovery implements TimerTask {
104 this.pktService = pktService; 104 this.pktService = pktService;
105 105
106 this.mastershipService = checkNotNull(masterService, "WTF!"); 106 this.mastershipService = checkNotNull(masterService, "WTF!");
107 - this.slowPorts = Collections.synchronizedSet(new HashSet<Long>()); 107 + this.slowPorts = Collections.synchronizedSet(new HashSet<>());
108 - this.fastPorts = Collections.synchronizedSet(new HashSet<Long>()); 108 + this.fastPorts = Collections.synchronizedSet(new HashSet<>());
109 this.portProbeCount = new HashMap<>(); 109 this.portProbeCount = new HashMap<>();
110 this.lldpPacket = new ONOSLLDP(); 110 this.lldpPacket = new ONOSLLDP();
111 this.lldpPacket.setChassisId(device.chassisId()); 111 this.lldpPacket.setChassisId(device.chassisId());
...@@ -296,14 +296,14 @@ public class LinkDiscovery implements TimerTask { ...@@ -296,14 +296,14 @@ public class LinkDiscovery implements TimerTask {
296 } 296 }
297 297
298 public synchronized void stop() { 298 public synchronized void stop() {
299 - timeout.cancel();
300 isStopped = true; 299 isStopped = true;
300 + timeout.cancel();
301 } 301 }
302 302
303 public synchronized void start() { 303 public synchronized void start() {
304 if (isStopped) { 304 if (isStopped) {
305 - timeout = Timer.getTimer().newTimeout(this, 0, MILLISECONDS);
306 isStopped = false; 305 isStopped = false;
306 + timeout = Timer.getTimer().newTimeout(this, 0, MILLISECONDS);
307 } else { 307 } else {
308 log.warn("LinkDiscovery started multiple times?"); 308 log.warn("LinkDiscovery started multiple times?");
309 } 309 }
...@@ -361,8 +361,8 @@ public class LinkDiscovery implements TimerTask { ...@@ -361,8 +361,8 @@ public class LinkDiscovery implements TimerTask {
361 return slowPorts.contains(portNumber) || fastPorts.contains(portNumber); 361 return slowPorts.contains(portNumber) || fastPorts.contains(portNumber);
362 } 362 }
363 363
364 - public boolean isStopped() { 364 + public synchronized boolean isStopped() {
365 - return isStopped; 365 + return isStopped || timeout.isCancelled();
366 } 366 }
367 367
368 } 368 }
......
...@@ -268,8 +268,8 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr ...@@ -268,8 +268,8 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
268 providerService.deviceConnected(did, description); 268 providerService.deviceConnected(did, description);
269 providerService.updatePorts(did, buildPortDescriptions(sw)); 269 providerService.updatePorts(did, buildPortDescriptions(sw));
270 270
271 - PortStatsCollector psc = new PortStatsCollector( 271 + PortStatsCollector psc =
272 - controller.getSwitch(dpid), POLL_INTERVAL); 272 + new PortStatsCollector(controller.getSwitch(dpid), POLL_INTERVAL);
273 psc.start(); 273 psc.start();
274 collectors.put(dpid, psc); 274 collectors.put(dpid, psc);
275 } 275 }
...@@ -314,7 +314,7 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr ...@@ -314,7 +314,7 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
314 /** 314 /**
315 * Translates a RoleState to the corresponding MastershipRole. 315 * Translates a RoleState to the corresponding MastershipRole.
316 * 316 *
317 - * @param response 317 + * @param response role state
318 * @return a MastershipRole 318 * @return a MastershipRole
319 */ 319 */
320 private MastershipRole roleOf(RoleState response) { 320 private MastershipRole roleOf(RoleState response) {
...@@ -334,7 +334,6 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr ...@@ -334,7 +334,6 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
334 /** 334 /**
335 * Builds a list of port descriptions for a given list of ports. 335 * Builds a list of port descriptions for a given list of ports.
336 * 336 *
337 - * @param ports the list of ports
338 * @return list of portdescriptions 337 * @return list of portdescriptions
339 */ 338 */
340 private List<PortDescription> buildPortDescriptions(OpenFlowSwitch sw) { 339 private List<PortDescription> buildPortDescriptions(OpenFlowSwitch sw) {
......
...@@ -45,8 +45,7 @@ public class PortStatsCollector implements TimerTask { ...@@ -45,8 +45,7 @@ public class PortStatsCollector implements TimerTask {
45 private final AtomicLong xidAtomic = new AtomicLong(1); 45 private final AtomicLong xidAtomic = new AtomicLong(1);
46 46
47 private Timeout timeout; 47 private Timeout timeout;
48 - 48 + private volatile boolean stopped;
49 - private boolean stopTimer = false;
50 49
51 /** 50 /**
52 * Creates a GroupStatsCollector object. 51 * Creates a GroupStatsCollector object.
...@@ -60,23 +59,22 @@ public class PortStatsCollector implements TimerTask { ...@@ -60,23 +59,22 @@ public class PortStatsCollector implements TimerTask {
60 } 59 }
61 60
62 @Override 61 @Override
63 - public void run(Timeout timeout) throws Exception { 62 + public void run(Timeout to) throws Exception {
63 + if (stopped || timeout.isCancelled()) {
64 + return;
65 + }
64 log.trace("Collecting stats for {}", sw.getStringId()); 66 log.trace("Collecting stats for {}", sw.getStringId());
65 67
66 sendPortStatistic(); 68 sendPortStatistic();
67 69
68 - if (!this.stopTimer) { 70 + if (!stopped && !timeout.isCancelled()) {
69 log.trace("Scheduling stats collection in {} seconds for {}", 71 log.trace("Scheduling stats collection in {} seconds for {}",
70 this.refreshInterval, this.sw.getStringId()); 72 this.refreshInterval, this.sw.getStringId());
71 - timeout.getTimer().newTimeout(this, refreshInterval, 73 + timeout.getTimer().newTimeout(this, refreshInterval, TimeUnit.SECONDS);
72 - TimeUnit.SECONDS);
73 } 74 }
74 } 75 }
75 76
76 private void sendPortStatistic() { 77 private void sendPortStatistic() {
77 - if (log.isTraceEnabled()) {
78 - log.trace("sendGroupStatistics {}:{}", sw.getStringId(), sw.getRole());
79 - }
80 if (sw.getRole() != RoleState.MASTER) { 78 if (sw.getRole() != RoleState.MASTER) {
81 return; 79 return;
82 } 80 }
...@@ -91,17 +89,18 @@ public class PortStatsCollector implements TimerTask { ...@@ -91,17 +89,18 @@ public class PortStatsCollector implements TimerTask {
91 /** 89 /**
92 * Starts the collector. 90 * Starts the collector.
93 */ 91 */
94 - public void start() { 92 + public synchronized void start() {
95 log.info("Starting Port Stats collection thread for {}", sw.getStringId()); 93 log.info("Starting Port Stats collection thread for {}", sw.getStringId());
94 + stopped = false;
96 timeout = timer.newTimeout(this, 1, TimeUnit.SECONDS); 95 timeout = timer.newTimeout(this, 1, TimeUnit.SECONDS);
97 } 96 }
98 97
99 /** 98 /**
100 * Stops the collector. 99 * Stops the collector.
101 */ 100 */
102 - public void stop() { 101 + public synchronized void stop() {
103 log.info("Stopping Port Stats collection thread for {}", sw.getStringId()); 102 log.info("Stopping Port Stats collection thread for {}", sw.getStringId());
104 - this.stopTimer = true; 103 + stopped = true;
105 timeout.cancel(); 104 timeout.cancel();
106 } 105 }
107 } 106 }
......
...@@ -43,6 +43,7 @@ import java.util.concurrent.CompletableFuture; ...@@ -43,6 +43,7 @@ import java.util.concurrent.CompletableFuture;
43 import java.util.concurrent.ConcurrentHashMap; 43 import java.util.concurrent.ConcurrentHashMap;
44 import java.util.concurrent.ConcurrentMap; 44 import java.util.concurrent.ConcurrentMap;
45 import java.util.concurrent.Executor; 45 import java.util.concurrent.Executor;
46 +import java.util.concurrent.RejectedExecutionException;
46 import java.util.concurrent.TimeUnit; 47 import java.util.concurrent.TimeUnit;
47 import java.util.concurrent.TimeoutException; 48 import java.util.concurrent.TimeoutException;
48 import java.util.concurrent.atomic.AtomicLong; 49 import java.util.concurrent.atomic.AtomicLong;
...@@ -301,7 +302,11 @@ public class NettyMessagingManager implements MessagingService { ...@@ -301,7 +302,11 @@ public class NettyMessagingManager implements MessagingService {
301 302
302 @Override 303 @Override
303 protected void channelRead0(ChannelHandlerContext ctx, InternalMessage message) throws Exception { 304 protected void channelRead0(ChannelHandlerContext ctx, InternalMessage message) throws Exception {
304 - dispatchLocally(message); 305 + try {
306 + dispatchLocally(message);
307 + } catch (RejectedExecutionException e) {
308 + log.warn("Unable to dispatch message due to {}", e.getMessage());
309 + }
305 } 310 }
306 311
307 @Override 312 @Override
......
...@@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.node.ArrayNode; ...@@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.node.ArrayNode;
20 import com.fasterxml.jackson.databind.node.ObjectNode; 20 import com.fasterxml.jackson.databind.node.ObjectNode;
21 import org.eclipse.jetty.websocket.WebSocket; 21 import org.eclipse.jetty.websocket.WebSocket;
22 import org.onlab.osgi.ServiceDirectory; 22 import org.onlab.osgi.ServiceDirectory;
23 +import org.onlab.osgi.ServiceNotFoundException;
23 import org.onosproject.cluster.ClusterService; 24 import org.onosproject.cluster.ClusterService;
24 import org.onosproject.cluster.ControllerNode; 25 import org.onosproject.cluster.ControllerNode;
25 import org.onosproject.ui.UiConnection; 26 import org.onosproject.ui.UiConnection;
...@@ -100,11 +101,19 @@ public class UiWebSocket ...@@ -100,11 +101,19 @@ public class UiWebSocket
100 101
101 @Override 102 @Override
102 public void onOpen(Connection connection) { 103 public void onOpen(Connection connection) {
103 - log.info("GUI client connected");
104 this.connection = connection; 104 this.connection = connection;
105 this.control = (FrameConnection) connection; 105 this.control = (FrameConnection) connection;
106 - createHandlers(); 106 + try {
107 - sendInstanceData(); 107 + createHandlers();
108 + sendInstanceData();
109 + log.info("GUI client connected");
110 +
111 + } catch (ServiceNotFoundException e) {
112 + log.warn("Unable to open GUI connection; services have been shut-down");
113 + this.connection.close();
114 + this.connection = null;
115 + this.control = null;
116 + }
108 } 117 }
109 118
110 @Override 119 @Override
......