diff --git a/core/api/src/main/java/org/onosproject/net/AnnotationKeys.java b/core/api/src/main/java/org/onosproject/net/AnnotationKeys.java
index ce3c094..2e92c53 100644
--- a/core/api/src/main/java/org/onosproject/net/AnnotationKeys.java
+++ b/core/api/src/main/java/org/onosproject/net/AnnotationKeys.java
@@ -89,16 +89,6 @@
     public static final String DRIVER = "driver";
 
     /**
-     * Annotation key for the device availability behavior. The value of this key
-     * is expected to be a boolean ({@code true} or {@code false}) and it is
-     * used to determine if the device can be marked online by the core when
-     * deemed appropriate (value is {@code false}, default behaviour if key is
-     * not present), or if marking online should be left to providers ({@code
-     * true}).
-     */
-    public static final String PROVIDER_MARK_ONLINE = "providerMarkOnline";
-
-    /**
      * Annotation key for durable links.
      */
     public static final String DURABLE = "durable";
diff --git a/core/api/src/main/java/org/onosproject/net/behaviour/PiPipelineProgrammable.java b/core/api/src/main/java/org/onosproject/net/behaviour/PiPipelineProgrammable.java
index 22367ba..be1c6c4 100644
--- a/core/api/src/main/java/org/onosproject/net/behaviour/PiPipelineProgrammable.java
+++ b/core/api/src/main/java/org/onosproject/net/behaviour/PiPipelineProgrammable.java
@@ -44,16 +44,17 @@
     CompletableFuture<Boolean> setPipeconf(PiPipeconf pipeconf);
 
     /**
-     * Returns true if the device is configured with the given pipeconf, false
-     * otherwise.
+     * Probes the device to verify that the given pipeconf is the one currently
+     * configured.
      * <p>
      * This method is expected to always return true after successfully calling
      * {@link #setPipeconf(PiPipeconf)} with the given pipeconf.
      *
      * @param pipeconf pipeconf
-     * @return true if the device has the given pipeconf set, false otherwise
+     * @return completable future eventually true if the device has the given
+     * pipeconf set, false otherwise
      */
-    boolean isPipeconfSet(PiPipeconf pipeconf);
+    CompletableFuture<Boolean> isPipeconfSet(PiPipeconf pipeconf);
 
     /**
      * Returns the default pipeconf for this device, to be used when any other
diff --git a/core/api/src/main/java/org/onosproject/net/config/basics/BasicDeviceConfig.java b/core/api/src/main/java/org/onosproject/net/config/basics/BasicDeviceConfig.java
index 0353bf2..1a7ffd6 100644
--- a/core/api/src/main/java/org/onosproject/net/config/basics/BasicDeviceConfig.java
+++ b/core/api/src/main/java/org/onosproject/net/config/basics/BasicDeviceConfig.java
@@ -29,6 +29,7 @@
     private static final String TYPE = "type";
     private static final String DRIVER = "driver";
     private static final String MANAGEMENT_ADDRESS = "managementAddress";
+    private static final String PIPECONF = "pipeconf";
     private static final String MANUFACTURER = "manufacturer";
     private static final String HW_VERSION = "hwVersion";
     private static final String SW_VERSION = "swVersion";
@@ -41,6 +42,7 @@
     private static final int SW_VERSION_MAX_LENGTH = 256;
     private static final int SERIAL_MAX_LENGTH = 256;
     private static final int MANAGEMENT_ADDRESS_MAX_LENGTH = 1024;
+    private static final int PIPECONF_MAX_LENGTH = 256;
 
     @Override
     public boolean isValid() {
@@ -52,13 +54,14 @@
                 && hasOnlyFields(ALLOWED, NAME, LOC_TYPE, LATITUDE, LONGITUDE,
                 GRID_Y, GRID_X, UI_TYPE, RACK_ADDRESS, OWNER, TYPE, DRIVER, ROLES,
                 MANUFACTURER, HW_VERSION, SW_VERSION, SERIAL,
-                MANAGEMENT_ADDRESS, DEVICE_KEY_ID)
+                MANAGEMENT_ADDRESS, PIPECONF, DEVICE_KEY_ID)
                 && isValidLength(DRIVER, DRIVER_MAX_LENGTH)
                 && isValidLength(MANUFACTURER, MANUFACTURER_MAX_LENGTH)
                 && isValidLength(HW_VERSION, MANUFACTURER_MAX_LENGTH)
                 && isValidLength(SW_VERSION, MANUFACTURER_MAX_LENGTH)
                 && isValidLength(SERIAL, MANUFACTURER_MAX_LENGTH)
-                && isValidLength(MANAGEMENT_ADDRESS, MANAGEMENT_ADDRESS_MAX_LENGTH);
+                && isValidLength(MANAGEMENT_ADDRESS, MANAGEMENT_ADDRESS_MAX_LENGTH)
+                && isValidLength(PIPECONF, PIPECONF_MAX_LENGTH);
     }
 
     /**
@@ -186,15 +189,25 @@
     }
 
     /**
-     * Returns the device management ip (ip:port).
+     * Returns the device management address (e.g, "ip:port" or full URI
+     * string).
      *
-     * @return device management address (ip:port) or null if not set
+     * @return device management address or null if not set
      */
     public String managementAddress() {
         return get(MANAGEMENT_ADDRESS, null);
     }
 
     /**
+     * Returns the device pipeconf.
+     *
+     * @return device pipeconf or null if not set
+     */
+    public String pipeconf() {
+        return get(PIPECONF, null);
+    }
+
+    /**
      * Sets the device management ip (ip:port).
      *
      * @param managementAddress new device management address (ip:port); null to clear
@@ -202,11 +215,23 @@
      */
     public BasicDeviceConfig managementAddress(String managementAddress) {
         checkArgument(managementAddress.length() <= MANAGEMENT_ADDRESS_MAX_LENGTH,
-                "serialNumber exceeds maximum length " + MANAGEMENT_ADDRESS_MAX_LENGTH);
+                "managementAddress exceeds maximum length " + MANAGEMENT_ADDRESS_MAX_LENGTH);
         return (BasicDeviceConfig) setOrClear(MANAGEMENT_ADDRESS, managementAddress);
     }
 
     /**
+     * Sets the device pipeconf.
+     *
+     * @param pipeconf new device pipeconf
+     * @return self
+     */
+    public BasicDeviceConfig pipeconf(String pipeconf) {
+        checkArgument(pipeconf.length() <= PIPECONF_MAX_LENGTH,
+                      "pipeconf exceeds maximum length " + MANAGEMENT_ADDRESS_MAX_LENGTH);
+        return (BasicDeviceConfig) setOrClear(PIPECONF, pipeconf);
+    }
+
+    /**
      * Returns the device key id.
      *
      * @return device key id or null if not set
diff --git a/core/api/src/main/java/org/onosproject/net/device/DeviceAgentEvent.java b/core/api/src/main/java/org/onosproject/net/device/DeviceAgentEvent.java
index d2520b0..b214070 100644
--- a/core/api/src/main/java/org/onosproject/net/device/DeviceAgentEvent.java
+++ b/core/api/src/main/java/org/onosproject/net/device/DeviceAgentEvent.java
@@ -54,19 +54,19 @@
         ROLE_MASTER,
 
         /**
-         * Signifies that the agent has standby/slave mastership role.
+         * Signifies that the agent has acquired standby/slave mastership role.
          */
         ROLE_STANDBY,
 
         /**
-         * Signifies that the agent cannot acquire any valid mastership role for
+         * Signifies that the agent doesn't have any valid mastership role for
          * the device.
          */
         ROLE_NONE,
 
         /**
-         * Signifies that the agent cannot perform operations on the device
-         * because its role is not master.
+         * Signifies that the agent tried to perform some operations on the
+         * device that requires master role.
          */
         NOT_MASTER,
 
@@ -81,15 +81,4 @@
     public DeviceAgentEvent(Type type, DeviceId deviceId) {
         super(type, deviceId);
     }
-
-    /**
-     * Creates a new device agent event for the given type, device ID and time.
-     *
-     * @param type     event type
-     * @param deviceId device ID
-     * @param time     occurrence time
-     */
-    public DeviceAgentEvent(Type type, DeviceId deviceId, long time) {
-        super(type, deviceId, time);
-    }
 }
diff --git a/core/api/src/main/java/org/onosproject/net/device/DeviceHandshaker.java b/core/api/src/main/java/org/onosproject/net/device/DeviceHandshaker.java
index bceaeff..8a941ca 100644
--- a/core/api/src/main/java/org/onosproject/net/device/DeviceHandshaker.java
+++ b/core/api/src/main/java/org/onosproject/net/device/DeviceHandshaker.java
@@ -31,13 +31,63 @@
 public interface DeviceHandshaker extends DeviceConnect {
 
     /**
-     * Checks the reachability (connectivity) of a device. Reachability, unlike
-     * availability, denotes whether THIS particular node can send messages and
-     * receive replies from the specified device.
+     * Returns true if this node is presumed to be able to send messages and
+     * receive replies from the device.
+     * <p>
+     * The implementation should not make any attempt at actively probing the
+     * device over the network, as such it should not block execution. Instead,
+     * it should return a result based solely on internal state (e.g. socket
+     * state). If it returns true, then this node is expected to communicate
+     * with the server successfully. In other words, if any message would be
+     * sent to the device immediately after this method is called and returns
+     * true, then such message is expected, but NOT guaranteed, to reach the
+     * device. If false, it means communication with the device is unlikely to
+     * happen soon.
+     * <p>
+     * Some implementations might require a connection to be created via {@link
+     * #connect()} before checking for reachability. Similarly, after invoking
+     * {@link #disconnect()}, this method might always return false.
      *
-     * @return CompletableFuture eventually true if reachable, false otherwise
+     * @return true if the device is deemed reachable, false otherwise
      */
-    CompletableFuture<Boolean> isReachable();
+    boolean isReachable();
+
+    /**
+     * Similar to {@link #isReachable()}, but performs probing of the device
+     * over the network. This method should be called if {@link #isReachable()}
+     * returns false and the caller wants to be sure this is not a transient
+     * failure state by actively probing the device.
+     *
+     * @return completable future eventually true if device responded to probe,
+     * false otherwise
+     */
+    CompletableFuture<Boolean> probeReachability();
+
+    /**
+     * Checks the availability of the device. Availability denotes whether the
+     * device is reachable and able to perform its functions as expected (e.g.,
+     * forward traffic). Similar to {@link #isReachable()}, implementations are
+     * not allowed to probe the device over the network, but the result should
+     * be based solely on internal state.
+     * <p>
+     * Implementation of this method is optional. If not supported, an exception
+     * should be thrown.
+     *
+     * @return true if the device is deemed available, false otherwise
+     * @throws UnsupportedOperationException if this method is not supported and
+     *                                       {@link #probeAvailability()} should
+     *                                       be used instead.
+     */
+    boolean isAvailable();
+
+    /**
+     * Similar to {@link #isAvailable()} but allows probing the device over the
+     * network. Differently from {@link #isAvailable()}, implementation of this
+     * method is mandatory.
+     *
+     * @return completable future eventually true if available, false otherwise
+     */
+    CompletableFuture<Boolean> probeAvailability();
 
     /**
      * Notifies the device a mastership role change as decided by the core. The
@@ -45,10 +95,48 @@
      * signaling the mastership role accepted by the device.
      *
      * @param newRole new mastership role
+     * @throws UnsupportedOperationException if the device does not support
+     *                                       mastership handling
      */
     void roleChanged(MastershipRole newRole);
 
     /**
+     * Notifies the device of a mastership role change as decided by the core.
+     * Differently from {@link #roleChanged(MastershipRole)}, the role is
+     * described by the given preference value, where {@code preference = 0}
+     * signifies {@link MastershipRole#MASTER} role and {@code preference > 0}
+     * signifies {@link MastershipRole#STANDBY}. Smaller preference values
+     * indicates higher mastership priority for different nodes.
+     * <p>
+     * This method does not permit notifying role {@link MastershipRole#NONE},
+     * in which case {@link #roleChanged(MastershipRole)} should be used
+     * instead.
+     * <p>
+     * Term is a monotonically increasing number, increased by one every time a
+     * new master is elected.
+     * <p>
+     * The implementation of this method should trigger a {@link
+     * DeviceAgentEvent} signaling the mastership role accepted by the device.
+     *
+     * @param preference preference value, where 0 signifies {@link
+     *                   MastershipRole#MASTER} and all other values {@link
+     *                   MastershipRole#STANDBY}
+     * @param term       term number
+     * @throws UnsupportedOperationException if the device does not support
+     *                                       mastership handling, or if it does
+     *                                       not support setting preference-based
+     *                                       mastership, and {@link #roleChanged(MastershipRole)}
+     *                                       should be used instead
+     */
+    default void roleChanged(int preference, long term) {
+        if (preference == 0) {
+            roleChanged(MastershipRole.MASTER);
+        } else {
+            roleChanged(MastershipRole.STANDBY);
+        }
+    }
+
+    /**
      * Returns the last known mastership role agreed by the device for this
      * node.
      *
diff --git a/core/api/src/main/java/org/onosproject/net/device/DeviceProvider.java b/core/api/src/main/java/org/onosproject/net/device/DeviceProvider.java
index ebdf7b1..771a8e4 100644
--- a/core/api/src/main/java/org/onosproject/net/device/DeviceProvider.java
+++ b/core/api/src/main/java/org/onosproject/net/device/DeviceProvider.java
@@ -41,8 +41,8 @@
      * Notifies the provider of a mastership role change for the specified
      * device as decided by the core.
      *
-     * @param deviceId  device identifier
-     * @param newRole newly determined mastership role
+     * @param deviceId device identifier
+     * @param newRole  newly determined mastership role
      */
     void roleChanged(DeviceId deviceId, MastershipRole newRole);
 
@@ -50,18 +50,40 @@
      * Checks the reachability (connectivity) of a device from this provider.
      * Reachability, unlike availability, denotes whether THIS particular node
      * can send messages and receive replies from the specified device.
+     * <p>
+     * Implementations are encouraged to check for reachability by using only
+     * internal provider state, i.e., without blocking execution.
      *
-     * @param deviceId  device identifier
+     * @param deviceId device identifier
      * @return true if reachable, false otherwise
      */
     boolean isReachable(DeviceId deviceId);
 
     /**
-     * Administratively enables or disables a port.
+     * Checks the availability of the device from the provider perspective.
+     * Availability denotes whether the device is reachable by
+     * this node and able to perform its functions as expected (e.g., forward
+     * traffic).
+     *
+     * <p>
+     * Implementations are encouraged to check for availability by using only
+     * internal provider state, i.e., without blocking execution.
      *
      * @param deviceId device identifier
+     * @return completable future eventually true if available, false otherwise
+     */
+    default boolean isAvailable(DeviceId deviceId) {
+        // For most implementations such as OpenFlow, reachability is equivalent
+        // to availability.
+        return isReachable(deviceId);
+    }
+
+    /**
+     * Administratively enables or disables a port.
+     *
+     * @param deviceId   device identifier
      * @param portNumber port number
-     * @param enable true if port is to be enabled, false to disable
+     * @param enable     true if port is to be enabled, false to disable
      */
     void changePortState(DeviceId deviceId, PortNumber portNumber,
                          boolean enable);
diff --git a/core/api/src/main/java/org/onosproject/net/device/DeviceProviderService.java b/core/api/src/main/java/org/onosproject/net/device/DeviceProviderService.java
index 80996eb..b00fa63 100644
--- a/core/api/src/main/java/org/onosproject/net/device/DeviceProviderService.java
+++ b/core/api/src/main/java/org/onosproject/net/device/DeviceProviderService.java
@@ -74,15 +74,30 @@
     void portStatusChanged(DeviceId deviceId, PortDescription portDescription);
 
     /**
-     * Notifies the core about the result of a RoleRequest sent to a device.
+     * Notifies the core about the result of a role request sent to a device.
+     * This method assumes that the provider knows the original role that was
+     * requested for a given response, if that is not the case, and only the
+     * response is known to the provider, then {@link #receivedRoleReply(DeviceId,
+     * MastershipRole)} should be used instead.
      *
-     * @param deviceId identity of the device
+     * @param deviceId  identity of the device
      * @param requested mastership role that was requested by the node
-     * @param response mastership role the switch accepted
+     * @param response  mastership role the switch accepted
      */
     void receivedRoleReply(DeviceId deviceId, MastershipRole requested, MastershipRole response);
 
     /**
+     * Notifies the core about a mastership role reported by the given device
+     * for this node.
+     *
+     * @param deviceId  identity of the device
+     * @param response  mastership role the switch accepted
+     */
+    default void receivedRoleReply(DeviceId deviceId, MastershipRole response) {
+        receivedRoleReply(deviceId, null, response);
+    }
+
+    /**
      * Updates statistics about all ports of a device.
      *
      * @param deviceId          identity of the device
diff --git a/core/api/src/main/java/org/onosproject/net/driver/DeviceConnect.java b/core/api/src/main/java/org/onosproject/net/driver/DeviceConnect.java
index f30386d..433be71 100644
--- a/core/api/src/main/java/org/onosproject/net/driver/DeviceConnect.java
+++ b/core/api/src/main/java/org/onosproject/net/driver/DeviceConnect.java
@@ -21,7 +21,9 @@
 
 /**
  * Abstraction of handler behaviour used to set-up and tear-down connections
- * with a device.
+ * with a device. A connection is intended as the presence of state (e.g. a
+ * transport session) required to carry messages between this node and the
+ * device.
  */
 @Beta
 public interface DeviceConnect extends HandlerBehaviour {
@@ -29,17 +31,33 @@
     /**
      * Connects to the device, for example by opening the transport session that
      * will be later used to send control messages. Returns true if the
-     * connection was initiated successfully, false otherwise.
+     * connection was initiated successfully, false otherwise. The
+     * implementation might require probing the device over the network to
+     * initiate the connection.
      * <p>
-     * Calling multiple times this method while a connection to the device is
-     * open should result in a no-op.
+     * When calling this method while a connection to the device already exists,
+     * the behavior is not defined. For example, some implementations might
+     * require to first call {@link #disconnect()}, while other might behave as
+     * a no-op.
      *
-     * @return CompletableFuture with true if the operation was successful
+     * @return CompletableFuture eventually true if a connection was created
+     * successfully, false otherwise
+     * @throws IllegalStateException if a connection already exists and the
+     *                               implementation requires to call {@link
+     *                               #disconnect()} first.
      */
-    CompletableFuture<Boolean> connect();
+    CompletableFuture<Boolean> connect() throws IllegalStateException;
 
     /**
-     * Returns true if a connection to the device is open, false otherwise.
+     * Returns true if a connection to the device exists, false otherwise. This
+     * method is NOT expected to send any message over the network to check for
+     * device reachability, but rather it should only give an indication if any
+     * internal connection state exists for the device. As such, it should NOT
+     * block execution.
+     * <p>
+     * In general, when called after {@link #connect()} it should always return
+     * true, while it is expected to always return false after calling {@link
+     * #disconnect()} or if {@link #connect()} was never called.
      *
      * @return true if the connection is open, false otherwise
      */
@@ -47,14 +65,10 @@
 
     /**
      * Disconnects from the device, for example closing the transport session
-     * previously opened. Returns true if the disconnection procedure was
-     * successful, false otherwise.
+     * previously opened.
      * <p>
      * Calling multiple times this method while a connection to the device is
-     * closed should result in a no-op.
-     *
-     * @return CompletableFuture with true if the operation was successful
+     * already closed should result in a no-op.
      */
-    CompletableFuture<Boolean> disconnect();
-
+    void disconnect();
 }
diff --git a/core/api/src/main/java/org/onosproject/net/pi/service/PiPipeconfConfig.java b/core/api/src/main/java/org/onosproject/net/pi/service/PiPipeconfConfig.java
deleted file mode 100644
index e8aa731..0000000
--- a/core/api/src/main/java/org/onosproject/net/pi/service/PiPipeconfConfig.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * Copyright 2017-present Open Networking Foundation
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.onosproject.net.pi.service;
-
-import com.google.common.annotations.Beta;
-import org.onosproject.net.DeviceId;
-import org.onosproject.net.config.Config;
-import org.onosproject.net.pi.model.PiPipeconfId;
-
-/**
- * Configuration for the PiPipeconf susbystem.
- */
-@Beta
-public final class PiPipeconfConfig extends Config<DeviceId> {
-
-    public static final String PIPIPECONFID = "piPipeconfId";
-
-    @Override
-    public boolean isValid() {
-        return hasOnlyFields(PIPIPECONFID);
-        //TODO will reinstate after synchonization of events
-        //&& !piPipeconfId().id().equals("");
-    }
-
-    public PiPipeconfId piPipeconfId() {
-        return new PiPipeconfId(get(PIPIPECONFID, ""));
-    }
-}
diff --git a/core/net/src/main/java/org/onosproject/net/device/impl/DeviceManager.java b/core/net/src/main/java/org/onosproject/net/device/impl/DeviceManager.java
index 6ce7c0c..0ffe615 100644
--- a/core/net/src/main/java/org/onosproject/net/device/impl/DeviceManager.java
+++ b/core/net/src/main/java/org/onosproject/net/device/impl/DeviceManager.java
@@ -23,13 +23,11 @@
 import org.onlab.util.Tools;
 import org.onosproject.cluster.ClusterService;
 import org.onosproject.cluster.NodeId;
-import org.onosproject.net.config.basics.PortDescriptionsConfig;
 import org.onosproject.mastership.MastershipEvent;
 import org.onosproject.mastership.MastershipListener;
 import org.onosproject.mastership.MastershipService;
 import org.onosproject.mastership.MastershipTerm;
 import org.onosproject.mastership.MastershipTermService;
-import org.onosproject.net.AnnotationKeys;
 import org.onosproject.net.ConnectPoint;
 import org.onosproject.net.Device;
 import org.onosproject.net.Device.Type;
@@ -46,6 +44,7 @@
 import org.onosproject.net.config.basics.BasicDeviceConfig;
 import org.onosproject.net.config.basics.DeviceAnnotationConfig;
 import org.onosproject.net.config.basics.PortAnnotationConfig;
+import org.onosproject.net.config.basics.PortDescriptionsConfig;
 import org.onosproject.net.device.DefaultPortDescription;
 import org.onosproject.net.device.DeviceAdminService;
 import org.onosproject.net.device.DeviceDescription;
@@ -499,7 +498,7 @@
             }
 
             // If this node is the master, ensure the device is marked online.
-            if (myRole == MASTER) {
+            if (myRole == MASTER && canMarkOnline(device)) {
                 post(store.markOnline(deviceId));
             }
 
@@ -788,15 +787,26 @@
                 return;
             }
 
