Committed by
Gerrit Code Review
log uncaught Exception
- Use execute instead of submit so that uncaught Exception will be dealt and logged by ExecutorService's handler. - Use component's own logger Change-Id: I761264aea00748980929b5048e111756776dd2f6
Showing
28 changed files
with
53 additions
and
50 deletions
... | @@ -128,7 +128,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { | ... | @@ -128,7 +128,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { |
128 | availableDeviceIdSet = Sets.newConcurrentHashSet(); | 128 | availableDeviceIdSet = Sets.newConcurrentHashSet(); |
129 | 129 | ||
130 | messageHandlingExecutor = Executors.newSingleThreadScheduledExecutor( | 130 | messageHandlingExecutor = Executors.newSingleThreadScheduledExecutor( |
131 | - groupedThreads("onos/app/cpman", "message-handlers")); | 131 | + groupedThreads("onos/app/cpman", "message-handlers", log)); |
132 | 132 | ||
133 | communicationService.addSubscriber(CONTROL_STATS, | 133 | communicationService.addSubscriber(CONTROL_STATS, |
134 | SERIALIZER::decode, this::handleRequest, messageHandlingExecutor); | 134 | SERIALIZER::decode, this::handleRequest, messageHandlingExecutor); | ... | ... |
... | @@ -72,7 +72,7 @@ public class HostMobility { | ... | @@ -72,7 +72,7 @@ public class HostMobility { |
72 | @Activate | 72 | @Activate |
73 | public void activate() { | 73 | public void activate() { |
74 | appId = coreService.registerApplication("org.onosproject.mobility"); | 74 | appId = coreService.registerApplication("org.onosproject.mobility"); |
75 | - eventHandler = newSingleThreadScheduledExecutor(groupedThreads("onos/app-mobility", "event-handler")); | 75 | + eventHandler = newSingleThreadScheduledExecutor(groupedThreads("onos/app-mobility", "event-handler", log)); |
76 | hostService.addListener(new InternalHostListener()); | 76 | hostService.addListener(new InternalHostListener()); |
77 | log.info("Started with Application ID {}", appId.id()); | 77 | log.info("Started with Application ID {}", appId.id()); |
78 | } | 78 | } | ... | ... |
... | @@ -143,7 +143,7 @@ public class OpenstackNodeManager implements OpenstackNodeService { | ... | @@ -143,7 +143,7 @@ public class OpenstackNodeManager implements OpenstackNodeService { |
143 | }; | 143 | }; |
144 | 144 | ||
145 | private final ExecutorService eventExecutor = | 145 | private final ExecutorService eventExecutor = |
146 | - newSingleThreadScheduledExecutor(groupedThreads("onos/openstacknode", "event-handler")); | 146 | + newSingleThreadScheduledExecutor(groupedThreads("onos/openstacknode", "event-handler", log)); |
147 | 147 | ||
148 | 148 | ||
149 | private final DeviceListener deviceListener = new InternalDeviceListener(); | 149 | private final DeviceListener deviceListener = new InternalDeviceListener(); | ... | ... |
... | @@ -120,7 +120,7 @@ public class IntentSynchronizer implements IntentSynchronizationService, | ... | @@ -120,7 +120,7 @@ public class IntentSynchronizer implements IntentSynchronizationService, |
120 | * @return executor service | 120 | * @return executor service |
121 | */ | 121 | */ |
122 | protected ExecutorService createExecutor() { | 122 | protected ExecutorService createExecutor() { |
123 | - return newSingleThreadExecutor(groupedThreads("onos/" + appId, "sync")); | 123 | + return newSingleThreadExecutor(groupedThreads("onos/" + appId, "sync", log)); |
124 | } | 124 | } |
125 | 125 | ||
126 | @Override | 126 | @Override | ... | ... |
... | @@ -57,7 +57,7 @@ public class CoreEventDispatcher extends DefaultEventSinkRegistry | ... | @@ -57,7 +57,7 @@ public class CoreEventDispatcher extends DefaultEventSinkRegistry |
57 | private final BlockingQueue<Event> events = new LinkedBlockingQueue<>(); | 57 | private final BlockingQueue<Event> events = new LinkedBlockingQueue<>(); |
58 | 58 | ||
59 | private final ExecutorService executor = | 59 | private final ExecutorService executor = |
60 | - newSingleThreadExecutor(groupedThreads("onos/event", "dispatch-%d")); | 60 | + newSingleThreadExecutor(groupedThreads("onos/event", "dispatch-%d", log)); |
61 | 61 | ||
62 | @SuppressWarnings("unchecked") | 62 | @SuppressWarnings("unchecked") |
63 | private static final Event KILL_PILL = new AbstractEvent(null, 0) { | 63 | private static final Event KILL_PILL = new AbstractEvent(null, 0) { | ... | ... |
... | @@ -125,7 +125,7 @@ public class FlowRuleManager | ... | @@ -125,7 +125,7 @@ public class FlowRuleManager |
125 | Executors.newFixedThreadPool(32, groupedThreads("onos/flowservice", "device-installer-%d", log)); | 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, log")); | 128 | + Executors.newFixedThreadPool(32, groupedThreads("onos/flowservice", "operations-%d", log)); |
129 | 129 | ||
130 | private IdGenerator idGenerator; | 130 | private IdGenerator idGenerator; |
131 | 131 | ||
... | @@ -294,7 +294,7 @@ public class FlowRuleManager | ... | @@ -294,7 +294,7 @@ public class FlowRuleManager |
294 | @Override | 294 | @Override |
295 | public void apply(FlowRuleOperations ops) { | 295 | public void apply(FlowRuleOperations ops) { |
296 | checkPermission(FLOWRULE_WRITE); | 296 | checkPermission(FLOWRULE_WRITE); |
297 | - operationsService.submit(new FlowOperationsProcessor(ops)); | 297 | + operationsService.execute(new FlowOperationsProcessor(ops)); |
298 | } | 298 | } |
299 | 299 | ||
300 | @Override | 300 | @Override |
... | @@ -623,14 +623,14 @@ public class FlowRuleManager | ... | @@ -623,14 +623,14 @@ public class FlowRuleManager |
623 | final FlowRuleBatchOperation b = new FlowRuleBatchOperation(perDeviceBatches.get(deviceId), | 623 | final FlowRuleBatchOperation b = new FlowRuleBatchOperation(perDeviceBatches.get(deviceId), |
624 | deviceId, id); | 624 | deviceId, id); |
625 | pendingFlowOperations.put(id, this); | 625 | pendingFlowOperations.put(id, this); |
626 | - deviceInstallers.submit(() -> store.storeBatch(b)); | 626 | + deviceInstallers.execute(() -> store.storeBatch(b)); |
627 | } | 627 | } |
628 | } | 628 | } |
629 | 629 | ||
630 | public void satisfy(DeviceId devId) { | 630 | public void satisfy(DeviceId devId) { |
631 | pendingDevices.remove(devId); | 631 | pendingDevices.remove(devId); |
632 | if (pendingDevices.isEmpty()) { | 632 | if (pendingDevices.isEmpty()) { |
633 | - operationsService.submit(this); | 633 | + operationsService.execute(this); |
634 | } | 634 | } |
635 | } | 635 | } |
636 | 636 | ||
... | @@ -640,7 +640,7 @@ public class FlowRuleManager | ... | @@ -640,7 +640,7 @@ public class FlowRuleManager |
640 | hasFailed.set(true); | 640 | hasFailed.set(true); |
641 | pendingDevices.remove(devId); | 641 | pendingDevices.remove(devId); |
642 | if (pendingDevices.isEmpty()) { | 642 | if (pendingDevices.isEmpty()) { |
643 | - operationsService.submit(this); | 643 | + operationsService.execute(this); |
644 | } | 644 | } |
645 | 645 | ||
646 | if (context != null) { | 646 | if (context != null) { | ... | ... |
... | @@ -134,7 +134,7 @@ public class FlowObjectiveManager implements FlowObjectiveService { | ... | @@ -134,7 +134,7 @@ public class FlowObjectiveManager implements FlowObjectiveService { |
134 | 134 | ||
135 | @Activate | 135 | @Activate |
136 | protected void activate() { | 136 | protected void activate() { |
137 | - executorService = newFixedThreadPool(4, groupedThreads("onos/objective-installer", "%d")); | 137 | + executorService = newFixedThreadPool(4, groupedThreads("onos/objective-installer", "%d", log)); |
138 | flowObjectiveStore.setDelegate(delegate); | 138 | flowObjectiveStore.setDelegate(delegate); |
139 | mastershipService.addListener(mastershipListener); | 139 | mastershipService.addListener(mastershipListener); |
140 | deviceService.addListener(deviceListener); | 140 | deviceService.addListener(deviceListener); |
... | @@ -191,7 +191,7 @@ public class FlowObjectiveManager implements FlowObjectiveService { | ... | @@ -191,7 +191,7 @@ public class FlowObjectiveManager implements FlowObjectiveService { |
191 | //Attempts to check if pipeliner is null for retry attempts | 191 | //Attempts to check if pipeliner is null for retry attempts |
192 | } else if (numAttempts < INSTALL_RETRY_ATTEMPTS) { | 192 | } else if (numAttempts < INSTALL_RETRY_ATTEMPTS) { |
193 | Thread.sleep(INSTALL_RETRY_INTERVAL); | 193 | Thread.sleep(INSTALL_RETRY_INTERVAL); |
194 | - executorService.submit(new ObjectiveInstaller(deviceId, objective, numAttempts + 1)); | 194 | + executorService.execute(new ObjectiveInstaller(deviceId, objective, numAttempts + 1)); |
195 | } else { | 195 | } else { |
196 | // Otherwise we've tried a few times and failed, report an | 196 | // Otherwise we've tried a few times and failed, report an |
197 | // error back to the user. | 197 | // error back to the user. |
... | @@ -208,7 +208,7 @@ public class FlowObjectiveManager implements FlowObjectiveService { | ... | @@ -208,7 +208,7 @@ public class FlowObjectiveManager implements FlowObjectiveService { |
208 | @Override | 208 | @Override |
209 | public void filter(DeviceId deviceId, FilteringObjective filteringObjective) { | 209 | public void filter(DeviceId deviceId, FilteringObjective filteringObjective) { |
210 | checkPermission(FLOWRULE_WRITE); | 210 | checkPermission(FLOWRULE_WRITE); |
211 | - executorService.submit(new ObjectiveInstaller(deviceId, filteringObjective)); | 211 | + executorService.execute(new ObjectiveInstaller(deviceId, filteringObjective)); |
212 | } | 212 | } |
213 | 213 | ||
214 | @Override | 214 | @Override |
... | @@ -217,14 +217,14 @@ public class FlowObjectiveManager implements FlowObjectiveService { | ... | @@ -217,14 +217,14 @@ public class FlowObjectiveManager implements FlowObjectiveService { |
217 | if (queueObjective(deviceId, forwardingObjective)) { | 217 | if (queueObjective(deviceId, forwardingObjective)) { |
218 | return; | 218 | return; |
219 | } | 219 | } |
220 | - executorService.submit(new ObjectiveInstaller(deviceId, forwardingObjective)); | 220 | + executorService.execute(new ObjectiveInstaller(deviceId, forwardingObjective)); |
221 | } | 221 | } |
222 | 222 | ||
223 | @Override | 223 | @Override |
224 | public void next(DeviceId deviceId, NextObjective nextObjective) { | 224 | public void next(DeviceId deviceId, NextObjective nextObjective) { |
225 | checkPermission(FLOWRULE_WRITE); | 225 | checkPermission(FLOWRULE_WRITE); |
226 | nextToDevice.put(nextObjective.id(), deviceId); | 226 | nextToDevice.put(nextObjective.id(), deviceId); |
227 | - executorService.submit(new ObjectiveInstaller(deviceId, nextObjective)); | 227 | + executorService.execute(new ObjectiveInstaller(deviceId, nextObjective)); |
228 | } | 228 | } |
229 | 229 | ||
230 | @Override | 230 | @Override | ... | ... |
... | @@ -201,7 +201,7 @@ public class FlowObjectiveCompositionManager implements FlowObjectiveService { | ... | @@ -201,7 +201,7 @@ public class FlowObjectiveCompositionManager implements FlowObjectiveService { |
201 | } | 201 | } |
202 | } else if (numAttempts < INSTALL_RETRY_ATTEMPTS) { | 202 | } else if (numAttempts < INSTALL_RETRY_ATTEMPTS) { |
203 | Thread.sleep(INSTALL_RETRY_INTERVAL); | 203 | Thread.sleep(INSTALL_RETRY_INTERVAL); |
204 | - executorService.submit(new ObjectiveInstaller(deviceId, objective, numAttempts + 1)); | 204 | + executorService.execute(new ObjectiveInstaller(deviceId, objective, numAttempts + 1)); |
205 | } else { | 205 | } else { |
206 | // Otherwise we've tried a few times and failed, report an | 206 | // Otherwise we've tried a few times and failed, report an |
207 | // error back to the user. | 207 | // error back to the user. |
... | @@ -221,7 +221,7 @@ public class FlowObjectiveCompositionManager implements FlowObjectiveService { | ... | @@ -221,7 +221,7 @@ public class FlowObjectiveCompositionManager implements FlowObjectiveService { |
221 | List<FilteringObjective> filteringObjectives | 221 | List<FilteringObjective> filteringObjectives |
222 | = this.deviceCompositionTreeMap.get(deviceId).updateFilter(filteringObjective); | 222 | = this.deviceCompositionTreeMap.get(deviceId).updateFilter(filteringObjective); |
223 | for (FilteringObjective tmp : filteringObjectives) { | 223 | for (FilteringObjective tmp : filteringObjectives) { |
224 | - executorService.submit(new ObjectiveInstaller(deviceId, tmp)); | 224 | + executorService.execute(new ObjectiveInstaller(deviceId, tmp)); |
225 | } | 225 | } |
226 | } | 226 | } |
227 | 227 | ||
... | @@ -235,7 +235,7 @@ public class FlowObjectiveCompositionManager implements FlowObjectiveService { | ... | @@ -235,7 +235,7 @@ public class FlowObjectiveCompositionManager implements FlowObjectiveService { |
235 | List<ForwardingObjective> forwardingObjectives | 235 | List<ForwardingObjective> forwardingObjectives |
236 | = this.deviceCompositionTreeMap.get(deviceId).updateForward(forwardingObjective); | 236 | = this.deviceCompositionTreeMap.get(deviceId).updateForward(forwardingObjective); |
237 | for (ForwardingObjective tmp : forwardingObjectives) { | 237 | for (ForwardingObjective tmp : forwardingObjectives) { |
238 | - executorService.submit(new ObjectiveInstaller(deviceId, tmp)); | 238 | + executorService.execute(new ObjectiveInstaller(deviceId, tmp)); |
239 | } | 239 | } |
240 | } | 240 | } |
241 | 241 | ||
... | @@ -245,7 +245,7 @@ public class FlowObjectiveCompositionManager implements FlowObjectiveService { | ... | @@ -245,7 +245,7 @@ public class FlowObjectiveCompositionManager implements FlowObjectiveService { |
245 | 245 | ||
246 | List<NextObjective> nextObjectives = this.deviceCompositionTreeMap.get(deviceId).updateNext(nextObjective); | 246 | List<NextObjective> nextObjectives = this.deviceCompositionTreeMap.get(deviceId).updateNext(nextObjective); |
247 | for (NextObjective tmp : nextObjectives) { | 247 | for (NextObjective tmp : nextObjectives) { |
248 | - executorService.submit(new ObjectiveInstaller(deviceId, tmp)); | 248 | + executorService.execute(new ObjectiveInstaller(deviceId, tmp)); |
249 | } | 249 | } |
250 | } | 250 | } |
251 | 251 | ... | ... |
... | @@ -90,7 +90,7 @@ public class IntentCleanup implements Runnable, IntentListener { | ... | @@ -90,7 +90,7 @@ public class IntentCleanup implements Runnable, IntentListener { |
90 | @Activate | 90 | @Activate |
91 | public void activate() { | 91 | public void activate() { |
92 | cfgService.registerProperties(getClass()); | 92 | cfgService.registerProperties(getClass()); |
93 | - executor = newSingleThreadExecutor(groupedThreads("onos/intent", "cleanup")); | 93 | + executor = newSingleThreadExecutor(groupedThreads("onos/intent", "cleanup", log)); |
94 | timer = new Timer("onos-intent-cleanup-timer"); | 94 | timer = new Timer("onos-intent-cleanup-timer"); |
95 | service.addListener(this); | 95 | service.addListener(this); |
96 | adjustRate(); | 96 | adjustRate(); |
... | @@ -149,7 +149,7 @@ public class IntentCleanup implements Runnable, IntentListener { | ... | @@ -149,7 +149,7 @@ public class IntentCleanup implements Runnable, IntentListener { |
149 | timerTask = new TimerTask() { | 149 | timerTask = new TimerTask() { |
150 | @Override | 150 | @Override |
151 | public void run() { | 151 | public void run() { |
152 | - executor.submit(IntentCleanup.this); | 152 | + executor.execute(IntentCleanup.this); |
153 | } | 153 | } |
154 | }; | 154 | }; |
155 | 155 | ... | ... |
... | @@ -149,8 +149,8 @@ public class IntentManager | ... | @@ -149,8 +149,8 @@ public class IntentManager |
149 | } | 149 | } |
150 | trackerService.setDelegate(topoDelegate); | 150 | trackerService.setDelegate(topoDelegate); |
151 | eventDispatcher.addSink(IntentEvent.class, listenerRegistry); | 151 | eventDispatcher.addSink(IntentEvent.class, listenerRegistry); |
152 | - batchExecutor = newSingleThreadExecutor(groupedThreads("onos/intent", "batch")); | 152 | + batchExecutor = newSingleThreadExecutor(groupedThreads("onos/intent", "batch", log)); |
153 | - workerExecutor = newFixedThreadPool(numThreads, groupedThreads("onos/intent", "worker-%d")); | 153 | + workerExecutor = newFixedThreadPool(numThreads, groupedThreads("onos/intent", "worker-%d", log)); |
154 | idGenerator = coreService.getIdGenerator("intent-ids"); | 154 | idGenerator = coreService.getIdGenerator("intent-ids"); |
155 | Intent.bindIdGenerator(idGenerator); | 155 | Intent.bindIdGenerator(idGenerator); |
156 | log.info("Started"); | 156 | log.info("Started"); | ... | ... |
... | @@ -116,7 +116,7 @@ public class ObjectiveTracker implements ObjectiveTrackerService { | ... | @@ -116,7 +116,7 @@ public class ObjectiveTracker implements ObjectiveTrackerService { |
116 | protected IntentPartitionService partitionService; | 116 | protected IntentPartitionService partitionService; |
117 | 117 | ||
118 | private ExecutorService executorService = | 118 | private ExecutorService executorService = |
119 | - newSingleThreadExecutor(groupedThreads("onos/intent", "objectivetracker")); | 119 | + newSingleThreadExecutor(groupedThreads("onos/intent", "objectivetracker", log)); |
120 | private ScheduledExecutorService executor = Executors | 120 | private ScheduledExecutorService executor = Executors |
121 | .newScheduledThreadPool(1); | 121 | .newScheduledThreadPool(1); |
122 | 122 | ... | ... |
... | @@ -120,7 +120,7 @@ public class PacketManager | ... | @@ -120,7 +120,7 @@ public class PacketManager |
120 | @Activate | 120 | @Activate |
121 | public void activate() { | 121 | public void activate() { |
122 | eventHandlingExecutor = Executors.newSingleThreadExecutor( | 122 | eventHandlingExecutor = Executors.newSingleThreadExecutor( |
123 | - groupedThreads("onos/net/packet", "event-handler")); | 123 | + groupedThreads("onos/net/packet", "event-handler", log)); |
124 | localNodeId = clusterService.getLocalNode().id(); | 124 | localNodeId = clusterService.getLocalNode().id(); |
125 | appId = coreService.getAppId(CoreService.CORE_APP_NAME); | 125 | appId = coreService.getAppId(CoreService.CORE_APP_NAME); |
126 | store.setDelegate(delegate); | 126 | store.setDelegate(delegate); | ... | ... |
... | @@ -124,7 +124,7 @@ public class DefaultTopologyProvider extends AbstractProvider | ... | @@ -124,7 +124,7 @@ public class DefaultTopologyProvider extends AbstractProvider |
124 | @Activate | 124 | @Activate |
125 | public synchronized void activate(ComponentContext context) { | 125 | public synchronized void activate(ComponentContext context) { |
126 | cfgService.registerProperties(DefaultTopologyProvider.class); | 126 | cfgService.registerProperties(DefaultTopologyProvider.class); |
127 | - executor = newFixedThreadPool(MAX_THREADS, groupedThreads("onos/topo", "build-%d")); | 127 | + executor = newFixedThreadPool(MAX_THREADS, groupedThreads("onos/topo", "build-%d", log)); |
128 | accumulator = new TopologyChangeAccumulator(); | 128 | accumulator = new TopologyChangeAccumulator(); |
129 | logConfig("Configured"); | 129 | logConfig("Configured"); |
130 | 130 | ... | ... |
... | @@ -107,9 +107,9 @@ public class DistributedClusterStore | ... | @@ -107,9 +107,9 @@ public class DistributedClusterStore |
107 | private final Map<NodeId, DateTime> nodeStateLastUpdatedTimes = Maps.newConcurrentMap(); | 107 | private final Map<NodeId, DateTime> nodeStateLastUpdatedTimes = Maps.newConcurrentMap(); |
108 | 108 | ||
109 | private ScheduledExecutorService heartBeatSender = Executors.newSingleThreadScheduledExecutor( | 109 | private ScheduledExecutorService heartBeatSender = Executors.newSingleThreadScheduledExecutor( |
110 | - groupedThreads("onos/cluster/membership", "heartbeat-sender")); | 110 | + groupedThreads("onos/cluster/membership", "heartbeat-sender", log)); |
111 | private ExecutorService heartBeatMessageHandler = Executors.newSingleThreadExecutor( | 111 | private ExecutorService heartBeatMessageHandler = Executors.newSingleThreadExecutor( |
112 | - groupedThreads("onos/cluster/membership", "heartbeat-receiver")); | 112 | + groupedThreads("onos/cluster/membership", "heartbeat-receiver", log)); |
113 | 113 | ||
114 | private PhiAccrualFailureDetector failureDetector; | 114 | private PhiAccrualFailureDetector failureDetector; |
115 | 115 | ||
... | @@ -377,7 +377,7 @@ public class DistributedClusterStore | ... | @@ -377,7 +377,7 @@ public class DistributedClusterStore |
377 | try { | 377 | try { |
378 | ScheduledExecutorService prevSender = heartBeatSender; | 378 | ScheduledExecutorService prevSender = heartBeatSender; |
379 | heartBeatSender = Executors.newSingleThreadScheduledExecutor( | 379 | heartBeatSender = Executors.newSingleThreadScheduledExecutor( |
380 | - groupedThreads("onos/cluster/membership", "heartbeat-sender-%d")); | 380 | + groupedThreads("onos/cluster/membership", "heartbeat-sender-%d", log)); |
381 | heartBeatSender.scheduleWithFixedDelay(this::heartbeat, 0, | 381 | heartBeatSender.scheduleWithFixedDelay(this::heartbeat, 0, |
382 | heartbeatInterval, TimeUnit.MILLISECONDS); | 382 | heartbeatInterval, TimeUnit.MILLISECONDS); |
383 | prevSender.shutdown(); | 383 | prevSender.shutdown(); | ... | ... |
... | @@ -167,7 +167,7 @@ public class NewDistributedFlowRuleStore | ... | @@ -167,7 +167,7 @@ public class NewDistributedFlowRuleStore |
167 | 167 | ||
168 | private ScheduledFuture<?> backupTask; | 168 | private ScheduledFuture<?> backupTask; |
169 | private final ScheduledExecutorService backupSenderExecutor = | 169 | private final ScheduledExecutorService backupSenderExecutor = |
170 | - Executors.newSingleThreadScheduledExecutor(groupedThreads("onos/flow", "backup-sender")); | 170 | + Executors.newSingleThreadScheduledExecutor(groupedThreads("onos/flow", "backup-sender", log)); |
171 | 171 | ||
172 | private EventuallyConsistentMap<DeviceId, List<TableStatisticsEntry>> deviceTableStats; | 172 | private EventuallyConsistentMap<DeviceId, List<TableStatisticsEntry>> deviceTableStats; |
173 | private final EventuallyConsistentMapListener<DeviceId, List<TableStatisticsEntry>> tableStatsListener = | 173 | private final EventuallyConsistentMapListener<DeviceId, List<TableStatisticsEntry>> tableStatsListener = |
... | @@ -203,7 +203,7 @@ public class NewDistributedFlowRuleStore | ... | @@ -203,7 +203,7 @@ public class NewDistributedFlowRuleStore |
203 | local = clusterService.getLocalNode().id(); | 203 | local = clusterService.getLocalNode().id(); |
204 | 204 | ||
205 | messageHandlingExecutor = Executors.newFixedThreadPool( | 205 | messageHandlingExecutor = Executors.newFixedThreadPool( |
206 | - msgHandlerPoolSize, groupedThreads("onos/store/flow", "message-handlers")); | 206 | + msgHandlerPoolSize, groupedThreads("onos/store/flow", "message-handlers", log)); |
207 | 207 | ||
208 | registerMessageHandlers(messageHandlingExecutor); | 208 | registerMessageHandlers(messageHandlingExecutor); |
209 | 209 | ... | ... |
... | @@ -185,7 +185,8 @@ public class DistributedGroupStore | ... | @@ -185,7 +185,8 @@ public class DistributedGroupStore |
185 | messageHandlingExecutor = Executors. | 185 | messageHandlingExecutor = Executors. |
186 | newFixedThreadPool(MESSAGE_HANDLER_THREAD_POOL_SIZE, | 186 | newFixedThreadPool(MESSAGE_HANDLER_THREAD_POOL_SIZE, |
187 | groupedThreads("onos/store/group", | 187 | groupedThreads("onos/store/group", |
188 | - "message-handlers")); | 188 | + "message-handlers", |
189 | + log)); | ||
189 | 190 | ||
190 | clusterCommunicator.addSubscriber(GroupStoreMessageSubjects.REMOTE_GROUP_OP_REQUEST, | 191 | clusterCommunicator.addSubscriber(GroupStoreMessageSubjects.REMOTE_GROUP_OP_REQUEST, |
191 | clusterMsgSerializer::deserialize, | 192 | clusterMsgSerializer::deserialize, | ... | ... |
... | @@ -118,7 +118,7 @@ public class DistributedPacketStore | ... | @@ -118,7 +118,7 @@ public class DistributedPacketStore |
118 | public void activate() { | 118 | public void activate() { |
119 | messageHandlingExecutor = Executors.newFixedThreadPool( | 119 | messageHandlingExecutor = Executors.newFixedThreadPool( |
120 | messageHandlerThreadPoolSize, | 120 | messageHandlerThreadPoolSize, |
121 | - groupedThreads("onos/store/packet", "message-handlers")); | 121 | + groupedThreads("onos/store/packet", "message-handlers", log)); |
122 | 122 | ||
123 | communicationService.<OutboundPacket>addSubscriber(PACKET_OUT_SUBJECT, | 123 | communicationService.<OutboundPacket>addSubscriber(PACKET_OUT_SUBJECT, |
124 | SERIALIZER::decode, | 124 | SERIALIZER::decode, | ... | ... |
... | @@ -76,7 +76,7 @@ public class DistributedProxyArpStore implements ProxyArpStore { | ... | @@ -76,7 +76,7 @@ public class DistributedProxyArpStore implements ProxyArpStore { |
76 | private Map<HostId, ArpResponseMessage> pendingMessages = Maps.newConcurrentMap(); | 76 | private Map<HostId, ArpResponseMessage> pendingMessages = Maps.newConcurrentMap(); |
77 | 77 | ||
78 | private ExecutorService executor = | 78 | private ExecutorService executor = |
79 | - newFixedThreadPool(4, groupedThreads("onos/arp", "sender-%d")); | 79 | + newFixedThreadPool(4, groupedThreads("onos/arp", "sender-%d", log)); |
80 | 80 | ||
81 | private NodeId localNodeId; | 81 | private NodeId localNodeId; |
82 | 82 | ... | ... |
... | @@ -120,7 +120,7 @@ public class DistributedStatisticStore implements StatisticStore { | ... | @@ -120,7 +120,7 @@ public class DistributedStatisticStore implements StatisticStore { |
120 | 120 | ||
121 | messageHandlingExecutor = Executors.newFixedThreadPool( | 121 | messageHandlingExecutor = Executors.newFixedThreadPool( |
122 | messageHandlerThreadPoolSize, | 122 | messageHandlerThreadPoolSize, |
123 | - groupedThreads("onos/store/statistic", "message-handlers")); | 123 | + groupedThreads("onos/store/statistic", "message-handlers", log)); |
124 | 124 | ||
125 | clusterCommunicator.<ConnectPoint, Set<FlowEntry>>addSubscriber(GET_CURRENT, | 125 | clusterCommunicator.<ConnectPoint, Set<FlowEntry>>addSubscriber(GET_CURRENT, |
126 | SERIALIZER::decode, | 126 | SERIALIZER::decode, | ... | ... |
... | @@ -218,7 +218,7 @@ public class EventuallyConsistentMapImpl<K, V> | ... | @@ -218,7 +218,7 @@ public class EventuallyConsistentMapImpl<K, V> |
218 | } else { | 218 | } else { |
219 | // should be a normal executor; it's used for receiving messages | 219 | // should be a normal executor; it's used for receiving messages |
220 | this.executor = | 220 | this.executor = |
221 | - Executors.newFixedThreadPool(8, groupedThreads("onos/ecm", mapName + "-fg-%d")); | 221 | + Executors.newFixedThreadPool(8, groupedThreads("onos/ecm", mapName + "-fg-%d", log)); |
222 | } | 222 | } |
223 | 223 | ||
224 | if (communicationExecutor != null) { | 224 | if (communicationExecutor != null) { |
... | @@ -227,7 +227,7 @@ public class EventuallyConsistentMapImpl<K, V> | ... | @@ -227,7 +227,7 @@ public class EventuallyConsistentMapImpl<K, V> |
227 | // sending executor; should be capped | 227 | // sending executor; should be capped |
228 | //TODO this probably doesn't need to be bounded anymore | 228 | //TODO this probably doesn't need to be bounded anymore |
229 | this.communicationExecutor = | 229 | this.communicationExecutor = |
230 | - newFixedThreadPool(8, groupedThreads("onos/ecm", mapName + "-publish-%d")); | 230 | + newFixedThreadPool(8, groupedThreads("onos/ecm", mapName + "-publish-%d", log)); |
231 | } | 231 | } |
232 | 232 | ||
233 | 233 | ||
... | @@ -235,7 +235,7 @@ public class EventuallyConsistentMapImpl<K, V> | ... | @@ -235,7 +235,7 @@ public class EventuallyConsistentMapImpl<K, V> |
235 | this.backgroundExecutor = backgroundExecutor; | 235 | this.backgroundExecutor = backgroundExecutor; |
236 | } else { | 236 | } else { |
237 | this.backgroundExecutor = | 237 | this.backgroundExecutor = |
238 | - newSingleThreadScheduledExecutor(groupedThreads("onos/ecm", mapName + "-bg-%d")); | 238 | + newSingleThreadScheduledExecutor(groupedThreads("onos/ecm", mapName + "-bg-%d", log)); |
239 | } | 239 | } |
240 | 240 | ||
241 | // start anti-entropy thread | 241 | // start anti-entropy thread |
... | @@ -718,7 +718,7 @@ public class EventuallyConsistentMapImpl<K, V> | ... | @@ -718,7 +718,7 @@ public class EventuallyConsistentMapImpl<K, V> |
718 | Map<K, UpdateEntry<K, V>> map = Maps.newHashMap(); | 718 | Map<K, UpdateEntry<K, V>> map = Maps.newHashMap(); |
719 | items.forEach(item -> map.compute(item.key(), (key, existing) -> | 719 | items.forEach(item -> map.compute(item.key(), (key, existing) -> |
720 | item.isNewerThan(existing) ? item : existing)); | 720 | item.isNewerThan(existing) ? item : existing)); |
721 | - communicationExecutor.submit(() -> { | 721 | + communicationExecutor.execute(() -> { |
722 | clusterCommunicator.unicast(ImmutableList.copyOf(map.values()), | 722 | clusterCommunicator.unicast(ImmutableList.copyOf(map.values()), |
723 | updateMessageSubject, | 723 | updateMessageSubject, |
724 | serializer::encode, | 724 | serializer::encode, | ... | ... |
... | @@ -140,7 +140,7 @@ public class CentecV350Pipeline extends AbstractHandlerBehaviour implements Pipe | ... | @@ -140,7 +140,7 @@ public class CentecV350Pipeline extends AbstractHandlerBehaviour implements Pipe |
140 | 140 | ||
141 | private ScheduledExecutorService groupChecker = | 141 | private ScheduledExecutorService groupChecker = |
142 | Executors.newScheduledThreadPool(2, groupedThreads("onos/pipeliner", | 142 | Executors.newScheduledThreadPool(2, groupedThreads("onos/pipeliner", |
143 | - "centec-V350-%d")); | 143 | + "centec-V350-%d", log)); |
144 | 144 | ||
145 | @Override | 145 | @Override |
146 | public void init(DeviceId deviceId, PipelinerContext context) { | 146 | public void init(DeviceId deviceId, PipelinerContext context) { | ... | ... |
... | @@ -124,7 +124,8 @@ public class OVSCorsaPipeline extends AbstractHandlerBehaviour implements Pipeli | ... | @@ -124,7 +124,8 @@ public class OVSCorsaPipeline extends AbstractHandlerBehaviour implements Pipeli |
124 | 124 | ||
125 | private ScheduledExecutorService groupChecker = | 125 | private ScheduledExecutorService groupChecker = |
126 | Executors.newScheduledThreadPool(2, groupedThreads("onos/pipeliner", | 126 | Executors.newScheduledThreadPool(2, groupedThreads("onos/pipeliner", |
127 | - "ovs-corsa-%d")); | 127 | + "ovs-corsa-%d", |
128 | + log)); | ||
128 | 129 | ||
129 | @Override | 130 | @Override |
130 | public void init(DeviceId deviceId, PipelinerContext context) { | 131 | public void init(DeviceId deviceId, PipelinerContext context) { | ... | ... |
... | @@ -117,7 +117,7 @@ public class Ofdpa2GroupHandler { | ... | @@ -117,7 +117,7 @@ public class Ofdpa2GroupHandler { |
117 | private Cache<GroupKey, List<OfdpaNextGroup>> pendingNextObjectives; | 117 | private Cache<GroupKey, List<OfdpaNextGroup>> pendingNextObjectives; |
118 | private ConcurrentHashMap<GroupKey, Set<GroupChainElem>> pendingGroups; | 118 | private ConcurrentHashMap<GroupKey, Set<GroupChainElem>> pendingGroups; |
119 | private ScheduledExecutorService groupChecker = | 119 | private ScheduledExecutorService groupChecker = |
120 | - Executors.newScheduledThreadPool(2, groupedThreads("onos/pipeliner", "ofdpa2-%d")); | 120 | + Executors.newScheduledThreadPool(2, groupedThreads("onos/pipeliner", "ofdpa2-%d", log)); |
121 | 121 | ||
122 | // index number for group creation | 122 | // index number for group creation |
123 | private AtomicCounter nextIndex; | 123 | private AtomicCounter nextIndex; | ... | ... |
... | @@ -132,7 +132,8 @@ public class SpringOpenTTP extends AbstractHandlerBehaviour | ... | @@ -132,7 +132,8 @@ public class SpringOpenTTP extends AbstractHandlerBehaviour |
132 | private ScheduledExecutorService groupChecker = Executors | 132 | private ScheduledExecutorService groupChecker = Executors |
133 | .newScheduledThreadPool(2, | 133 | .newScheduledThreadPool(2, |
134 | groupedThreads("onos/pipeliner", | 134 | groupedThreads("onos/pipeliner", |
135 | - "spring-open-%d")); | 135 | + "spring-open-%d", |
136 | + log)); | ||
136 | protected KryoNamespace appKryo = new KryoNamespace.Builder() | 137 | protected KryoNamespace appKryo = new KryoNamespace.Builder() |
137 | .register(KryoNamespaces.API) | 138 | .register(KryoNamespaces.API) |
138 | .register(GroupKey.class) | 139 | .register(GroupKey.class) | ... | ... |
... | @@ -151,13 +151,13 @@ public class Controller { | ... | @@ -151,13 +151,13 @@ public class Controller { |
151 | 151 | ||
152 | if (workerThreads == 0) { | 152 | if (workerThreads == 0) { |
153 | execFactory = new NioServerSocketChannelFactory( | 153 | execFactory = new NioServerSocketChannelFactory( |
154 | - Executors.newCachedThreadPool(groupedThreads("onos/of", "boss-%d")), | 154 | + Executors.newCachedThreadPool(groupedThreads("onos/of", "boss-%d", log)), |
155 | - Executors.newCachedThreadPool(groupedThreads("onos/of", "worker-%d"))); | 155 | + Executors.newCachedThreadPool(groupedThreads("onos/of", "worker-%d", log))); |
156 | return new ServerBootstrap(execFactory); | 156 | return new ServerBootstrap(execFactory); |
157 | } else { | 157 | } else { |
158 | execFactory = new NioServerSocketChannelFactory( | 158 | execFactory = new NioServerSocketChannelFactory( |
159 | - Executors.newCachedThreadPool(groupedThreads("onos/of", "boss-%d")), | 159 | + Executors.newCachedThreadPool(groupedThreads("onos/of", "boss-%d", log)), |
160 | - Executors.newCachedThreadPool(groupedThreads("onos/of", "worker-%d")), workerThreads); | 160 | + Executors.newCachedThreadPool(groupedThreads("onos/of", "worker-%d", log)), workerThreads); |
161 | return new ServerBootstrap(execFactory); | 161 | return new ServerBootstrap(execFactory); |
162 | } | 162 | } |
163 | } | 163 | } | ... | ... |
... | @@ -138,7 +138,7 @@ public class HostLocationProvider extends AbstractProvider implements HostProvid | ... | @@ -138,7 +138,7 @@ public class HostLocationProvider extends AbstractProvider implements HostProvid |
138 | cfgService.registerProperties(getClass()); | 138 | cfgService.registerProperties(getClass()); |
139 | appId = coreService.registerApplication("org.onosproject.provider.host"); | 139 | appId = coreService.registerApplication("org.onosproject.provider.host"); |
140 | eventHandler = newSingleThreadScheduledExecutor( | 140 | eventHandler = newSingleThreadScheduledExecutor( |
141 | - groupedThreads("onos/host-loc-provider", "event-handler")); | 141 | + groupedThreads("onos/host-loc-provider", "event-handler", log)); |
142 | providerService = providerRegistry.register(this); | 142 | providerService = providerRegistry.register(this); |
143 | packetService.addProcessor(processor, PacketProcessor.advisor(1)); | 143 | packetService.addProcessor(processor, PacketProcessor.advisor(1)); |
144 | deviceService.addListener(deviceListener); | 144 | deviceService.addListener(deviceListener); | ... | ... |
... | @@ -338,7 +338,7 @@ public class LldpLinkProvider extends AbstractProvider implements LinkProvider { | ... | @@ -338,7 +338,7 @@ public class LldpLinkProvider extends AbstractProvider implements LinkProvider { |
338 | 338 | ||
339 | loadDevices(); | 339 | loadDevices(); |
340 | 340 | ||
341 | - executor = newSingleThreadScheduledExecutor(groupedThreads("onos/link", "discovery-%d")); | 341 | + executor = newSingleThreadScheduledExecutor(groupedThreads("onos/link", "discovery-%d", log)); |
342 | executor.scheduleAtFixedRate(new SyncDeviceInfoTask(), | 342 | executor.scheduleAtFixedRate(new SyncDeviceInfoTask(), |
343 | DEVICE_SYNC_DELAY, DEVICE_SYNC_DELAY, SECONDS); | 343 | DEVICE_SYNC_DELAY, DEVICE_SYNC_DELAY, SECONDS); |
344 | executor.scheduleAtFixedRate(new LinkPrunerTask(), | 344 | executor.scheduleAtFixedRate(new LinkPrunerTask(), | ... | ... |
... | @@ -78,7 +78,7 @@ public class NewAdaptiveFlowStatsCollector { | ... | @@ -78,7 +78,7 @@ public class NewAdaptiveFlowStatsCollector { |
78 | private final OpenFlowSwitch sw; | 78 | private final OpenFlowSwitch sw; |
79 | 79 | ||
80 | private ScheduledExecutorService adaptiveFlowStatsScheduler = | 80 | private ScheduledExecutorService adaptiveFlowStatsScheduler = |
81 | - Executors.newScheduledThreadPool(4, groupedThreads("onos/flow", "device-stats-collector-%d")); | 81 | + Executors.newScheduledThreadPool(4, groupedThreads("onos/flow", "device-stats-collector-%d", log)); |
82 | private ScheduledFuture<?> calAndShortFlowsThread; | 82 | private ScheduledFuture<?> calAndShortFlowsThread; |
83 | private ScheduledFuture<?> midFlowsThread; | 83 | private ScheduledFuture<?> midFlowsThread; |
84 | private ScheduledFuture<?> longFlowsThread; | 84 | private ScheduledFuture<?> longFlowsThread; | ... | ... |
-
Please register or login to post a comment