Added ability to measure round-trip latency and to assure message integrity.
diff --git a/apps/foo/pom.xml b/apps/foo/pom.xml
index 860d70b..098a6ad 100644
--- a/apps/foo/pom.xml
+++ b/apps/foo/pom.xml
@@ -31,6 +31,16 @@
<groupId>org.livetribe.slp</groupId>
<artifactId>livetribe-slp</artifactId>
</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>
<groupId>org.apache.karaf.shell</groupId>
<artifactId>org.apache.karaf.shell.console</artifactId>
diff --git a/apps/foo/src/main/java/org/onlab/onos/ccc/ClusterDefinitionStore.java b/apps/foo/src/main/java/org/onlab/onos/ccc/ClusterDefinitionStore.java
new file mode 100644
index 0000000..74aae56
--- /dev/null
+++ b/apps/foo/src/main/java/org/onlab/onos/ccc/ClusterDefinitionStore.java
@@ -0,0 +1,75 @@
+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);
+ }
+
+}
diff --git a/apps/foo/src/main/java/org/onlab/onos/ccc/DistributedClusterStore.java b/apps/foo/src/main/java/org/onlab/onos/ccc/DistributedClusterStore.java
index 62eed8e..c0f74fa 100644
--- a/apps/foo/src/main/java/org/onlab/onos/ccc/DistributedClusterStore.java
+++ b/apps/foo/src/main/java/org/onlab/onos/ccc/DistributedClusterStore.java
@@ -21,12 +21,17 @@
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
import java.nio.channels.ByteChannel;
import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -47,21 +52,36 @@
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 int WORKERS = 3;
+ private static final int INITIATORS = 2;
private static final int COMM_BUFFER_SIZE = 16 * 1024;
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 final Map<NodeId, DefaultControllerNode> nodes = 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 =
- Executors.newSingleThreadExecutor(namedThreads("onos-listen"));
+ Executors.newSingleThreadExecutor(namedThreads("onos-comm-listen"));
private final ExecutorService commExecutors =
- Executors.newFixedThreadPool(WORKERS, namedThreads("onos-cluster"));
+ Executors.newFixedThreadPool(WORKERS, namedThreads("onos-comm-cluster"));
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 List<CommLoop> commLoops = new ArrayList<>(WORKERS);
@@ -71,9 +91,28 @@
establishIdentity();
startCommunications();
startListening();
+ startInitiating();
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() {
for (int i = 0; i < WORKERS; i++) {
try {
@@ -96,20 +135,26 @@
}
}
- // 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
@@ -154,7 +199,16 @@
@Override
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 @@
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;
}
}
diff --git a/cli/src/main/java/org/onlab/onos/cli/NodeAddCommand.java b/cli/src/main/java/org/onlab/onos/cli/NodeAddCommand.java
new file mode 100644
index 0000000..1c64742
--- /dev/null
+++ b/cli/src/main/java/org/onlab/onos/cli/NodeAddCommand.java
@@ -0,0 +1,34 @@
+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);
+ }
+
+}
diff --git a/cli/src/main/resources/OSGI-INF/blueprint/shell-config.xml b/cli/src/main/resources/OSGI-INF/blueprint/shell-config.xml
index 30fce6f..bf78531 100644
--- a/cli/src/main/resources/OSGI-INF/blueprint/shell-config.xml
+++ b/cli/src/main/resources/OSGI-INF/blueprint/shell-config.xml
@@ -5,6 +5,9 @@
<action class="org.onlab.onos.cli.NodesListCommand"/>
</command>
<command>
+ <action class="org.onlab.onos.cli.NodeAddCommand"/>
+ </command>
+ <command>
<action class="org.onlab.onos.cli.MastersListCommand"/>
<completers>
<ref component-id="clusterIdCompleter"/>
diff --git a/tools/test/bin/onos-config b/tools/test/bin/onos-config
index 9f1e3b0..19cb411 100755
--- a/tools/test/bin/onos-config
+++ b/tools/test/bin/onos-config
@@ -11,4 +11,5 @@
ssh $remote "
sudo perl -pi.bak -e \"s/ <interface>.*</ <interface>${ONOS_NIC:-192.168.56.*}</g\" \
$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
diff --git a/tools/test/cells/local b/tools/test/cells/local
index b04a5e3..6b9fea5 100644
--- a/tools/test/cells/local
+++ b/tools/test/cells/local
@@ -1,6 +1,8 @@
# Default virtual box ONOS instances 1,2 & ONOS mininet box
. $ONOS_ROOT/tools/test/cells/.reset
+export ONOS_NIC=192.168.56.*
+
export OC1="192.168.56.101"
export OC2="192.168.56.102"
diff --git a/tools/test/cells/tom b/tools/test/cells/tom
new file mode 100644
index 0000000..c4482b7
--- /dev/null
+++ b/tools/test/cells/tom
@@ -0,0 +1,10 @@
+# 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"
+
+
diff --git a/utils/nio/src/main/java/org/onlab/nio/IOLoop.java b/utils/nio/src/main/java/org/onlab/nio/IOLoop.java
index 1309330..38c9cf6 100644
--- a/utils/nio/src/main/java/org/onlab/nio/IOLoop.java
+++ b/utils/nio/src/main/java/org/onlab/nio/IOLoop.java
@@ -54,6 +54,15 @@
}
/**
+ * 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.
*
* @param byteChannel backing byte channel
@@ -182,9 +191,10 @@
* with a pending accept operation.
*
* @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 @@
* with a pending connect operation.
*
* @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 @@
* @param channel socket channel
* @param op pending operations mask to be applied to the selection
* 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);
streams.add(stream);
newStreamRequests.add(new NewStreamRequest(stream, channel, op));
selector.wakeup();
+ return stream;
}
/**