+            final MastershipRole expected = mastershipService.getLocalRole(deviceId);
+
+            if (requested == null) {
+                // Provider is not able to reconcile role responses with
+                // requests. We assume what was requested is what we expect.
+                // This will work only if mastership doesn't change too often,
+                // and devices are left enough time to provide responses before
+                // a different role is requested.
+                requested = expected;
+            }
+
             if (Objects.equals(requested, response)) {
-                if (Objects.equals(requested, mastershipService.getLocalRole(deviceId))) {
+                if (Objects.equals(requested, expected)) {
                     return;
                 } else {
-                    log.warn("Role mismatch on {}. set to {}, but store demands {}",
-                             deviceId, response, mastershipService.getLocalRole(deviceId));
+                    log.warn("Role mismatch on {}. Set to {}, but store demands {}",
+                             deviceId, response, expected);
                     // roleManager got the device to comply, but doesn't agree with
                     // the store; use the store's view, then try to reassert.
-                    backgroundService.execute(() -> reassertRole(deviceId, mastershipService.getLocalRole(deviceId)));
+                    backgroundService.execute(() -> reassertRole(deviceId, expected));
                     return;
                 }
             } else {
@@ -830,9 +840,12 @@
     }
 
     private boolean canMarkOnline(Device device) {
-        final boolean providerMarkOnline = Boolean.parseBoolean(
-                device.annotations().value(AnnotationKeys.PROVIDER_MARK_ONLINE));
-        return !providerMarkOnline;
+        DeviceProvider provider = getProvider(device.id());
+        if (provider == null) {
+            log.warn("Provider for {} was not found. Cannot evaluate availability", device.id());
+            return false;
+        }
+        return provider.isAvailable(device.id());
     }
 
     // Applies the specified role to the device; ignores NONE
diff --git a/core/net/src/main/java/org/onosproject/net/pi/impl/PiPipeconfManager.java b/core/net/src/main/java/org/onosproject/net/pi/impl/PiPipeconfManager.java
index 27f6e4d..9653091 100644
--- a/core/net/src/main/java/org/onosproject/net/pi/impl/PiPipeconfManager.java
+++ b/core/net/src/main/java/org/onosproject/net/pi/impl/PiPipeconfManager.java
@@ -23,10 +23,8 @@
 import org.onlab.util.ItemNotFoundException;
 import org.onlab.util.SharedExecutors;
 import org.onosproject.net.DeviceId;
-import org.onosproject.net.config.ConfigFactory;
 import org.onosproject.net.config.NetworkConfigRegistry;
 import org.onosproject.net.config.basics.BasicDeviceConfig;
-import org.onosproject.net.config.basics.SubjectFactories;
 import org.onosproject.net.driver.Behaviour;
 import org.onosproject.net.driver.DefaultDriver;
 import org.onosproject.net.driver.Driver;
@@ -36,7 +34,6 @@
 import org.onosproject.net.driver.DriverProvider;
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.net.pi.model.PiPipeconfId;
-import org.onosproject.net.pi.service.PiPipeconfConfig;
 import org.onosproject.net.pi.service.PiPipeconfMappingStore;
 import org.onosproject.net.pi.service.PiPipeconfService;
 import org.osgi.service.component.annotations.Activate;
@@ -73,7 +70,6 @@
     private final Logger log = getLogger(getClass());
 
     private static final String MERGED_DRIVER_SEPARATOR = ":";
-    private static final String CFG_SCHEME = "piPipeconf";
 
     private static final int MISSING_DRIVER_WATCHDOG_INTERVAL = 5; // Seconds.
 
@@ -98,19 +94,8 @@
     protected ExecutorService executor = Executors.newFixedThreadPool(
             10, groupedThreads("onos/pipeconf-manager", "%d", log));
 
-    protected final ConfigFactory configFactory =
-            new ConfigFactory<DeviceId, PiPipeconfConfig>(
-                    SubjectFactories.DEVICE_SUBJECT_FACTORY,
-                    PiPipeconfConfig.class, CFG_SCHEME) {
-                @Override
-                public PiPipeconfConfig createConfig() {
-                    return new PiPipeconfConfig();
-                }
-            };
-
     @Activate
     public void activate() {
-        cfgService.registerConfigFactory(configFactory);
         driverAdminService.addListener(driverListener);
         checkMissingMergedDrivers();
         if (!missingMergedDrivers.isEmpty()) {
@@ -127,7 +112,6 @@
     @Deactivate
     public void deactivate() {
         executor.shutdown();
-        cfgService.unregisterConfigFactory(configFactory);
         driverAdminService.removeListener(driverListener);
         pipeconfs.clear();
         missingMergedDrivers.clear();
@@ -231,7 +215,7 @@
             if (getDriver(newDriverName) != null) {
                 return newDriverName;
             }
-            log.info("Creating merged driver {}...", newDriverName);
+            log.debug("Creating merged driver {}...", newDriverName);
             final Driver mergedDriver = buildMergedDriver(
                     pipeconfId, baseDriverName, newDriverName);
             if (mergedDriver == null) {
diff --git a/core/net/src/main/java/org/onosproject/net/pi/impl/PiPipeconfWatchdogManager.java b/core/net/src/main/java/org/onosproject/net/pi/impl/PiPipeconfWatchdogManager.java
index 62fb157..eafb778 100644
--- a/core/net/src/main/java/org/onosproject/net/pi/impl/PiPipeconfWatchdogManager.java
+++ b/core/net/src/main/java/org/onosproject/net/pi/impl/PiPipeconfWatchdogManager.java
@@ -17,6 +17,7 @@
 package org.onosproject.net.pi.impl;
 
 import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.Striped;
 import org.onlab.util.KryoNamespace;
 import org.onlab.util.Tools;
@@ -250,7 +251,7 @@
         if (!handshaker.isConnected()) {
             return false;
         }
-        if (pipelineProg.isPipeconfSet(pipeconf)) {
+        if (Futures.getUnchecked(pipelineProg.isPipeconfSet(pipeconf))) {
             log.debug("Pipeconf {} already configured on {}",
                       pipeconf.id(), device.id());
             return true;
diff --git a/core/net/src/test/java/org/onosproject/net/pi/impl/PiPipeconfManagerTest.java b/core/net/src/test/java/org/onosproject/net/pi/impl/PiPipeconfManagerTest.java
index 20ded6a..de778a0 100644
--- a/core/net/src/test/java/org/onosproject/net/pi/impl/PiPipeconfManagerTest.java
+++ b/core/net/src/test/java/org/onosproject/net/pi/impl/PiPipeconfManagerTest.java
@@ -44,7 +44,6 @@
 import org.onosproject.net.driver.DriverProvider;
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.net.pi.model.PiPipeconfId;
-import org.onosproject.net.pi.service.PiPipeconfConfig;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -55,7 +54,6 @@
 import java.util.Set;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.onosproject.pipelines.basic.PipeconfLoader.BASIC_PIPECONF;
 
@@ -77,9 +75,6 @@
     private final Set<NetworkConfigListener> netCfgListeners = new HashSet<>();
     private final Set<DriverProvider> providers = new HashSet<>();
 
-    private final PiPipeconfConfig piPipeconfConfig = new PiPipeconfConfig();
-    private final InputStream jsonStream = PiPipeconfManagerTest.class
-            .getResourceAsStream("/org/onosproject/net/pi/impl/piPipeconfId.json");
     private final BasicDeviceConfig basicDeviceConfig = new BasicDeviceConfig();
     private final InputStream jsonStreamBasic = PiPipeconfManagerTest.class
             .getResourceAsStream("/org/onosproject/net/pi/impl/basic.json");
@@ -95,11 +90,8 @@
         piPipeconf = BASIC_PIPECONF;
         piPipeconfService.cfgService = cfgService;
         piPipeconfService.driverAdminService = driverAdminService;
-        String key = "piPipeconf";
         ObjectMapper mapper = new ObjectMapper();
-        JsonNode jsonNode = mapper.readTree(jsonStream);
         ConfigApplyDelegate delegate = new MockDelegate();
-        piPipeconfConfig.init(DEVICE_ID, key, jsonNode, mapper, delegate);
         String keyBasic = "basic";
         JsonNode jsonNodeBasic = mapper.readTree(jsonStreamBasic);
         basicDeviceConfig.init(DEVICE_ID, keyBasic, jsonNodeBasic, mapper, delegate);
@@ -111,7 +103,6 @@
         assertEquals("Incorrect driver admin service", driverAdminService, piPipeconfService.driverAdminService);
         assertEquals("Incorrect driverAdminService service", driverAdminService, piPipeconfService.driverAdminService);
         assertEquals("Incorrect configuration service", cfgService, piPipeconfService.cfgService);
-        assertTrue("Incorrect config factory", cfgFactories.contains(piPipeconfService.configFactory));
     }
 
     @Test
@@ -120,7 +111,6 @@
         assertEquals("Incorrect driver admin service", null, piPipeconfService.driverAdminService);
         assertEquals("Incorrect driverAdminService service", null, piPipeconfService.driverAdminService);
         assertEquals("Incorrect configuration service", null, piPipeconfService.cfgService);
-        assertFalse("Config factory should be unregistered", cfgFactories.contains(piPipeconfService.configFactory));
     }
 
     @Test
@@ -139,7 +129,8 @@
 
     @Test
     public void mergeDriver() {
-        PiPipeconfId piPipeconfId = cfgService.getConfig(DEVICE_ID, PiPipeconfConfig.class).piPipeconfId();
+        PiPipeconfId piPipeconfId = new PiPipeconfId(cfgService.getConfig(
+                DEVICE_ID, BasicDeviceConfig.class).pipeconf());
         assertEquals(piPipeconf.id(), piPipeconfId);
 
         String baseDriverName = cfgService.getConfig(DEVICE_ID, BasicDeviceConfig.class).driver();
@@ -196,10 +187,7 @@
         @Override
         public <S, C extends Config<S>> C getConfig(S subject, Class<C> configClass) {
             DeviceId did = (DeviceId) subject;
-            if (configClass.equals(PiPipeconfConfig.class)
-                    && did.equals(DEVICE_ID)) {
-                return (C) piPipeconfConfig;
-            } else if (configClass.equals(BasicDeviceConfig.class)
+            if (configClass.equals(BasicDeviceConfig.class)
                     && did.equals(DEVICE_ID)) {
                 return (C) basicDeviceConfig;
             }
diff --git a/core/net/src/test/resources/org/onosproject/net/pi/impl/basic.json b/core/net/src/test/resources/org/onosproject/net/pi/impl/basic.json
index 42de9ad..ad2a205 100644
--- a/core/net/src/test/resources/org/onosproject/net/pi/impl/basic.json
+++ b/core/net/src/test/resources/org/onosproject/net/pi/impl/basic.json
@@ -1,3 +1,4 @@
 {
-    "driver": "baseDriver"
-}
\ No newline at end of file
+    "driver": "baseDriver",
+    "pipeconf": "org.onosproject.pipelines.basic"
+}
diff --git a/core/net/src/test/resources/org/onosproject/net/pi/impl/piPipeconfId.json b/core/net/src/test/resources/org/onosproject/net/pi/impl/piPipeconfId.json
deleted file mode 100644
index 7f42958..0000000
--- a/core/net/src/test/resources/org/onosproject/net/pi/impl/piPipeconfId.json
+++ /dev/null
@@ -1,3 +0,0 @@
-{
-    "piPipeconfId": "org.onosproject.pipelines.basic"
-}
\ No newline at end of file
diff --git a/core/store/serializers/src/main/java/org/onosproject/store/serializers/KryoNamespaces.java b/core/store/serializers/src/main/java/org/onosproject/store/serializers/KryoNamespaces.java
index 7dc903a..e2b826d 100644
--- a/core/store/serializers/src/main/java/org/onosproject/store/serializers/KryoNamespaces.java
+++ b/core/store/serializers/src/main/java/org/onosproject/store/serializers/KryoNamespaces.java
@@ -212,13 +212,13 @@
 import org.onosproject.net.pi.model.PiActionId;
 import org.onosproject.net.pi.model.PiActionParamId;
 import org.onosproject.net.pi.model.PiActionProfileId;
-import org.onosproject.net.pi.model.PiPacketMetadataId;
 import org.onosproject.net.pi.model.PiCounterId;
 import org.onosproject.net.pi.model.PiCounterType;
 import org.onosproject.net.pi.model.PiMatchFieldId;
 import org.onosproject.net.pi.model.PiMatchType;
 import org.onosproject.net.pi.model.PiMeterId;
 import org.onosproject.net.pi.model.PiMeterType;
+import org.onosproject.net.pi.model.PiPacketMetadataId;
 import org.onosproject.net.pi.model.PiPacketOperationType;
 import org.onosproject.net.pi.model.PiPipeconfId;
 import org.onosproject.net.pi.model.PiTableId;
@@ -231,7 +231,6 @@
 import org.onosproject.net.pi.runtime.PiActionProfileMember;
 import org.onosproject.net.pi.runtime.PiActionProfileMemberHandle;
 import org.onosproject.net.pi.runtime.PiActionProfileMemberId;
-import org.onosproject.net.pi.runtime.PiPacketMetadata;
 import org.onosproject.net.pi.runtime.PiCounterCell;
 import org.onosproject.net.pi.runtime.PiCounterCellData;
 import org.onosproject.net.pi.runtime.PiCounterCellId;
@@ -244,13 +243,13 @@
 import org.onosproject.net.pi.runtime.PiLpmFieldMatch;
 import org.onosproject.net.pi.runtime.PiMatchKey;
 import org.onosproject.net.pi.runtime.PiMeterCellId;
+import org.onosproject.net.pi.runtime.PiPacketMetadata;
 import org.onosproject.net.pi.runtime.PiPacketOperation;
 import org.onosproject.net.pi.runtime.PiRangeFieldMatch;
 import org.onosproject.net.pi.runtime.PiTableAction;
 import org.onosproject.net.pi.runtime.PiTableEntry;
 import org.onosproject.net.pi.runtime.PiTableEntryHandle;
 import org.onosproject.net.pi.runtime.PiTernaryFieldMatch;
-import org.onosproject.net.pi.service.PiPipeconfConfig;
 import org.onosproject.net.pi.service.PiTranslatable;
 import org.onosproject.net.pi.service.PiTranslatedEntity;
 import org.onosproject.net.provider.ProviderId;
@@ -708,7 +707,6 @@
                     PiLpmFieldMatch.class,
                     PiMatchKey.class,
                     PiPacketOperation.class,
-                    PiPipeconfConfig.class,
                     PiRangeFieldMatch.class,
                     PiTableAction.class,
                     PiTableEntry.class,
diff --git a/drivers/gnmi/src/main/java/org/onosproject/drivers/gnmi/AbstractGnmiHandlerBehaviour.java b/drivers/gnmi/src/main/java/org/onosproject/drivers/gnmi/AbstractGnmiHandlerBehaviour.java
index 3bfca46..4ce9f18 100644
--- a/drivers/gnmi/src/main/java/org/onosproject/drivers/gnmi/AbstractGnmiHandlerBehaviour.java
+++ b/drivers/gnmi/src/main/java/org/onosproject/drivers/gnmi/AbstractGnmiHandlerBehaviour.java
@@ -16,17 +16,22 @@
 
 package org.onosproject.drivers.gnmi;
 
+import com.google.common.base.Strings;
 import io.grpc.StatusRuntimeException;
 import org.onosproject.gnmi.api.GnmiClient;
 import org.onosproject.gnmi.api.GnmiClientKey;
 import org.onosproject.gnmi.api.GnmiController;
 import org.onosproject.net.Device;
 import org.onosproject.net.DeviceId;
+import org.onosproject.net.config.NetworkConfigService;
+import org.onosproject.net.config.basics.BasicDeviceConfig;
 import org.onosproject.net.device.DeviceService;
 import org.onosproject.net.driver.AbstractHandlerBehaviour;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -41,9 +46,6 @@
     private static final String DEVICE_REQ_TIMEOUT = "deviceRequestTimeout";
     private static final int DEFAULT_DEVICE_REQ_TIMEOUT = 60;
 
-    private static final String GNMI_SERVER_ADDR_KEY = "gnmi_ip";
-    private static final String GNMI_SERVER_PORT_KEY = "gnmi_port";
-
     protected final Logger log = LoggerFactory.getLogger(getClass());
     protected DeviceId deviceId;
     protected DeviceService deviceService;
@@ -65,32 +67,33 @@
         return true;
     }
 
-    GnmiClient createClient() {
+    GnmiClient getClientByKey() {
+        final GnmiClientKey clientKey = clientKey();
+        if (clientKey == null) {
+            return null;
+        }
+        return handler().get(GnmiController.class).getClient(clientKey);
+    }
+
+    protected GnmiClientKey clientKey() {
         deviceId = handler().data().deviceId();
-        controller = handler().get(GnmiController.class);
 
-        final String serverAddr = this.data().value(GNMI_SERVER_ADDR_KEY);
-        final String serverPortString = this.data().value(GNMI_SERVER_PORT_KEY);
-
-        if (serverAddr == null || serverPortString == null) {
-            log.warn("Unable to create client for {}, missing driver data key (required is {} and {})",
-                    deviceId, GNMI_SERVER_ADDR_KEY, GNMI_SERVER_PORT_KEY);
+        final BasicDeviceConfig cfg = handler().get(NetworkConfigService.class)
+                .getConfig(deviceId, BasicDeviceConfig.class);
+        if (cfg == null || Strings.isNullOrEmpty(cfg.managementAddress())) {
+            log.error("Missing or invalid config for {}, cannot derive " +
+                              "gNMI server endpoints", deviceId);
             return null;
         }
 
-        final int serverPort;
         try {
-            serverPort = Integer.parseUnsignedInt(serverPortString);
-        } catch (NumberFormatException e) {
-            log.error("{} is not a valid port number", serverPortString);
+            return new GnmiClientKey(
+                    deviceId, new URI(cfg.managementAddress()));
+        } catch (URISyntaxException e) {
+            log.error("Management address of {} is not a valid URI: {}",
+                      deviceId, cfg.managementAddress());
             return null;
         }
-        GnmiClientKey clientKey = new GnmiClientKey(deviceId, serverAddr, serverPort);
-        if (!controller.createClient(clientKey)) {
-            log.warn("Unable to create client for {}, aborting operation", deviceId);
-            return null;
-        }
-        return controller.getClient(deviceId);
     }
 
     /**
diff --git a/drivers/gnmi/src/main/java/org/onosproject/drivers/gnmi/GnmiHandshaker.java b/drivers/gnmi/src/main/java/org/onosproject/drivers/gnmi/GnmiHandshaker.java
index c443460..a6d0c04 100644
--- a/drivers/gnmi/src/main/java/org/onosproject/drivers/gnmi/GnmiHandshaker.java
+++ b/drivers/gnmi/src/main/java/org/onosproject/drivers/gnmi/GnmiHandshaker.java
@@ -17,32 +17,43 @@
 package org.onosproject.drivers.gnmi;
 
 import org.onosproject.gnmi.api.GnmiClient;
+import org.onosproject.gnmi.api.GnmiClientKey;
 import org.onosproject.gnmi.api.GnmiController;
-import org.onosproject.net.DeviceId;
 import org.onosproject.net.MastershipRole;
 import org.onosproject.net.device.DeviceHandshaker;
 
 import java.util.concurrent.CompletableFuture;
 
+import static java.util.concurrent.CompletableFuture.completedFuture;
+
 /**
  * Implementation of DeviceHandshaker for gNMI.
  */
 public class GnmiHandshaker extends AbstractGnmiHandlerBehaviour implements DeviceHandshaker {
 
     @Override
-    public CompletableFuture<Boolean> isReachable() {
-        return CompletableFuture
-                // gNMI requires a client to be created to
-                // check for reachability.
-                .supplyAsync(super::createClient)
-                .thenApplyAsync(client -> {
-                    if (client == null) {
-                        return false;
-                    }
-                    return handler()
-                            .get(GnmiController.class)
-                            .isReachable(handler().data().deviceId());
-                });
+    public boolean isReachable() {
+        final GnmiClient client = getClientByKey();
+        return client != null && client.isServerReachable();
+    }
+
+    @Override
+    public CompletableFuture<Boolean> probeReachability() {
+        final GnmiClient client = getClientByKey();
+        if (client == null) {
+            return completedFuture(false);
+        }
+        return client.probeService();
+    }
+
+    @Override
+    public boolean isAvailable() {
+        return isReachable();
+    }
+
+    @Override
+    public CompletableFuture<Boolean> probeAvailability() {
+        return probeReachability();
     }
 
     @Override
@@ -57,43 +68,30 @@
 
     @Override
     public CompletableFuture<Boolean> connect() {
-        return CompletableFuture
-                .supplyAsync(super::createClient)
-                .thenComposeAsync(client -> {
-                    if (client == null) {
-                        return CompletableFuture.completedFuture(false);
-                    }
-                    return CompletableFuture.completedFuture(true);
-                });
+        return CompletableFuture.supplyAsync(this::createClient);
+    }
+
+    private boolean createClient() {
+        GnmiClientKey clientKey = clientKey();
+        if (clientKey == null) {
+            return false;
+        }
+        if (!handler().get(GnmiController.class).createClient(clientKey)) {
+            log.warn("Unable to create client for {}",
+                     handler().data().deviceId());
+            return false;
+        }
+        return true;
     }
 
     @Override
     public boolean isConnected() {
-        final GnmiController controller = handler().get(GnmiController.class);
-        final DeviceId deviceId = handler().data().deviceId();
-        final GnmiClient client = controller.getClient(deviceId);
-
-        if (client == null) {
-            return false;
-        }
-
-        return getFutureWithDeadline(
-                client.isServiceAvailable(),
-                "checking if gNMI service is available", false);
+        return getClientByKey() != null;
     }
 
     @Override
-    public CompletableFuture<Boolean> disconnect() {
-        final GnmiController controller = handler().get(GnmiController.class);
-        final DeviceId deviceId = handler().data().deviceId();
-        final GnmiClient client = controller.getClient(deviceId);
-        if (client == null) {
-            return CompletableFuture.completedFuture(true);
-        }
-        return client.shutdown()
-                .thenApplyAsync(v -> {
-                    controller.removeClient(deviceId);
-                    return true;
-                });
+    public void disconnect() {
+        handler().get(GnmiController.class)
+                .removeClient(handler().data().deviceId());
     }
 }
diff --git a/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/AbstractP4RuntimeHandlerBehaviour.java b/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/AbstractP4RuntimeHandlerBehaviour.java
index 8a9a403..5326861 100644
--- a/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/AbstractP4RuntimeHandlerBehaviour.java
+++ b/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/AbstractP4RuntimeHandlerBehaviour.java
@@ -16,13 +16,14 @@
 
 package org.onosproject.drivers.p4runtime;
 
-import io.grpc.StatusRuntimeException;
+import com.google.common.base.Strings;
 import org.onosproject.net.Device;
 import org.onosproject.net.DeviceId;
+import org.onosproject.net.config.NetworkConfigService;
+import org.onosproject.net.config.basics.BasicDeviceConfig;
 import org.onosproject.net.device.DeviceService;
 import org.onosproject.net.driver.AbstractHandlerBehaviour;
 import org.onosproject.net.pi.model.PiPipeconf;
-import org.onosproject.net.pi.model.PiPipeconfId;
 import org.onosproject.net.pi.model.PiPipelineInterpreter;
 import org.onosproject.net.pi.service.PiPipeconfService;
 import org.onosproject.net.pi.service.PiTranslationService;
@@ -32,10 +33,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.net.URI;
+import java.net.URISyntaxException;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
@@ -44,14 +43,6 @@
  */
 public class AbstractP4RuntimeHandlerBehaviour extends AbstractHandlerBehaviour {
 
-    // Default timeout in seconds for device operations.
-    private static final String DEVICE_REQ_TIMEOUT = "deviceRequestTimeout";
-    private static final int DEFAULT_DEVICE_REQ_TIMEOUT = 60;
-
-    private static final String P4RUNTIME_SERVER_ADDR_KEY = "p4runtime_ip";
-    private static final String P4RUNTIME_SERVER_PORT_KEY = "p4runtime_port";
-    private static final String P4RUNTIME_DEVICE_ID_KEY = "p4runtime_deviceId";
-
     protected final Logger log = LoggerFactory.getLogger(getClass());
 
     // Initialized by setupBehaviour()
@@ -77,30 +68,25 @@
         deviceService = handler().get(DeviceService.class);
         device = deviceService.getDevice(deviceId);
         if (device == null) {
-            log.warn("Unable to find device with id {}, aborting operation", deviceId);
+            log.warn("Unable to find device with id {}", deviceId);
             return false;
         }
 
         controller = handler().get(P4RuntimeController.class);
         client = controller.getClient(deviceId);
         if (client == null) {
-            log.warn("Unable to find client for {}, aborting operation", deviceId);
+            log.warn("Unable to find client for {}", deviceId);
             return false;
         }
 
         PiPipeconfService piPipeconfService = handler().get(PiPipeconfService.class);
-        if (!piPipeconfService.ofDevice(deviceId).isPresent()) {
+        if (!piPipeconfService.getPipeconf(deviceId).isPresent()) {
             log.warn("Unable to get assigned pipeconf for {} (mapping " +
-                             "missing in PiPipeconfService), aborting operation",
+                             "missing in PiPipeconfService)",
                      deviceId);
             return false;
         }
-        PiPipeconfId pipeconfId = piPipeconfService.ofDevice(deviceId).get();
-        if (!piPipeconfService.getPipeconf(pipeconfId).isPresent()) {
-            log.warn("Cannot find any pipeconf with ID '{}' ({}), aborting operation", pipeconfId, deviceId);
-            return false;
-        }
-        pipeconf = piPipeconfService.getPipeconf(pipeconfId).get();
+        pipeconf = piPipeconfService.getPipeconf(deviceId).get();
 
         translationService = handler().get(PiTranslationService.class);
 
@@ -123,49 +109,38 @@
     }
 
     /**
-     * Returns a P4Runtime client for this device, null if such client cannot be
-     * created.
+     * Returns a P4Runtime client previsouly created for this device, null if
+     * such client does not exist.
      *
      * @return client or null
      */
-    P4RuntimeClient createClient() {
+    P4RuntimeClient getClientByKey() {
+        final P4RuntimeClientKey clientKey = clientKey();
+        if (clientKey == null) {
+            return null;
+        }
+        return handler().get(P4RuntimeController.class).getClient(clientKey);
+    }
+
+    protected P4RuntimeClientKey clientKey() {
         deviceId = handler().data().deviceId();
-        controller = handler().get(P4RuntimeController.class);
 
-        final String serverAddr = this.data().value(P4RUNTIME_SERVER_ADDR_KEY);
-        final String serverPortString = this.data().value(P4RUNTIME_SERVER_PORT_KEY);
-        final String p4DeviceIdString = this.data().value(P4RUNTIME_DEVICE_ID_KEY);
-
-        if (serverAddr == null || serverPortString == null || p4DeviceIdString == null) {
-            log.warn("Unable to create client for {}, missing driver data key (required is {}, {}, and {})",
-                     deviceId, P4RUNTIME_SERVER_ADDR_KEY, P4RUNTIME_SERVER_PORT_KEY, P4RUNTIME_DEVICE_ID_KEY);
+        final BasicDeviceConfig cfg = handler().get(NetworkConfigService.class)
+                .getConfig(deviceId, BasicDeviceConfig.class);
+        if (cfg == null || Strings.isNullOrEmpty(cfg.managementAddress())) {
+            log.error("Missing or invalid config for {}, cannot derive " +
+                              "P4Runtime server endpoints", deviceId);
             return null;
         }
 
-        final int serverPort;
-        final long p4DeviceId;
-
         try {
-            serverPort = Integer.parseUnsignedInt(serverPortString);
-        } catch (NumberFormatException e) {
-            log.error("{} is not a valid P4Runtime port number", serverPortString);
+            return new P4RuntimeClientKey(
+                    deviceId, new URI(cfg.managementAddress()));
+        } catch (URISyntaxException e) {
+            log.error("Management address of {} is not a valid URI: {}",
+                      deviceId, cfg.managementAddress());
             return null;
         }
-        try {
-            p4DeviceId = Long.parseUnsignedLong(p4DeviceIdString);
-        } catch (NumberFormatException e) {
-            log.error("{} is not a valid P4Runtime-internal device ID", p4DeviceIdString);
-            return null;
-        }
-
-        final P4RuntimeClientKey clientKey = new P4RuntimeClientKey(
-                deviceId, serverAddr, serverPort, p4DeviceId);
-        if (!controller.createClient(clientKey)) {
-            log.debug("Unable to create client for {}, aborting operation", deviceId);
-            return null;
-        }
-
-        return controller.getClient(deviceId);
     }
 
     /**
@@ -184,58 +159,4 @@
             return Boolean.parseBoolean(handler().driver().getProperty(propName));
         }
     }
-
-    /**
-     * Returns the device request timeout driver property, or a default value
-     * if the property is not present or cannot be parsed.
-     *
-     * @return timeout value
-     */
-    private int getDeviceRequestTimeout() {
-        final String timeout = handler().driver()
-                .getProperty(DEVICE_REQ_TIMEOUT);
-        if (timeout == null) {
-            return DEFAULT_DEVICE_REQ_TIMEOUT;
-        } else {
-            try {
-                return Integer.parseInt(timeout);
-            } catch (NumberFormatException e) {
-                log.error("{} driver property '{}' is not a number, using default value {}",
-                          DEVICE_REQ_TIMEOUT, timeout, DEFAULT_DEVICE_REQ_TIMEOUT);
-                return DEFAULT_DEVICE_REQ_TIMEOUT;
-            }
-        }
-    }
-
-    /**
-     * Convenience method to get the result of a completable future while
-     * setting a timeout and checking for exceptions.
-     *
-     * @param future        completable future
-     * @param opDescription operation description to use in log messages. Should
-     *                      be a sentence starting with a verb ending in -ing,
-     *                      e.g. "reading...", "writing...", etc.
-     * @param defaultValue  value to return if operation fails
-     * @param <U>           type of returned value
-     * @return future result or default value
-     */
-    <U> U getFutureWithDeadline(CompletableFuture<U> future, String opDescription,
-                                U defaultValue) {
-        try {
-            return future.get(getDeviceRequestTimeout(), TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-            log.error("Exception while {} on {}", opDescription, deviceId);
-        } catch (ExecutionException e) {
-            final Throwable cause = e.getCause();
-            if (cause instanceof StatusRuntimeException) {
-                final StatusRuntimeException grpcError = (StatusRuntimeException) cause;
-                log.warn("Error while {} on {}: {}", opDescription, deviceId, grpcError.getMessage());
-            } else {
-                log.error("Exception while {} on {}", opDescription, deviceId, e.getCause());
-            }
-        } catch (TimeoutException e) {
-            log.error("Operation TIMEOUT while {} on {}", opDescription, deviceId);
-        }
-        return defaultValue;
-    }
 }
diff --git a/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/AbstractP4RuntimePipelineProgrammable.java b/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/AbstractP4RuntimePipelineProgrammable.java
index 664bd59..916a945 100644
--- a/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/AbstractP4RuntimePipelineProgrammable.java
+++ b/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/AbstractP4RuntimePipelineProgrammable.java
@@ -16,7 +16,6 @@
 
 package org.onosproject.drivers.p4runtime;
 
-import org.onlab.util.SharedExecutors;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.behaviour.PiPipelineProgrammable;
 import org.onosproject.net.pi.model.PiPipeconf;
@@ -28,6 +27,7 @@
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
+import static java.util.concurrent.CompletableFuture.completedFuture;
 import static org.slf4j.LoggerFactory.getLogger;
 
 /**
@@ -49,59 +49,42 @@
 
     @Override
     public CompletableFuture<Boolean> setPipeconf(PiPipeconf pipeconf) {
-        return CompletableFuture.supplyAsync(
-                () -> doDeployConfig(pipeconf),
-                SharedExecutors.getPoolThreadExecutor());
-    }
-
-    private boolean doDeployConfig(PiPipeconf pipeconf) {
-
         DeviceId deviceId = handler().data().deviceId();
         P4RuntimeController controller = handler().get(P4RuntimeController.class);
 
         P4RuntimeClient client = controller.getClient(deviceId);
         if (client == null) {
             log.warn("Unable to find client for {}, aborting pipeconf deploy", deviceId);
-            return false;
+            return completedFuture(false);
         }
 
         ByteBuffer deviceDataBuffer = createDeviceDataBuffer(pipeconf);
         if (deviceDataBuffer == null) {
             // Hopefully the child class logged the problem.
-            return false;
+            return completedFuture(false);
         }
 
-        final Boolean deploySuccess = getFutureWithDeadline(
-                client.setPipelineConfig(pipeconf, deviceDataBuffer),
-                "deploying pipeconf", null);
-        if (deploySuccess == null) {
-            return false;
-        } else if (!deploySuccess) {
-            log.warn("Unable to deploy pipeconf {} to {}", pipeconf.id(), deviceId);
-            return false;
-        }
-
-        return true;
+        return client.setPipelineConfig(pipeconf, deviceDataBuffer);
     }
 
     @Override
-    public boolean isPipeconfSet(PiPipeconf pipeconf) {
+    public CompletableFuture<Boolean> isPipeconfSet(PiPipeconf pipeconf) {
         DeviceId deviceId = handler().data().deviceId();
         P4RuntimeController controller = handler().get(P4RuntimeController.class);
 
         P4RuntimeClient client = controller.getClient(deviceId);
         if (client == null) {
             log.warn("Unable to find client for {}, cannot check if pipeconf is set", deviceId);
-            return false;
+            return completedFuture(false);
         }
 
         ByteBuffer deviceDataBuffer = createDeviceDataBuffer(pipeconf);
         if (deviceDataBuffer == null) {
             // Hopefully the child class logged the problem.
-            return false;
+            return completedFuture(false);
         }
 
-        return client.isPipelineConfigSetSync(pipeconf, deviceDataBuffer);
+        return client.isPipelineConfigSet(pipeconf, deviceDataBuffer);
     }
 
     @Override
diff --git a/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/P4RuntimeDeviceDescriptionDiscovery.java b/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/P4RuntimeDeviceDescriptionDiscovery.java
new file mode 100644
index 0000000..337ce6a
--- /dev/null
+++ b/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/P4RuntimeDeviceDescriptionDiscovery.java
@@ -0,0 +1,60 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.drivers.p4runtime;
+
+import org.onlab.packet.ChassisId;
+import org.onosproject.net.AnnotationKeys;
+import org.onosproject.net.DefaultAnnotations;
+import org.onosproject.net.Device;
+import org.onosproject.net.device.DefaultDeviceDescription;
+import org.onosproject.net.device.DeviceDescription;
+import org.onosproject.net.device.DeviceDescriptionDiscovery;
+import org.onosproject.net.device.PortDescription;
+import org.onosproject.net.driver.AbstractHandlerBehaviour;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Implementation of DeviceDescriptionDiscovery for P4Runtime devices.
+ */
+public class P4RuntimeDeviceDescriptionDiscovery
+        extends AbstractHandlerBehaviour implements DeviceDescriptionDiscovery {
+
+    private static final String UNKNOWN = "unknown";
+
+    @Override
+    public DeviceDescription discoverDeviceDetails() {
+        return new DefaultDeviceDescription(
+                data().deviceId().uri(),
+                Device.Type.SWITCH,
+                data().driver().manufacturer(),
+                data().driver().hwVersion(),
+                data().driver().swVersion(),
+                UNKNOWN,
+                new ChassisId(),
+                false,
+                DefaultAnnotations.builder()
+                        .set(AnnotationKeys.PROTOCOL, "P4Runtime")
+                        .build());
+    }
+
+    @Override
+    public List<PortDescription> discoverPortDetails() {
+        return Collections.emptyList();
+    }
+}
diff --git a/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/P4RuntimeHandshaker.java b/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/P4RuntimeHandshaker.java
index 4d84b2a..60b0dbc 100644
--- a/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/P4RuntimeHandshaker.java
+++ b/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/P4RuntimeHandshaker.java
@@ -16,86 +16,142 @@
 
 package org.onosproject.drivers.p4runtime;
 
-import org.onosproject.net.DeviceId;
+import org.onosproject.cluster.ClusterService;
 import org.onosproject.net.MastershipRole;
 import org.onosproject.net.device.DeviceAgentListener;
 import org.onosproject.net.device.DeviceHandshaker;
+import org.onosproject.net.pi.service.PiPipeconfWatchdogService;
 import org.onosproject.net.provider.ProviderId;
 import org.onosproject.p4runtime.api.P4RuntimeClient;
+import org.onosproject.p4runtime.api.P4RuntimeClientKey;
 import org.onosproject.p4runtime.api.P4RuntimeController;
 
+import java.math.BigInteger;
 import java.util.concurrent.CompletableFuture;
 
+import static java.util.concurrent.CompletableFuture.completedFuture;
+
 /**
  * Implementation of DeviceHandshaker for P4Runtime.
  */
 public class P4RuntimeHandshaker extends AbstractP4RuntimeHandlerBehaviour implements DeviceHandshaker {
 
+    // This is needed to compute an election ID based on mastership term and
+    // preference. At the time of writing the practical maximum cluster size is
+    // 9. Since election IDs are 128bit numbers, we should'nt be too worried of
+    // being conservative when setting a static max size here. Making the
+    // cluster size dynamic would likely cause conflicts when generating
+    // election IDs (e.g. two nodes seeing different cluster sizes).
+    private static final int MAX_CLUSTER_SIZE = 20;
+
     @Override
     public CompletableFuture<Boolean> connect() {
         return CompletableFuture
-                .supplyAsync(super::createClient)
-                .thenApplyAsync(client -> {
-                    if (client == null) {
-                        return false;
-                    }
-                    client.openSession();
-                    return true;
-                });
+                .supplyAsync(this::createClient);
+    }
+
+    private boolean createClient() {
+        final P4RuntimeClientKey clientKey = clientKey();
+        if (clientKey == null) {
+            return false;
+        }
+        if (!handler().get(P4RuntimeController.class).createClient(clientKey)) {
+            log.debug("Unable to create client for {}", data().deviceId());
+            return false;
+        }
+        return true;
     }
 
     @Override
     public boolean isConnected() {
-        final P4RuntimeController controller = handler().get(P4RuntimeController.class);
-        final DeviceId deviceId = handler().data().deviceId();
-        final P4RuntimeClient client = controller.getClient(deviceId);
-        return client != null && client.isSessionOpen();
+        // This is based on the client key obtained from the current netcfg. If
+        // a client already exists for this device, but the netcfg with the
+        // server endpoints has changed, this will return false.
+        return getClientByKey() != null;
     }
 
     @Override
-    public CompletableFuture<Boolean> disconnect() {
-        final P4RuntimeController controller = handler().get(P4RuntimeController.class);
-        final DeviceId deviceId = handler().data().deviceId();
-        final P4RuntimeClient client = controller.getClient(deviceId);
+    public void disconnect() {
+        // This removes clients associated with this device ID, even if the
+        // netcfg has changed and so the client key for this device.
+        handler().get(P4RuntimeController.class).removeClient(data().deviceId());
+    }
+
+    @Override
+    public boolean isReachable() {
+        final P4RuntimeClient client = getClientByKey();
+        return client != null && client.isServerReachable();
+    }
+
+    @Override
+    public CompletableFuture<Boolean> probeReachability() {
+        final P4RuntimeClient client = getClientByKey();
         if (client == null) {
-            return CompletableFuture.completedFuture(true);
+            return completedFuture(false);
         }
-        return client.shutdown()
-                .thenApplyAsync(v -> {
-                    controller.removeClient(deviceId);
-                    return true;
-                });
+        return client.probeService();
     }
 
     @Override
-    public CompletableFuture<Boolean> isReachable() {
-        return CompletableFuture
-                // P4RuntimeController requires a client to be created to
-                // check for reachability.
-                .supplyAsync(super::createClient)
-                .thenApplyAsync(client -> {
-                    if (client == null) {
-                        return false;
-                    }
-                    return handler()
-                            .get(P4RuntimeController.class)
-                            .isReachable(handler().data().deviceId());
-                });
+    public boolean isAvailable() {
+        // To be available, we require a session open (for packet in/out) and a
+        // pipeline config set.
+        final P4RuntimeClient client = getClientByKey();
+        if (client == null || !client.isServerReachable() || !client.isSessionOpen()) {
+            return false;
+        }
+        // Since we cannot probe the device, we rely on what's known by the
+        // pipeconf watchdog service.
+        return PiPipeconfWatchdogService.PipelineStatus.READY.equals(
+                handler().get(PiPipeconfWatchdogService.class)
+                        .getStatus(data().deviceId()));
+    }
+
+    @Override
+    public CompletableFuture<Boolean> probeAvailability() {
+        // To be available, we require a session open (for packet in/out) and a
+        // pipeline config set.
+        final P4RuntimeClient client = getClientByKey();
+        if (client == null || !client.isServerReachable() || !client.isSessionOpen()) {
+            return completedFuture(false);
+        }
+        return client.isAnyPipelineConfigSet();
     }
 
     @Override
     public void roleChanged(MastershipRole newRole) {
-        if (setupBehaviour() && newRole.equals(MastershipRole.MASTER)) {
-            client.runForMastership();
+        if (newRole.equals(MastershipRole.NONE)) {
+            final P4RuntimeClient client = getClientByKey();
+            if (client != null) {
+                log.info("Notified role NONE, closing session...");
+                client.closeSession();
+            }
+        } else {
+            throw new UnsupportedOperationException(
+                    "Use preference-based way for setting MASTER or STANDBY roles");
+        }
+    }
+
+    @Override
+    public void roleChanged(int preference, long term) {
+        if (setupBehaviour()) {
+            final int clusterSize = handler().get(ClusterService.class)
+                    .getNodes().size();
+            if (clusterSize > MAX_CLUSTER_SIZE) {
+                throw new IllegalStateException(
+                        "Cluster too big! Maz size supported is " + MAX_CLUSTER_SIZE);
+            }
+            BigInteger electionId = BigInteger.valueOf(term)
+                    .multiply(BigInteger.valueOf(MAX_CLUSTER_SIZE))
+                    .subtract(BigInteger.valueOf(preference));
+            client.setMastership(preference == 0, electionId);
         }
     }
 
     @Override
     public MastershipRole getRole() {
-        final P4RuntimeController controller = handler().get(P4RuntimeController.class);
-        final DeviceId deviceId = handler().data().deviceId();
-        final P4RuntimeClient client = controller.getClient(deviceId);
-        if (client == null || !client.isSessionOpen()) {
+        final P4RuntimeClient client = getClientByKey();
+        if (client == null || !client.isServerReachable() || !client.isSessionOpen()) {
             return MastershipRole.NONE;
         }
         return client.isMaster() ? MastershipRole.MASTER : MastershipRole.STANDBY;
diff --git a/drivers/p4runtime/src/main/resources/p4runtime-drivers.xml b/drivers/p4runtime/src/main/resources/p4runtime-drivers.xml
index 78188a7..0750805 100644
--- a/drivers/p4runtime/src/main/resources/p4runtime-drivers.xml
+++ b/drivers/p4runtime/src/main/resources/p4runtime-drivers.xml
@@ -16,6 +16,8 @@
   -->
 <drivers>
     <driver name="p4runtime" manufacturer="p4.org" hwVersion="master" swVersion="master">
+        <behaviour api="org.onosproject.net.device.DeviceDescriptionDiscovery"
+                   impl="org.onosproject.drivers.p4runtime.P4RuntimeDeviceDescriptionDiscovery"/>
         <behaviour api="org.onosproject.net.behaviour.TableStatisticsDiscovery"
                    impl="org.onosproject.drivers.p4runtime.P4RuntimeTableStatisticsDiscovery"/>
         <behaviour api="org.onosproject.net.device.DeviceHandshaker"
diff --git a/drivers/stratum/src/main/java/org/onosproject/drivers/stratum/StratumHandshaker.java b/drivers/stratum/src/main/java/org/onosproject/drivers/stratum/StratumHandshaker.java
index 4e169a5..8762af0 100644
--- a/drivers/stratum/src/main/java/org/onosproject/drivers/stratum/StratumHandshaker.java
+++ b/drivers/stratum/src/main/java/org/onosproject/drivers/stratum/StratumHandshaker.java
@@ -16,10 +16,8 @@
 
 package org.onosproject.drivers.stratum;
 
-import io.grpc.StatusRuntimeException;
 import org.onosproject.drivers.gnmi.GnmiHandshaker;
 import org.onosproject.drivers.p4runtime.P4RuntimeHandshaker;
-import org.onosproject.net.DeviceId;
 import org.onosproject.net.MastershipRole;
 import org.onosproject.net.device.DeviceAgentListener;
 import org.onosproject.net.device.DeviceHandshaker;
@@ -27,109 +25,97 @@
 import org.onosproject.net.driver.DriverData;
 import org.onosproject.net.driver.DriverHandler;
 import org.onosproject.net.provider.ProviderId;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
 /**
  * Implementation of DeviceHandshaker for Stratum device.
  */
 public class StratumHandshaker extends AbstractHandlerBehaviour implements DeviceHandshaker {
 
-    private static final Logger log = LoggerFactory.getLogger(StratumHandshaker.class);
-    private static final int DEFAULT_DEVICE_REQ_TIMEOUT = 10;
-
-    private P4RuntimeHandshaker p4RuntimeHandshaker;
-    private GnmiHandshaker gnmiHandshaker;
-    private DeviceId deviceId;
+    private P4RuntimeHandshaker p4runtime;
+    private GnmiHandshaker gnmi;
 
     public StratumHandshaker() {
-        p4RuntimeHandshaker = new P4RuntimeHandshaker();
-        gnmiHandshaker = new GnmiHandshaker();
+        p4runtime = new P4RuntimeHandshaker();
+        gnmi = new GnmiHandshaker();
     }
 
     @Override
     public void setHandler(DriverHandler handler) {
         super.setHandler(handler);
-        p4RuntimeHandshaker.setHandler(handler);
-        gnmiHandshaker.setHandler(handler);
+        p4runtime.setHandler(handler);
+        gnmi.setHandler(handler);
     }
 
     @Override
     public void setData(DriverData data) {
         super.setData(data);
-        p4RuntimeHandshaker.setData(data);
-        gnmiHandshaker.setData(data);
-        deviceId = data.deviceId();
+        p4runtime.setData(data);
+        gnmi.setData(data);
     }
 
     @Override
-    public CompletableFuture<Boolean> isReachable() {
-        return p4RuntimeHandshaker.isReachable()
-                .thenCombine(gnmiHandshaker.isReachable(), Boolean::logicalAnd);
+    public boolean isReachable() {
+        return p4runtime.isReachable() && gnmi.isReachable();
+    }
+
+    @Override
+    public CompletableFuture<Boolean> probeReachability() {
+        return p4runtime.probeReachability();
+    }
+
+    @Override
+    public boolean isAvailable() {
+        // Availability concerns packet forwarding, hence we consider only
+        // P4Runtime.
+        return p4runtime.isAvailable();
+    }
+
+    @Override
+    public CompletableFuture<Boolean> probeAvailability() {
+        return p4runtime.probeAvailability();
     }
 
     @Override
     public void roleChanged(MastershipRole newRole) {
-        p4RuntimeHandshaker.roleChanged(newRole);
-        // gNMI doesn't support mastership handling.
+        p4runtime.roleChanged(newRole);
+    }
+
+    @Override
+    public void roleChanged(int preference, long term) {
+        p4runtime.roleChanged(preference, term);
     }
 
     @Override
     public MastershipRole getRole() {
-        return p4RuntimeHandshaker.getRole();
+        return p4runtime.getRole();
     }
 
     @Override
     public void addDeviceAgentListener(ProviderId providerId, DeviceAgentListener listener) {
-        p4RuntimeHandshaker.addDeviceAgentListener(providerId, listener);
+        p4runtime.addDeviceAgentListener(providerId, listener);
     }
 
     @Override
     public void removeDeviceAgentListener(ProviderId providerId) {
-        p4RuntimeHandshaker.removeDeviceAgentListener(providerId);
+        p4runtime.removeDeviceAgentListener(providerId);
     }
 
     @Override
     public CompletableFuture<Boolean> connect() {
-        return p4RuntimeHandshaker.connect()
-                .thenCombine(gnmiHandshaker.connect(), Boolean::logicalAnd);
+        // We should execute connections in parallel.
+        return p4runtime.connect().thenCombine(gnmi.connect(), Boolean::logicalAnd);
     }
 
     @Override
     public boolean isConnected() {
-        final CompletableFuture<Boolean> p4runtimeConnected =
-                CompletableFuture.supplyAsync(p4RuntimeHandshaker::isConnected);
-        final CompletableFuture<Boolean> gnmiConnected =
-                CompletableFuture.supplyAsync(gnmiHandshaker::isConnected);
-
-        try {
-            return p4runtimeConnected
-                    .thenCombine(gnmiConnected, Boolean::logicalAnd)
-                    .get(DEFAULT_DEVICE_REQ_TIMEOUT, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-            log.error("Exception while checking connectivity on {}", data().deviceId());
-        } catch (ExecutionException e) {
-            final Throwable cause = e.getCause();
-            if (cause instanceof StatusRuntimeException) {
-                final StatusRuntimeException grpcError = (StatusRuntimeException) cause;
-                log.warn("Error while checking connectivity on {}: {}", deviceId, grpcError.getMessage());
-            } else {
-                log.error("Exception while checking connectivity on {}", deviceId, e.getCause());
-            }
-        } catch (TimeoutException e) {
-            log.error("Operation TIMEOUT while checking connectivity on {}", deviceId);
-        }
-        return false;
+        return p4runtime.isConnected() && gnmi.isConnected();
     }
 
     @Override
-    public CompletableFuture<Boolean> disconnect() {
-        return p4RuntimeHandshaker.disconnect()
-                .thenCombine(gnmiHandshaker.disconnect(), Boolean::logicalAnd);
+    public void disconnect() {
+        p4runtime.disconnect();
+        gnmi.disconnect();
     }
 }
diff --git a/pipelines/fabric/src/main/java/org/onosproject/pipelines/fabric/FabricIntProgrammable.java b/pipelines/fabric/src/main/java/org/onosproject/pipelines/fabric/FabricIntProgrammable.java
index a2bf569..d593262 100644
--- a/pipelines/fabric/src/main/java/org/onosproject/pipelines/fabric/FabricIntProgrammable.java
+++ b/pipelines/fabric/src/main/java/org/onosproject/pipelines/fabric/FabricIntProgrammable.java
@@ -16,9 +16,6 @@
 package org.onosproject.pipelines.fabric;
 
 import com.google.common.collect.Sets;
-import org.osgi.service.component.annotations.Reference;
-import org.osgi.service.component.annotations.ReferenceCardinality;
-
 import org.onosproject.core.ApplicationId;
 import org.onosproject.core.CoreService;
 import org.onosproject.inbandtelemetry.api.IntConfig;
@@ -28,6 +25,7 @@
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.PortNumber;
 import org.onosproject.net.config.NetworkConfigService;
+import org.onosproject.net.device.DeviceService;
 import org.onosproject.net.flow.DefaultFlowRule;
 import org.onosproject.net.flow.DefaultTrafficSelector;
 import org.onosproject.net.flow.DefaultTrafficTreatment;
@@ -44,7 +42,8 @@
 import org.onosproject.net.pi.model.PiTableId;
 import org.onosproject.net.pi.runtime.PiAction;
 import org.onosproject.net.pi.runtime.PiActionParam;
-import org.onosproject.provider.general.device.api.GeneralProviderDeviceConfig;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
 
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -107,30 +106,27 @@
             return false;
         }
 
-        final GeneralProviderDeviceConfig cfg = cfgService.getConfig(
-                deviceId, GeneralProviderDeviceConfig.class);
-        if (cfg == null) {
-            log.warn("Missing GeneralProviderDevice config for {}", deviceId);
-            return false;
-        }
-        final String switchId = cfg.protocolsInfo().containsKey("int") ?
-                cfg.protocolsInfo().get("int").configValues().get("switchId")
-                : null;
-        if (switchId == null || switchId.isEmpty()) {
-            log.warn("Missing INT device config for {}", deviceId);
-            return false;
-        }
+        // FIXME: create config class for INT to allow specifying arbitrary
+        //  switch IDs. The one for the GeneralDeviceProvider was temporary and
+        //  now has been removed. For now we use the chassis ID.
+        // final GeneralProviderDeviceConfig cfg = cfgService.getConfig(
+        //         deviceId, GeneralProviderDeviceConfig.class);
+        // if (cfg == null) {
+        //     log.warn("Missing GeneralProviderDevice config for {}", deviceId);
+        //     return false;
+        // }
+        // final String switchId = cfg.protocolsInfo().containsKey("int") ?
+        //         cfg.protocolsInfo().get("int").configValues().get("switchId")
+        //         : null;
+        // if (switchId == null || switchId.isEmpty()) {
+        //     log.warn("Missing INT device config for {}", deviceId);
+        //     return false;
+        // }
 
-        PiActionParam transitIdParam;
-        try {
-            transitIdParam = new PiActionParam(
-                    FabricConstants.SWITCH_ID,
-                    // FIXME set switch ID from netcfg
-                    copyFrom(Integer.parseInt(switchId)));
-        } catch (NumberFormatException e) {
-            log.warn("Invalid INT switch ID for {}: {}", deviceId, switchId);
-            return false;
-        }
+        PiActionParam transitIdParam = new PiActionParam(
+                FabricConstants.SWITCH_ID,
+                copyFrom(handler().get(DeviceService.class)
+                                 .getDevice(deviceId).chassisId().id()));
 
         PiAction transitAction = PiAction.builder()
                 .withId(FabricConstants.FABRIC_EGRESS_PROCESS_INT_MAIN_PROCESS_INT_TRANSIT_INIT_METADATA)
diff --git a/protocols/gnmi/api/src/main/java/org/onosproject/gnmi/api/GnmiClient.java b/protocols/gnmi/api/src/main/java/org/onosproject/gnmi/api/GnmiClient.java
index 6e65dd3..796ef55 100644
--- a/protocols/gnmi/api/src/main/java/org/onosproject/gnmi/api/GnmiClient.java
+++ b/protocols/gnmi/api/src/main/java/org/onosproject/gnmi/api/GnmiClient.java
@@ -68,12 +68,4 @@
      * Terminates the subscription channel of this device.
      */
     void terminateSubscriptionChannel();
-
-    /**
-     * Check weather the gNMI service is available or not by sending a dummy get
-     * request message.
-     *
-     * @return true if gNMI service available; false otherwise
-     */
-    CompletableFuture<Boolean> isServiceAvailable();
 }
diff --git a/protocols/gnmi/api/src/main/java/org/onosproject/gnmi/api/GnmiClientKey.java b/protocols/gnmi/api/src/main/java/org/onosproject/gnmi/api/GnmiClientKey.java
index 137cbb8..3a66c18 100644
--- a/protocols/gnmi/api/src/main/java/org/onosproject/gnmi/api/GnmiClientKey.java
+++ b/protocols/gnmi/api/src/main/java/org/onosproject/gnmi/api/GnmiClientKey.java
@@ -20,6 +20,8 @@
 import org.onosproject.grpc.api.GrpcClientKey;
 import org.onosproject.net.DeviceId;
 
+import java.net.URI;
+
 /**
  * Key that uniquely identifies a gNMI client.
  */
@@ -31,11 +33,10 @@
     /**
      * Creates a new gNMI client key.
      *
-     * @param deviceId    ONOS device ID
-     * @param serverAddr  gNMI server address
-     * @param serverPort  gNMI server port
+     * @param deviceId  ONOS device ID
+     * @param serverUri gNMI server URI
      */
-    public GnmiClientKey(DeviceId deviceId, String serverAddr, int serverPort) {
-        super(GNMI, deviceId, serverAddr, serverPort);
+    public GnmiClientKey(DeviceId deviceId, URI serverUri) {
+        super(GNMI, deviceId, serverUri);
     }
 }
diff --git a/protocols/gnmi/ctl/src/main/java/org/onosproject/gnmi/ctl/GnmiClientImpl.java b/protocols/gnmi/ctl/src/main/java/org/onosproject/gnmi/ctl/GnmiClientImpl.java
index 3701ff7..6645f39 100644
--- a/protocols/gnmi/ctl/src/main/java/org/onosproject/gnmi/ctl/GnmiClientImpl.java
+++ b/protocols/gnmi/ctl/src/main/java/org/onosproject/gnmi/ctl/GnmiClientImpl.java
@@ -50,7 +50,7 @@
     private GnmiSubscriptionManager gnmiSubscriptionManager;
 
     GnmiClientImpl(GnmiClientKey clientKey, ManagedChannel managedChannel, GnmiControllerImpl controller) {
-        super(clientKey);
+        super(clientKey, managedChannel, false, controller);
         this.blockingStub = gNMIGrpc.newBlockingStub(managedChannel);
         this.gnmiSubscriptionManager =
                 new GnmiSubscriptionManager(managedChannel, deviceId, controller);
@@ -82,7 +82,7 @@
     }
 
     @Override
-    public CompletableFuture<Boolean> isServiceAvailable() {
+    public CompletableFuture<Boolean> probeService() {
         return supplyInContext(this::doServiceAvailable, "isServiceAvailable");
     }
 
diff --git a/protocols/gnmi/ctl/src/main/java/org/onosproject/gnmi/ctl/GnmiControllerImpl.java b/protocols/gnmi/ctl/src/main/java/org/onosproject/gnmi/ctl/GnmiControllerImpl.java
index c0bdcdd..9c4643a 100644
--- a/protocols/gnmi/ctl/src/main/java/org/onosproject/gnmi/ctl/GnmiControllerImpl.java
+++ b/protocols/gnmi/ctl/src/main/java/org/onosproject/gnmi/ctl/GnmiControllerImpl.java
@@ -23,47 +23,24 @@
 import org.onosproject.gnmi.api.GnmiEvent;
 import org.onosproject.gnmi.api.GnmiEventListener;
 import org.onosproject.grpc.ctl.AbstractGrpcClientController;
-import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Component;
-import org.osgi.service.component.annotations.Deactivate;
-import org.slf4j.Logger;
-
-import static org.slf4j.LoggerFactory.getLogger;
 
 /**
  * Implementation of gNMI controller.
  */
 @Component(immediate = true, service = GnmiController.class)
 public class GnmiControllerImpl
-        extends AbstractGrpcClientController<GnmiClientKey, GnmiClient, GnmiEvent, GnmiEventListener>
+        extends AbstractGrpcClientController
+        <GnmiClientKey, GnmiClient, GnmiEvent, GnmiEventListener>
         implements GnmiController {
 
-    private final Logger log = getLogger(getClass());
-
-    @Activate
-    public void activate() {
-        super.activate();
-        eventDispatcher.addSink(GnmiEvent.class, listenerRegistry);
-        log.info("Started");
-    }
-
-    @Deactivate
-    public void deactivate() {
-        super.deactivate();
-        log.info("Stopped");
+    public GnmiControllerImpl() {
+        super(GnmiEvent.class);
     }
 
     @Override
-    protected GnmiClient createClientInstance(GnmiClientKey clientKey, ManagedChannel channel) {
+    protected GnmiClient createClientInstance(
+            GnmiClientKey clientKey, ManagedChannel channel) {
         return new GnmiClientImpl(clientKey, channel, this);
     }
-
-    /**
-     * Handles event from gNMI client.
-     *
-     * @param event the gNMI event
-     */
-    void postEvent(GnmiEvent event) {
-        post(event);
-    }
 }
diff --git a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcChannelController.java b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcChannelController.java
index 5b07254..ad4324b 100644
--- a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcChannelController.java
+++ b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcChannelController.java
@@ -23,6 +23,7 @@
 
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * Abstraction of a gRPC controller that stores and manages gRPC channels.
@@ -33,10 +34,10 @@
     int CONNECTION_TIMEOUT_SECONDS = 20;
 
     /**
-     * Creates a gRPC managed channel from the given builder and opens a
-     * connection to it. If the connection is successful returns the managed
-     * channel object and stores the channel internally, associated with the
-     * given channel ID.
+     * Creates a gRPC managed channel from the given builder and opens the
+     * connection. If the connection is successful, returns the managed channel
+     * object and stores the channel internally, associated with the given
+     * channel ID.
      * <p>
      * This method blocks until the channel is open or a timeout expires. By
      * default the timeout is {@link #CONNECTION_TIMEOUT_SECONDS} seconds. If
@@ -72,15 +73,6 @@
     Map<GrpcChannelId, ManagedChannel> getChannels();
 
     /**
-     * Returns true if the channel associated with the given identifier is open,
-     * i.e. the server is able to successfully reply to RPCs, false otherwise.
-     *
-     * @param channelId channel ID
-     * @return true if channel is open, false otherwise.
-     */
-    boolean isChannelOpen(GrpcChannelId channelId);
-
-    /**
      * If present, returns the channel associated with the given ID.
      *
      * @param channelId channel ID
@@ -88,4 +80,14 @@
      */
     Optional<ManagedChannel> getChannel(GrpcChannelId channelId);
 
+    /**
+     * Probes the server at the endpoint of the given channel. Returns true if
+     * the server responded to the probe, false otherwise or if the channel does
+     * not exist.
+     *
+     * @param channelId channel ID
+     * @return completable future eventually true if the gRPC server responded
+     * to the probe; false otherwise
+     */
+    CompletableFuture<Boolean> probeChannel(GrpcChannelId channelId);
 }
diff --git a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClient.java b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClient.java
index d040a23..d529529 100644
--- a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClient.java
+++ b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClient.java
@@ -17,11 +17,11 @@
 package org.onosproject.grpc.api;
 
 import com.google.common.annotations.Beta;
+
 import java.util.concurrent.CompletableFuture;
 
 /**
  * Abstraction of a gRPC client.
- *
  */
 @Beta
 public interface GrpcClient {
@@ -33,4 +33,30 @@
      * procedure
      */
     CompletableFuture<Void> shutdown();
+
+    /**
+     * This method provides a coarse modelling of gRPC channel {@link
+     * io.grpc.ConnectivityState}. The implementation does not make any attempt
+     * at probing the server by sending messages over the network, as such it
+     * does not block execution. It returns true if this client is expected to
+     * communicate with the server successfully. In other words, if any RPC
+     * would be executed immediately after this method is called and returns
+     * true, then it is expected, but NOT guaranteed, for that RPC message to
+     * reach the server and be processed. If false, it means the channel is in a
+     * failure state and communication with the server is unlikely to happen
+     * soon.
+     *
+     * @return true if server is deemed reachable, false otherwise
+     */
+    boolean isServerReachable();
+
+    /**
+     * Similar to {@link #isServerReachable()}, but might involve sending
+     * packets over the network. This checks whether the specific gRPC
+     * service(s) required by this client is available or not on the server.
+     *
+     * @return completable future eventually true if the gRPC service(s) on the
+     * server was available during the probe; false otherwise
+     */
+    CompletableFuture<Boolean> probeService();
 }
diff --git a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientController.java b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientController.java
index 3cdbcd1..b47d499 100644
--- a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientController.java
+++ b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientController.java
@@ -18,10 +18,11 @@
 
 import com.google.common.annotations.Beta;
 import org.onosproject.net.DeviceId;
+import org.onosproject.net.device.DeviceAgentListener;
+import org.onosproject.net.provider.ProviderId;
 
 /**
- * Abstraction of a gRPC controller which controls specific gRPC client {@link
- * C} with specific client key {@link K}.
+ * Abstraction of controller that manages gRPC clients.
  *
  * @param <K> the gRPC client key
  * @param <C> the gRPC client type
@@ -34,20 +35,27 @@
      * given information. As a result of this method, a client can be later
      * obtained by invoking {@link #getClient(DeviceId)}.
      * <p>
+     * Upon creation, a connection to the server is automatically started, which
+     * blocks execution. If the connection is successful, the client is created
+     * and this method returns true, otherwise (e.g., socket error) any state
+     * associated with this client is destroyed and returns false.
+     * <p>
      * Only one client can exist for the same device ID. Calls to this method
      * are idempotent fot the same client key, i.e. returns true if such client
-     * already exists but a new one is not created. If there exists a client
-     * with same device ID but different address and port, removes old one and
-     * recreate new one.
+     * already exists. Otherwise, if a client for the same device ID but
+     * different client key already exists, throws an exception.
      *
      * @param clientKey the client key
      * @return true if the client was created and the channel to the server is
      * open; false otherwise
+     * @throws IllegalArgumentException if a client for the same device ID but
+     *                                  different client key already exists.
      */
     boolean createClient(K clientKey);
 
     /**
-     * Retrieves the gRPC client to operate on the given device.
+     * Returns the gRPC client previously created for the given device, or null
+     * if such client does not exist.
      *
      * @param deviceId the device identifier
      * @return the gRPC client of the device if exists; null otherwise
@@ -55,24 +63,47 @@
     C getClient(DeviceId deviceId);
 
     /**
-     * Removes the gRPC client for the given device. If no client exists for the
-     * given device, the result is a no-op.
+     * Returns the gRPC client previously created for the given client key, or
+     * null if such client does not exist.
+     *
+     * @param clientKey client key
+     * @return the gRPC client of the device if exists; null otherwise
+     */
+    C getClient(K clientKey);
+
+    /**
+     * Removes the gRPC client for the given device and any gRPC channel state
+     * associated to it. If no client exists for the given device, the result is
+     * a no-op.
      *
      * @param deviceId the device identifier
      */
     void removeClient(DeviceId deviceId);
 
     /**
-     * Check reachability of the gRPC server running on the given device.
-     * Reachability can be tested only if a client is previously created using
-     * {@link #createClient(GrpcClientKey)}. Note that this only checks the
-     * reachability instead of checking service availability, different
-     * service-specific gRPC clients might check service availability in a
-     * different way.
+     * Similar to {@link #removeClient(DeviceId)} but uses the client key to
+     * identify the client to remove.
      *
-     * @param deviceId the device identifier
-     * @return true if client was created and is able to contact the gNMI
-     * server; false otherwise
+     * @param clientKey the client key
      */
-    boolean isReachable(DeviceId deviceId);
+    void removeClient(K clientKey);
+
+    /**
+     * Adds a listener for device agent events for the given provider.
+     *
+     * @param deviceId device identifier
+     * @param providerId provider ID
+     * @param listener the device agent listener
+     */
+    void addDeviceAgentListener(DeviceId deviceId, ProviderId providerId,
+                                DeviceAgentListener listener);
+
+    /**
+     * Removes the listener for device agent events that was previously
+     * registered for the given provider.
+     *
+     * @param deviceId   device identifier
+     * @param providerId the provider ID
+     */
+    void removeDeviceAgentListener(DeviceId deviceId, ProviderId providerId);
 }
diff --git a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientKey.java b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientKey.java
index 99ea23f..d1d0b0f 100644
--- a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientKey.java
+++ b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientKey.java
@@ -20,8 +20,11 @@
 import com.google.common.base.Objects;
 import org.onosproject.net.DeviceId;
 
+import java.net.URI;
+
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Strings.isNullOrEmpty;
 import static java.lang.String.format;
 
 /**
@@ -29,32 +32,36 @@
  */
 @Beta
 public class GrpcClientKey {
+
+    private static final String GRPC = "grpc";
+    private static final String GRPCS = "grpcs";
+
     private final String serviceName;
     private final DeviceId deviceId;
-    private final String serverAddr;
-    private final int serverPort;
+    private final URI serverUri;
 
     /**
      * Creates a new client key.
      *
      * @param serviceName gRPC service name of the client
-     * @param deviceId ONOS device ID
-     * @param serverAddr gRPC server address
-     * @param serverPort gRPC server port
+     * @param deviceId    ONOS device ID
+     * @param serverUri   gRPC server URI
      */
-    public GrpcClientKey(String serviceName, DeviceId deviceId, String serverAddr, int serverPort) {
+    public GrpcClientKey(String serviceName, DeviceId deviceId, URI serverUri) {
         checkNotNull(serviceName);
         checkNotNull(deviceId);
-        checkNotNull(serverAddr);
+        checkNotNull(serverUri);
         checkArgument(!serviceName.isEmpty(),
-                "Service name can not be null");
-        checkArgument(!serverAddr.isEmpty(),
-                "Server address should not be empty");
-        checkArgument(serverPort > 0 && serverPort <= 65535, "Invalid server port");
+                      "Service name can not be null");
+        checkArgument(serverUri.getScheme().equals(GRPC)
+                              || serverUri.getScheme().equals(GRPCS),
+                      format("Server URI scheme must be %s or %s", GRPC, GRPCS));
+        checkArgument(!isNullOrEmpty(serverUri.getHost()),
+                      "Server host address should not be empty");
+        checkArgument(serverUri.getPort() > 0 && serverUri.getPort() <= 65535, "Invalid server port");
         this.serviceName = serviceName;
         this.deviceId = deviceId;
-        this.serverAddr = serverAddr;
-        this.serverPort = serverPort;
+        this.serverUri = serverUri;
     }
 
     /**
@@ -76,21 +83,21 @@
     }
 
     /**
-     * Gets the gRPC server address.
+     * Returns the gRPC server URI.
      *
-     * @return the gRPC server address.
+     * @return the gRPC server URI.
      */
-    public String serverAddr() {
-        return serverAddr;
+    public URI serveUri() {
+        return serverUri;
     }
 
     /**
-     * Gets the gRPC server port.
+     * Returns true if the client requires TLS/SSL, false otherwise.
      *
-     * @return the gRPC server port.
+     * @return boolean
      */
-    public int serverPort() {
-        return serverPort;
+    public boolean requiresSecureChannel() {
+        return serverUri.getScheme().equals(GRPCS);
     }
 
     @Override
@@ -102,19 +109,18 @@
             return false;
         }
         GrpcClientKey that = (GrpcClientKey) o;
-        return serverPort == that.serverPort &&
-                Objects.equal(serviceName, that.serviceName) &&
+        return Objects.equal(serviceName, that.serviceName) &&
                 Objects.equal(deviceId, that.deviceId) &&
-                Objects.equal(serverAddr, that.serverAddr);
+                Objects.equal(serverUri, that.serverUri);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hashCode(serviceName, deviceId, serverAddr, serverPort);
+        return Objects.hashCode(serviceName, deviceId, serverUri);
     }
 
     @Override
     public String toString() {
-        return format("%s/%s@%s:%s", deviceId, serviceName, serverAddr, serverPort);
+        return format("%s/%s@%s", deviceId, serviceName, serverUri);
     }
 }
diff --git a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClient.java b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClient.java
index 0b21ca2..117c6e3 100644
--- a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClient.java
+++ b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClient.java
@@ -16,11 +16,14 @@
 
 package org.onosproject.grpc.ctl;
 
+import io.grpc.ConnectivityState;
 import io.grpc.Context;
+import io.grpc.ManagedChannel;
 import io.grpc.StatusRuntimeException;
 import org.onosproject.grpc.api.GrpcClient;
 import org.onosproject.grpc.api.GrpcClientKey;
 import org.onosproject.net.DeviceId;
+import org.onosproject.net.device.DeviceAgentEvent;
 import org.slf4j.Logger;
 
 import java.util.concurrent.CompletableFuture;
@@ -28,6 +31,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Supplier;
@@ -54,13 +58,51 @@
 
     protected final ExecutorService executorService;
     protected final DeviceId deviceId;
+    protected final ManagedChannel channel;
+    private final boolean persistent;
+    private final AbstractGrpcClientController controller;
+    private final AtomicBoolean channelOpen = new AtomicBoolean(false);
 
-    protected AbstractGrpcClient(GrpcClientKey clientKey) {
+    /**
+     * Creates an new client for the given key and channel. Setting persistent
+     * to true avoids the gRPC channel to stay IDLE. The controller instance is
+     * needed to propagate channel events.
+     *
+     * @param clientKey  client key
+     * @param channel    channel
+     * @param persistent true if the gRPC should never stay IDLE
+     * @param controller controller
+     */
+    protected AbstractGrpcClient(GrpcClientKey clientKey, ManagedChannel channel,
+                                 boolean persistent, AbstractGrpcClientController controller) {
         checkNotNull(clientKey);
+        checkNotNull(channel);
         this.deviceId = clientKey.deviceId();
+        this.channel = channel;
+        this.persistent = persistent;
+        this.controller = controller;
         this.executorService = Executors.newFixedThreadPool(DEFAULT_THREAD_POOL_SIZE, groupedThreads(
                 "onos-grpc-" + clientKey.serviceName() + "-client-" + deviceId.toString(), "%d"));
         this.contextExecutor = this.cancellableContext.fixedContextExecutor(executorService);
+
+        setChannelCallback(clientKey.deviceId(), channel, ConnectivityState.CONNECTING);
+    }
+
+    @Override
+    public boolean isServerReachable() {
+        final ConnectivityState state = channel.getState(false);
+        switch (state) {
+            case READY:
+            case IDLE:
+                return true;
+            case CONNECTING:
+            case TRANSIENT_FAILURE:
+            case SHUTDOWN:
+                return false;
+            default:
+                log.error("Unrecognized channel connectivity state {}", state);
+                return false;
+        }
     }
 
     @Override
@@ -171,4 +213,85 @@
             }
         }, executor);
     }
+
+    private void setChannelCallback(DeviceId deviceId, ManagedChannel channel,
+                                    ConnectivityState sourceState) {
+        if (log.isTraceEnabled()) {
+            log.trace("Setting channel callback for {} with source state {}...",
+                      deviceId, sourceState);
+        }
+        channel.notifyWhenStateChanged(
+                sourceState, new ChannelConnectivityCallback(deviceId, channel));
+    }
+
+    /**
+     * Runnable task invoked at each change of the channel connectivity state.
+     * New callbacks are created as long as the channel is not shut down.
+     */
+    private final class ChannelConnectivityCallback implements Runnable {
+
+        private final DeviceId deviceId;
+        private final ManagedChannel channel;
+
+        private ChannelConnectivityCallback(
+                DeviceId deviceId, ManagedChannel channel) {
+            this.deviceId = deviceId;
+            this.channel = channel;
+        }
+
+        @Override
+        public void run() {
+            final ConnectivityState newState = channel.getState(false);
+            final DeviceAgentEvent.Type eventType;
+            switch (newState) {
+                // On gRPC connectivity states:
+                // https://github.com/grpc/grpc/blob/master/doc/connectivity-semantics-and-api.md
+                case READY:
+                    eventType = DeviceAgentEvent.Type.CHANNEL_OPEN;
+                    break;
+                case TRANSIENT_FAILURE:
+                    eventType = DeviceAgentEvent.Type.CHANNEL_ERROR;
+                    break;
+                case SHUTDOWN:
+                    eventType = DeviceAgentEvent.Type.CHANNEL_CLOSED;
+                    break;
+                case IDLE:
+                    // IDLE and CONNECTING are transient states that will
+                    // eventually move to READY or TRANSIENT_FAILURE. Do not
+                    // generate an event for now.
+                    if (persistent) {
+                        log.debug("Forcing channel for {} to exist state IDLE...", deviceId);
+                        channel.getState(true);
+                    }
+                    eventType = null;
+                    break;
+                case CONNECTING:
+                    eventType = null;
+                    break;
+                default:
+                    log.error("Unrecognized connectivity state {}", newState);
+                    eventType = null;
+            }
+
+            if (log.isTraceEnabled()) {
+                log.trace("Detected channel connectivity change for {}, new state is {}",
+                          deviceId, newState);
+            }
+
+            if (eventType != null) {
+                // Avoid sending consecutive duplicate events.
+                final boolean present = eventType == DeviceAgentEvent.Type.CHANNEL_OPEN;
+                final boolean past = channelOpen.getAndSet(present);
+                if (present != past) {
+                    log.debug("Notifying event {} for {}", eventType, deviceId);
+                    controller.postEvent(new DeviceAgentEvent(eventType, deviceId));
+                }
+            }
+
+            if (newState != ConnectivityState.SHUTDOWN) {
+                // Channels never leave SHUTDOWN state, no need for a new callback.
+                setChannelCallback(deviceId, channel, newState);
+            }
+        }
+    }
 }
diff --git a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClientController.java b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClientController.java
index c0a3c6b..3c38209 100644
--- a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClientController.java
+++ b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClientController.java
@@ -21,7 +21,6 @@
 import io.grpc.ManagedChannel;
 import io.grpc.netty.GrpcSslContexts;
 import io.grpc.netty.NettyChannelBuilder;
-import io.netty.handler.ssl.NotSslRecordException;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
 import org.onosproject.event.AbstractListenerManager;
@@ -33,6 +32,9 @@
 import org.onosproject.grpc.api.GrpcClientController;
 import org.onosproject.grpc.api.GrpcClientKey;
 import org.onosproject.net.DeviceId;
+import org.onosproject.net.device.DeviceAgentEvent;
+import org.onosproject.net.device.DeviceAgentListener;
+import org.onosproject.net.provider.ProviderId;
 import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Deactivate;
 import org.osgi.service.component.annotations.Reference;
@@ -41,11 +43,11 @@
 
 import javax.net.ssl.SSLException;
 import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.locks.Lock;
 import java.util.function.Supplier;
 
 import static com.google.common.base.Preconditions.checkNotNull;
-import static com.google.common.base.Preconditions.checkState;
 import static java.lang.String.format;
 import static org.slf4j.LoggerFactory.getLogger;
 
@@ -74,40 +76,44 @@
     private final Map<DeviceId, K> clientKeys = Maps.newHashMap();
     private final Map<K, C> clients = Maps.newHashMap();
     private final Map<DeviceId, GrpcChannelId> channelIds = Maps.newHashMap();
+    private final ConcurrentMap<DeviceId, ConcurrentMap<ProviderId, DeviceAgentListener>>
+            deviceAgentListeners = Maps.newConcurrentMap();
+    private final Class<E> eventClass;
     private final Striped<Lock> stripedLocks = Striped.lock(DEFAULT_DEVICE_LOCK_SIZE);
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
     protected GrpcChannelController grpcChannelController;
 
+    public AbstractGrpcClientController(Class<E> eventClass) {
+        this.eventClass = eventClass;
+    }
+
     @Activate
     public void activate() {
+        eventDispatcher.addSink(eventClass, listenerRegistry);
         log.info("Started");
     }
 
     @Deactivate
     public void deactivate() {
+        eventDispatcher.removeSink(eventClass);
         clientKeys.keySet().forEach(this::removeClient);
         clientKeys.clear();
         clients.clear();
         channelIds.clear();
+        deviceAgentListeners.clear();
         log.info("Stopped");
     }
 
     @Override
     public boolean createClient(K clientKey) {
         checkNotNull(clientKey);
-        /*
-            FIXME we might want to move "useTls" and "fallback" to properties of the netcfg and clientKey
-                  For now, we will first try to connect with TLS (accepting any cert), then fall back to
-                  plaintext for every device
-         */
-        return withDeviceLock(() -> doCreateClient(clientKey, true, true), clientKey.deviceId());
+        return withDeviceLock(() -> doCreateClient(clientKey),
+                              clientKey.deviceId());
     }
 
-    private boolean doCreateClient(K clientKey, boolean useTls, boolean fallbackToPlainText) {
+    private boolean doCreateClient(K clientKey) {
         DeviceId deviceId = clientKey.deviceId();
-        String serverAddr = clientKey.serverAddr();
-        int serverPort = clientKey.serverPort();
 
         if (clientKeys.containsKey(deviceId)) {
             final GrpcClientKey existingKey = clientKeys.get(deviceId);
@@ -116,19 +122,25 @@
                           clientName(clientKey), clientKey);
                 return true;
             } else {
-                log.info("Requested new {} with updated key, removing old client... (oldKey={})",
-                         clientName(clientKey), existingKey);
-                doRemoveClient(deviceId);
+                throw new IllegalArgumentException(format(
+                        "A client already exists for device %s (%s)",
+                        deviceId, clientKey));
             }
         }
 
-        log.info("Creating client for {} (server={}:{})...", deviceId, serverAddr, serverPort);
+        final String method = clientKey.requiresSecureChannel()
+                ? "TLS" : "plaintext TCP";
+
+        log.info("Connecting {} client for {} to server at {} using {}...",
+                 clientKey.serviceName(), deviceId, clientKey.serveUri(), method);
 
         SslContext sslContext = null;
-        if (useTls) {
+        if (clientKey.requiresSecureChannel()) {
             try {
-                // Accept any server certificate; this is insecure and should not be used in production
-                sslContext = GrpcSslContexts.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();
+                // FIXME: Accept any server certificate; this is insecure and
+                //  should not be used in production
+                sslContext = GrpcSslContexts.forClient().trustManager(
+                        InsecureTrustManagerFactory.INSTANCE).build();
             } catch (SSLException e) {
                 log.error("Failed to build SSL Context", e);
                 return false;
@@ -137,17 +149,15 @@
 
         GrpcChannelId channelId = GrpcChannelId.of(clientKey.toString());
         NettyChannelBuilder channelBuilder = NettyChannelBuilder
-                .forAddress(serverAddr, serverPort)
+                .forAddress(clientKey.serveUri().getHost(),
+                            clientKey.serveUri().getPort())
                 .maxInboundMessageSize(DEFAULT_MAX_INBOUND_MSG_SIZE * MEGABYTES);
+
         if (sslContext != null) {
-            log.debug("Using SSL for gRPC connection to {}", deviceId);
             channelBuilder
                     .sslContext(sslContext)
                     .useTransportSecurity();
         } else {
-            checkState(!useTls,
-                    "Not authorized to use plaintext for gRPC connection to {}", deviceId);
-            log.debug("Using plaintext TCP for gRPC connection to {}", deviceId);
             channelBuilder.usePlaintext();
         }
 
@@ -156,24 +166,9 @@
         try {
             channel = grpcChannelController.connectChannel(channelId, channelBuilder);
         } catch (Throwable e) {
-            for (Throwable cause = e; cause != null; cause = cause.getCause()) {
-                if (useTls && cause instanceof NotSslRecordException) {
-                    // Likely root cause is that server is using plaintext
-                    log.info("Failed to connect to server (device={}) using TLS", deviceId);
-                    log.debug("TLS connection exception", e);
-                    if (fallbackToPlainText) {
-                        log.info("Falling back to plaintext for connection to {}", deviceId);
-                        return doCreateClient(clientKey, false, false);
-                    }
-                }
-                if (!useTls && "Connection reset by peer".equals(cause.getMessage())) {
-                    // Not a great signal, but could indicate the server is expected a TLS connection
-                    log.error("Failed to connect to server (device={}) using plaintext TCP; is the server using TLS?",
-                            deviceId);
-                    break;
-                }
-            }
-            log.warn("Unable to connect to gRPC server for {}", deviceId, e);
+            log.warn("Failed to connect to {} ({}) using {}: {}",
+                     deviceId, clientKey.serveUri(), method, e.toString());
+            log.debug("gRPC client connection exception", e);
             return false;
         }
 
@@ -216,37 +211,79 @@
     }
 
     @Override
+    public C getClient(K clientKey) {
+        checkNotNull(clientKey);
+        return clients.get(clientKey);
+    }
+
+    @Override
     public void removeClient(DeviceId deviceId) {
         checkNotNull(deviceId);
         withDeviceLock(() -> doRemoveClient(deviceId), deviceId);
     }
 
+    @Override
+    public void removeClient(K clientKey) {
+        checkNotNull(clientKey);
+        withDeviceLock(() -> doRemoveClient(clientKey), clientKey.deviceId());
+    }
+
     private Void doRemoveClient(DeviceId deviceId) {
         if (clientKeys.containsKey(deviceId)) {
-            final K clientKey = clientKeys.get(deviceId);
-            clients.get(clientKey).shutdown();
-            grpcChannelController.disconnectChannel(channelIds.get(deviceId));
-            clientKeys.remove(deviceId);
-            clients.remove(clientKey);
-            channelIds.remove(deviceId);
+            doRemoveClient(clientKeys.get(deviceId));
         }
         return null;
     }
 
-    @Override
-    public boolean isReachable(DeviceId deviceId) {
-        checkNotNull(deviceId);
-        return withDeviceLock(() -> doIsReachable(deviceId), deviceId);
+    private Void doRemoveClient(K clientKey) {
+        if (clients.containsKey(clientKey)) {
+            clients.get(clientKey).shutdown();
+        }
+        if (channelIds.containsKey(clientKey.deviceId())) {
+            grpcChannelController.disconnectChannel(
+                    channelIds.get(clientKey.deviceId()));
+        }
+        clientKeys.remove(clientKey.deviceId());
+        clients.remove(clientKey);
+        channelIds.remove(clientKey.deviceId());
+        return null;
     }
 
-    private boolean doIsReachable(DeviceId deviceId) {
-        // Default behaviour checks only the gRPC channel, should
-        // check according to different gRPC service
-        if (!clientKeys.containsKey(deviceId)) {
-            log.debug("Missing client for {}, cannot check for reachability", deviceId);
-            return false;
+    @Override
+    public void addDeviceAgentListener(DeviceId deviceId, ProviderId providerId, DeviceAgentListener listener) {
+        checkNotNull(deviceId, "deviceId cannot be null");
+        checkNotNull(deviceId, "providerId cannot be null");
+        checkNotNull(listener, "listener cannot be null");
+        deviceAgentListeners.putIfAbsent(deviceId, Maps.newConcurrentMap());
+        deviceAgentListeners.get(deviceId).put(providerId, listener);
+    }
+
+    @Override
+    public void removeDeviceAgentListener(DeviceId deviceId, ProviderId providerId) {
+        checkNotNull(deviceId, "deviceId cannot be null");
+        checkNotNull(providerId, "listener cannot be null");
+        deviceAgentListeners.computeIfPresent(deviceId, (did, listeners) -> {
+            listeners.remove(providerId);
+            return listeners.isEmpty() ? null : listeners;
+        });
+    }
+
+    public void postEvent(E event) {
+        checkNotNull(event);
+        post(event);
+    }
+
+    public void postEvent(DeviceAgentEvent event) {
+        // We should have only one event delivery mechanism. We have two now
+        // because we have two different types of events, DeviceAgentEvent and
+        // controller/protocol specific ones (e.g. P4Runtime or gNMI).
+        // TODO: extend device agent event to allow delivery protocol-specific
+        //  events, e.g. packet-in
+        checkNotNull(event);
+        if (deviceAgentListeners.containsKey(event.subject())) {
+            deviceAgentListeners.get(event.subject()).values()
+                    .forEach(l -> l.event(event));
         }
-        return grpcChannelController.isChannelOpen(channelIds.get(deviceId));
     }
 
     private <U> U withDeviceLock(Supplier<U> task, DeviceId deviceId) {
diff --git a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/GrpcChannelControllerImpl.java b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/GrpcChannelControllerImpl.java
index ca0c8c3..87c9426 100644
--- a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/GrpcChannelControllerImpl.java
+++ b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/GrpcChannelControllerImpl.java
@@ -41,6 +41,7 @@
 import java.util.Dictionary;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -152,50 +153,24 @@
         }
     }
 
-    private boolean doDummyMessage(ManagedChannel channel) throws StatusRuntimeException {
+    private void doDummyMessage(ManagedChannel channel) throws StatusRuntimeException {
         DummyServiceGrpc.DummyServiceBlockingStub dummyStub = DummyServiceGrpc
                 .newBlockingStub(channel)
                 .withDeadlineAfter(CONNECTION_TIMEOUT_SECONDS, TimeUnit.SECONDS);
         try {
-            return dummyStub.sayHello(Dummy.DummyMessageThatNoOneWouldReallyUse
-                                              .getDefaultInstance()) != null;
+            //noinspection ResultOfMethodCallIgnored
+            dummyStub.sayHello(Dummy.DummyMessageThatNoOneWouldReallyUse
+                                       .getDefaultInstance());
         } catch (StatusRuntimeException e) {
-            if (e.getStatus().equals(Status.UNIMPLEMENTED)) {
+            if (!e.getStatus().equals(Status.UNIMPLEMENTED)) {
                 // UNIMPLEMENTED means that the server received our message but
                 // doesn't know how to handle it. Hence, channel is open.
-                return true;
-            } else {
                 throw e;
             }
         }
     }
 
     @Override
-    public boolean isChannelOpen(GrpcChannelId channelId) {
-        checkNotNull(channelId);
-
-        Lock lock = channelLocks.get(channelId);
-        lock.lock();
-
-        try {
-            if (!channels.containsKey(channelId)) {
-                log.warn("Unknown channel ID '{}', can't check if channel is open",
-                         channelId);
-                return false;
-            }
-            try {
-                return doDummyMessage(channels.get(channelId));
-            } catch (StatusRuntimeException e) {
-                log.debug("Unable to send dummy message to {}: {}",
-                          channelId, e.toString());
-                return false;
-            }
-        } finally {
-            lock.unlock();
-        }
-    }
-
-    @Override
     public void disconnectChannel(GrpcChannelId channelId) {
         checkNotNull(channelId);
 
@@ -240,7 +215,6 @@
 
         Lock lock = channelLocks.get(channelId);
         lock.lock();
-
         try {
             return Optional.ofNullable(channels.get(channelId));
         } finally {
@@ -248,4 +222,23 @@
         }
     }
 
+    @Override
+    public CompletableFuture<Boolean> probeChannel(GrpcChannelId channelId) {
+        final ManagedChannel channel = channels.get(channelId);
+        if (channel == null) {
+            log.warn("Unable to find any channel with ID {}, cannot send probe",
+                     channelId);
+            return CompletableFuture.completedFuture(false);
+        }
+        return CompletableFuture.supplyAsync(() -> {
+            try {
+                doDummyMessage(channel);
+                return true;
+            } catch (StatusRuntimeException e) {
+                log.debug("Probe for {} failed", channelId);
+                log.debug("", e);
+                return false;
+            }
+        });
+    }
 }
diff --git a/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeClientKey.java b/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeClientKey.java
index 4c23bbc..452e978 100644
--- a/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeClientKey.java
+++ b/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeClientKey.java
@@ -20,28 +20,58 @@
 import org.onosproject.grpc.api.GrpcClientKey;
 import org.onosproject.net.DeviceId;
 
-import java.util.Objects;
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URLDecoder;
+
+import static java.lang.String.format;
 
 /**
  * Key that uniquely identifies a P4Runtime client.
  */
 @Beta
 public final class P4RuntimeClientKey extends GrpcClientKey {
+
+    private static final String DEVICE_ID_PARAM = "device_id=";
+
     private static final String P4RUNTIME = "P4Runtime";
     private final long p4DeviceId;
 
     /**
-     * Creates a new client key.
+     * Creates a new client key. The server URI is expected to carry the
+     * P4runtime server-internal 'device_id' as a param in the query string. For
+     * example, grpc://10.0.0.1:5001?device_id=1
      *
-     * @param deviceId   ONOS device ID
-     * @param serverAddr P4Runtime server address
-     * @param serverPort P4Runtime server port
-     * @param p4DeviceId P4Runtime server-internal device ID
+     * @param deviceId  ONOS device ID
+     * @param serverUri P4Runtime server URI
      */
-    public P4RuntimeClientKey(DeviceId deviceId, String serverAddr,
-                              int serverPort, long p4DeviceId) {
-        super(P4RUNTIME, deviceId, serverAddr, serverPort);
-        this.p4DeviceId = p4DeviceId;
+    public P4RuntimeClientKey(DeviceId deviceId, URI serverUri) {
+        super(P4RUNTIME, deviceId, serverUri);
+        this.p4DeviceId = extractP4DeviceId(serverUri);
+    }
+
+    private static Long extractP4DeviceId(URI uri) {
+        String[] segments = uri.getRawQuery().split("&");
+        try {
+            for (String s : segments) {
+                if (s.startsWith(DEVICE_ID_PARAM)) {
+                    return Long.parseUnsignedLong(
+                            URLDecoder.decode(
+                                    s.substring(DEVICE_ID_PARAM.length()), "utf-8"));
+                }
+            }
+        } catch (UnsupportedEncodingException e) {
+            throw new IllegalArgumentException(format(
+                    "Unable to decode P4Runtime-interval device_id from URI %s: %s",
+                    uri, e.toString()));
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException(format(
+                    "Invalid P4Runtime-interval device_id in URI %s: %s",
+                    uri, e.toString()));
+        }
+        throw new IllegalArgumentException(format(
+                "Missing P4Runtime-interval device_id in URI %s",
+                uri));
     }
 
     /**
@@ -52,29 +82,4 @@
     public long p4DeviceId() {
         return p4DeviceId;
     }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) {
-            return true;
-        }
-        if (o == null || getClass() != o.getClass()) {
-            return false;
-        }
-        if (!super.equals(o)) {
-            return false;
-        }
-        P4RuntimeClientKey that = (P4RuntimeClientKey) o;
-        return p4DeviceId == that.p4DeviceId;
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(super.hashCode(), p4DeviceId);
-    }
-
-    @Override
-    public String toString() {
-        return super.toString() + "/" + p4DeviceId;
-    }
 }
diff --git a/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeController.java b/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeController.java
index 0bc8ed6..18e925a 100644
--- a/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeController.java
+++ b/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeController.java
@@ -19,33 +19,13 @@
 import com.google.common.annotations.Beta;
 import org.onosproject.event.ListenerService;
 import org.onosproject.grpc.api.GrpcClientController;
-import org.onosproject.net.DeviceId;
-import org.onosproject.net.device.DeviceAgentListener;
-import org.onosproject.net.provider.ProviderId;
 
 /**
- * Controller of P4Runtime devices.
+ * Controller of P4Runtime clients.
  */
 @Beta
 public interface P4RuntimeController
         extends GrpcClientController<P4RuntimeClientKey, P4RuntimeClient>,
                 ListenerService<P4RuntimeEvent, P4RuntimeEventListener> {
-    /**
-     * Adds a listener for device agent events for the given provider.
-     *
-     * @param deviceId device identifier
-     * @param providerId provider ID
-     * @param listener the device agent listener
-     */
-    void addDeviceAgentListener(DeviceId deviceId, ProviderId providerId,
-                                DeviceAgentListener listener);
 
-    /**
-     * Removes the listener for device agent events that was previously
-     * registered for the given provider.
-     *
-     * @param deviceId   device identifier
-     * @param providerId the provider ID
-     */
-    void removeDeviceAgentListener(DeviceId deviceId, ProviderId providerId);
 }
diff --git a/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeEvent.java b/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeEvent.java
index 4e0ea60b..2b3db65 100644
--- a/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeEvent.java
+++ b/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeEvent.java
@@ -33,21 +33,6 @@
          * A packet-in.
          */
         PACKET_IN,
-
-        /**
-         * Arbitration reply.
-         */
-        ARBITRATION_RESPONSE,
-
-        /**
-         * Channel Event.
-         */
-        CHANNEL_EVENT,
-
-        /**
-         * Permission denied (not master).
-         */
-        PERMISSION_DENIED,
     }
 
     public P4RuntimeEvent(Type type, P4RuntimeEventSubject subject) {
diff --git a/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimePipelineConfigClient.java b/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimePipelineConfigClient.java
index 34dac66..81928c1 100644
--- a/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimePipelineConfigClient.java
+++ b/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimePipelineConfigClient.java
@@ -89,4 +89,25 @@
             PiPipeconf pipeconf, ByteBuffer deviceData) {
         return Futures.getUnchecked(isPipelineConfigSet(pipeconf, deviceData));
     }
+
+    /**
+     * Returns true if the device has a pipeline config set, false otherwise.
+     * <p>
+     * This method is expected to return {@code true} if invoked after
+     * successfully calling {@link #setPipelineConfig(PiPipeconf, ByteBuffer)}
+     * with any parameter.
+     *
+     * @return completable future, true if the device has a pipeline config set,
+     * false otherwise.
+     */
+    CompletableFuture<Boolean> isAnyPipelineConfigSet();
+
+    /**
+     * Same as {@link #isAnyPipelineConfigSet()}, but blocks execution.
+     *
+     * @return true if the device has a pipeline config set, false otherwise.
+     */
+    default boolean isAnyPipelineConfigSetSync() {
+        return Futures.getUnchecked(isAnyPipelineConfigSet());
+    }
 }
diff --git a/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeStreamClient.java b/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeStreamClient.java
index 1317e29..555f906 100644
--- a/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeStreamClient.java
+++ b/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeStreamClient.java
@@ -19,6 +19,8 @@
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.net.pi.runtime.PiPacketOperation;
 
+import java.math.BigInteger;
+
 /**
  * P4Runtime client interface for the StreamChannel RPC. It allows management of
  * the P4Runtime session (open/close, mastership arbitration) as well as sending
@@ -30,23 +32,26 @@
 public interface P4RuntimeStreamClient {
 
     /**
-     * Opens a session to the server by starting the Stream RPC and sending a
-     * mastership arbitration update message with an election ID that is
-     * expected to be unique among all available clients. If a client has been
-     * requested to become master via {@link #runForMastership()}, then this
-     * method should pick an election ID that is lower than the one currently
-     * associated with the master client.
+     * Opportunistically opens a session with the server by starting a
+     * StreamChannel RPC and sends a {@code MasterArbitrationUpdate} message
+     * with the given election ID. The {@code master} boolean flag is used to
+     * indicated if we are trying to became master or not. If false, the
+     * implementation might delay sending the {@code MasterArbitrationUpdate}
+     * message until another node becomes master with a higher election ID.
      * <p>
-     * If the server acknowledges the session to this client as open, the {@link
+     * If the server acknowledges this client as master, the {@link
      * P4RuntimeController} is expected to generate a {@link
      * org.onosproject.net.device.DeviceAgentEvent} with type {@link
-     * org.onosproject.net.device.DeviceAgentEvent.Type#CHANNEL_OPEN}.
+     * org.onosproject.net.device.DeviceAgentEvent.Type#ROLE_MASTER}.
+     *
+     * @param master true if we are trying to become master
+     * @param electionId election ID
      */
-    void openSession();
+    void setMastership(boolean master, BigInteger electionId);
 
     /**
-     * Returns true if the Stream RPC is active and the P4Runtime session is
-     * open, false otherwise.
+     * Returns true if the StreamChannel RPC is active and hence the P4Runtime
+     * session is open, false otherwise.
      *
      * @return boolean
      */
@@ -58,17 +63,6 @@
     void closeSession();
 
     /**
-     * Sends a master arbitration update to the device with a new election ID
-     * that is expected to be the highest one between all clients.
-     * <p>
-     * If the server acknowledges this client as master, the {@link
-     * P4RuntimeController} is expected to generate a {@link
-     * org.onosproject.net.device.DeviceAgentEvent} with type {@link
-     * org.onosproject.net.device.DeviceAgentEvent.Type#ROLE_MASTER}.
-     */
-    void runForMastership();
-
-    /**
      * Returns true if this client is master for the server, false otherwise.
      *
      * @return boolean
diff --git a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/P4RuntimeClientImpl.java b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/P4RuntimeClientImpl.java
index 353d44e..d06e3fc 100644
--- a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/P4RuntimeClientImpl.java
+++ b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/P4RuntimeClientImpl.java
@@ -17,21 +17,25 @@
 package org.onosproject.p4runtime.ctl.client;
 
 import io.grpc.ManagedChannel;
+import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
 import org.onosproject.grpc.ctl.AbstractGrpcClient;
 import org.onosproject.net.DeviceId;
+import org.onosproject.net.device.DeviceAgentEvent;
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.net.pi.runtime.PiPacketOperation;
 import org.onosproject.net.pi.service.PiPipeconfService;
 import org.onosproject.p4runtime.api.P4RuntimeClient;
 import org.onosproject.p4runtime.api.P4RuntimeClientKey;
-import org.onosproject.p4runtime.api.P4RuntimeEvent;
-import org.onosproject.p4runtime.ctl.controller.BaseEventSubject;
-import org.onosproject.p4runtime.ctl.controller.ChannelEvent;
+import org.onosproject.p4runtime.ctl.controller.MasterElectionIdStore;
 import org.onosproject.p4runtime.ctl.controller.P4RuntimeControllerImpl;
 import p4.v1.P4RuntimeGrpc;
 import p4.v1.P4RuntimeOuterClass;
+import p4.v1.P4RuntimeOuterClass.GetForwardingPipelineConfigRequest;
+import p4.v1.P4RuntimeOuterClass.GetForwardingPipelineConfigResponse;
 
+import java.math.BigInteger;
 import java.nio.ByteBuffer;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -39,6 +43,7 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static java.lang.String.format;
+import static p4.v1.P4RuntimeOuterClass.GetForwardingPipelineConfigRequest.ResponseType.COOKIE_ONLY;
 
 /**
  * Implementation of P4RuntimeClient.
@@ -59,7 +64,6 @@
     static final int LONG_TIMEOUT_SECONDS = 60;
 
     private final long p4DeviceId;
-    private final ManagedChannel channel;
     private final P4RuntimeControllerImpl controller;
     private final StreamClientImpl streamClient;
     private final PipelineConfigClientImpl pipelineConfigClient;
@@ -67,25 +71,27 @@
     /**
      * Instantiates a new client with the given arguments.
      *
-     * @param clientKey       client key
-     * @param channel         gRPC managed channel
-     * @param controller      P$Runtime controller instance
-     * @param pipeconfService pipeconf service instance
+     * @param clientKey             client key
+     * @param channel               gRPC managed channel
+     * @param controller            P$Runtime controller instance
+     * @param pipeconfService       pipeconf service instance
+     * @param masterElectionIdStore master election ID store
      */
     public P4RuntimeClientImpl(P4RuntimeClientKey clientKey,
                                ManagedChannel channel,
                                P4RuntimeControllerImpl controller,
-                               PiPipeconfService pipeconfService) {
-        super(clientKey);
+                               PiPipeconfService pipeconfService,
+                               MasterElectionIdStore masterElectionIdStore) {
+        super(clientKey, channel, true, controller);
         checkNotNull(channel);
         checkNotNull(controller);
         checkNotNull(pipeconfService);
+        checkNotNull(masterElectionIdStore);
 
         this.p4DeviceId = clientKey.p4DeviceId();
-        this.channel = channel;
         this.controller = controller;
         this.streamClient = new StreamClientImpl(
-                pipeconfService, this, controller);
+                pipeconfService, masterElectionIdStore, this, controller);
         this.pipelineConfigClient = new PipelineConfigClientImpl(this);
     }
 
@@ -108,13 +114,13 @@
     }
 
     @Override
-    public ReadRequest read(PiPipeconf pipeconf) {
-        return new ReadRequestImpl(this, pipeconf);
+    public CompletableFuture<Boolean> isAnyPipelineConfigSet() {
+        return pipelineConfigClient.isAnyPipelineConfigSet();
     }
 
     @Override
-    public void openSession() {
-        streamClient.openSession();
+    public ReadRequest read(PiPipeconf pipeconf) {
+        return new ReadRequestImpl(this, pipeconf);
     }
 
     @Override
@@ -128,8 +134,8 @@
     }
 
     @Override
-    public void runForMastership() {
-        streamClient.runForMastership();
+    public void setMastership(boolean master, BigInteger newElectionId) {
+        streamClient.setMastership(master, newElectionId);
     }
 
     @Override
@@ -147,6 +153,44 @@
         return new WriteRequestImpl(this, pipeconf);
     }
 
+    @Override
+    public CompletableFuture<Boolean> probeService() {
+        final CompletableFuture<Boolean> future = new CompletableFuture<>();
+        final StreamObserver<GetForwardingPipelineConfigResponse> responseObserver =
+                new StreamObserver<GetForwardingPipelineConfigResponse>() {
+                    @Override
+                    public void onNext(GetForwardingPipelineConfigResponse value) {
+                        future.complete(true);
+                    }
+
+                    @Override
+                    public void onError(Throwable t) {
+                        if (Status.fromThrowable(t).getCode() ==
+                                Status.Code.FAILED_PRECONDITION) {
+                            // Pipeline not set but service is available.
+                            future.complete(true);
+                        } else {
+                            log.debug("", t);
+                        }
+                        future.complete(false);
+                    }
+
+                    @Override
+                    public void onCompleted() {
+                        // Ignore, unary call.
+                    }
+                };
+        // Use long timeout as the device might return the full P4 blob
+        // (e.g. server does not support cookie), over a slow network.
+        execRpc(s -> s.getForwardingPipelineConfig(
+                GetForwardingPipelineConfigRequest.newBuilder()
+                        .setDeviceId(p4DeviceId)
+                        .setResponseType(COOKIE_ONLY)
+                        .build(), responseObserver),
+                SHORT_TIMEOUT_SECONDS);
+        return future;
+    }
+
     /**
      * Returns the P4Runtime-internal device ID associated with this client.
      *
@@ -184,7 +228,8 @@
      * @param stubConsumer P4Runtime stub consumer
      * @param timeout      timeout in seconds
      */
-    void execRpc(Consumer<P4RuntimeGrpc.P4RuntimeStub> stubConsumer, int timeout) {
+    void execRpc(Consumer<P4RuntimeGrpc.P4RuntimeStub> stubConsumer,
+                 int timeout) {
         if (log.isTraceEnabled()) {
             log.trace("Executing RPC with timeout {} seconds (context deadline {})...",
                       timeout, context().getDeadline());
@@ -235,21 +280,10 @@
     }
 
     private void checkGrpcException(StatusRuntimeException sre) {
-        switch (sre.getStatus().getCode()) {
-            case PERMISSION_DENIED:
-                // Notify upper layers that this node is not master.
-                controller.postEvent(new P4RuntimeEvent(
-                        P4RuntimeEvent.Type.PERMISSION_DENIED,
-                        new BaseEventSubject(deviceId)));
-                break;
-            case UNAVAILABLE:
-                // Channel might be closed.
-                controller.postEvent(new P4RuntimeEvent(
-                        P4RuntimeEvent.Type.CHANNEL_EVENT,
-                        new ChannelEvent(deviceId, ChannelEvent.Type.ERROR)));
-                break;
-            default:
-                break;
+        if (sre.getStatus().getCode() == Status.Code.PERMISSION_DENIED) {
+            // Notify upper layers that this node is not master.
+            controller.postEvent(new DeviceAgentEvent(
+                    DeviceAgentEvent.Type.NOT_MASTER, deviceId));
         }
     }
 }
diff --git a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/PipelineConfigClientImpl.java b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/PipelineConfigClientImpl.java
index c023f26..b0b6c57 100644
--- a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/PipelineConfigClientImpl.java
+++ b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/PipelineConfigClientImpl.java
@@ -18,6 +18,7 @@
 
 import com.google.protobuf.ByteString;
 import com.google.protobuf.TextFormat;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.p4runtime.api.P4RuntimePipelineConfigClient;
@@ -32,11 +33,13 @@
 import p4.v1.P4RuntimeOuterClass.SetForwardingPipelineConfigResponse;
 
 import java.nio.ByteBuffer;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static java.util.concurrent.CompletableFuture.completedFuture;
 import static org.onosproject.p4runtime.ctl.client.P4RuntimeClientImpl.LONG_TIMEOUT_SECONDS;
+import static org.onosproject.p4runtime.ctl.client.P4RuntimeClientImpl.SHORT_TIMEOUT_SECONDS;
 import static org.slf4j.LoggerFactory.getLogger;
 import static p4.v1.P4RuntimeOuterClass.GetForwardingPipelineConfigRequest.ResponseType.COOKIE_ONLY;
 import static p4.v1.P4RuntimeOuterClass.SetForwardingPipelineConfigRequest.Action.VERIFY_AND_COMMIT;
@@ -62,6 +65,12 @@
     public CompletableFuture<Boolean> setPipelineConfig(
             PiPipeconf pipeconf, ByteBuffer deviceData) {
 
+        if (!client.isSessionOpen()) {
+            log.warn("Dropping set pipeline config request for {}, session is CLOSED",
+                     client.deviceId());
+            return completedFuture(false);
+        }
+
         log.info("Setting pipeline config for {} to {}...",
                  client.deviceId(), pipeconf.id());
 
@@ -98,11 +107,13 @@
                         // All good, pipeline is set.
                         future.complete(true);
                     }
+
                     @Override
                     public void onError(Throwable t) {
                         client.handleRpcError(t, "SET-pipeline-config");
                         future.complete(false);
                     }
+
                     @Override
                     public void onCompleted() {
                         // Ignore, unary call.
@@ -154,6 +165,11 @@
                         pipeconf, expectedDeviceData, cfgFromDevice));
     }
 
+    @Override
+    public CompletableFuture<Boolean> isAnyPipelineConfigSet() {
+        return getPipelineCookieFromServer().thenApply(Objects::nonNull);
+    }
+
     private boolean comparePipelineConfig(
             PiPipeconf pipeconf, ByteBuffer expectedDeviceData,
             ForwardingPipelineConfig cfgFromDevice) {
@@ -209,17 +225,38 @@
                     public void onNext(GetForwardingPipelineConfigResponse value) {
                         if (value.hasConfig()) {
                             future.complete(value.getConfig());
+                            if (!value.getConfig().getP4DeviceConfig().isEmpty()) {
+                                log.warn("{} returned GetForwardingPipelineConfigResponse " +
+                                                 "with p4_device_config field set " +
+                                                 "({} bytes), but we requested COOKIE_ONLY",
+                                         client.deviceId(),
+                                         value.getConfig().getP4DeviceConfig().size());
+                            }
+                            if (value.getConfig().hasP4Info()) {
+                                log.warn("{} returned GetForwardingPipelineConfigResponse " +
+                                                 "with p4_info field set " +
+                                                 "({} bytes), but we requested COOKIE_ONLY",
+                                         client.deviceId(),
+                                         value.getConfig().getP4Info().getSerializedSize());
+                            }
                         } else {
+                            future.complete(null);
                             log.warn("{} returned {} with 'config' field unset",
                                      client.deviceId(), value.getClass().getSimpleName());
                         }
-                        future.complete(null);
                     }
 
                     @Override
                     public void onError(Throwable t) {
-                        client.handleRpcError(t, "GET-pipeline-config");
                         future.complete(null);
+                        if (Status.fromThrowable(t).getCode() ==
+                                Status.Code.FAILED_PRECONDITION) {
+                            // FAILED_PRECONDITION means that a pipeline
+                            // config was not set in the first place, don't
+                            // bother logging.
+                            return;
+                        }
+                        client.handleRpcError(t, "GET-pipeline-config");
                     }
 
                     @Override
@@ -231,7 +268,7 @@
         // (e.g. server does not support cookie), over a slow network.
         client.execRpc(
                 s -> s.getForwardingPipelineConfig(request, responseObserver),
-                LONG_TIMEOUT_SECONDS);
+                SHORT_TIMEOUT_SECONDS);
         return future;
     }
 }
diff --git a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/StreamClientImpl.java b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/StreamClientImpl.java
index 7afd97b..1bf195c 100644
--- a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/StreamClientImpl.java
+++ b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/StreamClientImpl.java
@@ -22,14 +22,15 @@
 import io.grpc.stub.ClientCallStreamObserver;
 import io.grpc.stub.StreamObserver;
 import org.onosproject.net.DeviceId;
+import org.onosproject.net.device.DeviceAgentEvent;
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.net.pi.runtime.PiPacketOperation;
 import org.onosproject.net.pi.service.PiPipeconfService;
 import org.onosproject.p4runtime.api.P4RuntimeEvent;
 import org.onosproject.p4runtime.api.P4RuntimeStreamClient;
 import org.onosproject.p4runtime.ctl.codec.CodecException;
-import org.onosproject.p4runtime.ctl.controller.ArbitrationUpdateEvent;
-import org.onosproject.p4runtime.ctl.controller.ChannelEvent;
+import org.onosproject.p4runtime.ctl.controller.MasterElectionIdStore;
+import org.onosproject.p4runtime.ctl.controller.MasterElectionIdStore.MasterElectionIdListener;
 import org.onosproject.p4runtime.ctl.controller.P4RuntimeControllerImpl;
 import org.onosproject.p4runtime.ctl.controller.PacketInEvent;
 import org.slf4j.Logger;
@@ -42,6 +43,8 @@
 import java.nio.ByteBuffer;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
 import static java.lang.String.format;
 import static org.onosproject.p4runtime.ctl.codec.Codecs.CODECS;
 import static org.slf4j.LoggerFactory.getLogger;
@@ -54,73 +57,95 @@
 
     private static final Logger log = getLogger(StreamClientImpl.class);
 
-    private static final BigInteger ONE_THOUSAND = BigInteger.valueOf(1000);
-
     private final P4RuntimeClientImpl client;
     private final DeviceId deviceId;
     private final long p4DeviceId;
     private final PiPipeconfService pipeconfService;
+    private final MasterElectionIdStore masterElectionIdStore;
     private final P4RuntimeControllerImpl controller;
+
     private final StreamChannelManager streamChannelManager = new StreamChannelManager();
+    private final MasterElectionIdListener masterElectionIdListener = new InternalMasterElectionIdListener();
 
-    private P4RuntimeOuterClass.Uint128 lastUsedElectionId = P4RuntimeOuterClass.Uint128
-            .newBuilder().setLow(1).build();
+    private final AtomicBoolean isMaster = new AtomicBoolean(false);
+    private final AtomicBoolean requestedToBeMaster = new AtomicBoolean(false);
 
-    private final AtomicBoolean isClientMaster = new AtomicBoolean(false);
+    private BigInteger pendingElectionId = null;
+    private BigInteger lastUsedElectionId = null;
 
     StreamClientImpl(
             PiPipeconfService pipeconfService,
+            MasterElectionIdStore masterElectionIdStore,
             P4RuntimeClientImpl client,
             P4RuntimeControllerImpl controller) {
         this.client = client;
         this.deviceId = client.deviceId();
         this.p4DeviceId = client.p4DeviceId();
         this.pipeconfService = pipeconfService;
+        this.masterElectionIdStore = masterElectionIdStore;
         this.controller = controller;
     }
 
     @Override
-    public void openSession() {
-        if (isSessionOpen()) {
-            log.debug("Dropping request to open session for {}, session is already open",
-                      deviceId);
-            return;
-        }
-        log.debug("Opening session for {}...", deviceId);
-        sendMasterArbitrationUpdate(controller.newMasterElectionId(deviceId));
-
-    }
-
-    @Override
     public boolean isSessionOpen() {
         return streamChannelManager.isOpen();
     }
 
     @Override
     public void closeSession() {
-        streamChannelManager.complete();
+        synchronized (requestedToBeMaster) {
+            this.masterElectionIdStore.unsetListener(deviceId);
+            streamChannelManager.teardown();
+            pendingElectionId = null;
+            requestedToBeMaster.set(false);
+            isMaster.set(false);
+        }
     }
 
     @Override
-    public void runForMastership() {
-        if (!isSessionOpen()) {
-            log.debug("Dropping mastership request for {}, session is closed",
-                      deviceId);
-            return;
+    public void setMastership(boolean master, BigInteger newElectionId) {
+        checkNotNull(newElectionId);
+        checkArgument(newElectionId.compareTo(BigInteger.ZERO) > 0,
+                      "newElectionId must be a non zero positive number");
+        synchronized (requestedToBeMaster) {
+            requestedToBeMaster.set(master);
+            pendingElectionId = newElectionId;
+            handlePendingElectionId(masterElectionIdStore.get(deviceId));
         }
-        // Becoming master is a race. Here we increase our chances of win, i.e.
-        // using the highest election ID, against other ONOS nodes in the
-        // cluster that are calling openSession() (which is used to start the
-        // stream RPC session, not to become master).
-        log.debug("Running for mastership on {}...", deviceId);
-        final BigInteger masterId = controller.newMasterElectionId(deviceId)
-                .add(ONE_THOUSAND);
-        sendMasterArbitrationUpdate(masterId);
+    }
+
+    private void handlePendingElectionId(BigInteger masterElectionId) {
+        synchronized (requestedToBeMaster) {
+            if (pendingElectionId == null) {
+                // No pending requests.
+                return;
+            }
+            if (!requestedToBeMaster.get() && masterElectionId != null
+                    && pendingElectionId.compareTo(masterElectionId) > 0) {
+                log.info("Deferring sending master arbitration update, master " +
+                                  "election ID of server ({}) is smaller than " +
+                                  "requested one ({}), but we do NOT want to be master...",
+                          masterElectionId, pendingElectionId);
+                // Will try again as soon as the server reports a new master
+                // election ID that is bigger than the pending non-master one.
+                masterElectionIdStore.setListener(deviceId, masterElectionIdListener);
+            } else {
+                // Send now.
+                log.info("Setting mastership on {}... " +
+                                  "master={}, newElectionId={}, masterElectionId={}",
+                          deviceId, requestedToBeMaster.get(),
+                          pendingElectionId, masterElectionId);
+                sendMasterArbitrationUpdate(pendingElectionId);
+                pendingElectionId = null;
+                // No need to listen for master election ID changes.
+                masterElectionIdStore.unsetListener(deviceId);
+            }
+        }
     }
 
     @Override
     public boolean isMaster() {
-        return streamChannelManager.isOpen() && isClientMaster.get();
+        return isMaster.get();
     }
 
     @Override
@@ -141,7 +166,7 @@
             final StreamMessageRequest packetOutRequest = StreamMessageRequest
                     .newBuilder().setPacket(packetOut).build();
             // Send.
-            streamChannelManager.sendIfOpen(packetOutRequest);
+            streamChannelManager.send(packetOutRequest);
         } catch (CodecException e) {
             log.error("Unable to send packet-out: {}", e.getMessage());
         }
@@ -160,7 +185,7 @@
                                         .setElectionId(idMsg)
                                         .build())
                         .build());
-        lastUsedElectionId = idMsg;
+        lastUsedElectionId = electionId;
     }
 
     private P4RuntimeOuterClass.Uint128 bigIntegerToUint128(BigInteger value) {
@@ -220,13 +245,22 @@
         if (!msg.hasElectionId() || !msg.hasStatus()) {
             return;
         }
-        final boolean isMaster = msg.getStatus().getCode() == Status.OK.getCode().value();
-        log.debug("Received arbitration update from {}: isMaster={}, electionId={}",
-                  deviceId, isMaster, uint128ToBigInteger(msg.getElectionId()));
-        controller.postEvent(new P4RuntimeEvent(
-                P4RuntimeEvent.Type.ARBITRATION_RESPONSE,
-                new ArbitrationUpdateEvent(deviceId, isMaster)));
-        isClientMaster.set(isMaster);
+        // Is this client master?
+        isMaster.set(msg.getStatus().getCode() == Status.OK.getCode().value());
+        // Notify new master election IDs to all nodes via distributed store.
+        // This is required for those nodes who do not have a Stream RPC open,
+        // and that otherwise would not be aware of changes, keeping their
+        // pending mastership operations forever.
+        final BigInteger masterElectionId = uint128ToBigInteger(msg.getElectionId());
+        masterElectionIdStore.set(deviceId, masterElectionId);
+
+        log.debug("Received arbitration update from {}: isMaster={}, masterElectionId={}",
+                  deviceId, isMaster.get(), masterElectionId);
+
+        // Post mastership event via controller.
+        controller.postEvent(new DeviceAgentEvent(
+                isMaster.get() ? DeviceAgentEvent.Type.ROLE_MASTER
+                        : DeviceAgentEvent.Type.ROLE_STANDBY, deviceId));
     }
 
     /**
@@ -236,7 +270,22 @@
      * @return election ID uint128 protobuf message
      */
     P4RuntimeOuterClass.Uint128 lastUsedElectionId() {
-        return lastUsedElectionId;
+        return lastUsedElectionId == null
+                ? P4RuntimeOuterClass.Uint128.getDefaultInstance()
+                : bigIntegerToUint128(lastUsedElectionId);
+    }
+
+    /**
+     * Handles updates of the master election ID by applying any pending
+     * mastership operation.
+     */
+    private class InternalMasterElectionIdListener
+            implements MasterElectionIdStore.MasterElectionIdListener {
+
+        @Override
+        public void updated(BigInteger masterElectionId) {
+            handlePendingElectionId(masterElectionId);
+        }
     }
 
     /**
@@ -258,23 +307,12 @@
             }
         }
 
-        void sendIfOpen(StreamMessageRequest value) {
-            // We do not lock here, but we ignore NPEs due to stream RPC not
-            // being active (null requestObserver). Good for frequent
-            // packet-outs.
-            try {
-                doSend(value);
-            } catch (NullPointerException e) {
-                if (requestObserver != null) {
-                    // Must be something else.
-                    throw e;
-                }
-            }
-        }
-
         private void doSend(StreamMessageRequest value) {
             try {
                 requestObserver.onNext(value);
+                // Optimistically set the session as open. In case of errors, it
+                // will be closed by the response stream observer.
+                streamChannelManager.signalOpen();
             } catch (Throwable ex) {
                 if (ex instanceof StatusRuntimeException) {
                     log.warn("Unable to send {} to {}: {}",
@@ -283,7 +321,7 @@
                     log.error("Exception while sending {} to {}: {}",
                               value.getUpdateCase().toString(), deviceId, ex);
                 }
-                complete();
+                teardown();
             }
         }
 
@@ -299,7 +337,7 @@
             }
         }
 
-        void complete() {
+        void teardown() {
             synchronized (this) {
                 signalClosed();
                 if (requestObserver != null) {
@@ -311,23 +349,16 @@
         }
 
         void signalOpen() {
-            synchronized (this) {
-                final boolean wasOpen = open.getAndSet(true);
-                if (!wasOpen) {
-                    controller.postEvent(new P4RuntimeEvent(
-                            P4RuntimeEvent.Type.CHANNEL_EVENT,
-                            new ChannelEvent(deviceId, ChannelEvent.Type.OPEN)));
-                }
-            }
+            open.set(true);
         }
 
         void signalClosed() {
             synchronized (this) {
                 final boolean wasOpen = open.getAndSet(false);
                 if (wasOpen) {
-                    controller.postEvent(new P4RuntimeEvent(
-                            P4RuntimeEvent.Type.CHANNEL_EVENT,
-                            new ChannelEvent(deviceId, ChannelEvent.Type.CLOSED)));
+                    // We lost any valid mastership role.
+                    controller.postEvent(new DeviceAgentEvent(
+                            DeviceAgentEvent.Type.ROLE_NONE, deviceId));
                 }
             }
         }
@@ -352,13 +383,11 @@
 
         @Override
         public void onNext(StreamMessageResponse message) {
-            streamChannelManager.signalOpen();
             try {
                 if (log.isTraceEnabled()) {
-                    log.trace(
-                            "Received {} from {}: {}",
-                            message.getUpdateCase(), deviceId,
-                            TextFormat.shortDebugString(message));
+                    log.trace("Received {} from {}: {}",
+                              message.getUpdateCase(), deviceId,
+                              TextFormat.shortDebugString(message));
                 }
                 switch (message.getUpdateCase()) {
                     case PACKET:
@@ -388,17 +417,18 @@
                     log.warn("Error on stream channel for {}: {}",
                              deviceId, throwable.getMessage());
                 }
+                log.debug("", throwable);
             } else {
                 log.error(format("Exception on stream channel for %s",
                                  deviceId), throwable);
             }
-            streamChannelManager.complete();
+            streamChannelManager.teardown();
         }
 
         @Override
         public void onCompleted() {
             log.warn("Stream channel for {} has completed", deviceId);
-            streamChannelManager.complete();
+            streamChannelManager.teardown();
         }
     }
 }
diff --git a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/ChannelEvent.java b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/ChannelEvent.java
deleted file mode 100644
index 0a77e46..0000000
--- a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/ChannelEvent.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Copyright 2019-present Open Networking Foundation
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.onosproject.p4runtime.ctl.controller;
-
-import org.onosproject.net.DeviceId;
-import org.onosproject.p4runtime.api.P4RuntimeEventSubject;
-
-/**
- * Channel event in P4Runtime.
- */
-public final class ChannelEvent implements P4RuntimeEventSubject {
-
-    public enum Type {
-        OPEN,
-        CLOSED,
-        ERROR
-    }
-
-    private DeviceId deviceId;
-    private Type type;
-
-    /**
-     * Creates channel event with given status and throwable.
-     *
-     * @param deviceId  the device
-     * @param type      error type
-     */
-    public ChannelEvent(DeviceId deviceId, Type type) {
-        this.deviceId = deviceId;
-        this.type = type;
-    }
-
-    /**
-     * Gets the type of this event.
-     *
-     * @return the error type
-     */
-    public Type type() {
-        return type;
-    }
-
-    @Override
-    public DeviceId deviceId() {
-        return deviceId;
-    }
-}
diff --git a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/DistributedElectionIdGenerator.java b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/DistributedElectionIdGenerator.java
deleted file mode 100644
index 980ab11..0000000
--- a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/DistributedElectionIdGenerator.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Copyright 2019-present Open Networking Foundation
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.onosproject.p4runtime.ctl.controller;
-
-import org.onlab.util.KryoNamespace;
-import org.onosproject.net.DeviceId;
-import org.onosproject.store.serializers.KryoNamespaces;
-import org.onosproject.store.service.AtomicCounterMap;
-import org.onosproject.store.service.Serializer;
-import org.onosproject.store.service.StorageService;
-import org.slf4j.Logger;
-
-import java.math.BigInteger;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import static org.slf4j.LoggerFactory.getLogger;
-
-/**
- * Distributed implementation of a generator of P4Runtime election IDs.
- */
-final class DistributedElectionIdGenerator {
-
-    private final Logger log = getLogger(this.getClass());
-
-    // FIXME: counter map use long, but P4Runtime accepts 128bit election IDs
-    private AtomicCounterMap<DeviceId> electionIds;
-
-    /**
-     * Creates a new election ID generator using the given storage service.
-     *
-     * @param storageService storage service
-     */
-    DistributedElectionIdGenerator(StorageService storageService) {
-        KryoNamespace serializer = KryoNamespace.newBuilder()
-                .register(KryoNamespaces.API)
-                .build();
-        this.electionIds = storageService.<DeviceId>atomicCounterMapBuilder()
-                .withName("p4runtime-election-ids")
-                .withSerializer(Serializer.using(serializer))
-                .build();
-    }
-
-    /**
-     * Returns an election ID for the given device ID. The first election ID for
-     * a given device ID is always 1.
-     *
-     * @param deviceId device ID
-     * @return new election ID
-     */
-    BigInteger generate(DeviceId deviceId) {
-        if (electionIds == null) {
-            return null;
-        }
-        // Default value is 0 for AtomicCounterMap.
-        return BigInteger.valueOf(electionIds.incrementAndGet(deviceId));
-    }
-
-    /**
-     * Destroy the backing distributed primitive of this generator.
-     */
-    void destroy() {
-        try {
-            electionIds.destroy().get(10, TimeUnit.SECONDS);
-        } catch (InterruptedException | ExecutionException | TimeoutException e) {
-            log.error("Exception while destroying distributed counter map", e);
-        } finally {
-            electionIds = null;
-        }
-    }
-}
diff --git a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/DistributedMasterElectionIdStore.java b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/DistributedMasterElectionIdStore.java
new file mode 100644
index 0000000..dff2330
--- /dev/null
+++ b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/DistributedMasterElectionIdStore.java
@@ -0,0 +1,143 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.p4runtime.ctl.controller;
+
+import com.google.common.collect.Maps;
+import org.onlab.util.KryoNamespace;
+import org.onosproject.net.DeviceId;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.EventuallyConsistentMap;
+import org.onosproject.store.service.EventuallyConsistentMapEvent;
+import org.onosproject.store.service.EventuallyConsistentMapListener;
+import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.WallClockTimestamp;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
+import org.slf4j.Logger;
+
+import java.math.BigInteger;
+import java.util.concurrent.ConcurrentMap;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Distributed implementation of MasterElectionIdStore.
+ */
+@Component(immediate = true, service = MasterElectionIdStore.class)
+public class DistributedMasterElectionIdStore implements MasterElectionIdStore {
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected StorageService storageService;
+
+    private static final KryoNamespace SERIALIZER = KryoNamespace.newBuilder()
+            .register(KryoNamespaces.API)
+            .register(BigInteger.class)
+            .build();
+
+    private final Logger log = getLogger(getClass());
+    private final EventuallyConsistentMapListener<DeviceId, BigInteger> mapListener =
+            new InternalMapListener();
+
+    private EventuallyConsistentMap<DeviceId, BigInteger> masterElectionIds;
+    private ConcurrentMap<DeviceId, MasterElectionIdListener> listeners =
+            Maps.newConcurrentMap();
+
+    @Activate
+    public void activate() {
+        this.listeners = Maps.newConcurrentMap();
+        this.masterElectionIds = storageService.<DeviceId, BigInteger>eventuallyConsistentMapBuilder()
+                .withName("p4runtime-master-election-ids")
+                .withSerializer(SERIALIZER)
+                .withTimestampProvider((k, v) -> new WallClockTimestamp())
+                .build();
+        this.masterElectionIds.addListener(mapListener);
+        log.info("Started");
+    }
+
+    @Deactivate
+    public void deactivate() {
+        this.masterElectionIds.removeListener(mapListener);
+        this.masterElectionIds.destroy();
+        this.masterElectionIds = null;
+        this.listeners.clear();
+        this.listeners = null;
+        log.info("Stopped");
+    }
+
+
+    @Override
+    public void set(DeviceId deviceId, BigInteger electionId) {
+        checkNotNull(deviceId);
+        checkNotNull(electionId);
+        this.masterElectionIds.put(deviceId, electionId);
+    }
+
+    @Override
+    public BigInteger get(DeviceId deviceId) {
+        checkNotNull(deviceId);
+        return this.masterElectionIds.get(deviceId);
+    }
+
+    @Override
+    public void remove(DeviceId deviceId) {
+        checkNotNull(deviceId);
+        this.masterElectionIds.remove(deviceId);
+    }
+
+    @Override
+    public void setListener(DeviceId deviceId, MasterElectionIdListener newListener) {
+        checkNotNull(deviceId);
+        checkNotNull(newListener);
+        listeners.compute(deviceId, (did, existingListener) -> {
+            if (existingListener == null || existingListener == newListener) {
+                return newListener;
+            } else {
+                log.error("Cannot add listener as one already exist for {}", deviceId);
+                return existingListener;
+            }
+        });
+    }
+
+    @Override
+    public void unsetListener(DeviceId deviceId) {
+        listeners.remove(deviceId);
+    }
+
+    private class InternalMapListener implements EventuallyConsistentMapListener<DeviceId, BigInteger> {
+        @Override
+        public void event(EventuallyConsistentMapEvent<DeviceId, BigInteger> event) {
+            final MasterElectionIdListener listener = listeners.get(event.key());
+            if (listener == null) {
+                return;
+            }
+            switch (event.type()) {
+                case PUT:
+                    listener.updated(event.value());
+                    break;
+                case REMOVE:
+                    listener.updated(null);
+                    break;
+                default:
+                    log.error("Unrecognized map event type {}", event.type());
+            }
+        }
+    }
+}
diff --git a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/MasterElectionIdStore.java b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/MasterElectionIdStore.java
new file mode 100644
index 0000000..f5393d1
--- /dev/null
+++ b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/MasterElectionIdStore.java
@@ -0,0 +1,80 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.p4runtime.ctl.controller;
+
+import org.onosproject.net.DeviceId;
+
+import java.math.BigInteger;
+
+/**
+ * Store that keeps track of master election IDs for each device.
+ */
+public interface MasterElectionIdStore {
+
+    /**
+     * Sets the master election ID for the given device.
+     *
+     * @param deviceId   device ID
+     * @param electionId election ID
+     */
+    void set(DeviceId deviceId, BigInteger electionId);
+
+    /**
+     * Returns the last known master election ID for the given device, or null.
+     *
+     * @param deviceId device ID
+     * @return election ID
+     */
+    BigInteger get(DeviceId deviceId);
+
+    /**
+     * Removes any state associated with the given device.
+     *
+     * @param deviceId device ID
+     */
+    void remove(DeviceId deviceId);
+
+    /**
+     * Sets a listener for the given device that will be invoked every time
+     * there will be changes to the master election ID.
+     *
+     * @param deviceId device ID
+     * @param listener listener
+     */
+    void setListener(DeviceId deviceId, MasterElectionIdListener listener);
+
+    /**
+     * Unset the listener for the given device.
+     *
+     * @param deviceId device ID
+     */
+    void unsetListener(DeviceId deviceId);
+
+    /**
+     * Listener of master election ID changes for a specific device.
+     */
+    interface MasterElectionIdListener {
+
+        /**
+         * Notifies that the master election ID has been updated to the given
+         * (nullable) value.
+         *
+         * @param masterElectionId new master election ID, or null
+         */
+        void updated(BigInteger masterElectionId);
+    }
+}
diff --git a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/P4RuntimeControllerImpl.java b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/P4RuntimeControllerImpl.java
index affbf7d..2256d5e 100644
--- a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/P4RuntimeControllerImpl.java
+++ b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/controller/P4RuntimeControllerImpl.java
@@ -16,33 +16,19 @@
 
 package org.onosproject.p4runtime.ctl.controller;
 
-import com.google.common.collect.Maps;
 import io.grpc.ManagedChannel;
 import org.onosproject.grpc.ctl.AbstractGrpcClientController;
 import org.onosproject.net.DeviceId;
-import org.onosproject.net.device.DeviceAgentEvent;
-import org.onosproject.net.device.DeviceAgentListener;
 import org.onosproject.net.pi.service.PiPipeconfService;
-import org.onosproject.net.provider.ProviderId;
 import org.onosproject.p4runtime.api.P4RuntimeClient;
 import org.onosproject.p4runtime.api.P4RuntimeClientKey;
 import org.onosproject.p4runtime.api.P4RuntimeController;
 import org.onosproject.p4runtime.api.P4RuntimeEvent;
 import org.onosproject.p4runtime.api.P4RuntimeEventListener;
 import org.onosproject.p4runtime.ctl.client.P4RuntimeClientImpl;
-import org.onosproject.store.service.StorageService;
-import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Component;
-import org.osgi.service.component.annotations.Deactivate;
 import org.osgi.service.component.annotations.Reference;
 import org.osgi.service.component.annotations.ReferenceCardinality;
-import org.slf4j.Logger;
-
-import java.math.BigInteger;
-import java.util.concurrent.ConcurrentMap;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-import static org.slf4j.LoggerFactory.getLogger;
 
 /**
  * P4Runtime controller implementation.
@@ -53,122 +39,34 @@
         <P4RuntimeClientKey, P4RuntimeClient, P4RuntimeEvent, P4RuntimeEventListener>
         implements P4RuntimeController {
 
-    private final Logger log = getLogger(getClass());
-
-    private final ConcurrentMap<DeviceId, ConcurrentMap<ProviderId, DeviceAgentListener>>
-            deviceAgentListeners = Maps.newConcurrentMap();
-
-    private DistributedElectionIdGenerator electionIdGenerator;
-
-    @Reference(cardinality = ReferenceCardinality.MANDATORY)
-    private StorageService storageService;
-
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
     private PiPipeconfService pipeconfService;
 
-    @Activate
-    public void activate() {
-        super.activate();
-        eventDispatcher.addSink(P4RuntimeEvent.class, listenerRegistry);
-        electionIdGenerator = new DistributedElectionIdGenerator(storageService);
-        log.info("Started");
-    }
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    private MasterElectionIdStore masterElectionIdStore;
 
-    @Deactivate
-    public void deactivate() {
-        super.deactivate();
-        deviceAgentListeners.clear();
-        electionIdGenerator.destroy();
-        electionIdGenerator = null;
-        log.info("Stopped");
+    public P4RuntimeControllerImpl() {
+        super(P4RuntimeEvent.class);
     }
 
     @Override
-    protected P4RuntimeClient createClientInstance(P4RuntimeClientKey clientKey, ManagedChannel channel) {
-        return new P4RuntimeClientImpl(clientKey, channel, this, pipeconfService);
+    public void removeClient(DeviceId deviceId) {
+        super.removeClient(deviceId);
+        // Assuming that when a client is removed, it is done so by all nodes,
+        // this is the best place to clear master election ID state.
+        masterElectionIdStore.remove(deviceId);
     }
 
     @Override
-    public void addDeviceAgentListener(DeviceId deviceId, ProviderId providerId, DeviceAgentListener listener) {
-        checkNotNull(deviceId, "deviceId cannot be null");
-        checkNotNull(deviceId, "providerId cannot be null");
-        checkNotNull(listener, "listener cannot be null");
-        deviceAgentListeners.putIfAbsent(deviceId, Maps.newConcurrentMap());
-        deviceAgentListeners.get(deviceId).put(providerId, listener);
+    public void removeClient(P4RuntimeClientKey clientKey) {
+        super.removeClient(clientKey);
+        masterElectionIdStore.remove(clientKey.deviceId());
     }
 
     @Override
-    public void removeDeviceAgentListener(DeviceId deviceId, ProviderId providerId) {
-        checkNotNull(deviceId, "deviceId cannot be null");
-        checkNotNull(providerId, "listener cannot be null");
-        deviceAgentListeners.computeIfPresent(deviceId, (did, listeners) -> {
-            listeners.remove(providerId);
-            return listeners;
-        });
-    }
-
-    public BigInteger newMasterElectionId(DeviceId deviceId) {
-        return electionIdGenerator.generate(deviceId);
-    }
-
-    public void postEvent(P4RuntimeEvent event) {
-        switch (event.type()) {
-            case CHANNEL_EVENT:
-                handleChannelEvent(event);
-                break;
-            case ARBITRATION_RESPONSE:
-                handleArbitrationReply(event);
-                break;
-            case PERMISSION_DENIED:
-                handlePermissionDenied(event);
-                break;
-            default:
-                post(event);
-                break;
-        }
-    }
-
-    private void handlePermissionDenied(P4RuntimeEvent event) {
-        postDeviceAgentEvent(event.subject().deviceId(), new DeviceAgentEvent(
-                DeviceAgentEvent.Type.NOT_MASTER, event.subject().deviceId()));
-    }
-
-    private void handleChannelEvent(P4RuntimeEvent event) {
-        final ChannelEvent channelEvent = (ChannelEvent) event.subject();
-        final DeviceId deviceId = channelEvent.deviceId();
-        final DeviceAgentEvent.Type agentEventType;
-        switch (channelEvent.type()) {
-            case OPEN:
-                agentEventType = DeviceAgentEvent.Type.CHANNEL_OPEN;
-                break;
-            case CLOSED:
-                agentEventType = DeviceAgentEvent.Type.CHANNEL_CLOSED;
-                break;
-            case ERROR:
-                agentEventType = !isReachable(deviceId)
-                        ? DeviceAgentEvent.Type.CHANNEL_CLOSED
-                        : DeviceAgentEvent.Type.CHANNEL_ERROR;
-                break;
-            default:
-                log.warn("Unrecognized channel event type {}", channelEvent.type());
-                return;
-        }
-        postDeviceAgentEvent(deviceId, new DeviceAgentEvent(agentEventType, deviceId));
-    }
-
-    private void handleArbitrationReply(P4RuntimeEvent event) {
-        final DeviceId deviceId = event.subject().deviceId();
-        final ArbitrationUpdateEvent response = (ArbitrationUpdateEvent) event.subject();
-        final DeviceAgentEvent.Type roleType = response.isMaster()
-                ? DeviceAgentEvent.Type.ROLE_MASTER
-                : DeviceAgentEvent.Type.ROLE_STANDBY;
-        postDeviceAgentEvent(deviceId, new DeviceAgentEvent(
-                roleType, response.deviceId()));
-    }
-
-    private void postDeviceAgentEvent(DeviceId deviceId, DeviceAgentEvent event) {
-        if (deviceAgentListeners.containsKey(deviceId)) {
-            deviceAgentListeners.get(deviceId).values().forEach(l -> l.event(event));
-        }
+    protected P4RuntimeClient createClientInstance(
+            P4RuntimeClientKey clientKey, ManagedChannel channel) {
+        return new P4RuntimeClientImpl(clientKey, channel, this,
+                                       pipeconfService, masterElectionIdStore);
     }
 }
diff --git a/protocols/p4runtime/ctl/src/test/java/org/onosproject/p4runtime/ctl/MockMasterElectionIdStore.java b/protocols/p4runtime/ctl/src/test/java/org/onosproject/p4runtime/ctl/MockMasterElectionIdStore.java
new file mode 100644
index 0000000..595e07d
--- /dev/null
+++ b/protocols/p4runtime/ctl/src/test/java/org/onosproject/p4runtime/ctl/MockMasterElectionIdStore.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.p4runtime.ctl;
+
+import org.onosproject.net.DeviceId;
+import org.onosproject.p4runtime.ctl.controller.MasterElectionIdStore;
+
+import java.math.BigInteger;
+
+public class MockMasterElectionIdStore implements MasterElectionIdStore {
+
+    @Override
+    public void set(DeviceId deviceId, BigInteger electionId) {
+
+    }
+
+    @Override
+    public BigInteger get(DeviceId deviceId) {
+        return null;
+    }
+
+    @Override
+    public void remove(DeviceId deviceId) {
+
+    }
+
+    @Override
+    public void setListener(DeviceId deviceId, MasterElectionIdListener listener) {
+
+    }
+
+    @Override
+    public void unsetListener(DeviceId deviceId) {
+
+    }
+}
diff --git a/protocols/p4runtime/ctl/src/test/java/org/onosproject/p4runtime/ctl/P4RuntimeGroupTest.java b/protocols/p4runtime/ctl/src/test/java/org/onosproject/p4runtime/ctl/P4RuntimeGroupTest.java
index c6a43cb..4c4dcdf 100644
--- a/protocols/p4runtime/ctl/src/test/java/org/onosproject/p4runtime/ctl/P4RuntimeGroupTest.java
+++ b/protocols/p4runtime/ctl/src/test/java/org/onosproject/p4runtime/ctl/P4RuntimeGroupTest.java
@@ -55,6 +55,8 @@
 import p4.v1.P4RuntimeOuterClass.WriteRequest;
 
 import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.net.URL;
 import java.util.Collection;
 import java.util.List;
@@ -62,6 +64,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
+import static java.lang.String.format;
 import static org.easymock.EasyMock.niceMock;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -105,7 +108,7 @@
     private static final int SET_EGRESS_PORT_ID = 16794308;
     private static final String GRPC_SERVER_NAME = "P4RuntimeGroupTest";
     private static final long DEFAULT_TIMEOUT_TIME = 10;
-    private static final Uint128 DEFAULT_ELECTION_ID = Uint128.newBuilder().setLow(1).build();
+    private static final Uint128 DEFAULT_ELECTION_ID = Uint128.getDefaultInstance();
     private static final String P4R_IP = "127.0.0.1";
     private static final int P4R_PORT = 50010;
 
@@ -157,10 +160,13 @@
 
 
     @Before
-    public void setup() {
+    public void setup() throws URISyntaxException {
         controller = niceMock(org.onosproject.p4runtime.ctl.controller.P4RuntimeControllerImpl.class);
-        P4RuntimeClientKey clientKey = new P4RuntimeClientKey(DEVICE_ID, P4R_IP, P4R_PORT, P4_DEVICE_ID);
-        client = new P4RuntimeClientImpl(clientKey, grpcChannel, controller, new MockPipeconfService());
+        P4RuntimeClientKey clientKey = new P4RuntimeClientKey(DEVICE_ID, new URI(
+                format("grpc://%s:%d?device_id=%d", P4R_IP, P4R_PORT, P4_DEVICE_ID)));
+        client = new P4RuntimeClientImpl(
+                clientKey, grpcChannel, controller, new MockPipeconfService(),
+                new MockMasterElectionIdStore());
     }
 
     @Test
diff --git a/providers/general/device/src/main/java/org/onosproject/provider/general/device/api/GeneralProviderDeviceConfig.java b/providers/general/device/src/main/java/org/onosproject/provider/general/device/api/GeneralProviderDeviceConfig.java
deleted file mode 100644
index 0675a4b..0000000
--- a/providers/general/device/src/main/java/org/onosproject/provider/general/device/api/GeneralProviderDeviceConfig.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Copyright 2017-present Open Networking Foundation
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.onosproject.provider.general.device.api;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.google.common.annotations.Beta;
-import org.onosproject.net.DeviceId;
-import org.onosproject.net.config.Config;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * Configuration for General device provider.
- */
-@Beta
-public class GeneralProviderDeviceConfig extends Config<DeviceId> {
-
-    private static final String DEVICEKEYID = "deviceKeyId";
-
-
-    @Override
-    public boolean isValid() {
-        return true;
-    }
-
-    /**
-     * Gets the information of all protocols associated to the device.
-     *
-     * @return map of protocol name and relative information
-     */
-    public Map<String, DeviceInfoConfig> protocolsInfo() {
-        return getProtocolInfoMap();
-    }
-
-    private Map<String, DeviceInfoConfig> getProtocolInfoMap() {
-        Map<String, DeviceInfoConfig> deviceMap = new HashMap<>();
-        node.fieldNames().forEachRemaining(name -> {
-
-            Map<String, String> configMap = new HashMap<>();
-            JsonNode protocol = node.get(name);
-            protocol.fieldNames().forEachRemaining(info -> configMap.put(info, protocol.get(info).asText()));
-
-            String deviceKeyId = "";
-            if (protocol.has(DEVICEKEYID)) {
-                deviceKeyId = protocol.get(DEVICEKEYID).asText("");
-            }
-
-            deviceMap.put(name, new DeviceInfoConfig(configMap, deviceKeyId));
-        });
-        return deviceMap;
-    }
-
-}
diff --git a/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/DeviceTaskExecutor.java b/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/DeviceTaskExecutor.java
new file mode 100644
index 0000000..6f90256
--- /dev/null
+++ b/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/DeviceTaskExecutor.java
@@ -0,0 +1,256 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.provider.general.device.impl;
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalListener;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.Striped;
+import org.onosproject.net.DeviceId;
+import org.slf4j.Logger;
+
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static java.lang.String.format;
+import static java.lang.System.currentTimeMillis;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Allows submitting tasks related to a specific device. Takes care of executing
+ * pending tasks sequentially for each device in a FIFO order, while using a
+ * delegate executor. It also avoids executing duplicate tasks when arriving
+ * back-to-back.
+ *
+ * @param <T> enum describing the type of task
+ */
+
+class DeviceTaskExecutor<T extends Enum> {
+    /**
+     * Minimum interval between duplicate back-to-back tasks.
+     */
+    private static final int DUPLICATE_MIN_INTERVAL_MILLIS = 1000;
+
+    private final Logger log = getLogger(getClass());
+
+    private final ExecutorService delegate;
+    private final AtomicBoolean canceled = new AtomicBoolean(false);
+    private final Set<DeviceId> busyDevices = Sets.newConcurrentHashSet();
+    private final Set<DeviceId> pendingDevices = Sets.newConcurrentHashSet();
+    private final Striped<Lock> deviceLocks = Striped.lock(30);
+    private final LoadingCache<DeviceId, TaskQueue> taskQueues = CacheBuilder.newBuilder()
+            .expireAfterAccess(1, TimeUnit.MINUTES)
+            .removalListener((RemovalListener<DeviceId, TaskQueue>) notification -> {
+                if (!notification.getValue().isEmpty()) {
+                    log.warn("Cache evicted non-empty task queue for {} ({} pending tasks)",
+                             notification.getKey(), notification.getValue().size());
+                }
+            })
+            .build(new CacheLoader<DeviceId, TaskQueue>() {
+                @SuppressWarnings("NullableProblems")
+                @Override
+                public TaskQueue load(DeviceId deviceId) {
+                    return new TaskQueue();
+                }
+            });
+
+    /**
+     * Creates a new executor with the given delegate executor service.
+     *
+     * @param delegate executor service
+     */
+    DeviceTaskExecutor(ExecutorService delegate) {
+        checkNotNull(delegate);
+        this.delegate = delegate;
+    }
+
+    /**
+     * Submit a tasks.
+     *
+     * @param deviceId device associated with the task
+     * @param type     type of task (used to remove eventual back-to-back
+     *                 duplicates)
+     * @param runnable runnable to execute
+     */
+    void submit(DeviceId deviceId, T type, Runnable runnable) {
+        checkNotNull(deviceId);
+        checkNotNull(type);
+        checkNotNull(runnable);
+
+        if (canceled.get()) {
+            log.warn("Executor was cancelled, cannot submit task {} for {}",
+                     type, deviceId);
+            return;
+        }
+
+        final DeviceTask task = new DeviceTask(deviceId, type, runnable);
+        deviceLocks.get(deviceId).lock();
+        try {
+            if (taskQueues.get(deviceId).isBackToBackDuplicate(type)) {
+                if (log.isDebugEnabled()) {
+                    log.debug("Dropping back-to-back duplicate task {} for {}",
+                              type, deviceId);
+                }
+                return;
+            }
+            if (taskQueues.get(deviceId).offer(task)) {
+                pendingDevices.add(deviceId);
+                if (!busyDevices.contains(deviceId)) {
+                    // The task was submitted to the queue and we are not
+                    // performing any other task for this device. There is at
+                    // least one task that is ready to be executed.
+                    delegate.execute(this::performTaskIfAny);
+                }
+            } else {
+                log.warn("Unable to submit task {} for {}",
+                         task.type, task.deviceId);
+            }
+        } catch (ExecutionException e) {
+            log.warn("Exception while accessing task queue cache", e);
+        } finally {
+            deviceLocks.get(task.deviceId).unlock();
+        }
+    }
+
+    /**
+     * Prevents the executor from executing any more tasks.
+     */
+    void cancel() {
+        canceled.set(true);
+    }
+
+    private void performTaskIfAny() {
+        final DeviceTask task = pollTask();
+        if (task == null) {
+            // No tasks.
+            return;
+        }
+        if (canceled.get()) {
+            log.warn("Executor was cancelled, dropping task {} for {}",
+                     task.type, task.deviceId);
+            return;
+        }
+        if (log.isTraceEnabled()) {
+            log.trace("STARTING task {} for {}...", task.type.name(), task.deviceId);
+        }
+        try {
+            task.runnable.run();
+        } catch (DeviceTaskException e) {
+            log.error("Unable to complete task {} for {}: {}",
+                      task.type, task.deviceId, e.getMessage());
+        } catch (Throwable t) {
+            log.error(format(
+                    "Uncaught exception when executing task %s for %s",
+                    task.type, task.deviceId), t);
+        }
+        if (log.isTraceEnabled()) {
+            log.trace("COMPLETED task {} for {}", task.type.name(), task.deviceId);
+        }
+        busyDevices.remove(task.deviceId);
+        delegate.execute(this::performTaskIfAny);
+    }
+
+    private DeviceTask pollTask() {
+        for (DeviceId deviceId : pendingDevices) {
+            final DeviceTask task;
+            deviceLocks.get(deviceId).lock();
+            try {
+                if (busyDevices.contains(deviceId)) {
+                    // Next device.
+                    continue;
+                }
+                task = taskQueues.get(deviceId).poll();
+                if (task == null) {
+                    // Next device.
+                    continue;
+                }
+                if (taskQueues.get(deviceId).isEmpty()) {
+                    pendingDevices.remove(deviceId);
+                }
+                busyDevices.add(deviceId);
+                return task;
+            } catch (ExecutionException e) {
+                log.warn("Exception while accessing task queue cache", e);
+            } finally {
+                deviceLocks.get(deviceId).unlock();
+            }
+        }
+        return null;
+    }
+
+    /**
+     * Device task as stored in the task queue.
+     */
+    private class DeviceTask {
+
+        private final DeviceId deviceId;
+        private final T type;
+        private final Runnable runnable;
+
+        DeviceTask(DeviceId deviceId, T type, Runnable runnable) {
+            this.deviceId = deviceId;
+            this.type = type;
+            this.runnable = runnable;
+        }
+    }
+
+    /**
+     * A queue that keeps track of the last task added to detects back-to-back
+     * duplicates.
+     */
+    private class TaskQueue extends ConcurrentLinkedQueue<DeviceTask> {
+
+        private T lastTaskAdded;
+        private long lastAddedMillis;
+
+        @Override
+        public boolean offer(DeviceTask deviceTask) {
+            lastTaskAdded = deviceTask.type;
+            lastAddedMillis = currentTimeMillis();
+            return super.offer(deviceTask);
+        }
+
+        boolean isBackToBackDuplicate(T taskType) {
+            return lastTaskAdded != null
+                    && lastTaskAdded.equals(taskType)
+                    && (currentTimeMillis() - lastAddedMillis) <= DUPLICATE_MIN_INTERVAL_MILLIS;
+        }
+    }
+
+    /**
+     * Signals an error that prevented normal execution of the task.
+     */
+    static class DeviceTaskException extends RuntimeException {
+
+        /**
+         * Creates a new exception.
+         *
+         * @param message explanation
+         */
+        DeviceTaskException(String message) {
+            super(message);
+        }
+    }
+}
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 1eaa808..481a310 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,18 +16,17 @@
 
 package org.onosproject.provider.general.device.impl;
 
-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.onlab.packet.ChassisId;
 import org.onlab.util.ItemNotFoundException;
 import org.onlab.util.Tools;
 import org.onosproject.cfg.ComponentConfigService;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.NodeId;
 import org.onosproject.core.CoreService;
 import org.onosproject.gnmi.api.GnmiController;
+import org.onosproject.mastership.MastershipInfo;
 import org.onosproject.mastership.MastershipService;
-import org.onosproject.net.AnnotationKeys;
 import org.onosproject.net.DefaultAnnotations;
 import org.onosproject.net.Device;
 import org.onosproject.net.DeviceId;
@@ -35,27 +34,21 @@
 import org.onosproject.net.PortNumber;
 import org.onosproject.net.behaviour.PiPipelineProgrammable;
 import org.onosproject.net.behaviour.PortAdmin;
-import org.onosproject.net.config.ConfigFactory;
 import org.onosproject.net.config.NetworkConfigEvent;
 import org.onosproject.net.config.NetworkConfigListener;
 import org.onosproject.net.config.NetworkConfigRegistry;
 import org.onosproject.net.config.basics.BasicDeviceConfig;
-import org.onosproject.net.config.basics.SubjectFactories;
 import org.onosproject.net.device.DefaultDeviceDescription;
 import org.onosproject.net.device.DeviceAgentEvent;
 import org.onosproject.net.device.DeviceAgentListener;
 import org.onosproject.net.device.DeviceDescription;
 import org.onosproject.net.device.DeviceDescriptionDiscovery;
-import org.onosproject.net.device.DeviceEvent;
 import org.onosproject.net.device.DeviceHandshaker;
-import org.onosproject.net.device.DeviceListener;
 import org.onosproject.net.device.DeviceProvider;
 import org.onosproject.net.device.DeviceProviderRegistry;
 import org.onosproject.net.device.DeviceProviderService;
 import org.onosproject.net.device.DeviceService;
 import org.onosproject.net.device.PortDescription;
-import org.onosproject.net.device.PortStatistics;
-import org.onosproject.net.device.PortStatisticsDiscovery;
 import org.onosproject.net.driver.Behaviour;
 import org.onosproject.net.driver.DefaultDriverData;
 import org.onosproject.net.driver.DefaultDriverHandler;
@@ -64,14 +57,13 @@
 import org.onosproject.net.driver.DriverService;
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.net.pi.model.PiPipeconfId;
-import org.onosproject.net.pi.service.PiPipeconfConfig;
 import org.onosproject.net.pi.service.PiPipeconfService;
 import org.onosproject.net.pi.service.PiPipeconfWatchdogEvent;
 import org.onosproject.net.pi.service.PiPipeconfWatchdogListener;
 import org.onosproject.net.pi.service.PiPipeconfWatchdogService;
 import org.onosproject.net.provider.AbstractProvider;
 import org.onosproject.net.provider.ProviderId;
-import org.onosproject.provider.general.device.api.GeneralProviderDeviceConfig;
+import org.onosproject.provider.general.device.impl.DeviceTaskExecutor.DeviceTaskException;
 import org.osgi.service.component.ComponentContext;
 import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Component;
@@ -81,50 +73,43 @@
 import org.osgi.service.component.annotations.ReferenceCardinality;
 import org.slf4j.Logger;
 
-import java.security.SecureRandom;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.Dictionary;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
-import java.util.StringJoiner;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.concurrent.locks.Lock;
-import java.util.function.Supplier;
 
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Strings.isNullOrEmpty;
+import static java.lang.System.currentTimeMillis;
 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.onosproject.provider.general.device.impl.OsgiPropertyConstants.OP_TIMEOUT_SHORT;
 import static org.onosproject.provider.general.device.impl.OsgiPropertyConstants.OP_TIMEOUT_SHORT_DEFAULT;
-import static org.onosproject.provider.general.device.impl.OsgiPropertyConstants.PROBE_FREQUENCY;
-import static org.onosproject.provider.general.device.impl.OsgiPropertyConstants.PROBE_FREQUENCY_DEFAULT;
-import static org.onosproject.provider.general.device.impl.OsgiPropertyConstants.STATS_POLL_FREQUENCY;
-import static org.onosproject.provider.general.device.impl.OsgiPropertyConstants.STATS_POLL_FREQUENCY_DEFAULT;
+import static org.onosproject.provider.general.device.impl.OsgiPropertyConstants.PROBE_INTERVAL;
+import static org.onosproject.provider.general.device.impl.OsgiPropertyConstants.PROBE_INTERVAL_DEFAULT;
+import static org.onosproject.provider.general.device.impl.OsgiPropertyConstants.STATS_POLL_INTERVAL;
+import static org.onosproject.provider.general.device.impl.OsgiPropertyConstants.STATS_POLL_INTERVAL_DEFAULT;
 import static org.slf4j.LoggerFactory.getLogger;
 
 /**
- * Provider which uses drivers to detect device and do initial handshake and
- * channel establishment with devices. Any other provider specific operation is
- * also delegated to the DeviceHandshaker driver.
+ * Provider which uses drivers to discover devices, perform initial handshake,
+ * and notify the core of disconnection events. The implementation listens for
+ * events from netcfg or the drivers (via {@link DeviceAgentListener}) andP
+ * schedules task for each event.
  */
-@Beta
 @Component(immediate = true,
         property = {
-                STATS_POLL_FREQUENCY + ":Integer=" + STATS_POLL_FREQUENCY_DEFAULT,
-                PROBE_FREQUENCY + ":Integer=" + PROBE_FREQUENCY_DEFAULT,
+                PROBE_INTERVAL + ":Integer=" + PROBE_INTERVAL_DEFAULT,
+                STATS_POLL_INTERVAL + ":Integer=" + STATS_POLL_INTERVAL_DEFAULT,
                 OP_TIMEOUT_SHORT + ":Integer=" + OP_TIMEOUT_SHORT_DEFAULT,
         })
 public class GeneralDeviceProvider extends AbstractProvider
@@ -132,16 +117,12 @@
 
     private final Logger log = getLogger(getClass());
 
-    private static final String APP_NAME = "org.onosproject.gdp";
+    private static final String APP_NAME = "org.onosproject.generaldeviceprovider";
     private static final String URI_SCHEME = "device";
-    private static final String CFG_SCHEME = "generalprovider";
     private static final String DEVICE_PROVIDER_PACKAGE =
             "org.onosproject.general.provider.device";
     private static final int CORE_POOL_SIZE = 10;
     private static final String UNKNOWN = "unknown";
-    private static final String DRIVER = "driver";
-    private static final Set<String> PIPELINE_CONFIGURABLE_PROTOCOLS =
-            ImmutableSet.of("p4runtime");
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
     private DeviceProviderRegistry providerRegistry;
@@ -165,6 +146,9 @@
     private MastershipService mastershipService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    private ClusterService clusterService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
     private PiPipeconfService pipeconfService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
@@ -180,68 +164,55 @@
 
     private GnmiDeviceStateSubscriber gnmiDeviceStateSubscriber;
 
-    /**
-     * Configure poll frequency for port status and statistics; default is 10 sec.
-     */
-    private int statsPollFrequency = STATS_POLL_FREQUENCY_DEFAULT;
+    /** Configure interval for checking device availability; default is 10 sec. */
+    private int probeInterval = PROBE_INTERVAL_DEFAULT;
 
-    /**
-     * Configure probe frequency for checking device availability; default is 10 sec.
-     */
-    private int probeFrequency = PROBE_FREQUENCY_DEFAULT;
+    /** Configure poll frequency for port status and stats; default is 10 sec. */
+    private int statsPollInterval = STATS_POLL_INTERVAL_DEFAULT;
 
-    /**
-     * Configure timeout in seconds for device operations that are supposed to take a short time
-     * (e.g. checking device reachability); default is 10 seconds.
-     */
+    /** Configure timeout in seconds for device operations; default is 10 sec. */
     private int opTimeoutShort = OP_TIMEOUT_SHORT_DEFAULT;
 
-    //FIXME to be removed when netcfg will issue device events in a bundle or
-    //ensures all configuration needed is present
-    private final Set<DeviceId> deviceConfigured = new CopyOnWriteArraySet<>();
-    private final Set<DeviceId> driverConfigured = new CopyOnWriteArraySet<>();
-    private final Set<DeviceId> pipelineConfigured = new CopyOnWriteArraySet<>();
-
-    private final Map<DeviceId, MastershipRole> requestedRoles = Maps.newConcurrentMap();
-    private final ConcurrentMap<DeviceId, ScheduledFuture<?>> statsPollingTasks = new ConcurrentHashMap<>();
     private final Map<DeviceId, DeviceHandshaker> handshakersWithListeners = Maps.newConcurrentMap();
-    private final InternalDeviceListener deviceListener = new InternalDeviceListener();
+    private final Map<DeviceId, Long> lastProbedAvailability = Maps.newConcurrentMap();
     private final InternalPipeconfWatchdogListener pipeconfWatchdogListener = new InternalPipeconfWatchdogListener();
     private final NetworkConfigListener cfgListener = new InternalNetworkConfigListener();
     private final DeviceAgentListener deviceAgentListener = new InternalDeviceAgentListener();
-    private final ConfigFactory factory = new InternalConfigFactory();
-    private final Striped<Lock> deviceLocks = Striped.lock(30);
 
-    private ExecutorService connectionExecutor;
-    private ScheduledExecutorService statsExecutor;
+    private ExecutorService mainExecutor;
+    private DeviceTaskExecutor<TaskType> taskExecutor;
     private ScheduledExecutorService probeExecutor;
     private ScheduledFuture<?> probeTask;
+    private StatsPoller statsPoller;
     private DeviceProviderService providerService;
 
     public GeneralDeviceProvider() {
         super(new ProviderId(URI_SCHEME, DEVICE_PROVIDER_PACKAGE));
     }
 
+    protected DeviceProviderService providerService() {
+        return providerService;
+    }
+
     @Activate
     public void activate(ComponentContext context) {
-        connectionExecutor = newFixedThreadPool(CORE_POOL_SIZE, groupedThreads(
-                "onos/gdp-connect", "%d", log));
-        statsExecutor = newScheduledThreadPool(CORE_POOL_SIZE, groupedThreads(
-                "onos/gdp-stats", "%d", log));
+        mainExecutor = newFixedThreadPool(CORE_POOL_SIZE, groupedThreads(
+                "onos/gdp-task", "%d", log));
+        taskExecutor = new DeviceTaskExecutor<>(mainExecutor);
         probeExecutor = newSingleThreadScheduledExecutor(groupedThreads(
                 "onos/gdp-probe", "%d", log));
         providerService = providerRegistry.register(this);
         componentConfigService.registerProperties(getClass());
         coreService.registerApplication(APP_NAME);
-        cfgService.registerConfigFactory(factory);
         cfgService.addListener(cfgListener);
-        deviceService.addListener(deviceListener);
         pipeconfWatchdogService.addListener(pipeconfWatchdogListener);
-        rescheduleProbeTask(false);
-        modified(context);
-        gnmiDeviceStateSubscriber = new GnmiDeviceStateSubscriber(gnmiController,
-                deviceService, mastershipService, providerService);
+        gnmiDeviceStateSubscriber = new GnmiDeviceStateSubscriber(
+                gnmiController, deviceService, mastershipService, providerService);
         gnmiDeviceStateSubscriber.activate();
+        startOrRescheduleProbeTask();
+        statsPoller = new StatsPoller(deviceService, mastershipService, providerService);
+        statsPoller.activate(statsPollInterval);
+        modified(context);
         log.info("Started");
     }
 
@@ -252,55 +223,35 @@
         }
 
         Dictionary<?, ?> properties = context.getProperties();
-        final int oldStatsPollFrequency = statsPollFrequency;
-        statsPollFrequency = Tools.getIntegerProperty(
-                properties, STATS_POLL_FREQUENCY, STATS_POLL_FREQUENCY_DEFAULT);
+        final int oldProbeFrequency = probeInterval;
+        probeInterval = Tools.getIntegerProperty(
+                properties, PROBE_INTERVAL, PROBE_INTERVAL_DEFAULT);
         log.info("Configured. {} is configured to {} seconds",
-                 STATS_POLL_FREQUENCY, statsPollFrequency);
-        final int oldProbeFrequency = probeFrequency;
-        probeFrequency = Tools.getIntegerProperty(
-                properties, PROBE_FREQUENCY, PROBE_FREQUENCY_DEFAULT);
+                 PROBE_INTERVAL, probeInterval);
+        final int oldStatsPollFrequency = statsPollInterval;
+        statsPollInterval = Tools.getIntegerProperty(
+                properties, STATS_POLL_INTERVAL, STATS_POLL_INTERVAL_DEFAULT);
         log.info("Configured. {} is configured to {} seconds",
-                 PROBE_FREQUENCY, probeFrequency);
+                 STATS_POLL_INTERVAL, statsPollInterval);
         opTimeoutShort = Tools.getIntegerProperty(
                 properties, OP_TIMEOUT_SHORT, OP_TIMEOUT_SHORT_DEFAULT);
         log.info("Configured. {} is configured to {} seconds",
                  OP_TIMEOUT_SHORT, opTimeoutShort);
 
-        if (oldStatsPollFrequency != statsPollFrequency) {
-            rescheduleStatsPollingTasks();
+        if (oldProbeFrequency != probeInterval) {
+            startOrRescheduleProbeTask();
         }
 
-        if (oldProbeFrequency != probeFrequency) {
-            rescheduleProbeTask(true);
-        }
-    }
-
-    private void rescheduleProbeTask(boolean deelay) {
-        synchronized (this) {
-            if (probeTask != null) {
-                probeTask.cancel(false);
-            }
-            probeTask = probeExecutor.scheduleAtFixedRate(
-                    this::triggerProbeAllDevices,
-                    deelay ? probeFrequency : 0,
-                    probeFrequency,
-                    TimeUnit.SECONDS);
+        if (oldStatsPollFrequency != statsPollInterval) {
+            statsPoller.reschedule(statsPollInterval);
         }
     }
 
     @Deactivate
     public void deactivate() {
-        // Shutdown stats polling tasks.
-        statsPollingTasks.keySet().forEach(this::cancelStatsPolling);
-        statsPollingTasks.clear();
-        statsExecutor.shutdownNow();
-        try {
-            statsExecutor.awaitTermination(5, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-            log.warn("statsExecutor not terminated properly");
-        }
-        statsExecutor = null;
+        // Shutdown stats poller.
+        statsPoller.deactivate();
+        statsPoller = null;
         // Shutdown probe executor.
         probeTask.cancel(true);
         probeTask = null;
@@ -311,87 +262,125 @@
             log.warn("probeExecutor not terminated properly");
         }
         probeExecutor = null;
-        // Shutdown connection executor.
-        connectionExecutor.shutdownNow();
+        // Shutdown main and task executor.
+        taskExecutor.cancel();
+        taskExecutor = null;
+        mainExecutor.shutdownNow();
         try {
-            connectionExecutor.awaitTermination(5, TimeUnit.SECONDS);
+            mainExecutor.awaitTermination(5, TimeUnit.SECONDS);
         } catch (InterruptedException e) {
             log.warn("connectionExecutor not terminated properly");
         }
-        connectionExecutor = null;
+        mainExecutor = null;
         // Remove all device agent listeners
         handshakersWithListeners.values().forEach(h -> h.removeDeviceAgentListener(id()));
         handshakersWithListeners.clear();
         // Other cleanup.
+        lastProbedAvailability.clear();
         componentConfigService.unregisterProperties(getClass(), false);
         cfgService.removeListener(cfgListener);
-        deviceService.removeListener(deviceListener);
         pipeconfWatchdogService.removeListener(pipeconfWatchdogListener);
         providerRegistry.unregister(this);
         providerService = null;
-        cfgService.unregisterConfigFactory(factory);
         gnmiDeviceStateSubscriber.deactivate();
         gnmiDeviceStateSubscriber = null;
         log.info("Stopped");
     }
 
-
     @Override
     public void triggerProbe(DeviceId deviceId) {
-        connectionExecutor.execute(withDeviceLock(
-                () -> doDeviceProbe(deviceId), deviceId));
+        checkNotNull(deviceId);
+        submitTask(deviceId, TaskType.PROBE_AVAILABILITY);
     }
 
     @Override
     public void roleChanged(DeviceId deviceId, MastershipRole newRole) {
-        log.info("Notifying role {} to device {}", newRole, deviceId);
-        requestedRoles.put(deviceId, newRole);
-        connectionExecutor.execute(() -> doRoleChanged(deviceId, newRole));
-    }
 
-    private void doRoleChanged(DeviceId deviceId, MastershipRole newRole) {
+        final MastershipInfo mastershipInfo = mastershipService.getMastershipFor(deviceId);
+        final NodeId localNodeId = clusterService.getLocalNode().id();
+
+        if (!mastershipInfo.getRole(localNodeId).equals(newRole)) {
+            log.warn("Inconsistent mastership info for {}! Requested {}, but " +
+                             "mastership service reports {}, will apply the latter...",
+                     deviceId, newRole, mastershipInfo.getRole(localNodeId));
+            newRole = mastershipInfo.getRole(localNodeId);
+        }
+
+        // Derive preference value.
+        final int preference;
+        switch (newRole) {
+            case MASTER:
+                preference = 0;
+                break;
+            case STANDBY:
+                preference = mastershipInfo.backups().indexOf(localNodeId) + 1;
+                if (preference == 0) {
+                    // Not found in list.
+                    log.error("Unable to derive mastership preference for {}, " +
+                                      "requested role {} but local node ID was " +
+                                      "not found among list of backup nodes " +
+                                      "reported by mastership service");
+                    return;
+                }
+                break;
+            case NONE:
+                // No preference for NONE, apply as is.
+                log.info("Notifying role {} to {}", newRole, deviceId);
+                roleChanged(deviceId, newRole);
+                return;
+            default:
+                log.error("Unrecognized mastership role {}", newRole);
+                return;
+        }
+
+        log.info("Notifying role {} (preference {}) for term {} to {}",
+                 newRole, preference, mastershipInfo.term(), deviceId);
+
         final DeviceHandshaker handshaker = getBehaviour(
                 deviceId, DeviceHandshaker.class);
         if (handshaker == null) {
-            log.error("Null handshaker. Unable to notify new role {} to {}",
+            log.error("Null handshaker. Unable to notify role {} to {}",
                       newRole, deviceId);
             return;
         }
-        handshaker.roleChanged(newRole);
+
+        try {
+            handshaker.roleChanged(preference, mastershipInfo.term());
+        } catch (UnsupportedOperationException e) {
+            // Preference-based method not supported.
+            handshaker.roleChanged(newRole);
+        }
     }
 
     @Override
     public boolean isReachable(DeviceId deviceId) {
-        log.debug("Testing reachability for device {}", deviceId);
         final DeviceHandshaker handshaker = getBehaviour(
                 deviceId, DeviceHandshaker.class);
         if (handshaker == null) {
             return false;
         }
-        return getFutureWithDeadline(
-                handshaker.isReachable(), "checking reachability",
-                deviceId, false, opTimeoutShort);
+        return handshaker.isReachable();
     }
 
-    private boolean isConnected(DeviceId deviceId) {
-        log.debug("Testing connection to device {}", deviceId);
+    @Override
+    public boolean isAvailable(DeviceId deviceId) {
         final DeviceHandshaker handshaker = getBehaviour(
                 deviceId, DeviceHandshaker.class);
         if (handshaker == null) {
             return false;
         }
-        return handshaker.isConnected();
+        try {
+            // Try without probing the device...
+            return handshaker.isAvailable();
+        } catch (UnsupportedOperationException e) {
+            // Driver does not support that.
+            return probeAvailability(handshaker);
+        }
     }
 
     @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);
@@ -409,9 +398,450 @@
 
     @Override
     public void triggerDisconnect(DeviceId deviceId) {
-        log.debug("Triggering disconnection of device {}", deviceId);
-        connectionExecutor.execute(withDeviceLock(
-                () -> doDisconnectDevice(deviceId), deviceId));
+        checkNotNull(deviceId);
+        log.info("Triggering disconnection of device {}", deviceId);
+        submitTask(deviceId, TaskType.CONNECTION_TEARDOWN);
+    }
+
+    /**
+     * Listener for configuration events.
+     */
+    private class InternalNetworkConfigListener implements NetworkConfigListener {
+        @Override
+        public void event(NetworkConfigEvent event) {
+            DeviceId deviceId = (DeviceId) event.subject();
+            switch (event.type()) {
+                case CONFIG_ADDED:
+                    if (configIsComplete(deviceId)) {
+                        submitTask(deviceId, TaskType.CONNECTION_SETUP);
+                    }
+                    break;
+                case CONFIG_UPDATED:
+                    if (configIsComplete(deviceId) && mgmtAddrUpdated(event)) {
+                        submitTask(deviceId, TaskType.CONNECTION_UPDATE);
+                    }
+                    break;
+                case CONFIG_REMOVED:
+                    if (event.configClass().equals(BasicDeviceConfig.class)) {
+                        submitTask(deviceId, TaskType.CONNECTION_TEARDOWN);
+                    }
+                    break;
+                default:
+                    // Ignore
+                    break;
+            }
+        }
+
+        private boolean mgmtAddrUpdated(NetworkConfigEvent event) {
+            if (!event.prevConfig().isPresent() || !event.config().isPresent()) {
+                return false;
+            }
+            final BasicDeviceConfig prev = (BasicDeviceConfig) event.prevConfig().get();
+            final BasicDeviceConfig current = (BasicDeviceConfig) event.config().get();
+            return !Objects.equals(prev.managementAddress(), current.managementAddress());
+        }
+
+        @Override
+        public boolean isRelevant(NetworkConfigEvent event) {
+            return event.configClass().equals(BasicDeviceConfig.class) &&
+                    (event.subject() instanceof DeviceId) &&
+                    myScheme((DeviceId) event.subject());
+        }
+    }
+
+    /**
+     * Listener for device agent events.
+     */
+    private class InternalDeviceAgentListener implements DeviceAgentListener {
+        @Override
+        public void event(DeviceAgentEvent event) {
+            DeviceId deviceId = event.subject();
+            switch (event.type()) {
+                case CHANNEL_OPEN:
+                    submitTask(deviceId, TaskType.CHANNEL_OPEN);
+                    break;
+                case CHANNEL_CLOSED:
+                case CHANNEL_ERROR:
+                    submitTask(deviceId, TaskType.CHANNEL_CLOSED);
+                    break;
+                case ROLE_MASTER:
+                    submitTask(deviceId, TaskType.ROLE_MASTER);
+                    break;
+                case ROLE_STANDBY:
+                    submitTask(deviceId, TaskType.ROLE_STANDBY);
+                    break;
+                case ROLE_NONE:
+                    submitTask(deviceId, TaskType.ROLE_NONE);
+                    break;
+                case NOT_MASTER:
+                    submitTask(deviceId, TaskType.NOT_MASTER);
+                    break;
+                default:
+                    log.warn("Unrecognized device agent event {}", event.type());
+            }
+        }
+    }
+
+    /**
+     * Pipeline event listener.
+     */
+    private class InternalPipeconfWatchdogListener implements PiPipeconfWatchdogListener {
+        @Override
+        public void event(PiPipeconfWatchdogEvent event) {
+            final DeviceId deviceId = event.subject();
+            switch (event.type()) {
+                case PIPELINE_READY:
+                    submitTask(deviceId, TaskType.PIPELINE_READY);
+                    break;
+                case PIPELINE_UNKNOWN:
+                    submitTask(deviceId, TaskType.PIPELINE_NOT_READY);
+                    break;
+                default:
+                    break;
+            }
+        }
+
+        @Override
+        public boolean isRelevant(PiPipeconfWatchdogEvent event) {
+            return myScheme(event.subject());
+        }
+    }
+
+    private void startOrRescheduleProbeTask() {
+        synchronized (this) {
+            if (probeTask != null) {
+                probeTask.cancel(false);
+            }
+            probeTask = probeExecutor.scheduleAtFixedRate(
+                    this::submitProbeTasks,
+                    0,
+                    probeInterval,
+                    TimeUnit.SECONDS);
+        }
+    }
+
+    private void submitProbeTasks() {
+        // Async trigger a task for all devices in the cfg.
+        log.debug("Starting probing for all devices");
+        cfgService.getSubjects(DeviceId.class).stream()
+                .filter(GeneralDeviceProvider::myScheme)
+                .forEach(this::submitProbeTask);
+    }
+
+    private void submitProbeTask(DeviceId deviceId) {
+        final DeviceHandshaker handshaker = handshakersWithListeners.get(deviceId);
+
+        if (handshaker == null) {
+            if (configIsComplete(deviceId)) {
+                // Device in config but we have not initiated a connection.
+                // Perhaps we missed the config event?
+                submitTask(deviceId, TaskType.CONNECTION_SETUP);
+            }
+            return;
+        }
+
+        if (!handshaker.isConnected()) {
+            // Device is in the core, but driver reports there is NOT a
+            // connection to it. Perhaps the netcfg changed and we didn't
+            // pick the event?
+            log.warn("Re-establishing lost connection to {}", deviceId);
+            submitTask(deviceId, TaskType.CONNECTION_TEARDOWN);
+            submitTask(deviceId, TaskType.CONNECTION_SETUP);
+            return;
+        }
+
+        // On probing offline devices, while we expect them to signal
+        // availability via CHANNEL_OPEN or similar events, periodic probing
+        // might be needed to stimulate some channel activity. We might consider
+        // requiring active probing of closed channels in the protocol layer.
+
+        final Long lastProbe = lastProbedAvailability.get(deviceId);
+        if (lastProbe != null &&
+                (currentTimeMillis() - lastProbe) < (probeInterval * 1000 / 3)) {
+            // This avoids overload of probe tasks which might involve sending
+            // messages over the network. We require a minimum interval of 1/3
+            // of the configured probeInterval between consecutive probe tasks.
+            if (log.isDebugEnabled()) {
+                log.debug("Dropping probe task for {} as it happened recently",
+                          deviceId);
+            }
+            return;
+        }
+
+        submitTask(deviceId, TaskType.PROBE_AVAILABILITY);
+    }
+
+    /**
+     * Type of tasks performed by this provider.
+     */
+    enum TaskType {
+        CONNECTION_SETUP,
+        CONNECTION_UPDATE,
+        CONNECTION_TEARDOWN,
+        PIPELINE_READY,
+        CHANNEL_OPEN,
+        CHANNEL_CLOSED,
+        PIPELINE_NOT_READY,
+        PROBE_AVAILABILITY,
+        ROLE_MASTER,
+        ROLE_NONE,
+        ROLE_STANDBY,
+        NOT_MASTER,
+    }
+
+    private void submitTask(DeviceId deviceId, TaskType taskType) {
+        taskExecutor.submit(deviceId, taskType, taskRunnable(deviceId, taskType));
+    }
+
+    private Runnable taskRunnable(DeviceId deviceId, TaskType taskType) {
+        switch (taskType) {
+            case CONNECTION_SETUP:
+                return () -> handleConnectionSetup(deviceId);
+            case CONNECTION_UPDATE:
+                return () -> handleConnectionUpdate(deviceId);
+            case CONNECTION_TEARDOWN:
+                return () -> handleConnectionTeardown(deviceId);
+            case CHANNEL_OPEN:
+                return () -> handleProbeAvailability(deviceId);
+            case CHANNEL_CLOSED:
+                return () -> markOfflineIfNeeded(deviceId);
+            case PIPELINE_NOT_READY:
+                return () -> markOfflineIfNeeded(deviceId);
+            case PIPELINE_READY:
+                return () -> handleProbeAvailability(deviceId);
+            case PROBE_AVAILABILITY:
+                return () -> handleProbeAvailability(deviceId);
+            case ROLE_MASTER:
+                return () -> handleMastershipResponse(deviceId, MastershipRole.MASTER);
+            case ROLE_STANDBY:
+                return () -> handleMastershipResponse(deviceId, MastershipRole.STANDBY);
+            case ROLE_NONE:
+                return () -> handleMastershipResponse(deviceId, MastershipRole.NONE);
+            case NOT_MASTER:
+                return () -> handleNotMaster(deviceId);
+            default:
+                throw new IllegalArgumentException("Unrecognized task type " + taskType);
+        }
+    }
+
+    private void handleConnectionSetup(DeviceId deviceId) {
+        assertConfig(deviceId);
+        // Bind pipeconf (if any and if device is capable).
+        bindPipeconfIfRequired(deviceId);
+        // Get handshaker.
+        final DeviceHandshaker handshaker = handshakerOrFail(deviceId);
+        if (handshaker.isConnected() || handshakersWithListeners.containsKey(deviceId)) {
+            throw new DeviceTaskException("connection already exists");
+        }
+        // Add device agent listener.
+        handshaker.addDeviceAgentListener(id(), deviceAgentListener);
+        handshakersWithListeners.put(deviceId, handshaker);
+        // Start connection via handshaker.
+        final Boolean connectSuccess = getFutureWithDeadline(
+                handshaker.connect(), "initiating connection",
+                deviceId, false, opTimeoutShort);
+        if (!connectSuccess) {
+            // Failed! Remove listeners.
+            handshaker.removeDeviceAgentListener(id());
+            handshakersWithListeners.remove(deviceId);
+            throw new DeviceTaskException("connection failed");
+        }
+        createOrUpdateDevice(deviceId, false);
+        final List<PortDescription> ports = getPortDetails(deviceId);
+        providerService.updatePorts(deviceId, ports);
+        // From here we expect a CHANNEL_OPEN event to update availability.
+    }
+
+    private void handleConnectionUpdate(DeviceId deviceId) {
+        assertConfig(deviceId);
+        final DeviceHandshaker handshaker = handshakerOrFail(deviceId);
+        if (!handshaker.isConnected()) {
+            // If driver reports that a connection still exists, perhaps the
+            // part of the netcfg that changed does not affect the connection.
+            // Otherwise, remove any previous connection state from the old
+            // netcfg and create a new one.
+            log.warn("Detected change of connection endpoints for {}, will " +
+                             "tear down existing connection and set up a new one...",
+                     deviceId);
+            handleConnectionTeardown(deviceId);
+            handleConnectionSetup(deviceId);
+        }
+    }
+
+    private void createOrUpdateDevice(DeviceId deviceId, boolean available) {
+        if (deviceService.getDevice(deviceId) != null
+                && deviceService.isAvailable(deviceId) == available) {
+            // Other nodes might have advertised this device before us.
+            return;
+        }
+        assertConfig(deviceId);
+        providerService.deviceConnected(deviceId, getDeviceDescription(
+                deviceId, available));
+    }
+
+    private boolean probeAvailability(DeviceHandshaker handshaker) {
+        lastProbedAvailability.put(handshaker.data().deviceId(), currentTimeMillis());
+        return getFutureWithDeadline(
+                handshaker.probeAvailability(), "probing availability",
+                handshaker.data().deviceId(), false, opTimeoutShort);
+    }
+
+    private boolean probeReachability(DeviceHandshaker handshaker) {
+        lastProbedAvailability.put(handshaker.data().deviceId(), currentTimeMillis());
+        return getFutureWithDeadline(
+                handshaker.probeReachability(), "probing reachability",
+                handshaker.data().deviceId(), false, opTimeoutShort);
+    }
+
+    private void markOfflineIfNeeded(DeviceId deviceId) {
+        assertDeviceRegistered(deviceId);
+        if (deviceService.isAvailable(deviceId)) {
+            providerService.deviceDisconnected(deviceId);
+        }
+    }
+
+    private void handleProbeAvailability(DeviceId deviceId) {
+        assertDeviceRegistered(deviceId);
+
+        // Make device has a valid mastership role.
+        final DeviceHandshaker handshaker = handshakerOrFail(deviceId);
+        final MastershipRole deviceRole = handshaker.getRole();
+        final MastershipRole expectedRole = mastershipService.getLocalRole(deviceId);
+        if (expectedRole == MastershipRole.NONE || expectedRole != deviceRole) {
+            // Device does NOT have a valid role...
+            if (!handshaker.isReachable() && !probeReachability(handshaker)) {
+                // ...but is not reachable. There isn't much we can do.
+                markOfflineIfNeeded(deviceId);
+                return;
+            }
+            // ...and is reachable, re-assert role.
+            roleChanged(deviceId, expectedRole == MastershipRole.NONE
+                    ? mastershipService.requestRoleForSync(deviceId)
+                    : expectedRole);
+            try {
+                // Wait for role to be notified and reachability state to be
+                // updated. This should be roughly equivalent to one RTT.
+                Thread.sleep(500);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                return;
+            }
+        }
+
+        // Check and update availability.
+        if (probeAvailability(handshakerOrFail(deviceId))) {
+            // Device ready to do its job.
+            createOrUpdateDevice(deviceId, true);
+        } else {
+            markOfflineIfNeeded(deviceId);
+            if (handshaker.isReachable() && isPipelineProgrammable(deviceId)) {
+                // If reachable, but not available, and pipeline programmable, there
+                // is a high chance it's because the pipeline is not READY
+                // (independently from what the pipeconf watchdog reports, as the
+                // status there might be outdated). Encourage pipeconf watchdog to
+                // perform a pipeline probe ASAP.
+                pipeconfWatchdogService.triggerProbe(deviceId);
+            }
+        }
+    }
+
+    private void handleMastershipResponse(DeviceId deviceId, MastershipRole response) {
+        assertDeviceRegistered(deviceId);
+        log.debug("Device {} asserted role {}", deviceId, response);
+        providerService.receivedRoleReply(deviceId, response);
+    }
+
+    private void handleNotMaster(DeviceId deviceId) {
+        assertDeviceRegistered(deviceId);
+        if (mastershipService.isLocalMaster(deviceId)) {
+            log.warn("Device {} notified that this node is not master, " +
+                             "relinquishing mastership...", deviceId);
+            mastershipService.relinquishMastership(deviceId);
+        }
+    }
+
+    private void assertDeviceRegistered(DeviceId deviceId) {
+        if (deviceService.getDevice(deviceId) == null) {
+            throw new DeviceTaskException("device not registered in the core");
+        }
+    }
+
+    private void handleConnectionTeardown(DeviceId deviceId) {
+        if (deviceService.getDevice(deviceId) != null
+                && deviceService.isAvailable(deviceId)) {
+            providerService.deviceDisconnected(deviceId);
+        }
+        final DeviceHandshaker handshaker = handshakerOrFail(deviceId);
+        handshaker.removeDeviceAgentListener(id());
+        handshakersWithListeners.remove(deviceId);
+        handshaker.disconnect();
+        lastProbedAvailability.remove(deviceId);
+    }
+
+    private void bindPipeconfIfRequired(DeviceId deviceId) {
+        if (pipeconfService.getPipeconf(deviceId).isPresent()
+                || !isPipelineProgrammable(deviceId)) {
+            // Nothing to do.
+            // Device has already a pipeconf or is not programmable.
+            return;
+        }
+        // Get pipeconf from netcfg or driver (default one).
+        final PiPipelineProgrammable pipelineProg = getBehaviour(
+                deviceId, PiPipelineProgrammable.class);
+        final PiPipeconfId pipeconfId = getPipeconfId(deviceId, pipelineProg);
+        if (pipeconfId == null) {
+            throw new DeviceTaskException("unable to find pipeconf");
+        }
+        // Store binding in pipeconf service.
+        pipeconfService.bindToDevice(pipeconfId, deviceId);
+    }
+
+    private PiPipeconfId getPipeconfId(DeviceId deviceId, PiPipelineProgrammable pipelineProg) {
+        // Places to look for a pipeconf ID (in priority order)):
+        // 1) netcfg
+        // 2) device driver (default one)
+        final PiPipeconfId pipeconfId = getPipeconfFromCfg(deviceId);
+        if (pipeconfId != null && !pipeconfId.id().isEmpty()) {
+            return pipeconfId;
+        }
+        if (pipelineProg != null
+                && pipelineProg.getDefaultPipeconf().isPresent()) {
+            final PiPipeconf defaultPipeconf = pipelineProg.getDefaultPipeconf().get();
+            log.info("Using default pipeconf {} for {}", defaultPipeconf.id(), deviceId);
+            return defaultPipeconf.id();
+        }
+        return null;
+    }
+
+    private PiPipeconfId getPipeconfFromCfg(DeviceId deviceId) {
+        BasicDeviceConfig config = cfgService.getConfig(
+                deviceId, BasicDeviceConfig.class);
+        if (config == null) {
+            return null;
+        }
+        return config.pipeconf() != null
+                ? new PiPipeconfId(config.pipeconf()) : null;
+    }
+
+    private DeviceHandshaker handshakerOrFail(DeviceId deviceId) {
+        final DeviceHandshaker handshaker = getBehaviour(
+                deviceId, DeviceHandshaker.class);
+        if (handshaker == null) {
+            throw new DeviceTaskException("missing handshaker behavior");
+        }
+        return handshaker;
+    }
+
+    private boolean configIsComplete(DeviceId deviceId) {
+        final BasicDeviceConfig basicDeviceCfg = cfgService.getConfig(
+                deviceId, BasicDeviceConfig.class);
+        return basicDeviceCfg != null && !isNullOrEmpty(basicDeviceCfg.driver());
+    }
+
+    private void assertConfig(DeviceId deviceId) {
+        if (!configIsComplete(deviceId)) {
+            throw new DeviceTaskException("configuration is not complete");
+        }
     }
 
     private Driver getDriver(DeviceId deviceId) {
@@ -425,8 +855,6 @@
     }
 
     private <T extends Behaviour> T getBehaviour(DeviceId deviceId, Class<T> type) {
-        // Get handshaker.
-
         Driver driver = getDriver(deviceId);
         if (driver == null) {
             return null;
@@ -435,83 +863,16 @@
             return null;
         }
         final DriverData data = new DefaultDriverData(driver, deviceId);
-        // Storing deviceKeyId and all other config values as data in the driver
-        // with protocol_<info> name as the key. e.g protocol_ip.
-        final GeneralProviderDeviceConfig providerConfig = cfgService.getConfig(
-                deviceId, GeneralProviderDeviceConfig.class);
-        if (providerConfig != null) {
-            providerConfig.protocolsInfo().forEach((protocol, info) -> {
-                info.configValues().forEach(
-                        (k, v) -> data.set(protocol + "_" + k, v));
-                data.set(protocol + "_key", info.deviceKeyId());
-            });
-        }
         final DefaultDriverHandler handler = new DefaultDriverHandler(data);
         return driver.createBehaviour(handler, type);
     }
 
-    private void doConnectDevice(DeviceId deviceId) {
-        log.debug("Initiating connection to device {}...", deviceId);
-        // Retrieve config
-        if (configIsMissing(deviceId)) {
-            return;
+    private boolean hasBehaviour(DeviceId deviceId, Class<? extends Behaviour> type) {
+        Driver driver = getDriver(deviceId);
+        if (driver == null) {
+            return false;
         }
-        // Bind pipeconf (if any and if device is capable).
-        if (!bindPipeconfIfRequired(deviceId)) {
-            // We already logged the error.
-            return;
-        }
-        // Get handshaker.
-        final DeviceHandshaker handshaker = getBehaviour(
-                deviceId, DeviceHandshaker.class);
-        if (handshaker == null) {
-            log.error("Missing handshaker behavior for {}, aborting connection",
-                      deviceId);
-            return;
-        }
-        // Add device agent listener.
-        handshaker.addDeviceAgentListener(id(), deviceAgentListener);
-        handshakersWithListeners.put(deviceId, handshaker);
-        // Start connection via handshaker.
-        final Boolean connectSuccess = getFutureWithDeadline(
-                handshaker.connect(), "initiating connection",
-                deviceId, false, opTimeoutShort);
-        if (!connectSuccess) {
-            log.warn("Unable to connect to {}", deviceId);
-        }
-    }
-
-    private void triggerAdvertiseDevice(DeviceId deviceId) {
-        connectionExecutor.execute(withDeviceLock(
-                () -> doAdvertiseDevice(deviceId), deviceId));
-    }
-
-    private void doAdvertiseDevice(DeviceId deviceId) {
-        // Retrieve config
-        if (configIsMissing(deviceId)) {
-            return;
-        }
-        // Obtain device and port description.
-        final boolean isPipelineReady = isPipelineReady(deviceId);
-        final DeviceDescription description = getDeviceDescription(
-                deviceId, isPipelineReady);
-        final List<PortDescription> ports = getPortDetails(deviceId);
-        // Advertise to core.
-        if (deviceService.getDevice(deviceId) == null ||
-                (description.isDefaultAvailable() &&
-                        !deviceService.isAvailable(deviceId))) {
-            if (!isPipelineReady) {
-                log.info("Advertising device {} to core with available={} as " +
-                                 "device pipeline is not ready yet",
-                         deviceId, description.isDefaultAvailable());
-            }
-            providerService.deviceConnected(deviceId, description);
-        }
-        providerService.updatePorts(deviceId, ports);
-        // If pipeline is not ready, encourage watchdog to perform probe ASAP.
-        if (!isPipelineReady) {
-            pipeconfWatchdogService.triggerProbe(deviceId);
-        }
+        return driver.hasBehaviour(type);
     }
 
     private DeviceDescription getDeviceDescription(
@@ -548,18 +909,6 @@
                 deviceId, DeviceHandshaker.class);
         final Driver driver = handshaker != null
                 ? handshaker.handler().driver() : null;
-        final GeneralProviderDeviceConfig cfg = cfgService.getConfig(
-                deviceId, GeneralProviderDeviceConfig.class);
-        final DefaultAnnotations.Builder annBuilder = DefaultAnnotations.builder();
-        // If device is pipeline programmable, let this provider decide when the
-        // device can be marked online.
-        annBuilder.set(AnnotationKeys.PROVIDER_MARK_ONLINE,
-                       String.valueOf(isPipelineProgrammable(deviceId)));
-        if (cfg != null) {
-            StringJoiner protoStringBuilder = new StringJoiner(", ");
-            cfg.protocolsInfo().keySet().forEach(protoStringBuilder::add);
-            annBuilder.set(AnnotationKeys.PROTOCOL, protoStringBuilder.toString());
-        }
         return new DefaultDeviceDescription(
                 deviceId.uri(),
                 Device.Type.SWITCH,
@@ -569,351 +918,15 @@
                 UNKNOWN,
                 new ChassisId(),
                 defaultAvailable,
-                annBuilder.build());
+                DefaultAnnotations.EMPTY);
     }
 
-    private void triggerMarkAvailable(DeviceId deviceId) {
-        connectionExecutor.execute(withDeviceLock(
-                () -> doMarkAvailable(deviceId), deviceId));
-    }
-
-    private void doMarkAvailable(DeviceId deviceId) {
-        if (deviceService.isAvailable(deviceId)) {
-            return;
-        }
-        final DeviceDescription descr = getDeviceDescription(deviceId, true);
-        // It has been observed that devices that were marked offline (e.g.
-        // after device disconnection) might end up with no master. Here we
-        // trigger a new master election (if device has no master).
-        mastershipService.requestRoleForSync(deviceId);
-        providerService.deviceConnected(deviceId, descr);
-    }
-
-    private boolean bindPipeconfIfRequired(DeviceId deviceId) {
-        if (pipeconfService.ofDevice(deviceId).isPresent()
-                || !isPipelineProgrammable(deviceId)) {
-            // Nothing to do, all good.
-            return true;
-        }
-        // Get pipeconf from netcfg or driver (default one).
-        final PiPipelineProgrammable pipelineProg = getBehaviour(
-                deviceId, PiPipelineProgrammable.class);
-        final PiPipeconfId pipeconfId = getPipeconfId(deviceId, pipelineProg);
-        if (pipeconfId == null) {
-            return false;
-        }
-        // Store binding in pipeconf service.
-        pipeconfService.bindToDevice(pipeconfId, deviceId);
-        return true;
-    }
-
-    private PiPipeconfId getPipeconfId(DeviceId deviceId, PiPipelineProgrammable pipelineProg) {
-        // Places to look for a pipeconf ID (in priority order)):
-        // 1) netcfg
-        // 2) device driver (default one)
-        final PiPipeconfId pipeconfId = getPipeconfFromCfg(deviceId);
-        if (pipeconfId != null && !pipeconfId.id().isEmpty()) {
-            return pipeconfId;
-        }
-        if (pipelineProg != null
-                && pipelineProg.getDefaultPipeconf().isPresent()) {
-            final PiPipeconf defaultPipeconf = pipelineProg.getDefaultPipeconf().get();
-            log.info("Using default pipeconf {} for {}", defaultPipeconf.id(), deviceId);
-            return defaultPipeconf.id();
-        } else {
-            log.warn("Unable to associate a pipeconf to {}", deviceId);
-            return null;
-        }
-    }
-
-    private void doDisconnectDevice(DeviceId deviceId) {
-        log.debug("Initiating disconnection from {}...", deviceId);
-        final DeviceHandshaker handshaker = getBehaviour(
-                deviceId, DeviceHandshaker.class);
-        final boolean isAvailable = deviceService.isAvailable(deviceId);
-        // Signal disconnection to core (if master).
-        if (isAvailable && mastershipService.isLocalMaster(deviceId)) {
-            providerService.deviceDisconnected(deviceId);
-        }
-        // Cancel tasks.
-        cancelStatsPolling(deviceId);
-        // Disconnect device.
-        if (handshaker == null) {
-            if (isAvailable) {
-                // If not available don't bother logging. We are probably
-                // invoking this method multiple times for the same device.
-                log.warn("Missing DeviceHandshaker behavior for {}, " +
-                                 "no guarantees of complete disconnection",
-                         deviceId);
-            }
-            return;
-        }
-        handshaker.removeDeviceAgentListener(id());
-        handshakersWithListeners.remove(deviceId);
-        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.
-    private Runnable exceptionSafe(Runnable runnable) {
-        return () -> {
-            try {
-                runnable.run();
-            } catch (Exception e) {
-                log.error("Unhandled Exception", e);
-            }
-        };
-    }
-
-    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 (device != null && deviceService.isAvailable(deviceId) &&
-                device.is(PortStatisticsDiscovery.class)) {
-            Collection<PortStatistics> statistics = device.as(PortStatisticsDiscovery.class)
-                    .discoverPortStatistics();
-            //updating statistcs only if not empty
-            if (!statistics.isEmpty()) {
-                providerService.updatePortStatistics(deviceId, statistics);
-            }
-        } else {
-            log.debug("Can't update port statistics for device {}", deviceId);
-        }
-    }
-
-    private boolean notMyScheme(DeviceId deviceId) {
-        return !deviceId.uri().getScheme().equals(URI_SCHEME);
-    }
-
-    private void triggerConnect(DeviceId deviceId) {
-        connectionExecutor.execute(withDeviceLock(
-                () -> doConnectDevice(deviceId), deviceId));
+    static boolean myScheme(DeviceId deviceId) {
+        return deviceId.uri().getScheme().equals(URI_SCHEME);
     }
 
     private boolean isPipelineProgrammable(DeviceId deviceId) {
-        final GeneralProviderDeviceConfig providerConfig = cfgService.getConfig(
-                deviceId, GeneralProviderDeviceConfig.class);
-        if (providerConfig == null) {
-            return false;
-        }
-        return !Collections.disjoint(
-                ImmutableSet.copyOf(providerConfig.node().fieldNames()),
-                PIPELINE_CONFIGURABLE_PROTOCOLS);
-    }
-
-    /**
-     * Listener for configuration events.
-     */
-    private class InternalNetworkConfigListener implements NetworkConfigListener {
-
-        @Override
-        public void event(NetworkConfigEvent event) {
-            connectionExecutor.execute(() -> consumeConfigEvent(event));
-        }
-
-        @Override
-        public boolean isRelevant(NetworkConfigEvent event) {
-            return (event.configClass().equals(GeneralProviderDeviceConfig.class) ||
-                    event.configClass().equals(BasicDeviceConfig.class) ||
-                    event.configClass().equals(PiPipeconfConfig.class)) &&
-                    (event.type() == NetworkConfigEvent.Type.CONFIG_ADDED ||
-                            event.type() == NetworkConfigEvent.Type.CONFIG_UPDATED);
-        }
-
-        private void consumeConfigEvent(NetworkConfigEvent event) {
-            DeviceId deviceId = (DeviceId) event.subject();
-            //Assuming that the deviceId comes with uri 'device:'
-            if (notMyScheme(deviceId)) {
-                // not under my scheme, skipping
-                log.debug("{} is not my scheme, skipping", deviceId);
-                return;
-            }
-            final boolean configComplete = withDeviceLock(
-                    () -> isDeviceConfigComplete(event, deviceId), deviceId);
-            if (!configComplete) {
-                // Still waiting for some configuration.
-                return;
-            }
-            // Good to go.
-            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
-            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)) {
-                    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;
-        }
-    }
-
-    private boolean isPipelineReady(DeviceId deviceId) {
-        final boolean isPipelineProg = isPipelineProgrammable(deviceId);
-        final boolean isPipeconfReady = pipeconfWatchdogService
-                .getStatus(deviceId)
-                .equals(PiPipeconfWatchdogService.PipelineStatus.READY);
-        return !isPipelineProg || isPipeconfReady;
-    }
-
-    private void cleanUpConfigInfo(DeviceId deviceId) {
-        deviceConfigured.remove(deviceId);
-        driverConfigured.remove(deviceId);
-        pipelineConfigured.remove(deviceId);
-    }
-
-    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 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(this::triggerDeviceProbe);
-    }
-
-    private PiPipeconfId getPipeconfFromCfg(DeviceId deviceId) {
-        PiPipeconfConfig config = cfgService.getConfig(
-                deviceId, PiPipeconfConfig.class);
-        if (config == null) {
-            return null;
-        }
-        return config.piPipeconfId();
-    }
-
-    private void triggerDeviceProbe(DeviceId deviceId) {
-        connectionExecutor.execute(withDeviceLock(
-                () -> doDeviceProbe(deviceId), deviceId));
-    }
-
-    private void doDeviceProbe(DeviceId deviceId) {
-        log.debug("Probing device {}...", deviceId);
-        if (configIsMissing(deviceId)) {
-            return;
-        }
-        if (!isConnected(deviceId)) {
-            if (deviceService.isAvailable(deviceId)) {
-                providerService.deviceDisconnected(deviceId);
-            }
-            triggerConnect(deviceId);
-        }
-    }
-
-    private boolean configIsMissing(DeviceId deviceId) {
-        final boolean present =
-                cfgService.getConfig(deviceId, GeneralProviderDeviceConfig.class) != null
-                        && cfgService.getConfig(deviceId, BasicDeviceConfig.class) != null;
-        if (!present) {
-            log.warn("Configuration for device {} is not complete", deviceId);
-        }
-        return !present;
-    }
-
-    private void handleMastershipResponse(DeviceId deviceId, MastershipRole response) {
-        // Notify core about mastership response.
-        final MastershipRole request = requestedRoles.get(deviceId);
-        final boolean isAvailable = deviceService.isAvailable(deviceId);
-        if (request == null || !isAvailable) {
-            return;
-        }
-        log.debug("Device {} asserted role {} (requested was {})",
-                  deviceId, response, request);
-        providerService.receivedRoleReply(deviceId, request, response);
-        // FIXME: this should be based on assigned mastership, not what returned by device
-        if (response.equals(MastershipRole.MASTER)) {
-            startStatsPolling(deviceId, false);
-        } else {
-            cancelStatsPolling(deviceId);
-        }
-    }
-
-    private void handleNotMaster(DeviceId deviceId) {
-        log.warn("Device {} notified that this node is not master, " +
-                         "relinquishing mastership...", deviceId);
-        mastershipService.relinquishMastership(deviceId);
+        return hasBehaviour(deviceId, PiPipelineProgrammable.class);
     }
 
     private <U> U getFutureWithDeadline(CompletableFuture<U> future, String opDescription,
@@ -930,86 +943,4 @@
         }
         return defaultValue;
     }
-
-    /**
-     * Listener for core device events.
-     */
-    private class InternalDeviceListener implements DeviceListener {
-        @Override
-        public void event(DeviceEvent event) {
-            DeviceId deviceId = event.subject().id();
-            // 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)) {
-                startStatsPolling(deviceId, true);
-            }
-        }
-
-        @Override
-        public boolean isRelevant(DeviceEvent event) {
-            return event.type() == Type.DEVICE_ADDED &&
-                    event.subject().id().toString().startsWith(URI_SCHEME.toLowerCase());
-        }
-    }
-
-    /**
-     * Listener for device agent events.
-     */
-    private class InternalDeviceAgentListener implements DeviceAgentListener {
-
-        @Override
-        public void event(DeviceAgentEvent event) {
-            DeviceId deviceId = event.subject();
-            switch (event.type()) {
-                case CHANNEL_OPEN:
-                    triggerAdvertiseDevice(deviceId);
-                    break;
-                case CHANNEL_CLOSED:
-                case CHANNEL_ERROR:
-                    triggerDeviceProbe(deviceId);
-                    break;
-                case ROLE_MASTER:
-                    handleMastershipResponse(deviceId, MastershipRole.MASTER);
-                    break;
-                case ROLE_STANDBY:
-                    handleMastershipResponse(deviceId, MastershipRole.STANDBY);
-                    break;
-                case ROLE_NONE:
-                    handleMastershipResponse(deviceId, MastershipRole.NONE);
-                    break;
-                case NOT_MASTER:
-                    handleNotMaster(deviceId);
-                    break;
-                default:
-                    log.warn("Unrecognized device agent event {}", event.type());
-            }
-        }
-
-    }
-
-    private class InternalPipeconfWatchdogListener implements PiPipeconfWatchdogListener {
-        @Override
-        public void event(PiPipeconfWatchdogEvent event) {
-            triggerMarkAvailable(event.subject());
-        }
-
-        @Override
-        public boolean isRelevant(PiPipeconfWatchdogEvent event) {
-            return event.type().equals(PiPipeconfWatchdogEvent.Type.PIPELINE_READY);
-        }
-    }
-
-    private class InternalConfigFactory
-            extends ConfigFactory<DeviceId, GeneralProviderDeviceConfig> {
-
-        InternalConfigFactory() {
-            super(SubjectFactories.DEVICE_SUBJECT_FACTORY,
-                  GeneralProviderDeviceConfig.class, CFG_SCHEME);
-        }
-
-        @Override
-        public GeneralProviderDeviceConfig createConfig() {
-            return new GeneralProviderDeviceConfig();
-        }
-    }
 }
diff --git a/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/OsgiPropertyConstants.java b/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/OsgiPropertyConstants.java
index 84419f4..358ce5f 100644
--- a/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/OsgiPropertyConstants.java
+++ b/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/OsgiPropertyConstants.java
@@ -23,11 +23,11 @@
 
     private OsgiPropertyConstants() {}
 
-    public static final String STATS_POLL_FREQUENCY = "deviceStatsPollFrequency";
-    public static final int STATS_POLL_FREQUENCY_DEFAULT = 10;
+    public static final String STATS_POLL_INTERVAL = "deviceStatsPollInterval";
+    public static final int STATS_POLL_INTERVAL_DEFAULT = 10;
 
-    public static final String PROBE_FREQUENCY = "deviceProbeFrequency";
-    public static final int PROBE_FREQUENCY_DEFAULT = 10;
+    public static final String PROBE_INTERVAL = "deviceProbeInterval";
+    public static final int PROBE_INTERVAL_DEFAULT = 10;
 
     public static final String OP_TIMEOUT_SHORT = "deviceOperationTimeoutShort";
     public static final int OP_TIMEOUT_SHORT_DEFAULT = 10;
diff --git a/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/StatsPoller.java b/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/StatsPoller.java
new file mode 100644
index 0000000..1426946
--- /dev/null
+++ b/providers/general/device/src/main/java/org/onosproject/provider/general/device/impl/StatsPoller.java
@@ -0,0 +1,209 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.provider.general.device.impl;
+
+import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.Striped;
+import org.onosproject.mastership.MastershipEvent;
+import org.onosproject.mastership.MastershipListener;
+import org.onosproject.mastership.MastershipService;
+import org.onosproject.net.Device;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.net.device.DeviceListener;
+import org.onosproject.net.device.DeviceProviderService;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.device.PortStatistics;
+import org.onosproject.net.device.PortStatisticsDiscovery;
+import org.slf4j.Logger;
+
+import java.security.SecureRandom;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.concurrent.Executors.newScheduledThreadPool;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.onosproject.provider.general.device.impl.GeneralDeviceProvider.myScheme;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Component devoted to polling stats from devices managed by the
+ * GeneralDeviceProvider.
+ */
+public class StatsPoller {
+
+    private static final int CORE_POOL_SIZE = 5;
+
+    private final Logger log = getLogger(getClass());
+
+    private final DeviceService deviceService;
+    private final MastershipService mastershipService;
+    private final DeviceProviderService providerService;
+
+    private final InternalDeviceListener deviceListener = new InternalDeviceListener();
+    private final MastershipListener mastershipListener = new InternalMastershipListener();
+    private final Striped<Lock> deviceLocks = Striped.lock(30);
+
+    private ScheduledExecutorService statsExecutor;
+    private Map<DeviceId, ScheduledFuture<?>> statsPollingTasks;
+    private Map<DeviceId, Integer> pollFrequencies;
+    private int statsPollInterval;
+
+    StatsPoller(DeviceService deviceService, MastershipService mastershipService,
+                DeviceProviderService providerService) {
+        this.deviceService = deviceService;
+        this.mastershipService = mastershipService;
+        this.providerService = providerService;
+    }
+
+
+    void activate(int statsPollInterval) {
+        checkArgument(statsPollInterval > 0, "statsPollInterval must be greater than 0");
+        this.statsPollInterval = statsPollInterval;
+        statsExecutor = newScheduledThreadPool(CORE_POOL_SIZE, groupedThreads(
+                "onos/gdp-stats", "%d", log));
+        statsPollingTasks = Maps.newHashMap();
+        pollFrequencies = Maps.newHashMap();
+        deviceService.getDevices().forEach(d -> updatePollingTask(d.id()));
+        deviceService.addListener(deviceListener);
+        mastershipService.addListener(mastershipListener);
+        log.info("Started");
+    }
+
+    void reschedule(int statsPollInterval) {
+        checkArgument(statsPollInterval > 0, "statsPollInterval must be greater than 0");
+        this.statsPollInterval = statsPollInterval;
+        statsPollingTasks.keySet().forEach(this::updatePollingTask);
+    }
+
+    void deactivate() {
+        deviceService.removeListener(deviceListener);
+        mastershipService.removeListener(mastershipListener);
+
+        statsPollingTasks.values().forEach(t -> t.cancel(false));
+        statsPollingTasks.clear();
+        pollFrequencies.clear();
+        statsPollingTasks = null;
+        pollFrequencies = null;
+
+        statsExecutor.shutdownNow();
+        try {
+            statsExecutor.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            log.warn("statsExecutor not terminated properly");
+        }
+        statsExecutor = null;
+
+        log.info("Stopped");
+    }
+
+
+    private void updatePollingTask(DeviceId deviceId) {
+        deviceLocks.get(deviceId).lock();
+        try {
+            final ScheduledFuture<?> existingTask = statsPollingTasks.get(deviceId);
+            final boolean shouldHaveTask = deviceService.getDevice(deviceId) != null
+                    && deviceService.isAvailable(deviceId)
+                    && mastershipService.isLocalMaster(deviceId)
+                    && deviceService.getDevice(deviceId).is(PortStatisticsDiscovery.class);
+            final boolean pollFrequencyChanged = !Objects.equals(
+                    pollFrequencies.get(deviceId), statsPollInterval);
+
+            if (existingTask != null && (!shouldHaveTask || pollFrequencyChanged)) {
+                existingTask.cancel(false);
+                statsPollingTasks.remove(deviceId);
+                pollFrequencies.remove(deviceId);
+                log.info("Cancelled polling task for {}", deviceId);
+            }
+
+            if (shouldHaveTask) {
+                final int delay = new SecureRandom().nextInt(statsPollInterval);
+                statsPollingTasks.put(deviceId, statsExecutor.scheduleAtFixedRate(
+                        exceptionSafe(() -> updatePortStatistics(deviceId)),
+                        delay, statsPollInterval, TimeUnit.SECONDS));
+                pollFrequencies.put(deviceId, statsPollInterval);
+                log.info("Started polling task for {} with interval {} seconds",
+                         deviceId, statsPollInterval);
+            }
+        } finally {
+            deviceLocks.get(deviceId).unlock();
+        }
+    }
+
+    private void updatePortStatistics(DeviceId deviceId) {
+        final Device device = deviceService.getDevice(deviceId);
+        final Collection<PortStatistics> statistics = device.as(
+                PortStatisticsDiscovery.class).discoverPortStatistics();
+        if (!statistics.isEmpty()) {
+            providerService.updatePortStatistics(deviceId, statistics);
+        }
+    }
+
+    private Runnable exceptionSafe(Runnable runnable) {
+        return () -> {
+            try {
+                runnable.run();
+            } catch (Exception e) {
+                log.error("Unhandled exception in stats poller", e);
+            }
+        };
+    }
+
+    private class InternalMastershipListener implements MastershipListener {
+
+        @Override
+        public void event(MastershipEvent event) {
+            updatePollingTask(event.subject());
+        }
+
+        @Override
+        public boolean isRelevant(MastershipEvent event) {
+            return event.type() == MastershipEvent.Type.MASTER_CHANGED
+                    && myScheme(event.subject());
+        }
+    }
+
+    /**
+     * Listener for core device events.
+     */
+    private class InternalDeviceListener implements DeviceListener {
+        @Override
+        public void event(DeviceEvent event) {
+            updatePollingTask(event.subject().id());
+        }
+
+        @Override
+        public boolean isRelevant(DeviceEvent event) {
+            switch (event.type()) {
+                case DEVICE_ADDED:
+                case DEVICE_UPDATED:
+                case DEVICE_AVAILABILITY_CHANGED:
+                case DEVICE_REMOVED:
+                case DEVICE_SUSPENDED:
+                    return myScheme(event.subject().id());
+                default:
+                    return false;
+            }
+        }
+    }
+}
diff --git a/tools/dev/mininet/bmv2.py b/tools/dev/mininet/bmv2.py
index 56c7dcc..3dc42fd 100644
--- a/tools/dev/mininet/bmv2.py
+++ b/tools/dev/mininet/bmv2.py
@@ -35,6 +35,7 @@
 SWITCH_START_TIMEOUT = 5  # seconds
 BMV2_LOG_LINES = 5
 BMV2_DEFAULT_DEVICE_ID = 1
+DEFAULT_PIPECONF = "org.onosproject.pipelines.basic"
 
 # Stratum paths relative to stratum repo root
 STRATUM_BMV2 = 'stratum_bmv2'
@@ -119,9 +120,10 @@
 
     def __init__(self, name, json=None, debugger=False, loglevel="warn",
                  elogger=False, grpcport=None, cpuport=255, notifications=False,
-                 thriftport=None, netcfg=True, dryrun=False, pipeconf="",
-                 pktdump=False, valgrind=False, gnmi=False,
-                 portcfg=True, onosdevid=None, stratum=False, **kwargs):
+                 thriftport=None, netcfg=True, dryrun=False,
+                 pipeconf=DEFAULT_PIPECONF, pktdump=False, valgrind=False,
+                 gnmi=False, portcfg=True, onosdevid=None, stratum=False,
+                 **kwargs):
         Switch.__init__(self, name, **kwargs)
         self.grpcPort = grpcport
         self.thriftPort = thriftport
@@ -173,7 +175,10 @@
     def getDeviceConfig(self, srcIP):
 
         basicCfg = {
-            "driver": "bmv2"
+            "managementAddress": "grpc://%s:%d?device_id=%d" % (
+                srcIP, self.grpcPort, BMV2_DEFAULT_DEVICE_ID),
+            "driver": "bmv2",
+            "pipeconf": self.pipeconfId
         }
 
         if self.longitude and self.latitude:
@@ -181,30 +186,9 @@
             basicCfg["latitude"] = self.latitude
 
         cfgData = {
-            "generalprovider": {
-                "p4runtime": {
-                    "ip": srcIP,
-                    "port": self.grpcPort,
-                    "deviceId": BMV2_DEFAULT_DEVICE_ID,
-                    "deviceKeyId": "p4runtime:%s" % self.onosDeviceId
-                },
-                # "bmv2-thrift": {
-                #     "ip": srcIP,
-                #     "port": self.thriftPort
-                # }
-            },
-            "piPipeconf": {
-                "piPipeconfId": self.pipeconfId
-            },
             "basic": basicCfg
         }
 
-        if self.withGnmi:
-            cfgData["generalprovider"]["gnmi"] = {
-                "ip": srcIP,
-                "port": self.grpcPort
-            }
-
         if self.injectPorts:
             portData = {}
             portId = 1
@@ -327,7 +311,8 @@
             '-persistent_config_dir=' + config_dir,
             '-initial_pipeline=' + stratumRoot + STRATUM_INIT_PIPELINE,
             '-cpu_port=%s' % self.cpuPort,
-            '-external_hercules_urls=0.0.0.0:%d' % self.grpcPort
+            '-external_hercules_urls=0.0.0.0:%d' % self.grpcPort,
+            '-max_num_controllers_per_node=10'
         ]
         for port, intf in self.intfs.items():
             if not intf.IP():