Added SDN-IP Leader election mechanism based on the Global Lock service.

NOTE: This is a short-term solution until the Leadership Service is
implemented.

Change-Id: Id1f21c06e67e08efb444ebec4e45eeca360e613d

@@ -69,7 +69,7 @@ public class IntentSynchronizer {
 
 
         bgpIntentsSynchronizerExecutor = Executors.newSingleThreadExecutor(
         bgpIntentsSynchronizerExecutor = Executors.newSingleThreadExecutor(
                 new ThreadFactoryBuilder()
                 new ThreadFactoryBuilder()
-                        .setNameFormat("bgp-intents-synchronizer-%d").build());
+                .setNameFormat("sdnip-intents-synchronizer-%d").build());
     }
     }
 
 
     /**
     /**

@@ -114,7 +114,8 @@ public class Router implements RouteListener {
                 HashMultimap.<Ip4Address, RouteEntry>create());
                 HashMultimap.<Ip4Address, RouteEntry>create());
 
 
         bgpUpdatesExecutor = Executors.newSingleThreadExecutor(
         bgpUpdatesExecutor = Executors.newSingleThreadExecutor(
-                new ThreadFactoryBuilder().setNameFormat("bgp-updates-%d").build());
+                new ThreadFactoryBuilder()
+                .setNameFormat("sdnip-bgp-updates-%d").build());
     }
     }
 
 
     /**
     /**

@@ -18,6 +18,8 @@ package org.onlab.onos.sdnip;
 import static org.slf4j.LoggerFactory.getLogger;
 import static org.slf4j.LoggerFactory.getLogger;
 
 
 import java.util.Collection;
 import java.util.Collection;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Component;
@@ -33,8 +35,12 @@ import org.onlab.onos.sdnip.bgp.BgpRouteEntry;
 import org.onlab.onos.sdnip.bgp.BgpSession;
 import org.onlab.onos.sdnip.bgp.BgpSession;
 import org.onlab.onos.sdnip.bgp.BgpSessionManager;
 import org.onlab.onos.sdnip.bgp.BgpSessionManager;
 import org.onlab.onos.sdnip.config.SdnIpConfigReader;
 import org.onlab.onos.sdnip.config.SdnIpConfigReader;
+import org.onlab.onos.store.service.Lock;
+import org.onlab.onos.store.service.LockService;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
 /**
 /**
  * Component for the SDN-IP peering application.
  * Component for the SDN-IP peering application.
  */
  */
