diff --git a/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/GeneralDeviceProvider.java b/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/GeneralDeviceProvider.java
index ad0d388..b341149 100644
--- a/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/GeneralDeviceProvider.java
+++ b/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/GeneralDeviceProvider.java
@@ -16,9 +16,13 @@
 
 package org.onosproject.provider.general.device.impl;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.annotations.Beta;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.Striped;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
@@ -32,7 +36,6 @@
 import org.onosproject.cfg.ComponentConfigService;
 import org.onosproject.cluster.ClusterService;
 import org.onosproject.cluster.LeadershipService;
-import org.onosproject.cluster.NodeId;
 import org.onosproject.core.CoreService;
 import org.onosproject.mastership.MastershipService;
 import org.onosproject.net.AnnotationKeys;
@@ -87,7 +90,6 @@
 import java.util.Dictionary;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
@@ -100,10 +102,11 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Supplier;
 
 import static java.util.concurrent.Executors.newFixedThreadPool;
 import static java.util.concurrent.Executors.newScheduledThreadPool;
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
 import static org.onlab.util.Tools.groupedThreads;
 import static org.onosproject.net.device.DeviceEvent.Type;
 import static org.slf4j.LoggerFactory.getLogger;
@@ -118,13 +121,7 @@
 public class GeneralDeviceProvider extends AbstractProvider
         implements DeviceProvider {
 
-    // Timeout in seconds for operations on devices.
-    private static final int DEVICE_OP_TIMEOUT = 10;
-
     private static final String DRIVER = "driver";
-    public static final String FIRST_CONNECTION_TOPIC = "first-connection-";
-    private static final String CHECK_CONNECTION_TOPIC = "check-connection-";
-    private static final String POLL_FREQUENCY = "pollFrequency";
 
     private final Logger log = getLogger(getClass());
 
@@ -158,17 +155,36 @@
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     private LeadershipService leadershipService;
 
-    private static final int DEFAULT_POLL_FREQUENCY_SECONDS = 10;
-    @Property(name = POLL_FREQUENCY, intValue = DEFAULT_POLL_FREQUENCY_SECONDS,
+    private static final String STATS_POLL_FREQUENCY = "deviceStatsPollFrequency";
+    private static final int DEFAULT_STATS_POLL_FREQUENCY = 10;
+    @Property(name = STATS_POLL_FREQUENCY, intValue = DEFAULT_STATS_POLL_FREQUENCY,
             label = "Configure poll frequency for port status and statistics; " +
                     "default is 10 sec")
-    private int pollFrequency = DEFAULT_POLL_FREQUENCY_SECONDS;
+    private int statsPollFrequency = DEFAULT_STATS_POLL_FREQUENCY;
 
-    private static final int DEVICE_AVAILABILITY_POLL_FREQUENCY_SECONDS = 10;
-    @Property(name = "deviceAvailabilityPollFrequency", intValue = DEVICE_AVAILABILITY_POLL_FREQUENCY_SECONDS,
-            label = "Configure poll frequency for checking device availability; " +
+    private static final String PROBE_FREQUENCY = "deviceProbeFrequency";
+    private static final int DEFAULT_PROBE_FREQUENCY = 10;
+    @Property(name = PROBE_FREQUENCY, intValue = DEFAULT_PROBE_FREQUENCY,
+            label = "Configure probe frequency for checking device availability; " +
                     "default is 10 sec")
-    private int deviceAvailabilityPollFrequency = DEVICE_AVAILABILITY_POLL_FREQUENCY_SECONDS;
+    private int probeFrequency = DEFAULT_PROBE_FREQUENCY;
+
+    private static final String OP_TIMEOUT_SHORT = "deviceOperationTimeoutShort";
+    private static final int DEFAULT_OP_TIMEOUT_SHORT = 10;
+    @Property(name = OP_TIMEOUT_SHORT, intValue = DEFAULT_OP_TIMEOUT_SHORT,
+            label = "Configure timeout in seconds for device operations " +
+                    "that are supposed to take a short time " +
+                    "(e.g. checking device reachability); default is 10 seconds")
+    private int opTimeoutShort = DEFAULT_OP_TIMEOUT_SHORT;
+
+    private static final String OP_TIMEOUT_LONG = "deviceOperationTimeoutLong";
+    private static final int DEFAULT_OP_TIMEOUT_LONG = 60;
+    @Property(name = OP_TIMEOUT_LONG, intValue = DEFAULT_OP_TIMEOUT_LONG,
+            label = "Configure timeout in seconds for device operations " +
+                    "that are supposed to take a relatively long time " +
+                    "(e.g. pushing a large pipeline configuration with slow " +
+                    "network); default is 60 seconds")
+    private int opTimeoutLong = DEFAULT_OP_TIMEOUT_LONG;
 
     private static final String APP_NAME = "org.onosproject.generaldeviceprovider";
     private static final String URI_SCHEME = "device";
@@ -180,7 +196,6 @@
     //FIXME this will be removed when the configuration is synced at the source.
     private static final Set<String> PIPELINE_CONFIGURABLE_PROTOCOLS = ImmutableSet.of("p4runtime");
 
-    private static final ConcurrentMap<DeviceId, Lock> DEVICE_LOCKS = Maps.newConcurrentMap();
     //FIXME to be removed when netcfg will issue device events in a bundle or
     //ensures all configuration needed is present
     private Set<DeviceId> deviceConfigured = new CopyOnWriteArraySet<>();
@@ -188,20 +203,20 @@
     private Set<DeviceId> pipelineConfigured = new CopyOnWriteArraySet<>();
 
     private final Map<DeviceId, DeviceHandshaker> handshakers = Maps.newConcurrentMap();
-
     private final Map<DeviceId, MastershipRole> requestedRoles = Maps.newConcurrentMap();
-
+    private final Striped<Lock> deviceLocks = Striped.lock(30);
 
     private ExecutorService connectionExecutor
             = newFixedThreadPool(CORE_POOL_SIZE, groupedThreads(
             "onos/generaldeviceprovider-device-connect", "%d", log));
-    private ScheduledExecutorService portStatsExecutor
+    private ScheduledExecutorService statsExecutor
             = newScheduledThreadPool(CORE_POOL_SIZE, groupedThreads(
-            "onos/generaldeviceprovider-port-stats", "%d", log));
-    private ScheduledExecutorService availabilityCheckExecutor
-            = newScheduledThreadPool(CORE_POOL_SIZE, groupedThreads(
-            "onos/generaldeviceprovider-availability-check", "%d", log));
-    private ConcurrentMap<DeviceId, ScheduledFuture<?>> scheduledTasks = new ConcurrentHashMap<>();
+            "onos/generaldeviceprovider-stats-poll", "%d", log));
+    private ConcurrentMap<DeviceId, ScheduledFuture<?>> statsPollingTasks = new ConcurrentHashMap<>();
+    private ScheduledExecutorService probeExecutor
+            = newSingleThreadScheduledExecutor(groupedThreads(
+            "onos/generaldeviceprovider-probe-", "%d", log));
+    private ScheduledFuture<?> probeTask = null;
 
     private DeviceProviderService providerService;
     private InternalDeviceListener deviceListener = new InternalDeviceListener();
@@ -232,46 +247,61 @@
         //This will fail if ONOS has CFG and drivers which depend on this provider
         // are activated, failing due to not finding the driver.
         cfgService.getSubjects(DeviceId.class, GeneralProviderDeviceConfig.class)
-                .forEach(did -> triggerConnectWithLeadership(
-                        did, FIRST_CONNECTION_TOPIC + did.toString()));
+                .forEach(this::triggerConnect);
         //Initiating a periodic check to see if any device is available again and reconnect it.
-        availabilityCheckExecutor.scheduleAtFixedRate(
-                this::scheduleDevicePolling, deviceAvailabilityPollFrequency,
-                deviceAvailabilityPollFrequency, TimeUnit.SECONDS);
+        rescheduleProbeTask();
         modified(context);
         log.info("Started");
     }
 
     @Modified
     public void modified(ComponentContext context) {
-        if (context != null) {
-            Dictionary<?, ?> properties = context.getProperties();
-            pollFrequency = Tools.getIntegerProperty(properties, POLL_FREQUENCY,
-                                                     DEFAULT_POLL_FREQUENCY_SECONDS);
-            log.info("Configured. Poll frequency is configured to {} seconds", pollFrequency);
+        if (context == null) {
+            return;
         }
 
-        if (!scheduledTasks.isEmpty()) {
-            //cancel all previous tasks
-            scheduledTasks.values().forEach(task -> task.cancel(false));
-            //resubmit task with new timeout.
-            Set<DeviceId> deviceSubjects =
-                    cfgService.getSubjects(DeviceId.class, GeneralProviderDeviceConfig.class);
-            deviceSubjects.forEach(deviceId -> {
-                if (notMyScheme(deviceId)) {
-                    // not under my scheme, skipping
-                    log.debug("{} is not my scheme, skipping", deviceId);
-                    return;
-                }
-                scheduledTasks.put(deviceId, scheduleStatsPolling(deviceId, true));
-            });
+        Dictionary<?, ?> properties = context.getProperties();
+        final int oldStatsPollFrequency = statsPollFrequency;
+        statsPollFrequency = Tools.getIntegerProperty(
+                properties, STATS_POLL_FREQUENCY, DEFAULT_STATS_POLL_FREQUENCY);
+        log.info("Configured. {} is configured to {} seconds",
+                 STATS_POLL_FREQUENCY, statsPollFrequency);
+        final int oldProbeFrequency = probeFrequency;
+        probeFrequency = Tools.getIntegerProperty(
+                properties, PROBE_FREQUENCY, DEFAULT_PROBE_FREQUENCY);
+        log.info("Configured. {} is configured to {} seconds",
+                 PROBE_FREQUENCY, probeFrequency);
+        opTimeoutShort = Tools.getIntegerProperty(
+                properties, OP_TIMEOUT_SHORT, DEFAULT_OP_TIMEOUT_SHORT);
+        log.info("Configured. {} is configured to {} seconds",
+                 OP_TIMEOUT_SHORT, opTimeoutShort);
+        opTimeoutLong = Tools.getIntegerProperty(
+                properties, OP_TIMEOUT_LONG, DEFAULT_OP_TIMEOUT_LONG);
+        log.info("Configured. {} is configured to {} seconds",
+                 OP_TIMEOUT_LONG, opTimeoutLong);
+
+        if (oldStatsPollFrequency != statsPollFrequency) {
+            rescheduleStatsPollingTasks();
         }
+
+        if (oldProbeFrequency != probeFrequency) {
+            rescheduleProbeTask();
+        }
+    }
+
+    private synchronized void rescheduleProbeTask() {
+        if (probeTask != null) {
+            probeTask.cancel(false);
+        }
+        probeTask = probeExecutor.scheduleAtFixedRate(
+                this::triggerProbeAllDevices, probeFrequency,
+                probeFrequency, TimeUnit.SECONDS);
     }
 
     @Deactivate
     public void deactivate() {
-        portStatsExecutor.shutdown();
-        availabilityCheckExecutor.shutdown();
+        statsExecutor.shutdown();
+        probeExecutor.shutdown();
         componentConfigService.unregisterProperties(getClass(), false);
         cfgService.removeListener(cfgListener);
         //Not Removing the device so they can still be used from other driver providers
@@ -293,18 +323,15 @@
 
     @Override
     public void triggerProbe(DeviceId deviceId) {
-        // TODO Really don't see the point of this in non OF Context,
-        // for now testing reachability, can be moved to no-op
-        log.debug("Triggering probe equals testing reachability on device {}", deviceId);
-        isReachable(deviceId);
+        connectionExecutor.execute(withDeviceLock(
+                () -> doDeviceProbe(deviceId), deviceId));
     }
 
     @Override
     public void roleChanged(DeviceId deviceId, MastershipRole newRole) {
         log.info("Received role {} for device {}", newRole, deviceId);
         requestedRoles.put(deviceId, newRole);
-        connectionExecutor.submit(exceptionSafe(
-                () -> doRoleChanged(deviceId, newRole)));
+        connectionExecutor.execute(() -> doRoleChanged(deviceId, newRole));
     }
 
     private void doRoleChanged(DeviceId deviceId, MastershipRole newRole) {
@@ -326,74 +353,63 @@
         }
         return getFutureWithDeadline(
                 handshaker.isReachable(), "checking reachability",
-                deviceId, false);
+                deviceId, false, opTimeoutShort);
     }
 
     @Override
     public void changePortState(DeviceId deviceId, PortNumber portNumber,
                                 boolean enable) {
+        connectionExecutor.execute(
+                () -> doChangePortState(deviceId, portNumber, enable));
+    }
+
+    private void doChangePortState(DeviceId deviceId, PortNumber portNumber,
+                                   boolean enable) {
         if (!deviceService.getDevice(deviceId).is(PortAdmin.class)) {
             log.warn("Missing PortAdmin behaviour on {}, aborting port state change",
                      deviceId);
             return;
         }
-        final PortAdmin portAdmin = getPortAdmin(deviceId);
-        final CompletableFuture<Boolean> modified = enable
+        final PortAdmin portAdmin = deviceService.getDevice(deviceId)
+                .as(PortAdmin.class);
+        final CompletableFuture<Boolean> modifyTask = enable
                 ? portAdmin.enable(portNumber)
                 : portAdmin.disable(portNumber);
-        modified.thenAcceptAsync(result -> {
-            if (!result) {
-                log.warn("Port {} status cannot be changed on {} (enable={})",
-                         portNumber, deviceId, enable);
-            }
-        });
+        final String descr = (enable ? "enabling" : "disabling") + " port " + portNumber;
+        getFutureWithDeadline(
+                modifyTask, descr, deviceId, null, opTimeoutShort);
     }
 
     @Override
     public void triggerDisconnect(DeviceId deviceId) {
         log.debug("Triggering disconnection of device {}", deviceId);
-        connectionExecutor.execute(
-                () -> disconnectDevice(deviceId)
-                        .thenRunAsync(() -> checkAndConnect(deviceId)));
+        connectionExecutor.execute(withDeviceLock(
+                () -> doDisconnectDevice(deviceId), deviceId));
     }
 
     private DeviceHandshaker getHandshaker(DeviceId deviceId) {
         return handshakers.computeIfAbsent(deviceId, id -> {
             Driver driver = getDriver(deviceId);
-            return driver == null ? null :
-                    getBehaviour(driver, DeviceHandshaker.class,
-                                 new DefaultDriverData(driver, deviceId));
+            return driver == null ? null : getBehaviour(
+                    driver, DeviceHandshaker.class,
+                    new DefaultDriverData(driver, deviceId));
         });
     }
 
-    private PortAdmin getPortAdmin(DeviceId deviceId) {
-        Driver driver = getDriver(deviceId);
-        return getBehaviour(driver, PortAdmin.class,
-                            new DefaultDriverData(driver, deviceId));
-
-    }
-
     private Driver getDriver(DeviceId deviceId) {
-        Driver driver = null;
         try {
-            driver = driverService.getDriver(deviceId);
+            // DriverManager checks first using basic device config.
+            return driverService.getDriver(deviceId);
         } catch (ItemNotFoundException e) {
-            log.debug("Falling back to configuration to fetch driver " +
-                              "for device {}", deviceId);
-            BasicDeviceConfig cfg = cfgService.getConfig(deviceId, BasicDeviceConfig.class);
-            if (cfg != null) {
-                driver = driverService.getDriver(cfg.driver());
-            }
+            log.error("Driver not found for {}", deviceId);
+            return null;
         }
-        return driver;
     }
 
-    //needed since the device manager will not return the driver through implementation()
-    // method since the device is not pushed to the core so for the connectDeviceAsMaster
-    // we need to work around that in order to test before calling
-    // store.createOrUpdateDevice
     private <T extends Behaviour> T getBehaviour(Driver driver, Class<T> type,
                                                  DriverData data) {
+        // Allows obtaining behavior implementations before the device is pushed
+        // to the core.
         if (driver != null && driver.hasBehaviour(type)) {
             DefaultDriverHandler handler = new DefaultDriverHandler(data);
             return driver.createBehaviour(handler, type);
@@ -402,7 +418,12 @@
         }
     }
 
-    private void doConnectDevice(DeviceId deviceId, boolean asMaster) {
+    private void doConnectDevice(DeviceId deviceId) {
+        // Some operations can be performed by one node only.
+        final boolean isLocalLeader = leadershipService.runForLeadership(
+                GeneralProviderDeviceConfig.class.getName() + deviceId)
+                .leader().nodeId().equals(clusterService.getLocalNode().id());
+
         if (deviceService.getDevice(deviceId) != null
                 && deviceService.isAvailable(deviceId)) {
             log.info("Device {} is already connected to ONOS and is available",
@@ -421,7 +442,7 @@
             return;
         }
         log.info("Initiating connection to device {} with driver {} ... asMaster={}",
-                 deviceId, basicDeviceConfig.driver(), asMaster);
+                 deviceId, basicDeviceConfig.driver(), isLocalLeader);
         // Get handshaker, driver and driver data.
         final DeviceHandshaker handshaker = getHandshaker(deviceId);
         if (handshaker == null) {
@@ -436,7 +457,7 @@
         // Start connection via handshaker.
         final Boolean connectSuccess = getFutureWithDeadline(
                 handshaker.connect(), "initiating connection",
-                deviceId, null);
+                deviceId, null, opTimeoutShort);
         if (connectSuccess == null) {
             // Error logged by getFutureWithDeadline().
             return;
@@ -445,15 +466,17 @@
             return;
         }
         // Handle pipeconf (if device is capable)
-        if (!handlePipeconf(deviceId, driver, driverData, asMaster)) {
+        if (!handlePipeconf(deviceId, driver, driverData, isLocalLeader)) {
             // We already logged the error.
-            handshaker.disconnect();
+            getFutureWithDeadline(
+                    handshaker.disconnect(), "performing disconnection",
+                    deviceId, null, opTimeoutShort);
             return;
         }
         // Add device agent listener.
         handshaker.addDeviceAgentListener(deviceAgentListener);
         // All good. Notify core (if master).
-        if (asMaster) {
+        if (isLocalLeader) {
             advertiseDevice(deviceId, driver, providerConfig, driverData);
         }
     }
@@ -504,7 +527,7 @@
      * core, false otherwise.
      */
     private boolean handlePipeconf(DeviceId deviceId, Driver driver,
-                                   DriverData driverData, boolean deployPipeconf) {
+                                   DriverData driverData, boolean asMaster) {
         final PiPipelineProgrammable pipelineProg = getBehaviour(
                 driver, PiPipelineProgrammable.class, driverData);
         if (pipelineProg == null) {
@@ -518,28 +541,41 @@
         }
         final PiPipeconfId pipeconfId = pipeconf.id();
 
-        if (deployPipeconf) {
-            final Boolean deploySuccess = getFutureWithDeadline(
-                    pipelineProg.deployPipeconf(pipeconf),
-                    "deploying pipeconf", deviceId, null);
-            if (deploySuccess == null) {
-                // Error logged by getFutureWithDeadline().
-                return false;
-            } else if (!deploySuccess) {
-                log.error("Unable to deploy pipeconf {} to {}, aborting device discovery",
-                          pipeconfId, deviceId);
-                return false;
-            }
+        final String mergedDriverName = piPipeconfService.mergeDriver(
+                deviceId, pipeconfId);
+        if (mergedDriverName == null) {
+            log.error("Unable to get merged driver for {} and {}, aborting device discovery",
+                      deviceId, pipeconfId);
+            return false;
         }
 
-        final Boolean mergeSuccess = getFutureWithDeadline(
-                piPipeconfService.bindToDevice(pipeconfId, deviceId),
-                "merging driver", deviceId, null);
-        if (mergeSuccess == null) {
+        if (!asMaster) {
+            // From now one only the master.
+            return true;
+        }
+
+        if (!setDriverViaCfg(deviceId, mergedDriverName)) {
+            return false;
+        }
+
+        // FIXME: we just introduced a race condition as it might happen that a
+        // node does not receive the new cfg (with the merged driver) before the
+        // device is advertised to the core. Perhaps we should be waiting for a
+        // NetworkConfig event signaling that the driver has been updated on all
+        // nodes? The effect is mitigated by deploying the pipeconf (slow
+        // operation), after calling setDriverViaCfg().
+
+        piPipeconfService.bindToDevice(pipeconfId, deviceId);
+
+        final Boolean deploySuccess = getFutureWithDeadline(
+                pipelineProg.deployPipeconf(pipeconf),
+                "deploying pipeconf", deviceId, null,
+                opTimeoutLong);
+        if (deploySuccess == null) {
             // Error logged by getFutureWithDeadline().
             return false;
-        } else if (!mergeSuccess) {
-            log.error("Unable to merge pipeconf driver for {}, aborting device discovery",
+        } else if (!deploySuccess) {
+            log.error("Unable to deploy pipeconf {} to {}, aborting device discovery",
                       pipeconfId, deviceId);
             return false;
         }
@@ -547,6 +583,21 @@
         return true;
     }
 
+    private boolean setDriverViaCfg(DeviceId deviceId, String driverName) {
+        BasicDeviceConfig cfg = cfgService.getConfig(deviceId, BasicDeviceConfig.class);
+        if (cfg == null) {
+            log.error("Unable to get basic device config for {}, aborting device discovery",
+                      deviceId);
+            return false;
+        }
+        ObjectNode newCfg = (ObjectNode) cfg.node();
+        newCfg = newCfg.put(DRIVER, driverName);
+        ObjectMapper mapper = new ObjectMapper();
+        JsonNode newCfgNode = mapper.convertValue(newCfg, JsonNode.class);
+        cfgService.applyConfig(deviceId, BasicDeviceConfig.class, newCfgNode);
+        return true;
+    }
+
     private PiPipeconf getPipeconf(DeviceId deviceId, PiPipelineProgrammable pipelineProg) {
         PiPipeconfId pipeconfId = getPipeconfFromCfg(deviceId);
         if (pipeconfId == null || pipeconfId.id().isEmpty()) {
@@ -569,16 +620,14 @@
         return piPipeconfService.getPipeconf(pipeconfId).get();
     }
 
-    private CompletableFuture<?> disconnectDevice(DeviceId deviceId) {
-        log.info("Disconnecting for device {}", deviceId);
+    private void doDisconnectDevice(DeviceId deviceId) {
+        log.info("Initiating disconnection from {}...", deviceId);
         // Remove from core (if master)
-        if (mastershipService.isLocalMaster(deviceId)) {
+        if (mastershipService.isLocalMaster(deviceId)
+                && deviceService.isAvailable(deviceId)) {
             providerService.deviceDisconnected(deviceId);
         }
-        // Cancel tasks
-        if (scheduledTasks.containsKey(deviceId)) {
-            scheduledTasks.remove(deviceId).cancel(true);
-        }
+        cancelStatsPolling(deviceId);
         // Perform disconnection with device.
         final DeviceHandshaker handshaker = handshakers.remove(deviceId);
         if (handshaker == null) {
@@ -586,18 +635,15 @@
             log.warn("Missing DeviceHandshaker behavior for {}, " +
                              "no guarantees of complete disconnection",
                      deviceId);
-            return CompletableFuture.completedFuture(false);
+            return;
         }
         handshaker.removeDeviceAgentListener(deviceAgentListener);
-        return handshaker.disconnect()
-                .thenApplyAsync(result -> {
-                    if (result) {
-                        log.info("Disconnected device {}", deviceId);
-                    } else {
-                        log.warn("Device {} was unable to disconnect", deviceId);
-                    }
-                    return result;
-                });
+        final boolean disconnectSuccess = getFutureWithDeadline(
+                handshaker.disconnect(), "performing disconnection",
+                deviceId, false, opTimeoutShort);
+        if (!disconnectSuccess) {
+            log.warn("Unable to disconnect from {}", deviceId);
+        }
     }
 
     // Needed to catch the exception in the executors since are not rethrown otherwise.
@@ -611,21 +657,27 @@
         };
     }
 
-    private Runnable withDeviceLock(Runnable runnable, DeviceId deviceId) {
-        return () -> {
-            Lock lock = DEVICE_LOCKS.computeIfAbsent(deviceId, key -> new ReentrantLock());
-            lock.lock();
-            try {
-                runnable.run();
-            } finally {
-                lock.unlock();
-            }
-        };
+    private <U> U withDeviceLock(Supplier<U> task, DeviceId deviceId) {
+        final Lock lock = deviceLocks.get(deviceId);
+        lock.lock();
+        try {
+            return task.get();
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private Runnable withDeviceLock(Runnable task, DeviceId deviceId) {
+        // Wrapper of withDeviceLock(Supplier, ...) for void tasks.
+        return () -> withDeviceLock(() -> {
+            task.run();
+            return null;
+        }, deviceId);
     }
 
     private void updatePortStatistics(DeviceId deviceId) {
         Device device = deviceService.getDevice(deviceId);
-        if (!Objects.isNull(device) && deviceService.isAvailable(deviceId) &&
+        if (device != null && deviceService.isAvailable(deviceId) &&
                 device.is(PortStatisticsDiscovery.class)) {
             Collection<PortStatistics> statistics = device.as(PortStatisticsDiscovery.class)
                     .discoverPortStatistics();
@@ -642,14 +694,9 @@
         return !deviceId.uri().getScheme().equals(URI_SCHEME);
     }
 
-    private void triggerConnectWithLeadership(DeviceId deviceId,
-                                              String leadershipTopic) {
-        final NodeId leaderNodeId = leadershipService.runForLeadership(
-                leadershipTopic).leader().nodeId();
-        final boolean thisNodeMaster = clusterService
-                .getLocalNode().id().equals(leaderNodeId);
-        connectionExecutor.submit(withDeviceLock(exceptionSafe(
-                () -> doConnectDevice(deviceId, thisNodeMaster)), deviceId));
+    private void triggerConnect(DeviceId deviceId) {
+        connectionExecutor.execute(withDeviceLock(
+                () -> doConnectDevice(deviceId), deviceId));
     }
 
     /**
@@ -666,67 +713,62 @@
                 log.debug("{} is not my scheme, skipping", deviceId);
                 return;
             }
-            if (!isDeviceConfigComplete(event, deviceId)) {
+            final boolean configComplete = withDeviceLock(
+                    () -> isDeviceConfigComplete(event, deviceId), deviceId);
+            if (!configComplete) {
                 // Still waiting for some configuration.
                 return;
             }
             // Good to go.
-            triggerConnectWithLeadership(
-                    deviceId, FIRST_CONNECTION_TOPIC + deviceId.toString());
+            triggerConnect(deviceId);
             cleanUpConfigInfo(deviceId);
         }
 
         private boolean isDeviceConfigComplete(NetworkConfigEvent event, DeviceId deviceId) {
             // FIXME to be removed when netcfg will issue device events in a bundle or
             // ensure all configuration needed is present
-            Lock lock = DEVICE_LOCKS.computeIfAbsent(deviceId, key -> new ReentrantLock());
-            lock.lock();
-            try {
-                if (event.configClass().equals(GeneralProviderDeviceConfig.class)) {
-                    //FIXME we currently assume that p4runtime devices are pipeline configurable.
-                    //If we want to connect a p4runtime device with no pipeline
-                    if (event.config().isPresent()) {
-                        deviceConfigured.add(deviceId);
-                        final boolean isNotPipelineConfigurable = Collections.disjoint(
-                                ImmutableSet.copyOf(event.config().get().node().fieldNames()),
-                                PIPELINE_CONFIGURABLE_PROTOCOLS);
-                        if (isNotPipelineConfigurable) {
-                            // Skip waiting for a pipeline if we can't support it.
-                            pipelineConfigured.add(deviceId);
-                        }
-                    }
-                } else if (event.configClass().equals(BasicDeviceConfig.class)) {
-                    if (event.config().isPresent() && event.config().get().node().has(DRIVER)) {
-                        driverConfigured.add(deviceId);
-                    }
-                } else if (event.configClass().equals(PiPipeconfConfig.class)) {
-                    if (event.config().isPresent()
-                            && event.config().get().node().has(PiPipeconfConfig.PIPIPECONFID)) {
+            if (event.configClass().equals(GeneralProviderDeviceConfig.class)) {
+                //FIXME we currently assume that p4runtime devices are pipeline configurable.
+                //If we want to connect a p4runtime device with no pipeline
+                if (event.config().isPresent()) {
+                    deviceConfigured.add(deviceId);
+                    final boolean isNotPipelineConfigurable = Collections.disjoint(
+                            ImmutableSet.copyOf(event.config().get().node().fieldNames()),
+                            PIPELINE_CONFIGURABLE_PROTOCOLS);
+                    if (isNotPipelineConfigurable) {
+                        // Skip waiting for a pipeline if we can't support it.
                         pipelineConfigured.add(deviceId);
                     }
                 }
-
-                if (deviceConfigured.contains(deviceId)
-                        && driverConfigured.contains(deviceId)
-                        && pipelineConfigured.contains(deviceId)) {
-                    return true;
-                } else {
-                    if (deviceConfigured.contains(deviceId) && driverConfigured.contains(deviceId)) {
-                        log.debug("Waiting for pipeline configuration for device {}", deviceId);
-                    } else if (pipelineConfigured.contains(deviceId) && driverConfigured.contains(deviceId)) {
-                        log.debug("Waiting for device configuration for device {}", deviceId);
-                    } else if (pipelineConfigured.contains(deviceId) && deviceConfigured.contains(deviceId)) {
-                        log.debug("Waiting for driver configuration for device {}", deviceId);
-                    } else if (driverConfigured.contains(deviceId)) {
-                        log.debug("Only driver configuration for device {}", deviceId);
-                    } else if (deviceConfigured.contains(deviceId)) {
-                        log.debug("Only device configuration for device {}", deviceId);
-                    }
+            } else if (event.configClass().equals(BasicDeviceConfig.class)) {
+                if (event.config().isPresent() && event.config().get().node().has(DRIVER)) {
+                    driverConfigured.add(deviceId);
                 }
-                return false;
-            } finally {
-                lock.unlock();
+            } else if (event.configClass().equals(PiPipeconfConfig.class)) {
+                if (event.config().isPresent()
+                        && event.config().get().node().has(PiPipeconfConfig.PIPIPECONFID)) {
+                    pipelineConfigured.add(deviceId);
+                }
             }
+
+            if (deviceConfigured.contains(deviceId)
+                    && driverConfigured.contains(deviceId)
+                    && pipelineConfigured.contains(deviceId)) {
+                return true;
+            } else {
+                if (deviceConfigured.contains(deviceId) && driverConfigured.contains(deviceId)) {
+                    log.debug("Waiting for pipeline configuration for device {}", deviceId);
+                } else if (pipelineConfigured.contains(deviceId) && driverConfigured.contains(deviceId)) {
+                    log.debug("Waiting for device configuration for device {}", deviceId);
+                } else if (pipelineConfigured.contains(deviceId) && deviceConfigured.contains(deviceId)) {
+                    log.debug("Waiting for driver configuration for device {}", deviceId);
+                } else if (driverConfigured.contains(deviceId)) {
+                    log.debug("Only driver configuration for device {}", deviceId);
+                } else if (deviceConfigured.contains(deviceId)) {
+                    log.debug("Only device configuration for device {}", deviceId);
+                }
+            }
+            return false;
         }
 
         @Override
@@ -757,18 +799,38 @@
         pipelineConfigured.remove(deviceId);
     }
 
-    private ScheduledFuture<?> scheduleStatsPolling(DeviceId deviceId, boolean randomize) {
-        int delay = 0;
-        if (randomize) {
-            delay = new SecureRandom().nextInt(10);
-        }
-        return portStatsExecutor.scheduleAtFixedRate(
-                exceptionSafe(() -> updatePortStatistics(deviceId)),
-                delay, pollFrequency, TimeUnit.SECONDS);
+    private void startStatsPolling(DeviceId deviceId, boolean withRandomDelay) {
+        statsPollingTasks.compute(deviceId, (did, oldTask) -> {
+            if (oldTask != null) {
+                oldTask.cancel(false);
+            }
+            final int delay = withRandomDelay
+                    ? new SecureRandom().nextInt(10) : 0;
+            return statsExecutor.scheduleAtFixedRate(
+                    exceptionSafe(() -> updatePortStatistics(deviceId)),
+                    delay, statsPollFrequency, TimeUnit.SECONDS);
+        });
     }
 
-    private void scheduleDevicePolling() {
-        cfgService.getSubjects(DeviceId.class, GeneralProviderDeviceConfig.class).forEach(this::checkAndConnect);
+    private void cancelStatsPolling(DeviceId deviceId) {
+        statsPollingTasks.computeIfPresent(deviceId, (did, task) -> {
+            task.cancel(false);
+            return null;
+        });
+    }
+
+    private void rescheduleStatsPollingTasks() {
+        statsPollingTasks.keySet().forEach(deviceId -> {
+            // startStatsPolling cancels old one if present.
+            startStatsPolling(deviceId, true);
+        });
+    }
+
+    private void triggerProbeAllDevices() {
+        // Async trigger a task for all devices in the cfg.
+        cfgService.getSubjects(DeviceId.class, GeneralProviderDeviceConfig.class)
+                .forEach(deviceId -> connectionExecutor.execute(withDeviceLock(
+                        () -> doDeviceProbe(deviceId), deviceId)));
     }
 
     private PiPipeconfId getPipeconfFromCfg(DeviceId deviceId) {
@@ -780,38 +842,28 @@
         return config.piPipeconfId();
     }
 
-    private void checkAndConnect(DeviceId deviceId) {
-        // Let's try and reconnect to a device which is stored in cfg.
-        // One of the following conditions must be satisfied:
-        // 1) device is null in the store meaning that is was never connected or
-        // it was administratively removed
-        // 2) the device is not available and there is no MASTER instance,
-        // meaning the device lost it's connection to ONOS at some point in the
-        // past.
-        // We also check that the general device provider config and the driver
-        // config are present. We do not check for reachability using
-        // isReachable(deviceId) since the behaviour of this method can vary
-        // depending on protocol nuances. We leave this check to the device
-        // handshaker at later stages of the connection process. IF the
-        // conditions are not met but instead the device is present in the
-        // store, available and this instance is MASTER but is not reachable we
-        // remove it from the store.
-
-        if ((deviceService.getDevice(deviceId) == null
-                || (!deviceService.isAvailable(deviceId)
-                && mastershipService.getMasterFor(deviceId) == null))
-                && configIsPresent(deviceId)) {
-            log.debug("Trying to re-connect to device {}", deviceId);
-            triggerConnectWithLeadership(
-                    deviceId, CHECK_CONNECTION_TOPIC + deviceId.toString());
-            cleanUpConfigInfo(deviceId);
-        } else if (deviceService.getDevice(deviceId) != null
-                && deviceService.isAvailable(deviceId)
-                && mastershipService.isLocalMaster(deviceId)
-                && !isReachable(deviceId)
-                && configIsPresent(deviceId)) {
-            log.info("Removing available but unreachable device {}", deviceId);
-            disconnectDevice(deviceId);
+    private void doDeviceProbe(DeviceId deviceId) {
+        if (!configIsPresent(deviceId)) {
+            return;
+        }
+        final boolean isAvailable = deviceService.getDevice(deviceId) != null
+                && deviceService.isAvailable(deviceId);
+        final boolean isLocalMaster = mastershipService.isLocalMaster(deviceId);
+        if (isAvailable) {
+            if (!isLocalMaster) {
+                return;
+            }
+            if (!isReachable(deviceId)) {
+                log.info("Disconnecting available but unreachable device {}...",
+                         deviceId);
+                triggerDisconnect(deviceId);
+            }
+        } else {
+            // We do not check for reachability using isReachable()
+            // since the behaviour of this method can vary depending on protocol
+            // nuances. We leave this check to the device handshaker at later
+            // stages of the connection process.
+            triggerConnect(deviceId);
         }
     }
 
@@ -828,7 +880,7 @@
     private void handleChannelClosed(DeviceId deviceId) {
         log.info("Disconnecting device {}, due to channel closed event",
                  deviceId);
-        disconnectDevice(deviceId);
+        triggerDisconnect(deviceId);
     }
 
     private void handleMastershipResponse(DeviceId deviceId, MastershipRole response) {
@@ -837,21 +889,17 @@
             return;
         }
         providerService.receivedRoleReply(deviceId, requestedRoles.get(deviceId), response);
-        // If not master, cancel polling tasks, otherwise start them.
-        if (!response.equals(MastershipRole.MASTER)
-                && scheduledTasks.get(deviceId) != null) {
-            scheduledTasks.remove(deviceId).cancel(false);
-        } else if (response.equals(MastershipRole.MASTER)
-                && scheduledTasks.get(deviceId) == null) {
-            scheduledTasks.put(deviceId, scheduleStatsPolling(deviceId, false));
-            updatePortStatistics(deviceId);
+        if (response.equals(MastershipRole.MASTER)) {
+            startStatsPolling(deviceId, false);
+        } else {
+            cancelStatsPolling(deviceId);
         }
     }
 
     private <U> U getFutureWithDeadline(CompletableFuture<U> future, String opDescription,
-                                        DeviceId deviceId, U defaultValue) {
+                                        DeviceId deviceId, U defaultValue, int timeout) {
         try {
-            return future.get(DEVICE_OP_TIMEOUT, TimeUnit.SECONDS);
+            return future.get(timeout, TimeUnit.SECONDS);
         } catch (InterruptedException e) {
             log.error("Thread interrupted while {} on {}", opDescription, deviceId);
             Thread.currentThread().interrupt();
@@ -873,7 +921,7 @@
             // For now this is scheduled periodically, when streaming API will
             // be available we check and base it on the streaming API (e.g. gNMI)
             if (mastershipService.isLocalMaster(deviceId)) {
-                scheduledTasks.put(deviceId, scheduleStatsPolling(deviceId, false));
+                startStatsPolling(deviceId, true);
             }
         }