Added ability to measure round-trip latency and to assure message integrity.

@@ -31,6 +31,16 @@
             <groupId>org.livetribe.slp</groupId>
             <groupId>org.livetribe.slp</groupId>
             <artifactId>livetribe-slp</artifactId>
             <artifactId>livetribe-slp</artifactId>
         </dependency>
         </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+        </dependency>
+
         <dependency>
         <dependency>
             <groupId>org.apache.karaf.shell</groupId>
             <groupId>org.apache.karaf.shell</groupId>
             <artifactId>org.apache.karaf.shell.console</artifactId>
             <artifactId>org.apache.karaf.shell.console</artifactId>

+package org.onlab.onos.ccc;
+
+import com.fasterxml.jackson.core.JsonEncoding;
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.onlab.onos.cluster.DefaultControllerNode;
+import org.onlab.onos.cluster.NodeId;
+import org.onlab.packet.IpPrefix;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+
+/**
+ * Allows for reading and writing cluster definition as a JSON file.
+ */
+public class ClusterDefinitionStore {
+
+    private final File file;
+
+    /**
+     * Creates a reader/writer of the cluster definition file.
+     *
+     * @param filePath location of the definition file
+     */
+    public ClusterDefinitionStore(String filePath) {
+        file = new File(filePath);
+    }
+
+    /**
+     * Returns set of the controller nodes, including self.
+     *
+     * @return set of controller nodes
+     */
+    public Set<DefaultControllerNode> read() throws IOException {
+        Set<DefaultControllerNode> nodes = new HashSet<>();
+        ObjectMapper mapper = new ObjectMapper();
+        ObjectNode clusterNodeDef = (ObjectNode) mapper.readTree(file);
+        Iterator<JsonNode> it = ((ArrayNode) clusterNodeDef.get("nodes")).elements();
+        while (it.hasNext()) {
+            ObjectNode nodeDef = (ObjectNode) it.next();
+            nodes.add(new DefaultControllerNode(new NodeId(nodeDef.get("id").asText()),
+                                                IpPrefix.valueOf(nodeDef.get("ip").asText()),
+                                                nodeDef.get("tcpPort").asInt(9876)));
+        }
+        return nodes;
+    }
+
+    /**
+     * Writes the given set of the controller nodes.
+     *
+     * @param nodes set of controller nodes
+     */
+    public void write(Set<DefaultControllerNode> nodes) throws IOException {
+        ObjectMapper mapper = new ObjectMapper();
+        ObjectNode clusterNodeDef = mapper.createObjectNode();
+        ArrayNode nodeDefs = mapper.createArrayNode();
+        clusterNodeDef.set("nodes", nodeDefs);
+        for (DefaultControllerNode node : nodes) {
+            ObjectNode nodeDef = mapper.createObjectNode();
+            nodeDef.put("id", node.id().toString())
+                    .put("ip", node.ip().toString())
+                    .put("tcpPort", node.tcpPort());
+            nodeDefs.add(nodeDef);
+        }
+        mapper.writeTree(new JsonFactory().createGenerator(file, JsonEncoding.UTF8),
+                         clusterNodeDef);
+    }
+
+}

@@ -21,12 +21,17 @@ import org.slf4j.LoggerFactory;
 
 
 import java.io.IOException;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
 import java.nio.channels.ByteChannel;
 import java.nio.channels.ByteChannel;
 import java.nio.channels.ServerSocketChannel;
 import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
 import java.util.ArrayList;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.List;
 import java.util.Map;
 import java.util.Map;
 import java.util.Set;
 import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Executors;
@@ -47,21 +52,36 @@ public class DistributedClusterStore
 
 
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final Logger log = LoggerFactory.getLogger(getClass());
 
 
+    private static final long CONNECTION_CUSTODIAN_DELAY = 100L;
+    private static final long CONNECTION_CUSTODIAN_FREQUENCY = 5000;
+
     private static final long SELECT_TIMEOUT = 50;
     private static final long SELECT_TIMEOUT = 50;
     private static final int WORKERS = 3;
     private static final int WORKERS = 3;
+    private static final int INITIATORS = 2;
     private static final int COMM_BUFFER_SIZE = 16 * 1024;
     private static final int COMM_BUFFER_SIZE = 16 * 1024;
     private static final int COMM_IDLE_TIME = 500;
     private static final int COMM_IDLE_TIME = 500;
 
 
