Split the thread pool to local thread pool for parallel stream on netconf provider
Basically, parallelStream() use the common thread pool to works.
So, using the common thread pool will affect to other code or be affected by other code.
This patch avoid above situation so that improve response time.
Change-Id: I7b5a3749e5a536ed098dc71a5c450d292f53615c
diff --git a/providers/netconf/device/src/main/java/org/onosproject/provider/netconf/device/impl/NetconfDeviceProvider.java b/providers/netconf/device/src/main/java/org/onosproject/provider/netconf/device/impl/NetconfDeviceProvider.java
index fe6060d..d09a352 100644
--- a/providers/netconf/device/src/main/java/org/onosproject/provider/netconf/device/impl/NetconfDeviceProvider.java
+++ b/providers/netconf/device/src/main/java/org/onosproject/provider/netconf/device/impl/NetconfDeviceProvider.java
@@ -85,9 +85,11 @@
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@ -193,6 +195,7 @@
private InternalDeviceListener deviceListener = new InternalDeviceListener();
private boolean active;
+ private ForkJoinPool scheduledTaskPool = new ForkJoinPool(CORE_POOL_SIZE);
@Activate
public void activate(ComponentContext context) {
@@ -412,12 +415,20 @@
//updating keys and device info
private void checkAndUpdateDevices() {
Set<DeviceId> deviceSubjects = cfgService.getSubjects(DeviceId.class, NetconfDeviceConfig.class);
- deviceSubjects.parallelStream().forEach(deviceId -> {
- log.debug("check and update {}", deviceId);
- NetconfDeviceConfig config = cfgService.getConfig(deviceId, NetconfDeviceConfig.class);
- storeDeviceKey(config.sshKey(), config.username(), config.password(), deviceId);
- discoverOrUpdatePorts(deviceId);
- });
+ try {
+ scheduledTaskPool.submit(() -> {
+ deviceSubjects.parallelStream().forEach(deviceId -> {
+ log.debug("check and update {}", deviceId);
+ NetconfDeviceConfig config = cfgService.getConfig(deviceId, NetconfDeviceConfig.class);
+ storeDeviceKey(config.sshKey(), config.username(), config.password(), deviceId);
+ discoverOrUpdatePorts(deviceId);
+ });
+ }).get();
+ } catch (ExecutionException e) {
+ log.error("Can't update the devices due to {}", e.getMessage());
+ } catch (InterruptedException | CancellationException e) {
+ log.info("Update device is cancelled due to {}", e.getMessage());
+ }
}
//Saving device keys in the store