@@ -43,6 +49,9 @@ import org.slf4j.Logger;
 public class SdnIp implements SdnIpService {
 public class SdnIp implements SdnIpService {
 
 
     private static final String SDN_IP_APP = "org.onlab.onos.sdnip";
     private static final String SDN_IP_APP = "org.onlab.onos.sdnip";
+    // NOTE: Must be 5s for now
+    private static final int LEASE_DURATION_MS = 5 * 1000;
+    private static final int LEASE_EXTEND_RETRY_MAX = 3;
 
 
     private final Logger log = getLogger(getClass());
     private final Logger log = getLogger(getClass());
 
 
@@ -55,15 +64,23 @@ public class SdnIp implements SdnIpService {
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected HostService hostService;
     protected HostService hostService;
 
 
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected LockService lockService;
+
     private IntentSynchronizer intentSynchronizer;
     private IntentSynchronizer intentSynchronizer;
     private SdnIpConfigReader config;
     private SdnIpConfigReader config;
     private PeerConnectivityManager peerConnectivity;
     private PeerConnectivityManager peerConnectivity;
     private Router router;
     private Router router;
     private BgpSessionManager bgpSessionManager;
     private BgpSessionManager bgpSessionManager;
 
 
+    private ExecutorService leaderElectionExecutor;
+    private Lock leaderLock;
+    private volatile boolean isShutdown = true;
+
     @Activate
     @Activate
     protected void activate() {
     protected void activate() {
-        log.debug("SDN-IP started");
+        log.info("SDN-IP started");
+        isShutdown = false;
 
 
         ApplicationId appId = coreService.registerApplication(SDN_IP_APP);
         ApplicationId appId = coreService.registerApplication(SDN_IP_APP);
         config = new SdnIpConfigReader();
         config = new SdnIpConfigReader();
@@ -83,9 +100,20 @@ public class SdnIp implements SdnIpService {
                             interfaceService);
                             interfaceService);
         router.start();
         router.start();
 
 
+        leaderLock = lockService.create(SDN_IP_APP + "/sdnIpLeaderLock");
+        leaderElectionExecutor = Executors.newSingleThreadExecutor(
+                new ThreadFactoryBuilder()
+                .setNameFormat("sdnip-leader-election-%d").build());
+        leaderElectionExecutor.execute(new Runnable() {
+                @Override
+                public void run() {
+                    doLeaderElectionThread();
+                }
+            });
+
         // Manually set the instance as the leader to allow testing
         // Manually set the instance as the leader to allow testing
         // TODO change this when we get a leader election
         // TODO change this when we get a leader election
-        intentSynchronizer.leaderChanged(true);
+        // intentSynchronizer.leaderChanged(true);
 
 
         bgpSessionManager = new BgpSessionManager(router);
         bgpSessionManager = new BgpSessionManager(router);
         // TODO: the local BGP listen port number should be configurable
         // TODO: the local BGP listen port number should be configurable
@@ -96,11 +124,16 @@ public class SdnIp implements SdnIpService {
 
 
     @Deactivate
     @Deactivate
     protected void deactivate() {
     protected void deactivate() {
+        isShutdown = true;
+
         bgpSessionManager.stop();
         bgpSessionManager.stop();
         router.stop();
         router.stop();
         peerConnectivity.stop();
         peerConnectivity.stop();
         intentSynchronizer.stop();
         intentSynchronizer.stop();
 
 
+        // Stop the thread(s)
+        leaderElectionExecutor.shutdownNow();
+
         log.info("Stopped");
         log.info("Stopped");
     }
     }
 
 
@@ -127,4 +160,66 @@ public class SdnIp implements SdnIpService {
     static String dpidToUri(String dpid) {
     static String dpidToUri(String dpid) {
         return "of:" + dpid.replace(":", "");
         return "of:" + dpid.replace(":", "");
     }
     }
+
+    /**
+     * Performs the leader election.
+     */
+    private void doLeaderElectionThread() {
+
+        //
+        // Try to acquire the lock and keep extending it until the instance
+        // is shutdown.
+        //
+        while (!isShutdown) {
+            log.debug("SDN-IP Leader Election begin");
+
+            // Block until it becomes the leader
+            leaderLock.lock(LEASE_DURATION_MS);
+
+            // This instance is the leader
+            log.info("SDN-IP Leader Elected");
+            intentSynchronizer.leaderChanged(true);
+
+            // Keep extending the expiration until shutdown
+            int extensionFailedCountdown = LEASE_EXTEND_RETRY_MAX - 1;
+
+            //
+            // Keep periodically extending the lock expiration.
+            // If there are multiple back-to-back failures to extend (with
+            // extra sleep time between retrials), then release the lock.
+            //
+            while (!isShutdown) {
+                try {
+                    Thread.sleep(LEASE_DURATION_MS / LEASE_EXTEND_RETRY_MAX);
+                    if (leaderLock.extendExpiration(LEASE_DURATION_MS)) {
+                        log.trace("SDN-IP Leader Extended");
+                        extensionFailedCountdown = LEASE_EXTEND_RETRY_MAX;
+                    } else {
+                        log.debug("SDN-IP Leader Cannot Extend Election");
+                        if (!leaderLock.isLocked()) {
+                            log.debug("SDN-IP Leader Lock Lost");
+                            intentSynchronizer.leaderChanged(false);
+                            break;              // Try again to get the lock
+                        }
+                        extensionFailedCountdown--;
+                        if (extensionFailedCountdown <= 0) {
+                            // Failed too many times to extend.
+                            // Release the lock.
+                            log.debug("SDN-IP Leader Lock Released");
+                            intentSynchronizer.leaderChanged(false);
+                            leaderLock.unlock();
+                            break;              // Try again to get the lock
+                        }
+                    }
+                } catch (InterruptedException e) {
+                    // Thread interrupted. Time to shutdown
+                    log.debug("SDN-IP Leader Interrupted");
+                }
+            }
+        }
+
+        // If we reach here, the instance was shutdown
+        intentSynchronizer.leaderChanged(false);
+        leaderLock.unlock();
+    }
 }
 }