+    private static final boolean SO_NO_DELAY = false;
+    private static final int SO_SEND_BUFFER_SIZE = 128 * 1024;
+    private static final int SO_RCV_BUFFER_SIZE = 128 * 1024;
+
     private DefaultControllerNode self;
     private DefaultControllerNode self;
     private final Map<NodeId, DefaultControllerNode> nodes = new ConcurrentHashMap<>();
     private final Map<NodeId, DefaultControllerNode> nodes = new ConcurrentHashMap<>();
     private final Map<NodeId, State> states = new ConcurrentHashMap<>();
     private final Map<NodeId, State> states = new ConcurrentHashMap<>();
+    private final Map<NodeId, TLVMessageStream> streams = new ConcurrentHashMap<>();
+    private final Map<SocketChannel, DefaultControllerNode> nodesByChannel = new ConcurrentHashMap<>();
 
 
     private final ExecutorService listenExecutor =
     private final ExecutorService listenExecutor =
-            Executors.newSingleThreadExecutor(namedThreads("onos-listen"));
+            Executors.newSingleThreadExecutor(namedThreads("onos-comm-listen"));
     private final ExecutorService commExecutors =
     private final ExecutorService commExecutors =
-            Executors.newFixedThreadPool(WORKERS, namedThreads("onos-cluster"));
+            Executors.newFixedThreadPool(WORKERS, namedThreads("onos-comm-cluster"));
     private final ExecutorService heartbeatExecutor =
     private final ExecutorService heartbeatExecutor =
-            Executors.newSingleThreadExecutor(namedThreads("onos-heartbeat"));
+            Executors.newSingleThreadExecutor(namedThreads("onos-comm-heartbeat"));
+    private final ExecutorService initiatorExecutors =
+            Executors.newFixedThreadPool(INITIATORS, namedThreads("onos-comm-initiator"));
+
+    private final Timer timer = new Timer();
+    private final TimerTask connectionCustodian = new ConnectionCustodian();
 
 
     private ListenLoop listenLoop;
     private ListenLoop listenLoop;
     private List<CommLoop> commLoops = new ArrayList<>(WORKERS);
     private List<CommLoop> commLoops = new ArrayList<>(WORKERS);
@@ -71,9 +91,28 @@ public class DistributedClusterStore
         establishIdentity();
         establishIdentity();
         startCommunications();
         startCommunications();
         startListening();
         startListening();
+        startInitiating();
         log.info("Started");
         log.info("Started");
     }
     }
 
 
+    @Deactivate
+    public void deactivate() {
+        listenLoop.shutdown();
+        for (CommLoop loop : commLoops) {
+            loop.shutdown();
+        }
+        log.info("Stopped");
+    }
+
+
+    // Establishes the controller's own identity.
+    private void establishIdentity() {
+        IpPrefix ip = valueOf(System.getProperty("onos.ip", "127.0.1.1"));
+        self = new DefaultControllerNode(new NodeId(ip.toString()), ip);
+        nodes.put(self.id(), self);
+    }
+
+    // Kicks off the IO loops.
     private void startCommunications() {
     private void startCommunications() {
         for (int i = 0; i < WORKERS; i++) {
         for (int i = 0; i < WORKERS; i++) {
             try {
             try {
@@ -96,20 +135,26 @@ public class DistributedClusterStore
         }
         }
     }
     }
 
 
-    // Establishes the controller's own identity.
-    private void establishIdentity() {
-        // For now rely on env. variable.
-        IpPrefix ip = valueOf(System.getenv("ONOS_NIC"));
-        self = new DefaultControllerNode(new NodeId(ip.toString()), ip);
+    /**
+     * Initiates open connection request and registers the pending socket
+     * channel with the given IO loop.
+     *
+     * @param loop loop with which the channel should be registered
+     * @throws java.io.IOException if the socket could not be open or connected
+     */
+    private void openConnection(DefaultControllerNode node, CommLoop loop) throws IOException {
+        SocketAddress sa = new InetSocketAddress(getByAddress(node.ip().toOctets()), node.tcpPort());
+        SocketChannel ch = SocketChannel.open();
+        nodesByChannel.put(ch, node);
+        ch.configureBlocking(false);
+        loop.connectStream(ch);
+        ch.connect(sa);
     }
     }
 
 
-    @Deactivate
-    public void deactivate() {
-        listenLoop.shutdown();
-        for (CommLoop loop : commLoops) {
-            loop.shutdown();
-        }
-        log.info("Stopped");
+
+    // Attempts to connect to any nodes that do not have an associated connection.
+    private void startInitiating() {
+        timer.schedule(connectionCustodian, CONNECTION_CUSTODIAN_DELAY, CONNECTION_CUSTODIAN_FREQUENCY);
     }
     }
 
 
     @Override
     @Override
@@ -154,7 +199,16 @@ public class DistributedClusterStore
 
 
         @Override
         @Override
         protected void acceptConnection(ServerSocketChannel channel) throws IOException {
         protected void acceptConnection(ServerSocketChannel channel) throws IOException {
+            SocketChannel sc = channel.accept();
+            sc.configureBlocking(false);
+
+            Socket so = sc.socket();
+            so.setTcpNoDelay(SO_NO_DELAY);
+            so.setReceiveBufferSize(SO_RCV_BUFFER_SIZE);
+            so.setSendBufferSize(SO_SEND_BUFFER_SIZE);
 
 
+            findLeastUtilizedLoop().acceptStream(sc);
+            log.info("Connected client");
         }
         }
     }
     }
 
 
