홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
tom
73094839 1 parent defed6ff

Initial check-in for new p2p cluster messaging; to be refactored.

Inline Side-by-side
Showing 8 changed files with 352 additions and 40 deletions
......@@ -28,10 +28,6 @@
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.livetribe.slp</groupId>
<artifactId>livetribe-slp</artifactId>
</dependency>
<dependency>
<groupId>org.apache.karaf.shell</groupId>
<artifactId>org.apache.karaf.shell.console</artifactId>
</dependency>
......
......@@ -26,6 +26,23 @@
<artifactId>onos-core-serializers</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.onlab.onos</groupId>
<artifactId>onlab-nio</artifactId>
<version>${project.version}</version>
</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.felix</groupId>
<artifactId>org.apache.felix.scr.annotations</artifactId>
......
package org.onlab.onos.store.cluster.impl;
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);
}
}
package org.onlab.onos.ccc;
package org.onlab.onos.store.cluster.impl;
import com.google.common.collect.ImmutableSet;
import org.apache.felix.scr.annotations.Activate;
......@@ -21,12 +21,18 @@ 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.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
......@@ -45,35 +51,88 @@ public class DistributedClusterStore
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 = 16 * 1024;
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 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<>();
// Executor pools for listening and managing connections to other nodes.
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 Timer timer = new Timer("onos-comm-initiator");
private final TimerTask connectionCustodian = new ConnectionCustodian();
private ListenLoop listenLoop;
private List<CommLoop> commLoops = new ArrayList<>(WORKERS);
@Activate
public void activate() {
establishIdentity();
loadClusterDefinition();
startCommunications();
startListening();
startInitiating();
log.info("Started");
}
@Deactivate
public void deactivate() {
listenLoop.shutdown();
for (CommLoop loop : commLoops) {
loop.shutdown();
}
log.info("Stopped");
}
// 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);
}
}
// Kicks off the IO loops.
private void startCommunications() {
for (int i = 0; i < WORKERS; i++) {
try {
......@@ -84,6 +143,13 @@ public class DistributedClusterStore
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.
......@@ -91,25 +157,34 @@ public class DistributedClusterStore
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...");
}
} catch (IOException e) {
log.error("Unable to listen for cluster connections", e);
}
}
// 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);
ch.connect(sa);
loop.connectStream(ch);
}
@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
......@@ -144,6 +219,7 @@ public class DistributedClusterStore
@Override
public void removeNode(NodeId nodeId) {
nodes.remove(nodeId);
streams.remove(nodeId);
}
// Listens and accepts inbound connections from other cluster nodes.
......@@ -154,7 +230,15 @@ public class DistributedClusterStore
@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);
}
}
......@@ -170,7 +254,109 @@ public class DistributedClusterStore
@Override
protected void processMessages(List<TLVMessage> messages, MessageStream<TLVMessage> stream) {
TLVMessageStream tlvStream = (TLVMessageStream) stream;
for (TLVMessage message : messages) {
// TODO: add type-based dispatching here...
log.info("Got message {}", message.type());
// FIXME: hack to get going
if (message.type() == HELLO_MSG) {
processHello(message, tlvStream);
}
}
}
@Override
public TLVMessageStream acceptStream(SocketChannel channel) {
TLVMessageStream stream = super.acceptStream(channel);
try {
InetSocketAddress sa = (InetSocketAddress) channel.getRemoteAddress();
log.info("Accepted a new connection from node {}", IpPrefix.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.info("Opened connection to node {}", node.id());
nodesByChannel.remove(channel);
}
return stream;
}
@Override
protected void connect(SelectionKey key) {
super.connect(key);
TLVMessageStream stream = (TLVMessageStream) key.attachment();
send(stream, createHello(self));
}
}
// FIXME: pure hack for now
private void processHello(TLVMessage message, TLVMessageStream stream) {
String data = new String(message.data());
log.info("Processing hello with data [{}]", data);
String[] fields = new String(data).split(":");
DefaultControllerNode node = new DefaultControllerNode(new NodeId(fields[0]),
IpPrefix.valueOf(fields[1]),
Integer.parseInt(fields[2]));
stream.setNode(node);
nodes.put(node.id(), node);
streams.put(node.id(), stream);
}
// 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());
}
}
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.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.ccc;
package org.onlab.onos.store.cluster.impl;
import org.onlab.nio.AbstractMessage;
......@@ -12,17 +12,16 @@ import static com.google.common.base.MoreObjects.toStringHelper;
public class TLVMessage extends AbstractMessage {
private final int type;
private final Object data;
private final byte[] data;
/**
* Creates an immutable TLV message.
*
* @param type message type
* @param length message length
* @param data message data
* @param data message data bytes
*/
public TLVMessage(int type, int length, Object data) {
this.length = length;
public TLVMessage(int type, byte[] data) {
this.length = data.length + TLVMessageStream.METADATA_LENGTH;
this.type = type;
this.data = data;
}
......@@ -37,11 +36,11 @@ public class TLVMessage extends AbstractMessage {
}
/**
* Returns the data object.
* Returns the data bytes.
*
* @return message data
*/
public Object data() {
public byte[] data() {
return data;
}
......
package org.onlab.onos.ccc;
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;
......@@ -13,8 +14,13 @@ import static com.google.common.base.Preconditions.checkState;
*/
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.
......@@ -29,17 +35,51 @@ public class TLVMessageStream extends MessageStream<TLVMessage> {
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();
int length = buffer.getInt();
length = buffer.getInt();
// TODO: add deserialization hook here
byte[] data = new byte[length - METADATA_LENGTH];
buffer.get(data);
return new TLVMessage(type, length, null);
return new TLVMessage(type, data);
}
@Override
......@@ -49,5 +89,7 @@ public class TLVMessageStream extends MessageStream<TLVMessage> {
buffer.putInt(message.length());
// TODO: add serialization hook here
buffer.put(message.data());
}
}
......
......@@ -48,14 +48,11 @@
description="ONOS core components">
<feature>onos-api</feature>
<bundle>mvn:org.onlab.onos/onos-core-net/1.0.0-SNAPSHOT</bundle>
<bundle>mvn:org.onlab.onos/onos-core-hz-common/1.0.0-SNAPSHOT</bundle>
<bundle>mvn:org.onlab.onos/onos-core-serializers/1.0.0-SNAPSHOT</bundle>
<bundle>mvn:org.onlab.onos/onos-core-hz-cluster/1.0.0-SNAPSHOT</bundle>
<bundle>mvn:org.onlab.onos/onos-core-hz-net/1.0.0-SNAPSHOT</bundle>
<bundle>mvn:org.onlab.onos/onos-core-dist/1.0.0-SNAPSHOT</bundle>
</feature>
<feature name="onos-core-dist" version="1.0.0"
description="ONOS core components">
<feature name="onos-core-hazelcast" version="1.0.0"
description="ONOS core components built on hazelcast">
<feature>onos-api</feature>
<bundle>mvn:org.onlab.onos/onos-core-net/1.0.0-SNAPSHOT</bundle>
<bundle>mvn:org.onlab.onos/onos-core-hz-common/1.0.0-SNAPSHOT</bundle>
......
......@@ -170,7 +170,7 @@ public abstract class MessageStream<M extends Message> {
}
/**
* Reads, withouth blocking, a list of messages from the stream.
* Reads, without blocking, a list of messages from the stream.
* The list will be empty if there were not messages pending.
*
* @return list of messages or null if backing channel has been closed
......