Merge branch 'master' of ssh://gerrit.onlab.us:29418/onos-next
Conflicts:
apps/pom.xml
Change-Id: I5dc2c751086118b5137ea8a0a2b81e9de9d48fac
diff --git a/apps/config/pom.xml b/apps/config/pom.xml
new file mode 100644
index 0000000..9a2d540
--- /dev/null
+++ b/apps/config/pom.xml
@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.onlab.onos</groupId>
+ <artifactId>onos-apps</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>onos-app-config</artifactId>
+ <packaging>bundle</packaging>
+
+ <description>ONOS simple network configuration reader</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-core-asl</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-mapper-asl</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ <version>2.4.2</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git a/apps/config/src/main/java/org/onlab/onos/config/AddressConfiguration.java b/apps/config/src/main/java/org/onlab/onos/config/AddressConfiguration.java
new file mode 100644
index 0000000..580e950
--- /dev/null
+++ b/apps/config/src/main/java/org/onlab/onos/config/AddressConfiguration.java
@@ -0,0 +1,34 @@
+package org.onlab.onos.config;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.codehaus.jackson.annotate.JsonProperty;
+
+/**
+ * Object to store address configuration read from a JSON file.
+ */
+public class AddressConfiguration {
+
+ private List<AddressEntry> addresses;
+
+ /**
+ * Gets a list of addresses in the system.
+ *
+ * @return the list of addresses
+ */
+ public List<AddressEntry> getAddresses() {
+ return Collections.unmodifiableList(addresses);
+ }
+
+ /**
+ * Sets a list of addresses in the system.
+ *
+ * @param addresses the list of addresses
+ */
+ @JsonProperty("addresses")
+ public void setAddresses(List<AddressEntry> addresses) {
+ this.addresses = addresses;
+ }
+
+}
diff --git a/apps/config/src/main/java/org/onlab/onos/config/AddressEntry.java b/apps/config/src/main/java/org/onlab/onos/config/AddressEntry.java
new file mode 100644
index 0000000..318aebd
--- /dev/null
+++ b/apps/config/src/main/java/org/onlab/onos/config/AddressEntry.java
@@ -0,0 +1,53 @@
+package org.onlab.onos.config;
+
+import java.util.List;
+
+import org.codehaus.jackson.annotate.JsonProperty;
+import org.onlab.packet.IpPrefix;
+import org.onlab.packet.MacAddress;
+
+/**
+ * Represents a set of addresses bound to a port.
+ */
+public class AddressEntry {
+ private String dpid;
+ private short portNumber;
+ private List<IpPrefix> ipAddresses;
+ private MacAddress macAddress;
+
+ public String getDpid() {
+ return dpid;
+ }
+
+ @JsonProperty("dpid")
+ public void setDpid(String strDpid) {
+ this.dpid = strDpid;
+ }
+
+ public short getPortNumber() {
+ return portNumber;
+ }
+
+ @JsonProperty("port")
+ public void setPortNumber(short portNumber) {
+ this.portNumber = portNumber;
+ }
+
+ public List<IpPrefix> getIpAddresses() {
+ return ipAddresses;
+ }
+
+ @JsonProperty("ips")
+ public void setIpAddresses(List<IpPrefix> ipAddresses) {
+ this.ipAddresses = ipAddresses;
+ }
+
+ public MacAddress getMacAddress() {
+ return macAddress;
+ }
+
+ @JsonProperty("mac")
+ public void setMacAddress(MacAddress macAddress) {
+ this.macAddress = macAddress;
+ }
+}
diff --git a/apps/config/src/main/java/org/onlab/onos/config/NetworkConfigReader.java b/apps/config/src/main/java/org/onlab/onos/config/NetworkConfigReader.java
new file mode 100644
index 0000000..985c4a2
--- /dev/null
+++ b/apps/config/src/main/java/org/onlab/onos/config/NetworkConfigReader.java
@@ -0,0 +1,90 @@
+package org.onlab.onos.config;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.onlab.onos.net.ConnectPoint;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.PortNumber;
+import org.onlab.onos.net.host.HostAdminService;
+import org.onlab.onos.net.host.PortAddresses;
+import org.slf4j.Logger;
+
+import com.google.common.collect.Sets;
+
+/**
+ * Simple configuration module to read in supplementary network configuration
+ * from a file.
+ */
+@Component(immediate = true)
+public class NetworkConfigReader {
+
+ private final Logger log = getLogger(getClass());
+
+ private static final String DEFAULT_CONFIG_FILE = "config/addresses.json";
+ private String configFileName = DEFAULT_CONFIG_FILE;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected HostAdminService hostAdminService;
+
+ @Activate
+ protected void activate() {
+ log.info("Started network config reader");
+
+ log.info("Config file set to {}", configFileName);
+
+ AddressConfiguration config = readNetworkConfig();
+
+ if (config != null) {
+ for (AddressEntry entry : config.getAddresses()) {
+
+ ConnectPoint cp = new ConnectPoint(
+ DeviceId.deviceId(dpidToUri(entry.getDpid())),
+ PortNumber.portNumber(entry.getPortNumber()));
+
+ PortAddresses addresses = new PortAddresses(cp,
+ Sets.newHashSet(entry.getIpAddresses()),
+ entry.getMacAddress());
+
+ hostAdminService.bindAddressesToPort(addresses);
+ }
+ }
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ log.info("Stopped");
+ }
+
+ private AddressConfiguration readNetworkConfig() {
+ File configFile = new File(configFileName);
+
+ ObjectMapper mapper = new ObjectMapper();
+
+ try {
+ AddressConfiguration config =
+ mapper.readValue(configFile, AddressConfiguration.class);
+
+ return config;
+ } catch (FileNotFoundException e) {
+ log.warn("Configuration file not found: {}", configFileName);
+ } catch (IOException e) {
+ log.error("Unable to read config from file:", e);
+ }
+
+ return null;
+ }
+
+ private static String dpidToUri(String dpid) {
+ return "of:" + dpid.replace(":", "");
+ }
+}
diff --git a/apps/config/src/main/java/org/onlab/onos/config/package-info.java b/apps/config/src/main/java/org/onlab/onos/config/package-info.java
new file mode 100644
index 0000000..514a47a
--- /dev/null
+++ b/apps/config/src/main/java/org/onlab/onos/config/package-info.java
@@ -0,0 +1,5 @@
+/**
+ * Simple configuration module to read in supplementary network configuration
+ * from a file.
+ */
+package org.onlab.onos.config;
diff --git a/apps/config/src/main/resources/config.json b/apps/config/src/main/resources/config.json
new file mode 100644
index 0000000..ca4be83
--- /dev/null
+++ b/apps/config/src/main/resources/config.json
@@ -0,0 +1,21 @@
+{
+ "interfaces" : [
+ {
+ "dpid" : "00:00:00:00:00:00:01",
+ "port" : "1",
+ "ips" : ["192.168.10.101/24"],
+ "mac" : "00:00:00:11:22:33"
+ },
+ {
+ "dpid" : "00:00:00:00:00:00:02",
+ "port" : "1",
+ "ips" : ["192.168.20.101/24", "192.168.30.101/24"]
+ },
+ {
+ "dpid" : "00:00:00:00:00:00:03",
+ "port" : "1",
+ "ips" : ["10.1.0.1/16"],
+ "mac" : "00:00:00:00:00:01"
+ }
+ ]
+}
diff --git a/apps/pom.xml b/apps/pom.xml
index f16783f..010c531 100644
--- a/apps/pom.xml
+++ b/apps/pom.xml
@@ -21,6 +21,7 @@
<module>fwd</module>
<module>foo</module>
<module>mobility</module>
+ <module>config</module>
</modules>
<properties>
diff --git a/core/api/src/main/java/org/onlab/onos/net/host/PortAddresses.java b/core/api/src/main/java/org/onlab/onos/net/host/PortAddresses.java
index 7d22877..728d922 100644
--- a/core/api/src/main/java/org/onlab/onos/net/host/PortAddresses.java
+++ b/core/api/src/main/java/org/onlab/onos/net/host/PortAddresses.java
@@ -9,6 +9,8 @@
import org.onlab.packet.IpPrefix;
import org.onlab.packet.MacAddress;
+import com.google.common.base.MoreObjects;
+
/**
* Represents address information bound to a port.
*/
@@ -83,4 +85,13 @@
public int hashCode() {
return Objects.hash(connectPoint, ipAddresses, macAddress);
}
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(getClass())
+ .add("connect-point", connectPoint)
+ .add("ip-addresses", ipAddresses)
+ .add("mac-address", macAddress)
+ .toString();
+ }
}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterConnectionListener.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterConnectionListener.java
new file mode 100644
index 0000000..ae4a76f
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterConnectionListener.java
@@ -0,0 +1,47 @@
+package org.onlab.onos.store.cluster.impl;
+
+import org.onlab.nio.AcceptorLoop;
+import org.onlab.packet.IpPrefix;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+
+import static java.net.InetAddress.getByAddress;
+
+/**
+ * Listens to inbound connection requests and accepts them.
+ */
+public class ClusterConnectionListener extends AcceptorLoop {
+
+ private static final long SELECT_TIMEOUT = 50;
+ private static final int COMM_BUFFER_SIZE = 32 * 1024;
+
+ private static final boolean SO_NO_DELAY = false;
+ private static final int SO_SEND_BUFFER_SIZE = COMM_BUFFER_SIZE;
+ private static final int SO_RCV_BUFFER_SIZE = COMM_BUFFER_SIZE;
+
+ private final WorkerFinder workerFinder;
+
+ ClusterConnectionListener(IpPrefix ip, int tcpPort,
+ WorkerFinder workerFinder) throws IOException {
+ super(SELECT_TIMEOUT, new InetSocketAddress(getByAddress(ip.toOctets()), tcpPort));
+ this.workerFinder = workerFinder;
+ }
+
+ @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);
+
+ workerFinder.findWorker().acceptStream(sc);
+ }
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterIOWorker.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterIOWorker.java
new file mode 100644
index 0000000..0e93985
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterIOWorker.java
@@ -0,0 +1,107 @@
+package org.onlab.onos.store.cluster.impl;
+
+import org.onlab.nio.IOLoop;
+import org.onlab.nio.MessageStream;
+import org.onlab.onos.cluster.DefaultControllerNode;
+import org.onlab.onos.store.cluster.messaging.ClusterMessage;
+import org.onlab.onos.store.cluster.messaging.ClusterMessageStream;
+import org.onlab.onos.store.cluster.messaging.SerializationService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.channels.ByteChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.SocketChannel;
+import java.util.List;
+import java.util.Objects;
+
+import static org.onlab.packet.IpPrefix.valueOf;
+
+/**
+ * Performs the IO operations related to a cluster-wide communications.
+ */
+public class ClusterIOWorker extends
+ IOLoop<ClusterMessage, ClusterMessageStream> {
+
+ private final Logger log = LoggerFactory.getLogger(getClass());
+
+ private static final long SELECT_TIMEOUT = 50;
+
+ private final ConnectionManager connectionManager;
+ private final CommunicationsDelegate commsDelegate;
+ private final SerializationService serializationService;
+ private final ClusterMessage helloMessage;
+
+ /**
+ * Creates a new cluster IO worker.
+ *
+ * @param connectionManager parent connection manager
+ * @param commsDelegate communications delegate for dispatching
+ * @param serializationService serialization service for encode/decode
+ * @param helloMessage hello message for greeting peers
+ * @throws IOException if errors occur during IO loop ignition
+ */
+ ClusterIOWorker(ConnectionManager connectionManager,
+ CommunicationsDelegate commsDelegate,
+ SerializationService serializationService,
+ ClusterMessage helloMessage) throws IOException {
+ super(SELECT_TIMEOUT);
+ this.connectionManager = connectionManager;
+ this.commsDelegate = commsDelegate;
+ this.serializationService = serializationService;
+ this.helloMessage = helloMessage;
+ }
+
+ @Override
+ protected ClusterMessageStream createStream(ByteChannel byteChannel) {
+ return new ClusterMessageStream(serializationService, this, byteChannel);
+ }
+
+ @Override
+ protected void processMessages(List<ClusterMessage> messages, MessageStream<ClusterMessage> stream) {
+ for (ClusterMessage message : messages) {
+ commsDelegate.dispatch(message);
+ }
+ }
+
+ @Override
+ public ClusterMessageStream acceptStream(SocketChannel channel) {
+ ClusterMessageStream stream = super.acceptStream(channel);
+ try {
+ InetSocketAddress sa = (InetSocketAddress) channel.getRemoteAddress();
+ log.info("Accepted connection from node {}", valueOf(sa.getAddress().getAddress()));
+ stream.write(helloMessage);
+
+ } catch (IOException e) {
+ log.warn("Unable to accept connection from an unknown end-point", e);
+ }
+ return stream;
+ }
+
+ @Override
+ protected void connect(SelectionKey key) throws IOException {
+ try {
+ super.connect(key);
+ ClusterMessageStream stream = (ClusterMessageStream) key.attachment();
+ stream.write(helloMessage);
+
+ } catch (IOException e) {
+ if (!Objects.equals(e.getMessage(), "Connection refused")) {
+ throw e;
+ }
+ }
+ }
+
+ @Override
+ protected void removeStream(MessageStream<ClusterMessage> stream) {
+ DefaultControllerNode node = ((ClusterMessageStream) stream).node();
+ if (node != null) {
+ log.info("Closed connection to node {}", node.id());
+ connectionManager.removeNodeStream(node);
+ }
+ super.removeStream(stream);
+ }
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterNodesDelegate.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterNodesDelegate.java
new file mode 100644
index 0000000..b822304
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterNodesDelegate.java
@@ -0,0 +1,25 @@
+package org.onlab.onos.store.cluster.impl;
+
+import org.onlab.onos.cluster.DefaultControllerNode;
+
+/**
+ * Simple back interface through which connection manager can interact with
+ * the cluster store.
+ */
+public interface ClusterNodesDelegate {
+
+ /**
+ * Notifies about a new cluster node being detected.
+ *
+ * @param node newly detected cluster node
+ */
+ void nodeDetected(DefaultControllerNode node);
+
+ /**
+ * Notifies about cluster node going offline.
+ *
+ * @param node cluster node that vanished
+ */
+ void nodeVanished(DefaultControllerNode node);
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/CommunicationsDelegate.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/CommunicationsDelegate.java
new file mode 100644
index 0000000..e74d14b
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/CommunicationsDelegate.java
@@ -0,0 +1,24 @@
+package org.onlab.onos.store.cluster.impl;
+
+import org.onlab.onos.store.cluster.messaging.ClusterMessage;
+
+/**
+ * Simple back interface for interacting with the communications service.
+ */
+public interface CommunicationsDelegate {
+
+ /**
+ * Dispatches the specified message to all registered subscribers.
+ *
+ * @param message message to be dispatched
+ */
+ void dispatch(ClusterMessage message);
+
+ /**
+ * Sets the sender.
+ *
+ * @param messageSender message sender
+ */
+ void setSender(MessageSender messageSender);
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ConnectionManager.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ConnectionManager.java
new file mode 100644
index 0000000..fac3c21
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ConnectionManager.java
@@ -0,0 +1,255 @@
+package org.onlab.onos.store.cluster.impl;
+
+import org.onlab.onos.cluster.DefaultControllerNode;
+import org.onlab.onos.cluster.NodeId;
+import org.onlab.onos.store.cluster.messaging.ClusterMessage;
+import org.onlab.onos.store.cluster.messaging.ClusterMessageStream;
+import org.onlab.onos.store.cluster.messaging.HelloMessage;
+import org.onlab.onos.store.cluster.messaging.SerializationService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
+import java.util.HashSet;
+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;
+
+import static java.net.InetAddress.getByAddress;
+import static org.onlab.util.Tools.namedThreads;
+
+/**
+ * Manages connections to other controller cluster nodes.
+ */
+public class ConnectionManager implements MessageSender {
+
+ private final Logger log = LoggerFactory.getLogger(getClass());
+
+ private static final long CONNECTION_CUSTODIAN_DELAY = 1000L;
+ private static final long CONNECTION_CUSTODIAN_FREQUENCY = 5000;
+
+ private static final long START_TIMEOUT = 1000;
+ private static final int WORKERS = 3;
+
+ private ClusterConnectionListener connectionListener;
+ private List<ClusterIOWorker> workers = new ArrayList<>(WORKERS);
+
+ private final DefaultControllerNode localNode;
+ private final ClusterNodesDelegate nodesDelegate;
+ private final CommunicationsDelegate commsDelegate;
+ private final SerializationService serializationService;
+
+ // Nodes to be monitored to make sure they have a connection.
+ private final Set<DefaultControllerNode> nodes = new HashSet<>();
+
+ // Means to track message streams to other nodes.
+ private final Map<NodeId, ClusterMessageStream> streams = new ConcurrentHashMap<>();
+
+ // Executor pools for listening and managing connections to other nodes.
+ private final ExecutorService listenExecutor =
+ Executors.newSingleThreadExecutor(namedThreads("onos-comm-listen"));
+ private final ExecutorService commExecutors =
+ Executors.newFixedThreadPool(WORKERS, namedThreads("onos-comm-cluster"));
+ private final ExecutorService heartbeatExecutor =
+ Executors.newSingleThreadExecutor(namedThreads("onos-comm-heartbeat"));
+
+ private final Timer timer = new Timer("onos-comm-initiator");
+ private final TimerTask connectionCustodian = new ConnectionCustodian();
+
+ private final WorkerFinder workerFinder = new LeastUtilitiedWorkerFinder();
+
+
+ /**
+ * Creates a new connection manager.
+ */
+ ConnectionManager(DefaultControllerNode localNode,
+ ClusterNodesDelegate nodesDelegate,
+ CommunicationsDelegate commsDelegate,
+ SerializationService serializationService) {
+ this.localNode = localNode;
+ this.nodesDelegate = nodesDelegate;
+ this.commsDelegate = commsDelegate;
+ this.serializationService = serializationService;
+
+ commsDelegate.setSender(this);
+ startCommunications();
+ startListening();
+ startInitiating();
+ log.info("Started");
+ }
+
+ /**
+ * Shuts down the connection manager.
+ */
+ void shutdown() {
+ connectionListener.shutdown();
+ for (ClusterIOWorker worker : workers) {
+ worker.shutdown();
+ }
+ log.info("Stopped");
+ }
+
+ /**
+ * Adds the node to the list of monitored nodes.
+ *
+ * @param node node to be added
+ */
+ void addNode(DefaultControllerNode node) {
+ nodes.add(node);
+ }
+
+ /**
+ * Removes the node from the list of monitored nodes.
+ *
+ * @param node node to be removed
+ */
+ void removeNode(DefaultControllerNode node) {
+ nodes.remove(node);
+ ClusterMessageStream stream = streams.remove(node.id());
+ if (stream != null) {
+ stream.close();
+ }
+ }
+
+ /**
+ * Removes the stream associated with the specified node.
+ *
+ * @param node node whose stream to remove
+ */
+ void removeNodeStream(DefaultControllerNode node) {
+ nodesDelegate.nodeVanished(node);
+ streams.remove(node.id());
+ }
+
+ @Override
+ public boolean send(NodeId nodeId, ClusterMessage message) {
+ ClusterMessageStream stream = streams.get(nodeId);
+ if (stream != null) {
+ try {
+ stream.write(message);
+ return true;
+ } catch (IOException e) {
+ log.warn("Unable to send a message about {} to node {}",
+ message.subject(), nodeId);
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Kicks off the IO loops and waits for them to startup.
+ */
+ private void startCommunications() {
+ HelloMessage hello = new HelloMessage(localNode.id(), localNode.ip(),
+ localNode.tcpPort());
+ for (int i = 0; i < WORKERS; i++) {
+ try {
+ ClusterIOWorker worker =
+ new ClusterIOWorker(this, commsDelegate,
+ serializationService, hello);
+ workers.add(worker);
+ commExecutors.execute(worker);
+ } catch (IOException e) {
+ log.warn("Unable to start communication worker", e);
+ }
+ }
+
+ // Wait for the IO loops to start
+ for (ClusterIOWorker loop : workers) {
+ if (!loop.awaitStart(START_TIMEOUT)) {
+ log.warn("Comm loop did not start on-time; moving on...");
+ }
+ }
+ }
+
+ /**
+ * Starts listening for connections from peer cluster members.
+ */
+ private void startListening() {
+ try {
+ connectionListener =
+ new ClusterConnectionListener(localNode.ip(), localNode.tcpPort(),
+ workerFinder);
+ listenExecutor.execute(connectionListener);
+ if (!connectionListener.awaitStart(START_TIMEOUT)) {
+ log.warn("Listener did not start on-time; moving on...");
+ }
+ } catch (IOException e) {
+ log.error("Unable to listen for cluster connections", e);
+ }
+ }
+
+ /**
+ * 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 initiateConnection(DefaultControllerNode node,
+ ClusterIOWorker loop) throws IOException {
+ SocketAddress sa = new InetSocketAddress(getByAddress(node.ip().toOctets()), node.tcpPort());
+ SocketChannel ch = SocketChannel.open();
+ ch.configureBlocking(false);
+ ch.connect(sa);
+ loop.connectStream(ch);
+ }
+
+
+ /**
+ * 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);
+ }
+
+ // 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) {
+ if (node != localNode && !streams.containsKey(node.id())) {
+ try {
+ initiateConnection(node, workerFinder.findWorker());
+ } catch (IOException e) {
+ log.debug("Unable to connect", e);
+ }
+ }
+ }
+ }
+ }
+
+ // Finds the least utilitied IO loop.
+ private class LeastUtilitiedWorkerFinder implements WorkerFinder {
+
+ @Override
+ public ClusterIOWorker findWorker() {
+ ClusterIOWorker leastUtilized = null;
+ int minCount = Integer.MAX_VALUE;
+ for (ClusterIOWorker worker : workers) {
+ int count = worker.streamCount();
+ if (count == 0) {
+ return worker;
+ }
+
+ if (count < minCount) {
+ leastUtilized = worker;
+ minCount = count;
+ }
+ }
+ return leastUtilized;
+ }
+ }
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/DistributedClusterStore.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/DistributedClusterStore.java
index 5cd9d9e..ae04226 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/DistributedClusterStore.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/DistributedClusterStore.java
@@ -4,10 +4,9 @@
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
-import org.onlab.nio.AcceptorLoop;
-import org.onlab.nio.IOLoop;
-import org.onlab.nio.MessageStream;
import org.onlab.onos.cluster.ClusterEvent;
import org.onlab.onos.cluster.ClusterStore;
import org.onlab.onos.cluster.ClusterStoreDelegate;
@@ -15,33 +14,18 @@
import org.onlab.onos.cluster.DefaultControllerNode;
import org.onlab.onos.cluster.NodeId;
import org.onlab.onos.store.AbstractStore;
+import org.onlab.onos.store.cluster.messaging.SerializationService;
import org.onlab.packet.IpPrefix;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.SelectionKey;
-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.Objects;
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;
-import static java.net.InetAddress.getByAddress;
import static org.onlab.onos.cluster.ControllerNode.State;
import static org.onlab.packet.IpPrefix.valueOf;
-import static org.onlab.util.Tools.namedThreads;
/**
* Distributed implementation of the cluster nodes store.
@@ -52,146 +36,69 @@
extends AbstractStore<ClusterEvent, ClusterStoreDelegate>
implements ClusterStore {
- private static final int HELLO_MSG = 1;
- private static final int ECHO_MSG = 2;
-
private final Logger log = LoggerFactory.getLogger(getClass());
- private static final long CONNECTION_CUSTODIAN_DELAY = 1000L;
- private static final long CONNECTION_CUSTODIAN_FREQUENCY = 5000;
-
- private static final long START_TIMEOUT = 1000;
- private static final long SELECT_TIMEOUT = 50;
- private static final int WORKERS = 3;
- private static final int COMM_BUFFER_SIZE = 32 * 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 = COMM_BUFFER_SIZE;
- private static final int SO_RCV_BUFFER_SIZE = COMM_BUFFER_SIZE;
-
- private DefaultControllerNode self;
+ private DefaultControllerNode localNode;
private final Map<NodeId, DefaultControllerNode> nodes = new ConcurrentHashMap<>();
private final Map<NodeId, State> states = new ConcurrentHashMap<>();
- // Means to track message streams to other nodes.
- private final Map<NodeId, TLVMessageStream> streams = new ConcurrentHashMap<>();
- private final Map<SocketChannel, DefaultControllerNode> nodesByChannel = new ConcurrentHashMap<>();
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ private CommunicationsDelegate commsDelegate;
- // Executor pools for listening and managing connections to other nodes.
- private final ExecutorService listenExecutor =
- Executors.newSingleThreadExecutor(namedThreads("onos-comm-listen"));
- private final ExecutorService commExecutors =
- Executors.newFixedThreadPool(WORKERS, namedThreads("onos-comm-cluster"));
- private final ExecutorService heartbeatExecutor =
- Executors.newSingleThreadExecutor(namedThreads("onos-comm-heartbeat"));
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ private SerializationService serializationService;
- private final Timer timer = new Timer("onos-comm-initiator");
- private final TimerTask connectionCustodian = new ConnectionCustodian();
-
- private ListenLoop listenLoop;
- private List<CommLoop> commLoops = new ArrayList<>(WORKERS);
+ private final ClusterNodesDelegate nodesDelegate = new InnerNodesDelegate();
+ private ConnectionManager connectionManager;
@Activate
public void activate() {
loadClusterDefinition();
- startCommunications();
- startListening();
- startInitiating();
+ establishSelfIdentity();
+ connectionManager = new ConnectionManager(localNode, nodesDelegate,
+ commsDelegate, serializationService);
log.info("Started");
}
@Deactivate
public void deactivate() {
- listenLoop.shutdown();
- for (CommLoop loop : commLoops) {
- loop.shutdown();
- }
log.info("Stopped");
}
- // Loads the cluster definition file
+ /**
+ * Loads the cluster definition file.
+ */
private void loadClusterDefinition() {
-// ClusterDefinitionStore cds = new ClusterDefinitionStore("../config/cluster.json");
-// try {
-// Set<DefaultControllerNode> storedNodes = cds.read();
-// for (DefaultControllerNode node : storedNodes) {
-// nodes.put(node.id(), node);
-// }
-// } catch (IOException e) {
-// log.error("Unable to read cluster definitions", e);
-// }
-
- // Establishes the controller's own identity.
- IpPrefix ip = valueOf(System.getProperty("onos.ip", "127.0.1.1"));
- self = nodes.get(new NodeId(ip.toString()));
-
- // As a fall-back, let's make sure we at least know who we are.
- if (self == null) {
- self = new DefaultControllerNode(new NodeId(ip.toString()), ip);
- nodes.put(self.id(), self);
- states.put(self.id(), State.ACTIVE);
- }
- }
-
- // Kicks off the IO loops.
- private void startCommunications() {
- for (int i = 0; i < WORKERS; i++) {
- try {
- CommLoop loop = new CommLoop();
- commLoops.add(loop);
- commExecutors.execute(loop);
- } catch (IOException e) {
- log.warn("Unable to start comm IO loop", e);
- }
- }
-
- // Wait for the IO loops to start
- for (CommLoop loop : commLoops) {
- if (!loop.awaitStart(START_TIMEOUT)) {
- log.warn("Comm loop did not start on-time; moving on...");
- }
- }
- }
-
- // Starts listening for connections from peer cluster members.
- private void startListening() {
+ ClusterDefinitionStore cds = new ClusterDefinitionStore("../config/cluster.json");
try {
- listenLoop = new ListenLoop(self.ip(), self.tcpPort());
- listenExecutor.execute(listenLoop);
- if (!listenLoop.awaitStart(START_TIMEOUT)) {
- log.warn("Listen loop did not start on-time; moving on...");
+ Set<DefaultControllerNode> storedNodes = cds.read();
+ for (DefaultControllerNode node : storedNodes) {
+ nodes.put(node.id(), node);
}
} catch (IOException e) {
- log.error("Unable to listen for cluster connections", e);
+ log.error("Unable to read cluster definitions", e);
}
}
/**
- * 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
+ * Determines who the local controller node is.
*/
- 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);
- ch.connect(sa);
- loop.connectStream(ch);
- }
+ private void establishSelfIdentity() {
+ // Establishes the controller's own identity.
+ IpPrefix ip = valueOf(System.getProperty("onos.ip", "127.0.1.1"));
+ localNode = nodes.get(new NodeId(ip.toString()));
-
- // 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);
+ // As a fall-back, let's make sure we at least know who we are.
+ if (localNode == null) {
+ localNode = new DefaultControllerNode(new NodeId(ip.toString()), ip);
+ nodes.put(localNode.id(), localNode);
+ states.put(localNode.id(), State.ACTIVE);
+ }
}
@Override
public ControllerNode getLocalNode() {
- return self;
+ return localNode;
}
@Override
@@ -215,179 +122,29 @@
public ControllerNode addNode(NodeId nodeId, IpPrefix ip, int tcpPort) {
DefaultControllerNode node = new DefaultControllerNode(nodeId, ip, tcpPort);
nodes.put(nodeId, node);
+ connectionManager.addNode(node);
return node;
}
@Override
public void removeNode(NodeId nodeId) {
- nodes.remove(nodeId);
- TLVMessageStream stream = streams.remove(nodeId);
- if (stream != null) {
- stream.close();
+ DefaultControllerNode node = nodes.remove(nodeId);
+ if (node != null) {
+ connectionManager.removeNode(node);
}
}
- // Listens and accepts inbound connections from other cluster nodes.
- private class ListenLoop extends AcceptorLoop {
- ListenLoop(IpPrefix ip, int tcpPort) throws IOException {
- super(SELECT_TIMEOUT, new InetSocketAddress(getByAddress(ip.toOctets()), tcpPort));
+ // Entity to handle back calls from the connection manager.
+ private class InnerNodesDelegate implements ClusterNodesDelegate {
+ @Override
+ public void nodeDetected(DefaultControllerNode node) {
+ nodes.put(node.id(), node);
+ states.put(node.id(), State.ACTIVE);
}
@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);
+ public void nodeVanished(DefaultControllerNode node) {
+ states.put(node.id(), State.INACTIVE);
}
}
-
- private class CommLoop extends IOLoop<TLVMessage, TLVMessageStream> {
- CommLoop() throws IOException {
- super(SELECT_TIMEOUT);
- }
-
- @Override
- protected TLVMessageStream createStream(ByteChannel byteChannel) {
- return new TLVMessageStream(this, byteChannel, COMM_BUFFER_SIZE, COMM_IDLE_TIME);
- }
-
- @Override
- protected void processMessages(List<TLVMessage> messages, MessageStream<TLVMessage> stream) {
- TLVMessageStream tlvStream = (TLVMessageStream) stream;
- for (TLVMessage message : messages) {
- // TODO: add type-based dispatching here... this is just a hack to get going
- if (message.type() == HELLO_MSG) {
- processHello(message, tlvStream);
- } else if (message.type() == ECHO_MSG) {
- processEcho(message, tlvStream);
- } else {
- log.info("Deal with other messages");
- }
- }
- }
-
- @Override
- public TLVMessageStream acceptStream(SocketChannel channel) {
- TLVMessageStream stream = super.acceptStream(channel);
- try {
- InetSocketAddress sa = (InetSocketAddress) channel.getRemoteAddress();
- log.info("Accepted connection from node {}", valueOf(sa.getAddress().getAddress()));
- stream.write(createHello(self));
-
- } 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.debug("Opened connection to node {}", node.id());
- nodesByChannel.remove(channel);
- }
- return stream;
- }
-
- @Override
- protected void connect(SelectionKey key) throws IOException {
- try {
- super.connect(key);
- TLVMessageStream stream = (TLVMessageStream) key.attachment();
- send(stream, createHello(self));
- } catch (IOException e) {
- if (!Objects.equals(e.getMessage(), "Connection refused")) {
- throw e;
- }
- }
- }
-
- @Override
- protected void removeStream(MessageStream<TLVMessage> stream) {
- DefaultControllerNode node = ((TLVMessageStream) stream).node();
- if (node != null) {
- log.info("Closed connection to node {}", node.id());
- states.put(node.id(), State.INACTIVE);
- streams.remove(node.id());
- }
- super.removeStream(stream);
- }
- }
-
- // Processes a HELLO message from a peer controller node.
- private void processHello(TLVMessage message, TLVMessageStream stream) {
- // FIXME: pure hack for now
- String data = new String(message.data());
- String[] fields = data.split(":");
- DefaultControllerNode node = new DefaultControllerNode(new NodeId(fields[0]),
- valueOf(fields[1]),
- Integer.parseInt(fields[2]));
- stream.setNode(node);
- nodes.put(node.id(), node);
- streams.put(node.id(), stream);
- states.put(node.id(), State.ACTIVE);
- }
-
- // Processes an ECHO message from a peer controller node.
- private void processEcho(TLVMessage message, TLVMessageStream tlvStream) {
- // TODO: implement heart-beat refresh
- log.info("Dealing with echoes...");
- }
-
- // Sends message to the specified stream.
- private void send(TLVMessageStream stream, TLVMessage message) {
- try {
- stream.write(message);
- } catch (IOException e) {
- log.warn("Unable to send message to {}", stream.node().id());
- }
- }
-
- // Creates a hello message to be sent to a peer controller node.
- private TLVMessage createHello(DefaultControllerNode self) {
- return new TLVMessage(HELLO_MSG, (self.id() + ":" + self.ip() + ":" + self.tcpPort()).getBytes());
- }
-
- // 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.debug("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/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/MessageSender.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/MessageSender.java
new file mode 100644
index 0000000..55f868c
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/MessageSender.java
@@ -0,0 +1,21 @@
+package org.onlab.onos.store.cluster.impl;
+
+import org.onlab.onos.cluster.NodeId;
+import org.onlab.onos.store.cluster.messaging.ClusterMessage;
+
+/**
+ * Created by tom on 9/29/14.
+ */
+public interface MessageSender {
+
+ /**
+ * Sends the specified message to the given cluster node.
+ *
+ * @param nodeId node identifier
+ * @param message mesage to send
+ * @return true if the message was sent sucessfully; false if there is
+ * no stream or if there was an error
+ */
+ boolean send(NodeId nodeId, ClusterMessage message);
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/TLVMessage.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/TLVMessage.java
deleted file mode 100644
index 246f8ee..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/TLVMessage.java
+++ /dev/null
@@ -1,70 +0,0 @@
-package org.onlab.onos.store.cluster.impl;
-
-import org.onlab.nio.AbstractMessage;
-
-import java.util.Objects;
-
-import static com.google.common.base.MoreObjects.toStringHelper;
-
-/**
- * Base message for cluster-wide communications using TLVs.
- */
-public class TLVMessage extends AbstractMessage {
-
- private final int type;
- private final byte[] data;
-
- /**
- * Creates an immutable TLV message.
- *
- * @param type message type
- * @param data message data bytes
- */
- public TLVMessage(int type, byte[] data) {
- this.length = data.length + TLVMessageStream.METADATA_LENGTH;
- this.type = type;
- this.data = data;
- }
-
- /**
- * Returns the message type indicator.
- *
- * @return message type
- */
- public int type() {
- return type;
- }
-
- /**
- * Returns the data bytes.
- *
- * @return message data
- */
- public byte[] data() {
- return data;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(type, data);
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null || getClass() != obj.getClass()) {
- return false;
- }
- final TLVMessage other = (TLVMessage) obj;
- return Objects.equals(this.type, other.type) &&
- Objects.equals(this.data, other.data);
- }
-
- @Override
- public String toString() {
- return toStringHelper(this).add("type", type).add("length", length).toString();
- }
-
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/TLVMessageStream.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/TLVMessageStream.java
deleted file mode 100644
index b003945..0000000
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/TLVMessageStream.java
+++ /dev/null
@@ -1,95 +0,0 @@
-package org.onlab.onos.store.cluster.impl;
-
-import org.onlab.nio.IOLoop;
-import org.onlab.nio.MessageStream;
-import org.onlab.onos.cluster.DefaultControllerNode;
-
-import java.nio.ByteBuffer;
-import java.nio.channels.ByteChannel;
-
-import static com.google.common.base.Preconditions.checkState;
-
-/**
- * Stream for transferring TLV messages between cluster members.
- */
-public class TLVMessageStream extends MessageStream<TLVMessage> {
-
- public static final int METADATA_LENGTH = 16; // 8 + 4 + 4
-
- private static final int LENGTH_OFFSET = 12;
- private static final long MARKER = 0xfeedcafecafefeedL;
-
- private DefaultControllerNode node;
-
- /**
- * Creates a message stream associated with the specified IO loop and
- * backed by the given byte channel.
- *
- * @param loop IO loop
- * @param byteChannel backing byte channel
- * @param bufferSize size of the backing byte buffers
- * @param maxIdleMillis maximum number of millis the stream can be idle
- */
- protected TLVMessageStream(IOLoop<TLVMessage, ?> loop, ByteChannel byteChannel,
- int bufferSize, int maxIdleMillis) {
- super(loop, byteChannel, bufferSize, maxIdleMillis);
- }
-
- /**
- * Returns the node with which this stream is associated.
- *
- * @return controller node
- */
- DefaultControllerNode node() {
- return node;
- }
-
- /**
- * Sets the node with which this stream is affiliated.
- *
- * @param node controller node
- */
- void setNode(DefaultControllerNode node) {
- checkState(this.node == null, "Stream is already bound to a node");
- this.node = node;
- }
-
- @Override
- protected TLVMessage read(ByteBuffer buffer) {
- // Do we have enough bytes to read the header? If not, bail.
- if (buffer.remaining() < METADATA_LENGTH) {
- return null;
- }
-
- // Peek at the length and if we have enough to read the entire message
- // go ahead, otherwise bail.
- int length = buffer.getInt(buffer.position() + LENGTH_OFFSET);
- if (buffer.remaining() < length) {
- return null;
- }
-
- // At this point, we have enough data to read a complete message.
- long marker = buffer.getLong();
- checkState(marker == MARKER, "Incorrect message marker");
-
- int type = buffer.getInt();
- length = buffer.getInt();
-
- // TODO: add deserialization hook here
- byte[] data = new byte[length - METADATA_LENGTH];
- buffer.get(data);
-
- return new TLVMessage(type, data);
- }
-
- @Override
- protected void write(TLVMessage message, ByteBuffer buffer) {
- buffer.putLong(MARKER);
- buffer.putInt(message.type());
- buffer.putInt(message.length());
-
- // TODO: add serialization hook here
- buffer.put(message.data());
- }
-
-}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/WorkerFinder.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/WorkerFinder.java
new file mode 100644
index 0000000..06f4f8a
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/WorkerFinder.java
@@ -0,0 +1,14 @@
+package org.onlab.onos.store.cluster.impl;
+
+/**
+ * Provides means to find a worker IO loop.
+ */
+public interface WorkerFinder {
+
+ /**
+ * Finds a suitable worker.
+ *
+ * @return available worker
+ */
+ ClusterIOWorker findWorker();
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterCommunicationService.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterCommunicationService.java
new file mode 100644
index 0000000..87ed221
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterCommunicationService.java
@@ -0,0 +1,46 @@
+package org.onlab.onos.store.cluster.messaging;
+
+import org.onlab.onos.cluster.NodeId;
+
+import java.util.Set;
+
+/**
+ * Service for assisting communications between controller cluster nodes.
+ */
+public interface ClusterCommunicationService {
+
+ /**
+ * Sends a message to the specified controller node.
+ *
+ * @param message message to send
+ * @param toNodeId node identifier
+ * @return true if the message was sent sucessfully; false if there is
+ * no stream or if there was an error
+ */
+ boolean send(ClusterMessage message, NodeId toNodeId);
+
+ /**
+ * Adds a new subscriber for the specified message subject.
+ *
+ * @param subject message subject
+ * @param subscriber message subscriber
+ */
+ void addSubscriber(MessageSubject subject, MessageSubscriber subscriber);
+
+ /**
+ * Removes the specified subscriber from the given message subject.
+ *
+ * @param subject message subject
+ * @param subscriber message subscriber
+ */
+ void removeSubscriber(MessageSubject subject, MessageSubscriber subscriber);
+
+ /**
+ * Returns the set of subscribers for the specified message subject.
+ *
+ * @param subject message subject
+ * @return set of message subscribers
+ */
+ Set<MessageSubscriber> getSubscribers(MessageSubject subject);
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessage.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessage.java
new file mode 100644
index 0000000..3033ac9
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessage.java
@@ -0,0 +1,37 @@
+package org.onlab.onos.store.cluster.messaging;
+
+import org.onlab.nio.AbstractMessage;
+
+import static com.google.common.base.MoreObjects.toStringHelper;
+
+/**
+ * Base message for cluster-wide communications.
+ */
+public abstract class ClusterMessage extends AbstractMessage {
+
+ private final MessageSubject subject;
+
+ /**
+ * Creates a cluster message.
+ *
+ * @param subject message subject
+ */
+ protected ClusterMessage(MessageSubject subject) {
+ this.subject = subject;
+ }
+
+ /**
+ * Returns the message subject indicator.
+ *
+ * @return message subject
+ */
+ public MessageSubject subject() {
+ return subject;
+ }
+
+ @Override
+ public String toString() {
+ return toStringHelper(this).add("subject", subject).add("length", length).toString();
+ }
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessageStream.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessageStream.java
new file mode 100644
index 0000000..0970726
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessageStream.java
@@ -0,0 +1,67 @@
+package org.onlab.onos.store.cluster.messaging;
+
+import org.onlab.nio.IOLoop;
+import org.onlab.nio.MessageStream;
+import org.onlab.onos.cluster.DefaultControllerNode;
+
+import java.nio.ByteBuffer;
+import java.nio.channels.ByteChannel;
+
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Stream for transferring messages between two cluster members.
+ */
+public class ClusterMessageStream extends MessageStream<ClusterMessage> {
+
+ private static final int COMM_BUFFER_SIZE = 32 * 1024;
+ private static final int COMM_IDLE_TIME = 500;
+
+ private DefaultControllerNode node;
+ private SerializationService serializationService;
+
+ /**
+ * Creates a message stream associated with the specified IO loop and
+ * backed by the given byte channel.
+ *
+ * @param serializationService service for encoding/decoding messages
+ * @param loop IO loop
+ * @param byteChannel backing byte channel
+ */
+ public ClusterMessageStream(SerializationService serializationService,
+ IOLoop<ClusterMessage, ?> loop,
+ ByteChannel byteChannel) {
+ super(loop, byteChannel, COMM_BUFFER_SIZE, COMM_IDLE_TIME);
+ this.serializationService = serializationService;
+ }
+
+ /**
+ * Returns the node with which this stream is associated.
+ *
+ * @return controller node
+ */
+ public DefaultControllerNode node() {
+ return node;
+ }
+
+ /**
+ * Sets the node with which this stream is affiliated.
+ *
+ * @param node controller node
+ */
+ public void setNode(DefaultControllerNode node) {
+ checkState(this.node == null, "Stream is already bound to a node");
+ this.node = node;
+ }
+
+ @Override
+ protected ClusterMessage read(ByteBuffer buffer) {
+ return serializationService.decode(buffer);
+ }
+
+ @Override
+ protected void write(ClusterMessage message, ByteBuffer buffer) {
+ serializationService.encode(message, buffer);
+ }
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/EchoMessage.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/EchoMessage.java
new file mode 100644
index 0000000..d25a341
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/EchoMessage.java
@@ -0,0 +1,37 @@
+package org.onlab.onos.store.cluster.messaging;
+
+import org.onlab.onos.cluster.NodeId;
+
+/**l
+ * Echo heart-beat message that nodes send to each other.
+ */
+public class EchoMessage extends ClusterMessage {
+
+ private NodeId nodeId;
+
+ // For serialization
+ private EchoMessage() {
+ super(MessageSubject.HELLO);
+ nodeId = null;
+ }
+
+ /**
+ * Creates a new heart-beat echo message.
+ *
+ * @param nodeId sending node identification
+ */
+ public EchoMessage(NodeId nodeId) {
+ super(MessageSubject.HELLO);
+ nodeId = nodeId;
+ }
+
+ /**
+ * Returns the sending node identifer.
+ *
+ * @return node identifier
+ */
+ public NodeId nodeId() {
+ return nodeId;
+ }
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/HelloMessage.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/HelloMessage.java
new file mode 100644
index 0000000..ddc79d3
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/HelloMessage.java
@@ -0,0 +1,63 @@
+package org.onlab.onos.store.cluster.messaging;
+
+import org.onlab.onos.cluster.NodeId;
+import org.onlab.packet.IpPrefix;
+
+/**
+ * Hello message that nodes use to greet each other.
+ */
+public class HelloMessage extends ClusterMessage {
+
+ private NodeId nodeId;
+ private IpPrefix ipAddress;
+ private int tcpPort;
+
+ // For serialization
+ private HelloMessage() {
+ super(MessageSubject.HELLO);
+ nodeId = null;
+ ipAddress = null;
+ tcpPort = 0;
+ }
+
+ /**
+ * Creates a new hello message for the specified end-point data.
+ *
+ * @param nodeId sending node identification
+ * @param ipAddress sending node IP address
+ * @param tcpPort sending node TCP port
+ */
+ public HelloMessage(NodeId nodeId, IpPrefix ipAddress, int tcpPort) {
+ super(MessageSubject.HELLO);
+ nodeId = nodeId;
+ ipAddress = ipAddress;
+ tcpPort = tcpPort;
+ }
+
+ /**
+ * Returns the sending node identifer.
+ *
+ * @return node identifier
+ */
+ public NodeId nodeId() {
+ return nodeId;
+ }
+
+ /**
+ * Returns the sending node IP address.
+ *
+ * @return node IP address
+ */
+ public IpPrefix ipAddress() {
+ return ipAddress;
+ }
+
+ /**
+ * Returns the sending node TCP listen port.
+ *
+ * @return TCP listen port
+ */
+ public int tcpPort() {
+ return tcpPort;
+ }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/MessageSubject.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/MessageSubject.java
new file mode 100644
index 0000000..3b888b3
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/MessageSubject.java
@@ -0,0 +1,14 @@
+package org.onlab.onos.store.cluster.messaging;
+
+/**
+ * Representation of a message subject.
+ */
+public enum MessageSubject {
+
+ /** Represents a first greeting message. */
+ HELLO,
+
+ /** Signifies a heart-beat message. */
+ ECHO
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/MessageSubscriber.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/MessageSubscriber.java
new file mode 100644
index 0000000..6b78fec
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/MessageSubscriber.java
@@ -0,0 +1,15 @@
+package org.onlab.onos.store.cluster.messaging;
+
+/**
+ * Represents a message consumer.
+ */
+public interface MessageSubscriber {
+
+ /**
+ * Receives the specified cluster message.
+ *
+ * @param message message to be received
+ */
+ void receive(ClusterMessage message);
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/SerializationService.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/SerializationService.java
new file mode 100644
index 0000000..79e054b
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/SerializationService.java
@@ -0,0 +1,26 @@
+package org.onlab.onos.store.cluster.messaging;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Service for serializing/deserializing intra-cluster messages.
+ */
+public interface SerializationService {
+
+ /**
+ * Decodes the specified byte buffer to obtain a message within.
+ *
+ * @param buffer byte buffer with message(s)
+ * @return parsed message
+ */
+ ClusterMessage decode(ByteBuffer buffer);
+
+ /**
+ * Encodes the specified message into the given byte buffer.
+ *
+ * @param message message to be encoded
+ * @param buffer byte buffer to receive the message data
+ */
+ void encode(ClusterMessage message, ByteBuffer buffer);
+
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/ClusterCommunicationManager.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/ClusterCommunicationManager.java
new file mode 100644
index 0000000..bafb2c3
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/ClusterCommunicationManager.java
@@ -0,0 +1,64 @@
+package org.onlab.onos.store.cluster.messaging.impl;
+
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Multimap;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.cluster.NodeId;
+import org.onlab.onos.store.cluster.impl.CommunicationsDelegate;
+import org.onlab.onos.store.cluster.impl.MessageSender;
+import org.onlab.onos.store.cluster.messaging.ClusterCommunicationService;
+import org.onlab.onos.store.cluster.messaging.ClusterMessage;
+import org.onlab.onos.store.cluster.messaging.MessageSubject;
+import org.onlab.onos.store.cluster.messaging.MessageSubscriber;
+
+import java.util.Set;
+
+/**
+ * Implements the cluster communication services to use by other stores.
+ */
+@Component(immediate = true)
+@Service
+public class ClusterCommunicationManager
+ implements ClusterCommunicationService, CommunicationsDelegate {
+
+ // TODO: use something different that won't require synchronization
+ private Multimap<MessageSubject, MessageSubscriber> subscribers = HashMultimap.create();
+ private MessageSender messageSender;
+
+ @Override
+ public boolean send(ClusterMessage message, NodeId toNodeId) {
+ return messageSender.send(toNodeId, message);
+ }
+
+ @Override
+ public synchronized void addSubscriber(MessageSubject subject, MessageSubscriber subscriber) {
+ subscribers.put(subject, subscriber);
+ }
+
+ @Override
+ public synchronized void removeSubscriber(MessageSubject subject, MessageSubscriber subscriber) {
+ subscribers.remove(subject, subscriber);
+ }
+
+ @Override
+ public Set<MessageSubscriber> getSubscribers(MessageSubject subject) {
+ return ImmutableSet.copyOf(subscribers.get(subject));
+ }
+
+ @Override
+ public void dispatch(ClusterMessage message) {
+ Set<MessageSubscriber> set = getSubscribers(message.subject());
+ if (set != null) {
+ for (MessageSubscriber subscriber : set) {
+ subscriber.receive(message);
+ }
+ }
+ }
+
+ @Override
+ public void setSender(MessageSender messageSender) {
+ this.messageSender = messageSender;
+ }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/MessageSerializer.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/MessageSerializer.java
new file mode 100644
index 0000000..93c8310
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/MessageSerializer.java
@@ -0,0 +1,69 @@
+package org.onlab.onos.store.cluster.messaging.impl;
+
+import org.onlab.onos.store.cluster.messaging.ClusterMessage;
+import org.onlab.onos.store.cluster.messaging.MessageSubject;
+import org.onlab.onos.store.cluster.messaging.SerializationService;
+
+import java.nio.ByteBuffer;
+
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Factory for parsing messages sent between cluster members.
+ */
+public class MessageSerializer implements SerializationService {
+
+ private static final int METADATA_LENGTH = 16; // 8 + 4 + 4
+ private static final int LENGTH_OFFSET = 12;
+
+ private static final long MARKER = 0xfeedcafebeaddeadL;
+
+ @Override
+ public ClusterMessage decode(ByteBuffer buffer) {
+ try {
+ // Do we have enough bytes to read the header? If not, bail.
+ if (buffer.remaining() < METADATA_LENGTH) {
+ return null;
+ }
+
+ // Peek at the length and if we have enough to read the entire message
+ // go ahead, otherwise bail.
+ int length = buffer.getInt(buffer.position() + LENGTH_OFFSET);
+ if (buffer.remaining() < length) {
+ return null;
+ }
+
+ // At this point, we have enough data to read a complete message.
+ long marker = buffer.getLong();
+ checkState(marker == MARKER, "Incorrect message marker");
+
+ int subjectOrdinal = buffer.getInt();
+ MessageSubject subject = MessageSubject.values()[subjectOrdinal];
+ length = buffer.getInt();
+
+ // TODO: sanity checking for length
+ byte[] data = new byte[length - METADATA_LENGTH];
+ buffer.get(data);
+
+ // TODO: add deserialization hook here; for now this hack
+ return null; // actually deserialize
+
+ } catch (Exception e) {
+ // TODO: recover from exceptions by forwarding stream to next marker
+ e.printStackTrace();
+ }
+ return null;
+ }
+
+ @Override
+ public void encode(ClusterMessage message, ByteBuffer buffer) {
+ try {
+ int i = 0;
+ // Type based lookup for proper encoder
+ } catch (Exception e) {
+ // TODO: recover from exceptions by forwarding stream to next marker
+ e.printStackTrace();
+ }
+ }
+
+}
diff --git a/features/features.xml b/features/features.xml
index 63511fd..d2d9567 100644
--- a/features/features.xml
+++ b/features/features.xml
@@ -19,6 +19,9 @@
<bundle>mvn:de.javakaffee/kryo-serializers/0.27</bundle>
<bundle>mvn:org.onlab.onos/onlab-nio/1.0.0-SNAPSHOT</bundle>
+
+ <bundle>mvn:org.codehaus.jackson/jackson-core-asl/1.9.13</bundle>
+ <bundle>mvn:org.codehaus.jackson/jackson-mapper-asl/1.9.13</bundle>
</feature>
<feature name="onos-thirdparty-web" version="1.0.0"
@@ -130,4 +133,10 @@
<bundle>mvn:org.onlab.onos/onos-app-foo/1.0.0-SNAPSHOT</bundle>
</feature>
+ <feature name="onos-app-config" version="1.0.0"
+ description="ONOS network config reader">
+ <feature>onos-api</feature>
+ <bundle>mvn:org.onlab.onos/onos-app-config/1.0.0-SNAPSHOT</bundle>
+ </feature>
+
</features>
diff --git a/pom.xml b/pom.xml
index b98a9e5..5cf67de 100644
--- a/pom.xml
+++ b/pom.xml
@@ -92,6 +92,17 @@
<version>3.3.2</version>
</dependency>
+ <dependency>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-core-asl</artifactId>
+ <version>1.9.13</version>
+ </dependency>
+ <dependency>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-mapper-asl</artifactId>
+ <version>1.9.13</version>
+ </dependency>
+
<!-- Web related -->
<dependency>