@@ -172,5 +226,64 @@ public class DistributedClusterStore
         protected void processMessages(List<TLVMessage> messages, MessageStream<TLVMessage> stream) {
         protected void processMessages(List<TLVMessage> messages, MessageStream<TLVMessage> stream) {
 
 
         }
         }
+
+        @Override
+        public TLVMessageStream acceptStream(SocketChannel channel) {
+            TLVMessageStream stream = super.acceptStream(channel);
+            try {
+                InetSocketAddress sa = (InetSocketAddress) channel.getRemoteAddress();
+                log.info("Accepted a new connection from {}", IpPrefix.valueOf(sa.getAddress().getAddress()));
+            } catch (IOException e) {
+                log.warn("Unable to accept connection from an unknown end-point", e);
+            }
+            return stream;
+        }
+
+        @Override
+        public TLVMessageStream connectStream(SocketChannel channel) {
+            TLVMessageStream stream = super.connectStream(channel);
+            DefaultControllerNode node = nodesByChannel.get(channel);
+            if (node != null) {
+                log.info("Opened connection to {}", node.id());
+                streams.put(node.id(), stream);
+            }
+            return stream;
+        }
+    }
+
+
+    // Sweeps through all controller nodes and attempts to open connection to
+    // those that presently do not have one.
+    private class ConnectionCustodian extends TimerTask {
+        @Override
+        public void run() {
+            for (DefaultControllerNode node : nodes.values()) {
+                if (node != self && !streams.containsKey(node.id())) {
+                    try {
+                        openConnection(node, findLeastUtilizedLoop());
+                    } catch (IOException e) {
+                        log.warn("Unable to connect", e);
+                    }
+                }
+            }
+        }
+    }
+
+    // Finds the least utilities IO loop.
+    private CommLoop findLeastUtilizedLoop() {
+        CommLoop leastUtilized = null;
+        int minCount = Integer.MAX_VALUE;
+        for (CommLoop loop : commLoops) {
+            int count = loop.streamCount();
+            if (count == 0) {
+                return loop;
+            }
+
+            if (count < minCount) {
+                leastUtilized = loop;
+                minCount = count;
+            }
+        }
+        return leastUtilized;
     }
     }
 }
 }

+package org.onlab.onos.cli;
+
+import org.apache.karaf.shell.commands.Argument;
+import org.apache.karaf.shell.commands.Command;
+import org.onlab.onos.cluster.ClusterAdminService;
+import org.onlab.onos.cluster.NodeId;
+import org.onlab.packet.IpPrefix;
+
+/**
+ * Lists all controller cluster nodes.
+ */
+@Command(scope = "onos", name = "add-node",
+         description = "Lists all controller cluster nodes")
+public class NodeAddCommand extends AbstractShellCommand {
+
+    @Argument(index = 0, name = "nodeId", description = "Node ID",
+              required = true, multiValued = false)
+    String nodeId = null;
+
+    @Argument(index = 1, name = "ip", description = "Node IP address",
+              required = true, multiValued = false)
+    String ip = null;
+
+    @Argument(index = 2, name = "tcpPort", description = "TCP port",
+              required = false, multiValued = false)
+    int tcpPort = 9876;
+
+    @Override
+    protected void execute() {
+        ClusterAdminService service = get(ClusterAdminService.class);
+        service.addNode(new NodeId(nodeId), IpPrefix.valueOf(ip), tcpPort);
+    }
+
+}

@@ -5,6 +5,9 @@
             <action class="org.onlab.onos.cli.NodesListCommand"/>
             <action class="org.onlab.onos.cli.NodesListCommand"/>
         </command>
         </command>
         <command>
         <command>
