Integrated Kryo serializers with the communications manager and IO loop stuff.
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterCommunicationAdminService.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterCommunicationAdminService.java
index 76e4167..22ed9de 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterCommunicationAdminService.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterCommunicationAdminService.java
@@ -29,4 +29,8 @@
*/
void startUp(DefaultControllerNode localNode, ClusterNodesDelegate delegate);
+ /**
+ * Clears all nodes and streams as part of leaving the cluster.
+ */
+ void clearAllNodesAndStreams();
}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterCommunicationManager.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterCommunicationManager.java
index 8acc8fe..bac32e9 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterCommunicationManager.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ClusterCommunicationManager.java
@@ -13,6 +13,7 @@
import org.onlab.onos.cluster.NodeId;
import org.onlab.onos.store.cluster.messaging.ClusterCommunicationService;
import org.onlab.onos.store.cluster.messaging.ClusterMessage;
+import org.onlab.onos.store.cluster.messaging.GoodbyeMessage;
import org.onlab.onos.store.cluster.messaging.HelloMessage;
import org.onlab.onos.store.cluster.messaging.MessageSubject;
import org.onlab.onos.store.cluster.messaging.MessageSubscriber;
@@ -83,9 +84,11 @@
private final Timer timer = new Timer("onos-comm-initiator");
private final TimerTask connectionCustodian = new ConnectionCustodian();
+ private GoodbyeSubscriber goodbyeSubscriber = new GoodbyeSubscriber();
@Activate
public void activate() {
+ addSubscriber(MessageSubject.GOODBYE, goodbyeSubscriber);
log.info("Activated but waiting for delegate");
}
@@ -102,9 +105,20 @@
}
@Override
+ public boolean send(ClusterMessage message) {
+ boolean ok = true;
+ for (DefaultControllerNode node : nodes) {
+ if (!node.equals(localNode)) {
+ ok = send(message, node.id()) && ok;
+ }
+ }
+ return ok;
+ }
+
+ @Override
public boolean send(ClusterMessage message, NodeId toNodeId) {
ClusterMessageStream stream = streams.get(toNodeId);
- if (stream != null) {
+ if (stream != null && !toNodeId.equals(localNode.id())) {
try {
stream.write(message);
return true;
@@ -140,6 +154,7 @@
@Override
public void removeNode(DefaultControllerNode node) {
+ send(new GoodbyeMessage(node.id()));
nodes.remove(node);
ClusterMessageStream stream = streams.remove(node.id());
if (stream != null) {
@@ -159,6 +174,16 @@
log.info("Started");
}
+ @Override
+ public void clearAllNodesAndStreams() {
+ nodes.clear();
+ send(new GoodbyeMessage(localNode.id()));
+ for (ClusterMessageStream stream : streams.values()) {
+ stream.close();
+ }
+ streams.clear();
+ }
+
/**
* Dispatches the specified message to all subscribers to its subject.
*
@@ -304,4 +329,11 @@
}
}
+ private class GoodbyeSubscriber implements MessageSubscriber {
+ @Override
+ public void receive(ClusterMessage message, NodeId fromNodeId) {
+ log.info("Received goodbye message from {}", fromNodeId);
+ nodesDelegate.nodeRemoved(fromNodeId);
+ }
+ }
}
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
index 0c47d18..b82a835 100644
--- 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
@@ -27,4 +27,11 @@
*/
void nodeVanished(NodeId nodeId);
+ /**
+ * Notifies about remote request to remove node from cluster.
+ *
+ * @param nodeId identifier of the cluster node that was removed
+ */
+ void nodeRemoved(NodeId nodeId);
+
}
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 025804a..647690e 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
@@ -127,9 +127,17 @@
@Override
public void removeNode(NodeId nodeId) {
- DefaultControllerNode node = nodes.remove(nodeId);
- if (node != null) {
- communicationAdminService.removeNode(node);
+ if (nodeId.equals(localNode.id())) {
+ // FIXME: this is still broken
+ // We are being ejected from the cluster, so remove all other nodes.
+ communicationAdminService.clearAllNodesAndStreams();
+ nodes.clear();
+ } else {
+ // Remove the other node.
+ DefaultControllerNode node = nodes.remove(nodeId);
+ if (node != null) {
+ communicationAdminService.removeNode(node);
+ }
}
}
@@ -148,6 +156,11 @@
public void nodeVanished(NodeId nodeId) {
states.put(nodeId, State.INACTIVE);
}
+
+ @Override
+ public void nodeRemoved(NodeId nodeId) {
+ removeNode(nodeId);
+ }
}
}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/MessageSerializer.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/MessageSerializer.java
index b54f0db..260d2b2 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/MessageSerializer.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/MessageSerializer.java
@@ -1,17 +1,50 @@
package org.onlab.onos.store.cluster.impl;
+import de.javakaffee.kryoserializers.URISerializer;
+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.Service;
+import org.onlab.onos.cluster.ControllerNode;
+import org.onlab.onos.cluster.DefaultControllerNode;
import org.onlab.onos.cluster.NodeId;
+import org.onlab.onos.net.ConnectPoint;
+import org.onlab.onos.net.DefaultDevice;
+import org.onlab.onos.net.DefaultLink;
+import org.onlab.onos.net.DefaultPort;
+import org.onlab.onos.net.Device;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.Element;
+import org.onlab.onos.net.Link;
+import org.onlab.onos.net.LinkKey;
+import org.onlab.onos.net.MastershipRole;
+import org.onlab.onos.net.Port;
+import org.onlab.onos.net.PortNumber;
+import org.onlab.onos.net.provider.ProviderId;
import org.onlab.onos.store.cluster.messaging.ClusterMessage;
+import org.onlab.onos.store.cluster.messaging.EchoMessage;
+import org.onlab.onos.store.cluster.messaging.GoodbyeMessage;
import org.onlab.onos.store.cluster.messaging.HelloMessage;
import org.onlab.onos.store.cluster.messaging.MessageSubject;
import org.onlab.onos.store.cluster.messaging.SerializationService;
+import org.onlab.onos.store.serializers.ConnectPointSerializer;
+import org.onlab.onos.store.serializers.DefaultLinkSerializer;
+import org.onlab.onos.store.serializers.DefaultPortSerializer;
+import org.onlab.onos.store.serializers.DeviceIdSerializer;
+import org.onlab.onos.store.serializers.IpPrefixSerializer;
+import org.onlab.onos.store.serializers.LinkKeySerializer;
+import org.onlab.onos.store.serializers.NodeIdSerializer;
+import org.onlab.onos.store.serializers.PortNumberSerializer;
+import org.onlab.onos.store.serializers.ProviderIdSerializer;
import org.onlab.packet.IpPrefix;
+import org.onlab.util.KryoPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.net.URI;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
import static com.google.common.base.Preconditions.checkState;
@@ -24,11 +57,64 @@
private final Logger log = LoggerFactory.getLogger(getClass());
- private static final int METADATA_LENGTH = 16; // 8 + 4 + 4
- private static final int LENGTH_OFFSET = 12;
+ private static final int METADATA_LENGTH = 12; // 8 + 4
+ private static final int LENGTH_OFFSET = 8;
private static final long MARKER = 0xfeedcafebeaddeadL;
+ private KryoPool serializerPool;
+
+ @Activate
+ public void activate() {
+ setupKryoPool();
+ log.info("Started");
+ }
+
+ @Deactivate
+ public void deactivate() {
+ log.info("Stopped");
+ }
+
+ /**
+ * Sets up the common serialzers pool.
+ */
+ protected void setupKryoPool() {
+ // FIXME Slice out types used in common to separate pool/namespace.
+ serializerPool = KryoPool.newBuilder()
+ .register(ArrayList.class,
+ HashMap.class,
+
+ ControllerNode.State.class,
+ Device.Type.class,
+
+ DefaultControllerNode.class,
+ DefaultDevice.class,
+ MastershipRole.class,
+ Port.class,
+ Element.class,
+
+ Link.Type.class,
+
+ MessageSubject.class,
+ HelloMessage.class,
+ GoodbyeMessage.class,
+ EchoMessage.class
+ )
+ .register(IpPrefix.class, new IpPrefixSerializer())
+ .register(URI.class, new URISerializer())
+ .register(NodeId.class, new NodeIdSerializer())
+ .register(ProviderId.class, new ProviderIdSerializer())
+ .register(DeviceId.class, new DeviceIdSerializer())
+ .register(PortNumber.class, new PortNumberSerializer())
+ .register(DefaultPort.class, new DefaultPortSerializer())
+ .register(LinkKey.class, new LinkKeySerializer())
+ .register(ConnectPoint.class, new ConnectPointSerializer())
+ .register(DefaultLink.class, new DefaultLinkSerializer())
+ .build()
+ .populate(1);
+ }
+
+
@Override
public ClusterMessage decode(ByteBuffer buffer) {
try {
@@ -47,18 +133,12 @@
// 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
- String[] fields = new String(data).split(":");
- return new HelloMessage(new NodeId(fields[0]), IpPrefix.valueOf(fields[1]), Integer.parseInt(fields[2]));
+ return (ClusterMessage) serializerPool.deserialize(data);
} catch (Exception e) {
// TODO: recover from exceptions by forwarding stream to next marker
@@ -70,12 +150,8 @@
@Override
public void encode(ClusterMessage message, ByteBuffer buffer) {
try {
- HelloMessage helloMessage = (HelloMessage) message;
+ byte[] data = serializerPool.serialize(message);
buffer.putLong(MARKER);
- buffer.putInt(message.subject().ordinal());
-
- String str = helloMessage.nodeId() + ":" + helloMessage.ipAddress() + ":" + helloMessage.tcpPort();
- byte[] data = str.getBytes();
buffer.putInt(data.length + METADATA_LENGTH);
buffer.put(data);
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
index 87ed221..fe7fcd3 100644
--- 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
@@ -10,6 +10,15 @@
public interface ClusterCommunicationService {
/**
+ * Sends a message to all controller nodes.
+ *
+ * @param message message to send
+ * @return true if the message was sent sucessfully to all nodes; false
+ * if there is no stream or if there was an error for some node
+ */
+ boolean send(ClusterMessage message);
+
+ /**
* Sends a message to the specified controller node.
*
* @param message message to send
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/GoodbyeMessage.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/GoodbyeMessage.java
new file mode 100644
index 0000000..e9326f3
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/GoodbyeMessage.java
@@ -0,0 +1,37 @@
+package org.onlab.onos.store.cluster.messaging;
+
+import org.onlab.onos.cluster.NodeId;
+
+/**
+ * Goodbye message that nodes use to leave the cluster for good.
+ */
+public class GoodbyeMessage extends ClusterMessage {
+
+ private NodeId nodeId;
+
+ // For serialization
+ private GoodbyeMessage() {
+ super(MessageSubject.GOODBYE);
+ nodeId = null;
+ }
+
+ /**
+ * Creates a new goodbye message.
+ *
+ * @param nodeId sending node identification
+ */
+ public GoodbyeMessage(NodeId nodeId) {
+ super(MessageSubject.HELLO);
+ this.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/MessageSubject.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/MessageSubject.java
index 3b888b3..bf86c5b 100644
--- 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
@@ -8,6 +8,9 @@
/** Represents a first greeting message. */
HELLO,
+ /** Signifies node's intent to leave the cluster. */
+ GOODBYE,
+
/** Signifies a heart-beat message. */
ECHO