홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
HIGUCHI Yuta
Committed by Gerrit Code Review
060da9a1 1 parent 4a24a3e0

Improve Executors related logging behavior 

- Specify Logger for the Executor
- Use Executor#execute instead of ExecutorService#submit for
  fire and forget type of usage.
   Note: submit() will swallow thrown Exception

Change-Id: I507b841dc3feedf4ad20a746c304518d68fb846a
Inline Side-by-side
Showing 12 changed files with 26 additions and 22 deletions
...@@ -115,7 +115,8 @@ public class EventHistoryManager ...@@ -115,7 +115,8 @@ public class EventHistoryManager
115 appId = coreService.registerApplication("org.onosproject.events"); 115 appId = coreService.registerApplication("org.onosproject.events");
116 log.debug("Registered as {}", appId); 116 log.debug("Registered as {}", appId);
117 117
118 - pruner = newSingleThreadScheduledExecutor(minPriority(groupedThreads("onos/events", "history-pruner"))); 118 + pruner = newSingleThreadScheduledExecutor(
119 + minPriority(groupedThreads("onos/events", "history-pruner", log)));
119 120
120 pruner.scheduleWithFixedDelay(this::pruneEventHistoryTask, 121 pruner.scheduleWithFixedDelay(this::pruneEventHistoryTask,
121 pruneInterval, pruneInterval, TimeUnit.SECONDS); 122 pruneInterval, pruneInterval, TimeUnit.SECONDS);
......
...@@ -125,7 +125,8 @@ public class DeviceManager ...@@ -125,7 +125,8 @@ public class DeviceManager
125 125
126 @Activate 126 @Activate
127 public void activate() { 127 public void activate() {
128 - backgroundService = newSingleThreadScheduledExecutor(groupedThreads("onos/device", "manager-background")); 128 + backgroundService = newSingleThreadScheduledExecutor(
129 + groupedThreads("onos/device", "manager-background", log));
129 localNodeId = clusterService.getLocalNode().id(); 130 localNodeId = clusterService.getLocalNode().id();
130 131
131 store.setDelegate(delegate); 132 store.setDelegate(delegate);
...@@ -499,7 +500,7 @@ public class DeviceManager ...@@ -499,7 +500,7 @@ public class DeviceManager
499 deviceId, response, mastershipService.getLocalRole(deviceId)); 500 deviceId, response, mastershipService.getLocalRole(deviceId));
500 // roleManager got the device to comply, but doesn't agree with 501 // roleManager got the device to comply, but doesn't agree with
501 // the store; use the store's view, then try to reassert. 502 // the store; use the store's view, then try to reassert.
502 - backgroundService.submit(() -> reassertRole(deviceId, mastershipService.getLocalRole(deviceId))); 503 + backgroundService.execute(() -> reassertRole(deviceId, mastershipService.getLocalRole(deviceId)));
503 return; 504 return;
504 } 505 }
505 } else { 506 } else {
...@@ -684,7 +685,7 @@ public class DeviceManager ...@@ -684,7 +685,7 @@ public class DeviceManager
684 685
685 @Override 686 @Override
686 public void event(MastershipEvent event) { 687 public void event(MastershipEvent event) {
687 - backgroundService.submit(() -> { 688 + backgroundService.execute(() -> {
688 try { 689 try {
689 handleMastershipEvent(event); 690 handleMastershipEvent(event);
690 } catch (Exception e) { 691 } catch (Exception e) {
......
...@@ -122,10 +122,10 @@ public class FlowRuleManager ...@@ -122,10 +122,10 @@ public class FlowRuleManager
122 private final FlowRuleDriverProvider defaultProvider = new FlowRuleDriverProvider(); 122 private final FlowRuleDriverProvider defaultProvider = new FlowRuleDriverProvider();
123 123
124 protected ExecutorService deviceInstallers = 124 protected ExecutorService deviceInstallers =
125 - Executors.newFixedThreadPool(32, groupedThreads("onos/flowservice", "device-installer-%d")); 125 + Executors.newFixedThreadPool(32, groupedThreads("onos/flowservice", "device-installer-%d", log));
126 126
127 protected ExecutorService operationsService = 127 protected ExecutorService operationsService =
128 - Executors.newFixedThreadPool(32, groupedThreads("onos/flowservice", "operations-%d")); 128 + Executors.newFixedThreadPool(32, groupedThreads("onos/flowservice", "operations-%d, log"));
129 129
130 private IdGenerator idGenerator; 130 private IdGenerator idGenerator;
131 131
......
...@@ -200,7 +200,7 @@ public class IntentManager ...@@ -200,7 +200,7 @@ public class IntentManager
200 if (newNumThreads != numThreads) { 200 if (newNumThreads != numThreads) {
201 numThreads = newNumThreads; 201 numThreads = newNumThreads;
202 ExecutorService oldWorkerExecutor = workerExecutor; 202 ExecutorService oldWorkerExecutor = workerExecutor;
203 - workerExecutor = newFixedThreadPool(numThreads, groupedThreads("onos/intent", "worker-%d")); 203 + workerExecutor = newFixedThreadPool(numThreads, groupedThreads("onos/intent", "worker-%d", log));
204 if (oldWorkerExecutor != null) { 204 if (oldWorkerExecutor != null) {
205 oldWorkerExecutor.shutdown(); 205 oldWorkerExecutor.shutdown();
206 } 206 }
......
...@@ -138,11 +138,11 @@ final class ResourceDeviceListener implements DeviceListener { ...@@ -138,11 +138,11 @@ final class ResourceDeviceListener implements DeviceListener {
138 } 138 }
139 139
140 private void registerDeviceResource(Device device) { 140 private void registerDeviceResource(Device device) {
141 - executor.submit(() -> adminService.register(Resources.discrete(device.id()).resource())); 141 + executor.execute(() -> adminService.register(Resources.discrete(device.id()).resource()));
142 } 142 }
143 143
144 private void unregisterDeviceResource(Device device) { 144 private void unregisterDeviceResource(Device device) {
145 - executor.submit(() -> { 145 + executor.execute(() -> {
146 DiscreteResource devResource = Resources.discrete(device.id()).resource(); 146 DiscreteResource devResource = Resources.discrete(device.id()).resource();
147 List<Resource> allResources = getDescendantResources(devResource); 147 List<Resource> allResources = getDescendantResources(devResource);
148 adminService.unregister(Lists.transform(allResources, Resource::id)); 148 adminService.unregister(Lists.transform(allResources, Resource::id));
...@@ -151,7 +151,7 @@ final class ResourceDeviceListener implements DeviceListener { ...@@ -151,7 +151,7 @@ final class ResourceDeviceListener implements DeviceListener {
151 151
152 private void registerPortResource(Device device, Port port) { 152 private void registerPortResource(Device device, Port port) {
153 Resource portPath = Resources.discrete(device.id(), port.number()).resource(); 153 Resource portPath = Resources.discrete(device.id(), port.number()).resource();
154 - executor.submit(() -> { 154 + executor.execute(() -> {
155 adminService.register(portPath); 155 adminService.register(portPath);
156 156
157 queryBandwidth(device.id(), port.number()) 157 queryBandwidth(device.id(), port.number())
...@@ -198,7 +198,7 @@ final class ResourceDeviceListener implements DeviceListener { ...@@ -198,7 +198,7 @@ final class ResourceDeviceListener implements DeviceListener {
198 } 198 }
199 199
200 private void unregisterPortResource(Device device, Port port) { 200 private void unregisterPortResource(Device device, Port port) {
201 - executor.submit(() -> { 201 + executor.execute(() -> {
202 DiscreteResource portResource = Resources.discrete(device.id(), port.number()).resource(); 202 DiscreteResource portResource = Resources.discrete(device.id(), port.number()).resource();
203 List<Resource> allResources = getDescendantResources(portResource); 203 List<Resource> allResources = getDescendantResources(portResource);
204 adminService.unregister(Lists.transform(allResources, Resource::id)); 204 adminService.unregister(Lists.transform(allResources, Resource::id));
......
...@@ -89,7 +89,7 @@ final class ResourceNetworkConfigListener implements NetworkConfigListener { ...@@ -89,7 +89,7 @@ final class ResourceNetworkConfigListener implements NetworkConfigListener {
89 @Override 89 @Override
90 public void event(NetworkConfigEvent event) { 90 public void event(NetworkConfigEvent event) {
91 if (event.configClass() == BandwidthCapacity.class) { 91 if (event.configClass() == BandwidthCapacity.class) {
92 - executor.submit(() -> { 92 + executor.execute(() -> {
93 try { 93 try {
94 handleBandwidthCapacity(event); 94 handleBandwidthCapacity(event);
95 } catch (Exception e) { 95 } catch (Exception e) {
......
...@@ -84,7 +84,7 @@ public final class ResourceRegistrar { ...@@ -84,7 +84,7 @@ public final class ResourceRegistrar {
84 private DeviceListener deviceListener; 84 private DeviceListener deviceListener;
85 85
86 private final ExecutorService executor = 86 private final ExecutorService executor =
87 - Executors.newSingleThreadExecutor(groupedThreads("onos/resource", "registrar")); 87 + Executors.newSingleThreadExecutor(groupedThreads("onos/resource", "registrar", log));
88 88
89 private NetworkConfigListener cfgListener; 89 private NetworkConfigListener cfgListener;
90 90
......
...@@ -68,6 +68,7 @@ import java.util.function.Function; ...@@ -68,6 +68,7 @@ import java.util.function.Function;
68 import static com.google.common.collect.Multimaps.newSetMultimap; 68 import static com.google.common.collect.Multimaps.newSetMultimap;
69 import static com.google.common.collect.Multimaps.synchronizedSetMultimap; 69 import static com.google.common.collect.Multimaps.synchronizedSetMultimap;
70 import static com.google.common.io.ByteStreams.toByteArray; 70 import static com.google.common.io.ByteStreams.toByteArray;
71 +import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
71 import static java.util.concurrent.TimeUnit.MILLISECONDS; 72 import static java.util.concurrent.TimeUnit.MILLISECONDS;
72 import static org.onlab.util.Tools.groupedThreads; 73 import static org.onlab.util.Tools.groupedThreads;
73 import static org.onlab.util.Tools.randomDelay; 74 import static org.onlab.util.Tools.randomDelay;
...@@ -138,10 +139,10 @@ public class GossipApplicationStore extends ApplicationArchive ...@@ -138,10 +139,10 @@ public class GossipApplicationStore extends ApplicationArchive
138 .register(MultiValuedTimestamp.class) 139 .register(MultiValuedTimestamp.class)
139 .register(InternalState.class); 140 .register(InternalState.class);
140 141
141 - executor = Executors.newSingleThreadScheduledExecutor(groupedThreads("onos/app", "store")); 142 + executor = newSingleThreadScheduledExecutor(groupedThreads("onos/app", "store", log));
142 143
143 messageHandlingExecutor = Executors.newSingleThreadExecutor( 144 messageHandlingExecutor = Executors.newSingleThreadExecutor(
144 - groupedThreads("onos/store/app", "message-handler")); 145 + groupedThreads("onos/store/app", "message-handler", log));
145 146
146 clusterCommunicator.<String, byte[]>addSubscriber(APP_BITS_REQUEST, 147 clusterCommunicator.<String, byte[]>addSubscriber(APP_BITS_REQUEST,
147 bytes -> new String(bytes, Charsets.UTF_8), 148 bytes -> new String(bytes, Charsets.UTF_8),
......
...@@ -96,13 +96,13 @@ import java.util.Objects; ...@@ -96,13 +96,13 @@ import java.util.Objects;
96 import java.util.Set; 96 import java.util.Set;
97 import java.util.concurrent.ConcurrentMap; 97 import java.util.concurrent.ConcurrentMap;
98 import java.util.concurrent.ExecutorService; 98 import java.util.concurrent.ExecutorService;
99 -import java.util.concurrent.Executors;
100 import java.util.concurrent.ScheduledExecutorService; 99 import java.util.concurrent.ScheduledExecutorService;
101 import java.util.concurrent.TimeUnit; 100 import java.util.concurrent.TimeUnit;
102 101
103 import static com.google.common.base.Preconditions.checkArgument; 102 import static com.google.common.base.Preconditions.checkArgument;
104 import static com.google.common.base.Predicates.notNull; 103 import static com.google.common.base.Predicates.notNull;
105 import static com.google.common.base.Verify.verify; 104 import static com.google.common.base.Verify.verify;
105 +import static java.util.concurrent.Executors.newCachedThreadPool;
106 import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor; 106 import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
107 import static org.apache.commons.lang3.concurrent.ConcurrentUtils.createIfAbsentUnchecked; 107 import static org.apache.commons.lang3.concurrent.ConcurrentUtils.createIfAbsentUnchecked;
108 import static org.onlab.util.Tools.groupedThreads; 108 import static org.onlab.util.Tools.groupedThreads;
...@@ -200,10 +200,10 @@ public class GossipDeviceStore ...@@ -200,10 +200,10 @@ public class GossipDeviceStore
200 200
201 @Activate 201 @Activate
202 public void activate() { 202 public void activate() {
203 - executor = Executors.newCachedThreadPool(groupedThreads("onos/device", "fg-%d")); 203 + executor = newCachedThreadPool(groupedThreads("onos/device", "fg-%d", log));
204 204
205 backgroundExecutor = 205 backgroundExecutor =
206 - newSingleThreadScheduledExecutor(minPriority(groupedThreads("onos/device", "bg-%d"))); 206 + newSingleThreadScheduledExecutor(minPriority(groupedThreads("onos/device", "bg-%d", log)));
207 207
208 clusterCommunicator.addSubscriber( 208 clusterCommunicator.addSubscriber(
209 GossipDeviceStoreMessageSubjects.DEVICE_UPDATE, new InternalDeviceEventListener(), executor); 209 GossipDeviceStoreMessageSubjects.DEVICE_UPDATE, new InternalDeviceEventListener(), executor);
......
...@@ -305,7 +305,7 @@ public class NewDistributedFlowRuleStore ...@@ -305,7 +305,7 @@ public class NewDistributedFlowRuleStore
305 msgHandlerPoolSize = newPoolSize; 305 msgHandlerPoolSize = newPoolSize;
306 ExecutorService oldMsgHandler = messageHandlingExecutor; 306 ExecutorService oldMsgHandler = messageHandlingExecutor;
307 messageHandlingExecutor = Executors.newFixedThreadPool( 307 messageHandlingExecutor = Executors.newFixedThreadPool(
308 - msgHandlerPoolSize, groupedThreads("onos/store/flow", "message-handlers")); 308 + msgHandlerPoolSize, groupedThreads("onos/store/flow", "message-handlers", log));
309 309
310 // replace previously registered handlers. 310 // replace previously registered handlers.
311 registerMessageHandlers(messageHandlingExecutor); 311 registerMessageHandlers(messageHandlingExecutor);
......
...@@ -123,10 +123,10 @@ public class ConsistentDeviceMastershipStore ...@@ -123,10 +123,10 @@ public class ConsistentDeviceMastershipStore
123 public void activate() { 123 public void activate() {
124 messageHandlingExecutor = 124 messageHandlingExecutor =
125 Executors.newSingleThreadExecutor( 125 Executors.newSingleThreadExecutor(
126 - groupedThreads("onos/store/device/mastership", "message-handler")); 126 + groupedThreads("onos/store/device/mastership", "message-handler", log));
127 transferExecutor = 127 transferExecutor =
128 Executors.newSingleThreadScheduledExecutor( 128 Executors.newSingleThreadScheduledExecutor(
129 - groupedThreads("onos/store/device/mastership", "mastership-transfer-executor")); 129 + groupedThreads("onos/store/device/mastership", "mastership-transfer-executor", log));
130 clusterCommunicator.addSubscriber(ROLE_RELINQUISH_SUBJECT, 130 clusterCommunicator.addSubscriber(ROLE_RELINQUISH_SUBJECT,
131 SERIALIZER::decode, 131 SERIALIZER::decode,
132 this::relinquishLocalRole, 132 this::relinquishLocalRole,
......
...@@ -118,7 +118,7 @@ public class DistributedFlowStatisticStore implements FlowStatisticStore { ...@@ -118,7 +118,7 @@ public class DistributedFlowStatisticStore implements FlowStatisticStore {
118 118
119 messageHandlingExecutor = Executors.newFixedThreadPool( 119 messageHandlingExecutor = Executors.newFixedThreadPool(
120 messageHandlerThreadPoolSize, 120 messageHandlerThreadPoolSize,
121 - groupedThreads("onos/store/statistic", "message-handlers")); 121 + groupedThreads("onos/store/statistic", "message-handlers", log));
122 122
123 clusterCommunicator.addSubscriber( 123 clusterCommunicator.addSubscriber(
124 GET_CURRENT, SERIALIZER::decode, this::getCurrentStatisticInternal, SERIALIZER::encode, 124 GET_CURRENT, SERIALIZER::decode, this::getCurrentStatisticInternal, SERIALIZER::encode,
...@@ -200,6 +200,7 @@ public class DistributedFlowStatisticStore implements FlowStatisticStore { ...@@ -200,6 +200,7 @@ public class DistributedFlowStatisticStore implements FlowStatisticStore {
200 previous.computeIfPresent(cp, (c, e) -> { e.remove(rule); return e; }); 200 previous.computeIfPresent(cp, (c, e) -> { e.remove(rule); return e; });
201 } 201 }
202 202
203 + @Override
203 public synchronized void updateFlowStatistic(FlowEntry rule) { 204 public synchronized void updateFlowStatistic(FlowEntry rule) {
204 ConnectPoint cp = buildConnectPoint(rule); 205 ConnectPoint cp = buildConnectPoint(rule);
205 if (cp == null) { 206 if (cp == null) {
......