+            <action class="org.onlab.onos.cli.NodeAddCommand"/>
+        </command>
+        <command>
             <action class="org.onlab.onos.cli.MastersListCommand"/>
             <action class="org.onlab.onos.cli.MastersListCommand"/>
             <completers>
             <completers>
                 <ref component-id="clusterIdCompleter"/>
                 <ref component-id="clusterIdCompleter"/>

@@ -11,4 +11,5 @@ remote=$ONOS_USER@${1:-$OCI}
 ssh $remote "
 ssh $remote "
     sudo perl -pi.bak -e \"s/            <interface>.*</            <interface>${ONOS_NIC:-192.168.56.*}</g\" \
     sudo perl -pi.bak -e \"s/            <interface>.*</            <interface>${ONOS_NIC:-192.168.56.*}</g\" \
         $ONOS_INSTALL_DIR/$KARAF_DIST/etc/hazelcast.xml
         $ONOS_INSTALL_DIR/$KARAF_DIST/etc/hazelcast.xml
+    echo \"onos.ip=\$(ifconfig  | grep $ONOS_NIC | cut -d: -f2 | cut -d\\  -f1)\" >> $ONOS_INSTALL_DIR/$KARAF_DIST/etc/system.properties
 "
 "
\ No newline at end of file

 # Default virtual box ONOS instances 1,2 & ONOS mininet box
 # Default virtual box ONOS instances 1,2 & ONOS mininet box
 . $ONOS_ROOT/tools/test/cells/.reset
 . $ONOS_ROOT/tools/test/cells/.reset
 
 
+export ONOS_NIC=192.168.56.*
+
 export OC1="192.168.56.101"
 export OC1="192.168.56.101"
 export OC2="192.168.56.102"
 export OC2="192.168.56.102"
 
 

+# Default virtual box ONOS instances 1,2 & ONOS mininet box
+
+export ONOS_NIC=192.168.56.*
+
+export OC1="192.168.56.101"
+export OC2="192.168.56.102"
+
+export OCN="192.168.56.105"
+
+

@@ -54,6 +54,15 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>>
     }
     }
 
 
     /**
     /**
+     * Returns the number of streams in custody of the IO loop.
+     *
+     * @return number of message streams using this loop
+     */
+    public int streamCount() {
+        return streams.size();
+    }
+
+    /**
      * Creates a new message stream backed by the specified socket channel.
      * Creates a new message stream backed by the specified socket channel.
      *
      *
      * @param byteChannel backing byte channel
      * @param byteChannel backing byte channel
@@ -182,9 +191,10 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>>
      * with a pending accept operation.
      * with a pending accept operation.
      *
      *
      * @param channel backing socket channel
      * @param channel backing socket channel
+     * @return newly accepted message stream
      */
      */
-    public void acceptStream(SocketChannel channel) {
-        createAndAdmit(channel, SelectionKey.OP_READ);
+    public S acceptStream(SocketChannel channel) {
+        return createAndAdmit(channel, SelectionKey.OP_READ);
     }
     }
 
 
 
 
@@ -193,9 +203,10 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>>
      * with a pending connect operation.
      * with a pending connect operation.
      *
      *
      * @param channel backing socket channel
      * @param channel backing socket channel
+     * @return newly connected message stream
      */
      */
-    public void connectStream(SocketChannel channel) {
-        createAndAdmit(channel, SelectionKey.OP_CONNECT);
+    public S connectStream(SocketChannel channel) {
+        return createAndAdmit(channel, SelectionKey.OP_CONNECT);
     }
     }
 
 
     /**
     /**
@@ -205,12 +216,14 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>>
      * @param channel socket channel
      * @param channel socket channel
      * @param op      pending operations mask to be applied to the selection
      * @param op      pending operations mask to be applied to the selection
      *                key as a set of initial interestedOps
      *                key as a set of initial interestedOps
+     * @return newly created message stream
      */
      */
-    private synchronized void createAndAdmit(SocketChannel channel, int op) {
+    private synchronized S createAndAdmit(SocketChannel channel, int op) {
         S stream = createStream(channel);
         S stream = createStream(channel);
         streams.add(stream);
         streams.add(stream);
         newStreamRequests.add(new NewStreamRequest(stream, channel, op));
         newStreamRequests.add(new NewStreamRequest(stream, channel, op));
         selector.wakeup();
         selector.wakeup();
+        return stream;
     }
     }
 
 
     /**
     /**