홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Ayaka Koshibe
78d16a4e 2 parents 65efaefe 1d416c5f

Merge branch 'master' of ssh://gerrit.onlab.us:29418/onos-next

Expand all
Inline Side-by-side
Showing 54 changed files with 1380 additions and 356 deletions
1 +<?xml version="1.0" encoding="UTF-8"?>
2 +<project xmlns="http://maven.apache.org/POM/4.0.0"
3 + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4 + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
5 + <modelVersion>4.0.0</modelVersion>
6 +
7 + <parent>
8 + <groupId>org.onlab.onos</groupId>
9 + <artifactId>onos-apps</artifactId>
10 + <version>1.0.0-SNAPSHOT</version>
11 + <relativePath>../pom.xml</relativePath>
12 + </parent>
13 +
14 + <artifactId>onos-app-config</artifactId>
15 + <packaging>bundle</packaging>
16 +
17 + <description>ONOS simple network configuration reader</description>
18 +
19 + <dependencies>
20 + <dependency>
21 + <groupId>org.codehaus.jackson</groupId>
22 + <artifactId>jackson-core-asl</artifactId>
23 + </dependency>
24 + <dependency>
25 + <groupId>org.codehaus.jackson</groupId>
26 + <artifactId>jackson-mapper-asl</artifactId>
27 + </dependency>
28 + <dependency>
29 + <groupId>com.fasterxml.jackson.core</groupId>
30 + <artifactId>jackson-annotations</artifactId>
31 + <version>2.4.2</version>
32 + <scope>provided</scope>
33 + </dependency>
34 + </dependencies>
35 +
36 +</project>
1 +package org.onlab.onos.config;
2 +
3 +import java.util.Collections;
4 +import java.util.List;
5 +
6 +import org.codehaus.jackson.annotate.JsonProperty;
7 +
8 +/**
9 + * Object to store address configuration read from a JSON file.
10 + */
11 +public class AddressConfiguration {
12 +
13 + private List<AddressEntry> addresses;
14 +
15 + /**
16 + * Gets a list of addresses in the system.
17 + *
18 + * @return the list of addresses
19 + */
20 + public List<AddressEntry> getAddresses() {
21 + return Collections.unmodifiableList(addresses);
22 + }
23 +
24 + /**
25 + * Sets a list of addresses in the system.
26 + *
27 + * @param addresses the list of addresses
28 + */
29 + @JsonProperty("addresses")
30 + public void setAddresses(List<AddressEntry> addresses) {
31 + this.addresses = addresses;
32 + }
33 +
34 +}
1 +package org.onlab.onos.config;
2 +
3 +import java.util.List;
4 +
5 +import org.codehaus.jackson.annotate.JsonProperty;
6 +import org.onlab.packet.IpPrefix;
7 +import org.onlab.packet.MacAddress;
8 +
9 +/**
10 + * Represents a set of addresses bound to a port.
11 + */
12 +public class AddressEntry {
13 + private String dpid;
14 + private short portNumber;
15 + private List<IpPrefix> ipAddresses;
16 + private MacAddress macAddress;
17 +
18 + public String getDpid() {
19 + return dpid;
20 + }
21 +
22 + @JsonProperty("dpid")
23 + public void setDpid(String strDpid) {
24 + this.dpid = strDpid;
25 + }
26 +
27 + public short getPortNumber() {
28 + return portNumber;
29 + }
30 +
31 + @JsonProperty("port")
32 + public void setPortNumber(short portNumber) {
33 + this.portNumber = portNumber;
34 + }
35 +
36 + public List<IpPrefix> getIpAddresses() {
37 + return ipAddresses;
38 + }
39 +
40 + @JsonProperty("ips")
41 + public void setIpAddresses(List<IpPrefix> ipAddresses) {
42 + this.ipAddresses = ipAddresses;
43 + }
44 +
45 + public MacAddress getMacAddress() {
46 + return macAddress;
47 + }
48 +
49 + @JsonProperty("mac")
50 + public void setMacAddress(MacAddress macAddress) {
51 + this.macAddress = macAddress;
52 + }
53 +}
1 +package org.onlab.onos.config;
2 +
3 +import static org.slf4j.LoggerFactory.getLogger;
4 +
5 +import java.io.File;
6 +import java.io.FileNotFoundException;
7 +import java.io.IOException;
8 +
9 +import org.apache.felix.scr.annotations.Activate;
10 +import org.apache.felix.scr.annotations.Component;
11 +import org.apache.felix.scr.annotations.Deactivate;
12 +import org.apache.felix.scr.annotations.Reference;
13 +import org.apache.felix.scr.annotations.ReferenceCardinality;
14 +import org.codehaus.jackson.map.ObjectMapper;
15 +import org.onlab.onos.net.ConnectPoint;
16 +import org.onlab.onos.net.DeviceId;
17 +import org.onlab.onos.net.PortNumber;
18 +import org.onlab.onos.net.host.HostAdminService;
19 +import org.onlab.onos.net.host.PortAddresses;
20 +import org.slf4j.Logger;
21 +
22 +import com.google.common.collect.Sets;
23 +
24 +/**
25 + * Simple configuration module to read in supplementary network configuration
26 + * from a file.
27 + */
28 +@Component(immediate = true)
29 +public class NetworkConfigReader {
30 +
31 + private final Logger log = getLogger(getClass());
32 +
33 + private static final String DEFAULT_CONFIG_FILE = "config/addresses.json";
34 + private String configFileName = DEFAULT_CONFIG_FILE;
35 +
36 + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
37 + protected HostAdminService hostAdminService;
38 +
39 + @Activate
40 + protected void activate() {
41 + log.info("Started network config reader");
42 +
43 + log.info("Config file set to {}", configFileName);
44 +
45 + AddressConfiguration config = readNetworkConfig();
46 +
47 + if (config != null) {
48 + for (AddressEntry entry : config.getAddresses()) {
49 +
50 + ConnectPoint cp = new ConnectPoint(
51 + DeviceId.deviceId(dpidToUri(entry.getDpid())),
52 + PortNumber.portNumber(entry.getPortNumber()));
53 +
54 + PortAddresses addresses = new PortAddresses(cp,
55 + Sets.newHashSet(entry.getIpAddresses()),
56 + entry.getMacAddress());
57 +
58 + hostAdminService.bindAddressesToPort(addresses);
59 + }
60 + }
61 + }
62 +
63 + @Deactivate
64 + protected void deactivate() {
65 + log.info("Stopped");
66 + }
67 +
68 + private AddressConfiguration readNetworkConfig() {
69 + File configFile = new File(configFileName);
70 +
71 + ObjectMapper mapper = new ObjectMapper();
72 +
73 + try {
74 + AddressConfiguration config =
75 + mapper.readValue(configFile, AddressConfiguration.class);
76 +
77 + return config;
78 + } catch (FileNotFoundException e) {
79 + log.warn("Configuration file not found: {}", configFileName);
80 + } catch (IOException e) {
81 + log.error("Unable to read config from file:", e);
82 + }
83 +
84 + return null;
85 + }
86 +
87 + private static String dpidToUri(String dpid) {
88 + return "of:" + dpid.replace(":", "");
89 + }
90 +}
1 +/**
2 + * Simple configuration module to read in supplementary network configuration
3 + * from a file.
4 + */
5 +package org.onlab.onos.config;
1 +{
2 + "interfaces" : [
3 + {
4 + "dpid" : "00:00:00:00:00:00:01",
5 + "port" : "1",
6 + "ips" : ["192.168.10.101/24"],
7 + "mac" : "00:00:00:11:22:33"
8 + },
9 + {
10 + "dpid" : "00:00:00:00:00:00:02",
11 + "port" : "1",
12 + "ips" : ["192.168.20.101/24", "192.168.30.101/24"]
13 + },
14 + {
15 + "dpid" : "00:00:00:00:00:00:03",
16 + "port" : "1",
17 + "ips" : ["10.1.0.1/16"],
18 + "mac" : "00:00:00:00:00:01"
19 + }
20 + ]
21 +}
...@@ -26,7 +26,9 @@ import org.onlab.onos.net.packet.InboundPacket; ...@@ -26,7 +26,9 @@ import org.onlab.onos.net.packet.InboundPacket;
26 import org.onlab.onos.net.packet.PacketContext; 26 import org.onlab.onos.net.packet.PacketContext;
27 import org.onlab.onos.net.packet.PacketProcessor; 27 import org.onlab.onos.net.packet.PacketProcessor;
28 import org.onlab.onos.net.packet.PacketService; 28 import org.onlab.onos.net.packet.PacketService;
29 +import org.onlab.onos.net.proxyarp.ProxyArpService;
29 import org.onlab.onos.net.topology.TopologyService; 30 import org.onlab.onos.net.topology.TopologyService;
31 +import org.onlab.packet.ARP;
30 import org.onlab.packet.Ethernet; 32 import org.onlab.packet.Ethernet;
31 import org.slf4j.Logger; 33 import org.slf4j.Logger;
32 34
...@@ -50,6 +52,9 @@ public class ReactiveForwarding { ...@@ -50,6 +52,9 @@ public class ReactiveForwarding {
50 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) 52 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
51 protected FlowRuleService flowRuleService; 53 protected FlowRuleService flowRuleService;
52 54
55 + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
56 + protected ProxyArpService proxyArpService;
57 +
53 private ReactivePacketProcessor processor = new ReactivePacketProcessor(); 58 private ReactivePacketProcessor processor = new ReactivePacketProcessor();
54 59
55 private ApplicationId appId; 60 private ApplicationId appId;
...@@ -85,6 +90,16 @@ public class ReactiveForwarding { ...@@ -85,6 +90,16 @@ public class ReactiveForwarding {
85 90
86 InboundPacket pkt = context.inPacket(); 91 InboundPacket pkt = context.inPacket();
87 Ethernet ethPkt = pkt.parsed(); 92 Ethernet ethPkt = pkt.parsed();
93 + if (ethPkt.getEtherType() == Ethernet.TYPE_ARP) {
94 + ARP arp = (ARP) ethPkt.getPayload();
95 + if (arp.getOpCode() == ARP.OP_REPLY) {
96 + proxyArpService.forward(ethPkt);
97 + } else if (arp.getOpCode() == ARP.OP_REQUEST) {
98 + proxyArpService.reply(ethPkt);
99 + }
100 + context.block();
101 + return;
102 + }
88 HostId id = HostId.hostId(ethPkt.getDestinationMAC()); 103 HostId id = HostId.hostId(ethPkt.getDestinationMAC());
89 104
90 // Do we know who this is for? If not, flood and bail. 105 // Do we know who this is for? If not, flood and bail.
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
20 <module>tvue</module> 20 <module>tvue</module>
21 <module>fwd</module> 21 <module>fwd</module>
22 <module>foo</module> 22 <module>foo</module>
23 + <module>config</module>
23 </modules> 24 </modules>
24 25
25 <properties> 26 <properties>
......
...@@ -9,6 +9,8 @@ import org.onlab.onos.net.ConnectPoint; ...@@ -9,6 +9,8 @@ import org.onlab.onos.net.ConnectPoint;
9 import org.onlab.packet.IpPrefix; 9 import org.onlab.packet.IpPrefix;
10 import org.onlab.packet.MacAddress; 10 import org.onlab.packet.MacAddress;
11 11
12 +import com.google.common.base.MoreObjects;
13 +
12 /** 14 /**
13 * Represents address information bound to a port. 15 * Represents address information bound to a port.
14 */ 16 */
...@@ -83,4 +85,13 @@ public class PortAddresses { ...@@ -83,4 +85,13 @@ public class PortAddresses {
83 public int hashCode() { 85 public int hashCode() {
84 return Objects.hash(connectPoint, ipAddresses, macAddress); 86 return Objects.hash(connectPoint, ipAddresses, macAddress);
85 } 87 }
88 +
89 + @Override
90 + public String toString() {
91 + return MoreObjects.toStringHelper(getClass())
92 + .add("connect-point", connectPoint)
93 + .add("ip-addresses", ipAddresses)
94 + .add("mac-address", macAddress)
95 + .toString();
96 + }
86 } 97 }
......
1 package org.onlab.onos.net.link; 1 package org.onlab.onos.net.link;
2 2
3 +import java.util.Set;
4 +
3 import org.onlab.onos.net.ConnectPoint; 5 import org.onlab.onos.net.ConnectPoint;
4 import org.onlab.onos.net.DeviceId; 6 import org.onlab.onos.net.DeviceId;
5 import org.onlab.onos.net.Link; 7 import org.onlab.onos.net.Link;
6 8
7 -import java.util.Set;
8 -
9 /** 9 /**
10 * Service for interacting with the inventory of infrastructure links. 10 * Service for interacting with the inventory of infrastructure links.
11 */ 11 */
......
...@@ -21,9 +21,16 @@ public interface ProxyArpService { ...@@ -21,9 +21,16 @@ public interface ProxyArpService {
21 * Sends a reply for a given request. If the host is not known then the arp 21 * Sends a reply for a given request. If the host is not known then the arp
22 * will be flooded at all edge ports. 22 * will be flooded at all edge ports.
23 * 23 *
24 - * @param request 24 + * @param eth
25 * an arp request 25 * an arp request
26 */ 26 */
27 - void reply(Ethernet request); 27 + void reply(Ethernet eth);
28 +
29 + /**
30 + * Forwards an ARP request to its destination. Floods at the edge the ARP request if the
31 + * destination is not known.
32 + * @param eth an ethernet frame containing an ARP request.
33 + */
34 + void forward(Ethernet eth);
28 35
29 } 36 }
......
1 package org.onlab.onos.net.link; 1 package org.onlab.onos.net.link;
2 2
3 +import java.util.Set;
4 +
3 import org.onlab.onos.net.ConnectPoint; 5 import org.onlab.onos.net.ConnectPoint;
4 import org.onlab.onos.net.DeviceId; 6 import org.onlab.onos.net.DeviceId;
5 import org.onlab.onos.net.Link; 7 import org.onlab.onos.net.Link;
6 8
7 -import java.util.Set;
8 -
9 /** 9 /**
10 * Test adapter for link service. 10 * Test adapter for link service.
11 */ 11 */
...@@ -63,4 +63,5 @@ public class LinkServiceAdapter implements LinkService { ...@@ -63,4 +63,5 @@ public class LinkServiceAdapter implements LinkService {
63 public void removeListener(LinkListener listener) { 63 public void removeListener(LinkListener listener) {
64 } 64 }
65 65
66 +
66 } 67 }
......
...@@ -53,7 +53,7 @@ public class LinkManager ...@@ -53,7 +53,7 @@ public class LinkManager
53 protected final AbstractListenerRegistry<LinkEvent, LinkListener> 53 protected final AbstractListenerRegistry<LinkEvent, LinkListener>
54 listenerRegistry = new AbstractListenerRegistry<>(); 54 listenerRegistry = new AbstractListenerRegistry<>();
55 55
56 - private LinkStoreDelegate delegate = new InternalStoreDelegate(); 56 + private final LinkStoreDelegate delegate = new InternalStoreDelegate();
57 57
58 private final DeviceListener deviceListener = new InternalDeviceListener(); 58 private final DeviceListener deviceListener = new InternalDeviceListener();
59 59
......
1 /** 1 /**
2 * Core subsystem for responding to arp requests. 2 * Core subsystem for responding to arp requests.
3 */ 3 */
4 -package org.onlab.onos.proxyarp.impl; 4 +package org.onlab.onos.net.proxyarp.impl;
......
1 -package org.onlab.onos.proxyarp.impl;
2 -
3 -import static com.google.common.base.Preconditions.checkArgument;
4 -import static com.google.common.base.Preconditions.checkNotNull;
5 -
6 -import java.nio.ByteBuffer;
7 -import java.util.Set;
8 -
9 -import org.apache.felix.scr.annotations.Reference;
10 -import org.apache.felix.scr.annotations.ReferenceCardinality;
11 -import org.onlab.onos.net.Host;
12 -import org.onlab.onos.net.flow.DefaultTrafficTreatment;
13 -import org.onlab.onos.net.flow.TrafficTreatment;
14 -import org.onlab.onos.net.host.HostService;
15 -import org.onlab.onos.net.packet.DefaultOutboundPacket;
16 -import org.onlab.onos.net.packet.PacketService;
17 -import org.onlab.onos.net.proxyarp.ProxyArpService;
18 -import org.onlab.onos.net.topology.TopologyService;
19 -import org.onlab.packet.ARP;
20 -import org.onlab.packet.Ethernet;
21 -import org.onlab.packet.IpPrefix;
22 -import org.onlab.packet.VlanId;
23 -
24 -public class ProxyArpManager implements ProxyArpService {
25 -
26 - private static final String MAC_ADDR_NULL = "Mac address cannot be null.";
27 - private static final String REQUEST_NULL = "Arp request cannot be null.";
28 - private static final String REQUEST_NOT_ARP = "Ethernet frame does not contain ARP request.";
29 - private static final String NOT_ARP_REQUEST = "ARP is not a request.";
30 -
31 - @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
32 - protected HostService hostService;
33 -
34 - @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
35 - protected PacketService packetService;
36 -
37 - @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
38 - protected TopologyService topologyService;
39 -
40 - @Override
41 - public boolean known(IpPrefix addr) {
42 - checkNotNull(MAC_ADDR_NULL, addr);
43 - Set<Host> hosts = hostService.getHostsByIp(addr);
44 - return !hosts.isEmpty();
45 - }
46 -
47 - @Override
48 - public void reply(Ethernet request) {
49 - checkNotNull(REQUEST_NULL, request);
50 - checkArgument(request.getEtherType() == Ethernet.TYPE_ARP,
51 - REQUEST_NOT_ARP);
52 - ARP arp = (ARP) request.getPayload();
53 - checkArgument(arp.getOpCode() == ARP.OP_REQUEST, NOT_ARP_REQUEST);
54 -
55 - VlanId vlan = VlanId.vlanId(request.getVlanID());
56 - Set<Host> hosts = hostService.getHostsByIp(IpPrefix.valueOf(arp
57 - .getTargetProtocolAddress()));
58 -
59 - Host h = null;
60 - for (Host host : hosts) {
61 - if (host.vlan().equals(vlan)) {
62 - h = host;
63 - break;
64 - }
65 - }
66 -
67 - if (h == null) {
68 - flood(request);
69 - return;
70 - }
71 -
72 - Ethernet arpReply = buildArpReply(h, request);
73 - // TODO: check send status with host service.
74 - TrafficTreatment.Builder builder = new DefaultTrafficTreatment.Builder();
75 - builder.setOutput(h.location().port());
76 - packetService.emit(new DefaultOutboundPacket(h.location().deviceId(),
77 - builder.build(), ByteBuffer.wrap(arpReply.serialize())));
78 - }
79 -
80 - private void flood(Ethernet request) {
81 - // TODO: flood on all edge ports.
82 - }
83 -
84 - private Ethernet buildArpReply(Host h, Ethernet request) {
85 - Ethernet eth = new Ethernet();
86 - eth.setDestinationMACAddress(request.getSourceMACAddress());
87 - eth.setSourceMACAddress(h.mac().getAddress());
88 - eth.setEtherType(Ethernet.TYPE_ARP);
89 - ARP arp = new ARP();
90 - arp.setOpCode(ARP.OP_REPLY);
91 - arp.setSenderHardwareAddress(h.mac().getAddress());
92 - arp.setTargetHardwareAddress(request.getSourceMACAddress());
93 -
94 - arp.setTargetProtocolAddress(((ARP) request.getPayload())
95 - .getSenderProtocolAddress());
96 - arp.setSenderProtocolAddress(h.ipAddresses().iterator().next().toInt());
97 - eth.setPayload(arp);
98 - return eth;
99 - }
100 -}
...@@ -33,8 +33,11 @@ import org.onlab.onos.net.device.PortDescription; ...@@ -33,8 +33,11 @@ import org.onlab.onos.net.device.PortDescription;
33 import org.onlab.onos.net.provider.AbstractProvider; 33 import org.onlab.onos.net.provider.AbstractProvider;
34 import org.onlab.onos.net.provider.ProviderId; 34 import org.onlab.onos.net.provider.ProviderId;
35 import org.onlab.onos.store.common.StoreManager; 35 import org.onlab.onos.store.common.StoreManager;
36 +import org.onlab.onos.store.common.StoreService;
36 import org.onlab.onos.store.common.TestStoreManager; 37 import org.onlab.onos.store.common.TestStoreManager;
37 import org.onlab.onos.store.device.impl.DistributedDeviceStore; 38 import org.onlab.onos.store.device.impl.DistributedDeviceStore;
39 +import org.onlab.onos.store.serializers.KryoSerializationManager;
40 +import org.onlab.onos.store.serializers.KryoSerializationService;
38 import org.onlab.packet.IpPrefix; 41 import org.onlab.packet.IpPrefix;
39 42
40 import java.util.ArrayList; 43 import java.util.ArrayList;
...@@ -92,6 +95,7 @@ public class DistributedDeviceManagerTest { ...@@ -92,6 +95,7 @@ public class DistributedDeviceManagerTest {
92 private DistributedDeviceStore dstore; 95 private DistributedDeviceStore dstore;
93 private TestMastershipManager masterManager; 96 private TestMastershipManager masterManager;
94 private EventDeliveryService eventService; 97 private EventDeliveryService eventService;
98 + private KryoSerializationManager serializationMgr;
95 99
96 @Before 100 @Before
97 public void setUp() { 101 public void setUp() {
...@@ -107,7 +111,10 @@ public class DistributedDeviceManagerTest { ...@@ -107,7 +111,10 @@ public class DistributedDeviceManagerTest {
107 storeManager = new TestStoreManager(Hazelcast.newHazelcastInstance(config)); 111 storeManager = new TestStoreManager(Hazelcast.newHazelcastInstance(config));
108 storeManager.activate(); 112 storeManager.activate();
109 113
110 - dstore = new TestDistributedDeviceStore(); 114 + serializationMgr = new KryoSerializationManager();
115 + serializationMgr.activate();
116 +
117 + dstore = new TestDistributedDeviceStore(storeManager, serializationMgr);
111 dstore.activate(); 118 dstore.activate();
112 119
113 mgr.store = dstore; 120 mgr.store = dstore;
...@@ -133,6 +140,7 @@ public class DistributedDeviceManagerTest { ...@@ -133,6 +140,7 @@ public class DistributedDeviceManagerTest {
133 mgr.deactivate(); 140 mgr.deactivate();
134 141
135 dstore.deactivate(); 142 dstore.deactivate();
143 + serializationMgr.deactivate();
136 storeManager.deactivate(); 144 storeManager.deactivate();
137 } 145 }
138 146
...@@ -298,8 +306,10 @@ public class DistributedDeviceManagerTest { ...@@ -298,8 +306,10 @@ public class DistributedDeviceManagerTest {
298 306
299 private class TestDistributedDeviceStore extends DistributedDeviceStore { 307 private class TestDistributedDeviceStore extends DistributedDeviceStore {
300 308
301 - public TestDistributedDeviceStore() { 309 + public TestDistributedDeviceStore(StoreService storeService,
302 - this.storeService = storeManager; 310 + KryoSerializationService kryoSerializationService) {
311 + this.storeService = storeService;
312 + this.kryoSerializationService = kryoSerializationService;
303 } 313 }
304 } 314 }
305 315
......
1 +package org.onlab.onos.store.cluster.impl;
2 +
3 +import org.onlab.nio.AcceptorLoop;
4 +import org.onlab.packet.IpPrefix;
5 +
6 +import java.io.IOException;
7 +import java.net.InetSocketAddress;
8 +import java.net.Socket;
9 +import java.nio.channels.ServerSocketChannel;
10 +import java.nio.channels.SocketChannel;
11 +
12 +import static java.net.InetAddress.getByAddress;
13 +
14 +/**
15 + * Listens to inbound connection requests and accepts them.
16 + */
17 +public class ClusterConnectionListener extends AcceptorLoop {
18 +
19 + private static final long SELECT_TIMEOUT = 50;
20 + private static final int COMM_BUFFER_SIZE = 32 * 1024;
21 +
22 + private static final boolean SO_NO_DELAY = false;
23 + private static final int SO_SEND_BUFFER_SIZE = COMM_BUFFER_SIZE;
24 + private static final int SO_RCV_BUFFER_SIZE = COMM_BUFFER_SIZE;
25 +
26 + private final WorkerFinder workerFinder;
27 +
28 + ClusterConnectionListener(IpPrefix ip, int tcpPort,
29 + WorkerFinder workerFinder) throws IOException {
30 + super(SELECT_TIMEOUT, new InetSocketAddress(getByAddress(ip.toOctets()), tcpPort));
31 + this.workerFinder = workerFinder;
32 + }
33 +
34 + @Override
35 + protected void acceptConnection(ServerSocketChannel channel) throws IOException {
36 + SocketChannel sc = channel.accept();
37 + sc.configureBlocking(false);
38 +
39 + Socket so = sc.socket();
40 + so.setTcpNoDelay(SO_NO_DELAY);
41 + so.setReceiveBufferSize(SO_RCV_BUFFER_SIZE);
42 + so.setSendBufferSize(SO_SEND_BUFFER_SIZE);
43 +
44 + workerFinder.findWorker().acceptStream(sc);
45 + }
46 +
47 +}
1 +package org.onlab.onos.store.cluster.impl;
2 +
3 +import org.onlab.nio.IOLoop;
4 +import org.onlab.nio.MessageStream;
5 +import org.onlab.onos.cluster.DefaultControllerNode;
6 +import org.onlab.onos.store.cluster.messaging.ClusterMessage;
7 +import org.onlab.onos.store.cluster.messaging.ClusterMessageStream;
8 +import org.onlab.onos.store.cluster.messaging.SerializationService;
9 +import org.slf4j.Logger;
10 +import org.slf4j.LoggerFactory;
11 +
12 +import java.io.IOException;
13 +import java.net.InetSocketAddress;
14 +import java.nio.channels.ByteChannel;
15 +import java.nio.channels.SelectionKey;
16 +import java.nio.channels.SocketChannel;
17 +import java.util.List;
18 +import java.util.Objects;
19 +
20 +import static org.onlab.packet.IpPrefix.valueOf;
21 +
22 +/**
23 + * Performs the IO operations related to a cluster-wide communications.
24 + */
25 +public class ClusterIOWorker extends
26 + IOLoop<ClusterMessage, ClusterMessageStream> {
27 +
28 + private final Logger log = LoggerFactory.getLogger(getClass());
29 +
30 + private static final long SELECT_TIMEOUT = 50;
31 +
32 + private final ConnectionManager connectionManager;
33 + private final CommunicationsDelegate commsDelegate;
34 + private final SerializationService serializationService;
35 + private final ClusterMessage helloMessage;
36 +
37 + /**
38 + * Creates a new cluster IO worker.
39 + *
40 + * @param connectionManager parent connection manager
41 + * @param commsDelegate communications delegate for dispatching
42 + * @param serializationService serialization service for encode/decode
43 + * @param helloMessage hello message for greeting peers
44 + * @throws IOException if errors occur during IO loop ignition
45 + */
46 + ClusterIOWorker(ConnectionManager connectionManager,
47 + CommunicationsDelegate commsDelegate,
48 + SerializationService serializationService,
49 + ClusterMessage helloMessage) throws IOException {
50 + super(SELECT_TIMEOUT);
51 + this.connectionManager = connectionManager;
52 + this.commsDelegate = commsDelegate;
53 + this.serializationService = serializationService;
54 + this.helloMessage = helloMessage;
55 + }
56 +
57 + @Override
58 + protected ClusterMessageStream createStream(ByteChannel byteChannel) {
59 + return new ClusterMessageStream(serializationService, this, byteChannel);
60 + }
61 +
62 + @Override
63 + protected void processMessages(List<ClusterMessage> messages, MessageStream<ClusterMessage> stream) {
64 + for (ClusterMessage message : messages) {
65 + commsDelegate.dispatch(message);
66 + }
67 + }
68 +
69 + @Override
70 + public ClusterMessageStream acceptStream(SocketChannel channel) {
71 + ClusterMessageStream stream = super.acceptStream(channel);
72 + try {
73 + InetSocketAddress sa = (InetSocketAddress) channel.getRemoteAddress();
74 + log.info("Accepted connection from node {}", valueOf(sa.getAddress().getAddress()));
75 + stream.write(helloMessage);
76 +
77 + } catch (IOException e) {
78 + log.warn("Unable to accept connection from an unknown end-point", e);
79 + }
80 + return stream;
81 + }
82 +
83 + @Override
84 + protected void connect(SelectionKey key) throws IOException {
85 + try {
86 + super.connect(key);
87 + ClusterMessageStream stream = (ClusterMessageStream) key.attachment();
88 + stream.write(helloMessage);
89 +
90 + } catch (IOException e) {
91 + if (!Objects.equals(e.getMessage(), "Connection refused")) {
92 + throw e;
93 + }
94 + }
95 + }
96 +
97 + @Override
98 + protected void removeStream(MessageStream<ClusterMessage> stream) {
99 + DefaultControllerNode node = ((ClusterMessageStream) stream).node();
100 + if (node != null) {
101 + log.info("Closed connection to node {}", node.id());
102 + connectionManager.removeNodeStream(node);
103 + }
104 + super.removeStream(stream);
105 + }
106 +
107 +}
1 +package org.onlab.onos.store.cluster.impl;
2 +
3 +import org.onlab.onos.cluster.DefaultControllerNode;
4 +
5 +/**
6 + * Simple back interface through which connection manager can interact with
7 + * the cluster store.
8 + */
9 +public interface ClusterNodesDelegate {
10 +
11 + /**
12 + * Notifies about a new cluster node being detected.
13 + *
14 + * @param node newly detected cluster node
15 + */
16 + void nodeDetected(DefaultControllerNode node);
17 +
18 + /**
19 + * Notifies about cluster node going offline.
20 + *
21 + * @param node cluster node that vanished
22 + */
23 + void nodeVanished(DefaultControllerNode node);
24 +
25 +}
1 +package org.onlab.onos.store.cluster.impl;
2 +
3 +import org.onlab.onos.store.cluster.messaging.ClusterMessage;
4 +
5 +/**
6 + * Simple back interface for interacting with the communications service.
7 + */
8 +public interface CommunicationsDelegate {
9 +
10 + /**
11 + * Dispatches the specified message to all registered subscribers.
12 + *
13 + * @param message message to be dispatched
14 + */
15 + void dispatch(ClusterMessage message);
16 +
17 + /**
18 + * Sets the sender.
19 + *
20 + * @param messageSender message sender
21 + */
22 + void setSender(MessageSender messageSender);
23 +
24 +}
1 +package org.onlab.onos.store.cluster.impl;
2 +
3 +import org.onlab.onos.cluster.DefaultControllerNode;
4 +import org.onlab.onos.cluster.NodeId;
5 +import org.onlab.onos.store.cluster.messaging.ClusterMessage;
6 +import org.onlab.onos.store.cluster.messaging.ClusterMessageStream;
7 +import org.onlab.onos.store.cluster.messaging.HelloMessage;
8 +import org.onlab.onos.store.cluster.messaging.SerializationService;
9 +import org.slf4j.Logger;
10 +import org.slf4j.LoggerFactory;
11 +
12 +import java.io.IOException;
13 +import java.net.InetSocketAddress;
14 +import java.net.SocketAddress;
15 +import java.nio.channels.SocketChannel;
16 +import java.util.ArrayList;
17 +import java.util.HashSet;
18 +import java.util.List;
19 +import java.util.Map;
20 +import java.util.Set;
21 +import java.util.Timer;
22 +import java.util.TimerTask;
23 +import java.util.concurrent.ConcurrentHashMap;
24 +import java.util.concurrent.ExecutorService;
25 +import java.util.concurrent.Executors;
26 +
27 +import static java.net.InetAddress.getByAddress;
28 +import static org.onlab.util.Tools.namedThreads;
29 +
30 +/**
31 + * Manages connections to other controller cluster nodes.
32 + */
33 +public class ConnectionManager implements MessageSender {
34 +
35 + private final Logger log = LoggerFactory.getLogger(getClass());
36 +
37 + private static final long CONNECTION_CUSTODIAN_DELAY = 1000L;
38 + private static final long CONNECTION_CUSTODIAN_FREQUENCY = 5000;
39 +
40 + private static final long START_TIMEOUT = 1000;
41 + private static final int WORKERS = 3;
42 +
43 + private ClusterConnectionListener connectionListener;
44 + private List<ClusterIOWorker> workers = new ArrayList<>(WORKERS);
45 +
46 + private final DefaultControllerNode localNode;
47 + private final ClusterNodesDelegate nodesDelegate;
48 + private final CommunicationsDelegate commsDelegate;
49 + private final SerializationService serializationService;
50 +
51 + // Nodes to be monitored to make sure they have a connection.
52 + private final Set<DefaultControllerNode> nodes = new HashSet<>();
53 +
54 + // Means to track message streams to other nodes.
55 + private final Map<NodeId, ClusterMessageStream> streams = new ConcurrentHashMap<>();
56 +
57 + // Executor pools for listening and managing connections to other nodes.
58 + private final ExecutorService listenExecutor =
59 + Executors.newSingleThreadExecutor(namedThreads("onos-comm-listen"));
60 + private final ExecutorService commExecutors =
61 + Executors.newFixedThreadPool(WORKERS, namedThreads("onos-comm-cluster"));
62 + private final ExecutorService heartbeatExecutor =
63 + Executors.newSingleThreadExecutor(namedThreads("onos-comm-heartbeat"));
64 +
65 + private final Timer timer = new Timer("onos-comm-initiator");
66 + private final TimerTask connectionCustodian = new ConnectionCustodian();
67 +
68 + private final WorkerFinder workerFinder = new LeastUtilitiedWorkerFinder();
69 +
70 +
71 + /**
72 + * Creates a new connection manager.
73 + */
74 + ConnectionManager(DefaultControllerNode localNode,
75 + ClusterNodesDelegate nodesDelegate,
76 + CommunicationsDelegate commsDelegate,
77 + SerializationService serializationService) {
78 + this.localNode = localNode;
79 + this.nodesDelegate = nodesDelegate;
80 + this.commsDelegate = commsDelegate;
81 + this.serializationService = serializationService;
82 +
83 + commsDelegate.setSender(this);
84 + startCommunications();
85 + startListening();
86 + startInitiating();
87 + log.info("Started");
88 + }
89 +
90 + /**
91 + * Shuts down the connection manager.
92 + */
93 + void shutdown() {
94 + connectionListener.shutdown();
95 + for (ClusterIOWorker worker : workers) {
96 + worker.shutdown();
97 + }
98 + log.info("Stopped");
99 + }
100 +
101 + /**
102 + * Adds the node to the list of monitored nodes.
103 + *
104 + * @param node node to be added
105 + */
106 + void addNode(DefaultControllerNode node) {
107 + nodes.add(node);
108 + }
109 +
110 + /**
111 + * Removes the node from the list of monitored nodes.
112 + *
113 + * @param node node to be removed
114 + */
115 + void removeNode(DefaultControllerNode node) {
116 + nodes.remove(node);
117 + ClusterMessageStream stream = streams.remove(node.id());
118 + if (stream != null) {
119 + stream.close();
120 + }
121 + }
122 +
123 + /**
124 + * Removes the stream associated with the specified node.
125 + *
126 + * @param node node whose stream to remove
127 + */
128 + void removeNodeStream(DefaultControllerNode node) {
129 + nodesDelegate.nodeVanished(node);
130 + streams.remove(node.id());
131 + }
132 +
133 + @Override
134 + public boolean send(NodeId nodeId, ClusterMessage message) {
135 + ClusterMessageStream stream = streams.get(nodeId);
136 + if (stream != null) {
137 + try {
138 + stream.write(message);
139 + return true;
140 + } catch (IOException e) {
141 + log.warn("Unable to send a message about {} to node {}",
142 + message.subject(), nodeId);
143 + }
144 + }
145 + return false;
146 + }
147 +
148 + /**
149 + * Kicks off the IO loops and waits for them to startup.
150 + */
151 + private void startCommunications() {
152 + HelloMessage hello = new HelloMessage(localNode.id(), localNode.ip(),
153 + localNode.tcpPort());
154 + for (int i = 0; i < WORKERS; i++) {
155 + try {
156 + ClusterIOWorker worker =
157 + new ClusterIOWorker(this, commsDelegate,
158 + serializationService, hello);
159 + workers.add(worker);
160 + commExecutors.execute(worker);
161 + } catch (IOException e) {
162 + log.warn("Unable to start communication worker", e);
163 + }
164 + }
165 +
166 + // Wait for the IO loops to start
167 + for (ClusterIOWorker loop : workers) {
168 + if (!loop.awaitStart(START_TIMEOUT)) {
169 + log.warn("Comm loop did not start on-time; moving on...");
170 + }
171 + }
172 + }
173 +
174 + /**
175 + * Starts listening for connections from peer cluster members.
176 + */
177 + private void startListening() {
178 + try {
179 + connectionListener =
180 + new ClusterConnectionListener(localNode.ip(), localNode.tcpPort(),
181 + workerFinder);
182 + listenExecutor.execute(connectionListener);
183 + if (!connectionListener.awaitStart(START_TIMEOUT)) {
184 + log.warn("Listener did not start on-time; moving on...");
185 + }
186 + } catch (IOException e) {
187 + log.error("Unable to listen for cluster connections", e);
188 + }
189 + }
190 +
191 + /**
192 + * Initiates open connection request and registers the pending socket
193 + * channel with the given IO loop.
194 + *
195 + * @param loop loop with which the channel should be registered
196 + * @throws java.io.IOException if the socket could not be open or connected
197 + */
198 + private void initiateConnection(DefaultControllerNode node,
199 + ClusterIOWorker loop) throws IOException {
200 + SocketAddress sa = new InetSocketAddress(getByAddress(node.ip().toOctets()), node.tcpPort());
201 + SocketChannel ch = SocketChannel.open();
202 + ch.configureBlocking(false);
203 + ch.connect(sa);
204 + loop.connectStream(ch);
205 + }
206 +
207 +
208 + /**
209 + * Attempts to connect to any nodes that do not have an associated connection.
210 + */
211 + private void startInitiating() {
212 + timer.schedule(connectionCustodian, CONNECTION_CUSTODIAN_DELAY,
213 + CONNECTION_CUSTODIAN_FREQUENCY);
214 + }
215 +
216 + // Sweeps through all controller nodes and attempts to open connection to
217 + // those that presently do not have one.
218 + private class ConnectionCustodian extends TimerTask {
219 + @Override
220 + public void run() {
221 + for (DefaultControllerNode node : nodes) {
222 + if (node != localNode && !streams.containsKey(node.id())) {
223 + try {
224 + initiateConnection(node, workerFinder.findWorker());
225 + } catch (IOException e) {
226 + log.debug("Unable to connect", e);
227 + }
228 + }
229 + }
230 + }
231 + }
232 +
233 + // Finds the least utilitied IO loop.
234 + private class LeastUtilitiedWorkerFinder implements WorkerFinder {
235 +
236 + @Override
237 + public ClusterIOWorker findWorker() {
238 + ClusterIOWorker leastUtilized = null;
239 + int minCount = Integer.MAX_VALUE;
240 + for (ClusterIOWorker worker : workers) {
241 + int count = worker.streamCount();
242 + if (count == 0) {
243 + return worker;
244 + }
245 +
246 + if (count < minCount) {
247 + leastUtilized = worker;
248 + minCount = count;
249 + }
250 + }
251 + return leastUtilized;
252 + }
253 + }
254 +
255 +}
1 +package org.onlab.onos.store.cluster.impl;
2 +
3 +import org.onlab.onos.cluster.NodeId;
4 +import org.onlab.onos.store.cluster.messaging.ClusterMessage;
5 +
6 +/**
7 + * Created by tom on 9/29/14.
8 + */
9 +public interface MessageSender {
10 +
11 + /**
12 + * Sends the specified message to the given cluster node.
13 + *
14 + * @param nodeId node identifier
15 + * @param message mesage to send
16 + * @return true if the message was sent sucessfully; false if there is
17 + * no stream or if there was an error
18 + */
19 + boolean send(NodeId nodeId, ClusterMessage message);
20 +
21 +}
1 -package org.onlab.onos.store.cluster.impl;
2 -
3 -import org.onlab.nio.AbstractMessage;
4 -
5 -import java.util.Objects;
6 -
7 -import static com.google.common.base.MoreObjects.toStringHelper;
8 -
9 -/**
10 - * Base message for cluster-wide communications using TLVs.
11 - */
12 -public class TLVMessage extends AbstractMessage {
13 -
14 - private final int type;
15 - private final byte[] data;
16 -
17 - /**
18 - * Creates an immutable TLV message.
19 - *
20 - * @param type message type
21 - * @param data message data bytes
22 - */
23 - public TLVMessage(int type, byte[] data) {
24 - this.length = data.length + TLVMessageStream.METADATA_LENGTH;
25 - this.type = type;
26 - this.data = data;
27 - }
28 -
29 - /**
30 - * Returns the message type indicator.
31 - *
32 - * @return message type
33 - */
34 - public int type() {
35 - return type;
36 - }
37 -
38 - /**
39 - * Returns the data bytes.
40 - *
41 - * @return message data
42 - */
43 - public byte[] data() {
44 - return data;
45 - }
46 -
47 - @Override
48 - public int hashCode() {
49 - return Objects.hash(type, data);
50 - }
51 -
52 - @Override
53 - public boolean equals(Object obj) {
54 - if (this == obj) {
55 - return true;
56 - }
57 - if (obj == null || getClass() != obj.getClass()) {
58 - return false;
59 - }
60 - final TLVMessage other = (TLVMessage) obj;
61 - return Objects.equals(this.type, other.type) &&
62 - Objects.equals(this.data, other.data);
63 - }
64 -
65 - @Override
66 - public String toString() {
67 - return toStringHelper(this).add("type", type).add("length", length).toString();
68 - }
69 -
70 -}
1 +package org.onlab.onos.store.cluster.impl;
2 +
3 +/**
4 + * Provides means to find a worker IO loop.
5 + */
6 +public interface WorkerFinder {
7 +
8 + /**
9 + * Finds a suitable worker.
10 + *
11 + * @return available worker
12 + */
13 + ClusterIOWorker findWorker();
14 +}
1 +package org.onlab.onos.store.cluster.messaging;
2 +
3 +import org.onlab.onos.cluster.NodeId;
4 +
5 +import java.util.Set;
6 +
7 +/**
8 + * Service for assisting communications between controller cluster nodes.
9 + */
10 +public interface ClusterCommunicationService {
11 +
12 + /**
13 + * Sends a message to the specified controller node.
14 + *
15 + * @param message message to send
16 + * @param toNodeId node identifier
17 + * @return true if the message was sent sucessfully; false if there is
18 + * no stream or if there was an error
19 + */
20 + boolean send(ClusterMessage message, NodeId toNodeId);
21 +
22 + /**
23 + * Adds a new subscriber for the specified message subject.
24 + *
25 + * @param subject message subject
26 + * @param subscriber message subscriber
27 + */
28 + void addSubscriber(MessageSubject subject, MessageSubscriber subscriber);
29 +
30 + /**
31 + * Removes the specified subscriber from the given message subject.
32 + *
33 + * @param subject message subject
34 + * @param subscriber message subscriber
35 + */
36 + void removeSubscriber(MessageSubject subject, MessageSubscriber subscriber);
37 +
38 + /**
39 + * Returns the set of subscribers for the specified message subject.
40 + *
41 + * @param subject message subject
42 + * @return set of message subscribers
43 + */
44 + Set<MessageSubscriber> getSubscribers(MessageSubject subject);
45 +
46 +}
1 +package org.onlab.onos.store.cluster.messaging;
2 +
3 +import org.onlab.nio.AbstractMessage;
4 +
5 +import static com.google.common.base.MoreObjects.toStringHelper;
6 +
7 +/**
8 + * Base message for cluster-wide communications.
9 + */
10 +public abstract class ClusterMessage extends AbstractMessage {
11 +
12 + private final MessageSubject subject;
13 +
14 + /**
15 + * Creates a cluster message.
16 + *
17 + * @param subject message subject
18 + */
19 + protected ClusterMessage(MessageSubject subject) {
20 + this.subject = subject;
21 + }
22 +
23 + /**
24 + * Returns the message subject indicator.
25 + *
26 + * @return message subject
27 + */
28 + public MessageSubject subject() {
29 + return subject;
30 + }
31 +
32 + @Override
33 + public String toString() {
34 + return toStringHelper(this).add("subject", subject).add("length", length).toString();
35 + }
36 +
37 +}
1 -package org.onlab.onos.store.cluster.impl; 1 +package org.onlab.onos.store.cluster.messaging;
2 2
3 import org.onlab.nio.IOLoop; 3 import org.onlab.nio.IOLoop;
4 import org.onlab.nio.MessageStream; 4 import org.onlab.nio.MessageStream;
...@@ -10,29 +10,29 @@ import java.nio.channels.ByteChannel; ...@@ -10,29 +10,29 @@ import java.nio.channels.ByteChannel;
10 import static com.google.common.base.Preconditions.checkState; 10 import static com.google.common.base.Preconditions.checkState;
11 11
12 /** 12 /**
13 - * Stream for transferring TLV messages between cluster members. 13 + * Stream for transferring messages between two cluster members.
14 */ 14 */
15 -public class TLVMessageStream extends MessageStream<TLVMessage> { 15 +public class ClusterMessageStream extends MessageStream<ClusterMessage> {
16 16
17 - public static final int METADATA_LENGTH = 16; // 8 + 4 + 4 17 + private static final int COMM_BUFFER_SIZE = 32 * 1024;
18 - 18 + private static final int COMM_IDLE_TIME = 500;
19 - private static final int LENGTH_OFFSET = 12;
20 - private static final long MARKER = 0xfeedcafecafefeedL;
21 19
22 private DefaultControllerNode node; 20 private DefaultControllerNode node;
21 + private SerializationService serializationService;
23 22
24 /** 23 /**
25 * Creates a message stream associated with the specified IO loop and 24 * Creates a message stream associated with the specified IO loop and
26 * backed by the given byte channel. 25 * backed by the given byte channel.
27 * 26 *
28 - * @param loop IO loop 27 + * @param serializationService service for encoding/decoding messages
29 - * @param byteChannel backing byte channel 28 + * @param loop IO loop
30 - * @param bufferSize size of the backing byte buffers 29 + * @param byteChannel backing byte channel
31 - * @param maxIdleMillis maximum number of millis the stream can be idle
32 */ 30 */
33 - protected TLVMessageStream(IOLoop<TLVMessage, ?> loop, ByteChannel byteChannel, 31 + public ClusterMessageStream(SerializationService serializationService,
34 - int bufferSize, int maxIdleMillis) { 32 + IOLoop<ClusterMessage, ?> loop,
35 - super(loop, byteChannel, bufferSize, maxIdleMillis); 33 + ByteChannel byteChannel) {
34 + super(loop, byteChannel, COMM_BUFFER_SIZE, COMM_IDLE_TIME);
35 + this.serializationService = serializationService;
36 } 36 }
37 37
38 /** 38 /**
...@@ -40,7 +40,7 @@ public class TLVMessageStream extends MessageStream<TLVMessage> { ...@@ -40,7 +40,7 @@ public class TLVMessageStream extends MessageStream<TLVMessage> {
40 * 40 *
41 * @return controller node 41 * @return controller node
42 */ 42 */
43 - DefaultControllerNode node() { 43 + public DefaultControllerNode node() {
44 return node; 44 return node;
45 } 45 }
46 46
...@@ -49,47 +49,19 @@ public class TLVMessageStream extends MessageStream<TLVMessage> { ...@@ -49,47 +49,19 @@ public class TLVMessageStream extends MessageStream<TLVMessage> {
49 * 49 *
50 * @param node controller node 50 * @param node controller node
51 */ 51 */
52 - void setNode(DefaultControllerNode node) { 52 + public void setNode(DefaultControllerNode node) {
53 checkState(this.node == null, "Stream is already bound to a node"); 53 checkState(this.node == null, "Stream is already bound to a node");
54 this.node = node; 54 this.node = node;
55 } 55 }
56 56
57 @Override 57 @Override
58 - protected TLVMessage read(ByteBuffer buffer) { 58 + protected ClusterMessage read(ByteBuffer buffer) {
59 - // Do we have enough bytes to read the header? If not, bail. 59 + return serializationService.decode(buffer);
60 - if (buffer.remaining() < METADATA_LENGTH) {
61 - return null;
62 - }
63 -
64 - // Peek at the length and if we have enough to read the entire message
65 - // go ahead, otherwise bail.
66 - int length = buffer.getInt(buffer.position() + LENGTH_OFFSET);
67 - if (buffer.remaining() < length) {
68 - return null;
69 - }
70 -
71 - // At this point, we have enough data to read a complete message.
72 - long marker = buffer.getLong();
73 - checkState(marker == MARKER, "Incorrect message marker");
74 -
75 - int type = buffer.getInt();
76 - length = buffer.getInt();
77 -
78 - // TODO: add deserialization hook here
79 - byte[] data = new byte[length - METADATA_LENGTH];
80 - buffer.get(data);
81 -
82 - return new TLVMessage(type, data);
83 } 60 }
84 61
85 @Override 62 @Override
86 - protected void write(TLVMessage message, ByteBuffer buffer) { 63 + protected void write(ClusterMessage message, ByteBuffer buffer) {
87 - buffer.putLong(MARKER); 64 + serializationService.encode(message, buffer);
88 - buffer.putInt(message.type());
89 - buffer.putInt(message.length());
90 -
91 - // TODO: add serialization hook here
92 - buffer.put(message.data());
93 } 65 }
94 66
95 } 67 }
......
1 +package org.onlab.onos.store.cluster.messaging;
2 +
3 +import org.onlab.onos.cluster.NodeId;
4 +
5 +/**l
6 + * Echo heart-beat message that nodes send to each other.
7 + */
8 +public class EchoMessage extends ClusterMessage {
9 +
10 + private NodeId nodeId;
11 +
12 + // For serialization
13 + private EchoMessage() {
14 + super(MessageSubject.HELLO);
15 + nodeId = null;
16 + }
17 +
18 + /**
19 + * Creates a new heart-beat echo message.
20 + *
21 + * @param nodeId sending node identification
22 + */
23 + public EchoMessage(NodeId nodeId) {
24 + super(MessageSubject.HELLO);
25 + nodeId = nodeId;
26 + }
27 +
28 + /**
29 + * Returns the sending node identifer.
30 + *
31 + * @return node identifier
32 + */
33 + public NodeId nodeId() {
34 + return nodeId;
35 + }
36 +
37 +}
1 +package org.onlab.onos.store.cluster.messaging;
2 +
3 +import org.onlab.onos.cluster.NodeId;
4 +import org.onlab.packet.IpPrefix;
5 +
6 +/**
7 + * Hello message that nodes use to greet each other.
8 + */
9 +public class HelloMessage extends ClusterMessage {
10 +
11 + private NodeId nodeId;
12 + private IpPrefix ipAddress;
13 + private int tcpPort;
14 +
15 + // For serialization
16 + private HelloMessage() {
17 + super(MessageSubject.HELLO);
18 + nodeId = null;
19 + ipAddress = null;
20 + tcpPort = 0;
21 + }
22 +
23 + /**
24 + * Creates a new hello message for the specified end-point data.
25 + *
26 + * @param nodeId sending node identification
27 + * @param ipAddress sending node IP address
28 + * @param tcpPort sending node TCP port
29 + */
30 + public HelloMessage(NodeId nodeId, IpPrefix ipAddress, int tcpPort) {
31 + super(MessageSubject.HELLO);
32 + nodeId = nodeId;
33 + ipAddress = ipAddress;
34 + tcpPort = tcpPort;
35 + }
36 +
37 + /**
38 + * Returns the sending node identifer.
39 + *
40 + * @return node identifier
41 + */
42 + public NodeId nodeId() {
43 + return nodeId;
44 + }
45 +
46 + /**
47 + * Returns the sending node IP address.
48 + *
49 + * @return node IP address
50 + */
51 + public IpPrefix ipAddress() {
52 + return ipAddress;
53 + }
54 +
55 + /**
56 + * Returns the sending node TCP listen port.
57 + *
58 + * @return TCP listen port
59 + */
60 + public int tcpPort() {
61 + return tcpPort;
62 + }
63 +}
1 +package org.onlab.onos.store.cluster.messaging;
2 +
3 +/**
4 + * Representation of a message subject.
5 + */
6 +public enum MessageSubject {
7 +
8 + /** Represents a first greeting message. */
9 + HELLO,
10 +
11 + /** Signifies a heart-beat message. */
12 + ECHO
13 +
14 +}
1 +package org.onlab.onos.store.cluster.messaging;
2 +
3 +/**
4 + * Represents a message consumer.
5 + */
6 +public interface MessageSubscriber {
7 +
8 + /**
9 + * Receives the specified cluster message.
10 + *
11 + * @param message message to be received
12 + */
13 + void receive(ClusterMessage message);
14 +
15 +}
1 +package org.onlab.onos.store.cluster.messaging;
2 +
3 +import java.nio.ByteBuffer;
4 +
5 +/**
6 + * Service for serializing/deserializing intra-cluster messages.
7 + */
8 +public interface SerializationService {
9 +
10 + /**
11 + * Decodes the specified byte buffer to obtain a message within.
12 + *
13 + * @param buffer byte buffer with message(s)
14 + * @return parsed message
15 + */
16 + ClusterMessage decode(ByteBuffer buffer);
17 +
18 + /**
19 + * Encodes the specified message into the given byte buffer.
20 + *
21 + * @param message message to be encoded
22 + * @param buffer byte buffer to receive the message data
23 + */
24 + void encode(ClusterMessage message, ByteBuffer buffer);
25 +
26 +}
1 +package org.onlab.onos.store.cluster.messaging.impl;
2 +
3 +import com.google.common.collect.HashMultimap;
4 +import com.google.common.collect.ImmutableSet;
5 +import com.google.common.collect.Multimap;
6 +import org.apache.felix.scr.annotations.Component;
7 +import org.apache.felix.scr.annotations.Service;
8 +import org.onlab.onos.cluster.NodeId;
9 +import org.onlab.onos.store.cluster.impl.CommunicationsDelegate;
10 +import org.onlab.onos.store.cluster.impl.MessageSender;
11 +import org.onlab.onos.store.cluster.messaging.ClusterCommunicationService;
12 +import org.onlab.onos.store.cluster.messaging.ClusterMessage;
13 +import org.onlab.onos.store.cluster.messaging.MessageSubject;
14 +import org.onlab.onos.store.cluster.messaging.MessageSubscriber;
15 +
16 +import java.util.Set;
17 +
18 +/**
19 + * Implements the cluster communication services to use by other stores.
20 + */
21 +@Component(immediate = true)
22 +@Service
23 +public class ClusterCommunicationManager
24 + implements ClusterCommunicationService, CommunicationsDelegate {
25 +
26 + // TODO: use something different that won't require synchronization
27 + private Multimap<MessageSubject, MessageSubscriber> subscribers = HashMultimap.create();
28 + private MessageSender messageSender;
29 +
30 + @Override
31 + public boolean send(ClusterMessage message, NodeId toNodeId) {
32 + return messageSender.send(toNodeId, message);
33 + }
34 +
35 + @Override
36 + public synchronized void addSubscriber(MessageSubject subject, MessageSubscriber subscriber) {
37 + subscribers.put(subject, subscriber);
38 + }
39 +
40 + @Override
41 + public synchronized void removeSubscriber(MessageSubject subject, MessageSubscriber subscriber) {
42 + subscribers.remove(subject, subscriber);
43 + }
44 +
45 + @Override
46 + public Set<MessageSubscriber> getSubscribers(MessageSubject subject) {
47 + return ImmutableSet.copyOf(subscribers.get(subject));
48 + }
49 +
50 + @Override
51 + public void dispatch(ClusterMessage message) {
52 + Set<MessageSubscriber> set = getSubscribers(message.subject());
53 + if (set != null) {
54 + for (MessageSubscriber subscriber : set) {
55 + subscriber.receive(message);
56 + }
57 + }
58 + }
59 +
60 + @Override
61 + public void setSender(MessageSender messageSender) {
62 + this.messageSender = messageSender;
63 + }
64 +}
1 +package org.onlab.onos.store.cluster.messaging.impl;
2 +
3 +import org.onlab.onos.store.cluster.messaging.ClusterMessage;
4 +import org.onlab.onos.store.cluster.messaging.MessageSubject;
5 +import org.onlab.onos.store.cluster.messaging.SerializationService;
6 +
7 +import java.nio.ByteBuffer;
8 +
9 +import static com.google.common.base.Preconditions.checkState;
10 +
11 +/**
12 + * Factory for parsing messages sent between cluster members.
13 + */
14 +public class MessageSerializer implements SerializationService {
15 +
16 + private static final int METADATA_LENGTH = 16; // 8 + 4 + 4
17 + private static final int LENGTH_OFFSET = 12;
18 +
19 + private static final long MARKER = 0xfeedcafebeaddeadL;
20 +
21 + @Override
22 + public ClusterMessage decode(ByteBuffer buffer) {
23 + try {
24 + // Do we have enough bytes to read the header? If not, bail.
25 + if (buffer.remaining() < METADATA_LENGTH) {
26 + return null;
27 + }
28 +
29 + // Peek at the length and if we have enough to read the entire message
30 + // go ahead, otherwise bail.
31 + int length = buffer.getInt(buffer.position() + LENGTH_OFFSET);
32 + if (buffer.remaining() < length) {
33 + return null;
34 + }
35 +
36 + // At this point, we have enough data to read a complete message.
37 + long marker = buffer.getLong();
38 + checkState(marker == MARKER, "Incorrect message marker");
39 +
40 + int subjectOrdinal = buffer.getInt();
41 + MessageSubject subject = MessageSubject.values()[subjectOrdinal];
42 + length = buffer.getInt();
43 +
44 + // TODO: sanity checking for length
45 + byte[] data = new byte[length - METADATA_LENGTH];
46 + buffer.get(data);
47 +
48 + // TODO: add deserialization hook here; for now this hack
49 + return null; // actually deserialize
50 +
51 + } catch (Exception e) {
52 + // TODO: recover from exceptions by forwarding stream to next marker
53 + e.printStackTrace();
54 + }
55 + return null;
56 + }
57 +
58 + @Override
59 + public void encode(ClusterMessage message, ByteBuffer buffer) {
60 + try {
61 + int i = 0;
62 + // Type based lookup for proper encoder
63 + } catch (Exception e) {
64 + // TODO: recover from exceptions by forwarding stream to next marker
65 + e.printStackTrace();
66 + }
67 + }
68 +
69 +}
...@@ -49,6 +49,7 @@ public class DistributedClusterStore ...@@ -49,6 +49,7 @@ public class DistributedClusterStore
49 private final MembershipListener listener = new InternalMembershipListener(); 49 private final MembershipListener listener = new InternalMembershipListener();
50 private final Map<NodeId, State> states = new ConcurrentHashMap<>(); 50 private final Map<NodeId, State> states = new ConcurrentHashMap<>();
51 51
52 + @Override
52 @Activate 53 @Activate
53 public void activate() { 54 public void activate() {
54 super.activate(); 55 super.activate();
...@@ -56,7 +57,7 @@ public class DistributedClusterStore ...@@ -56,7 +57,7 @@ public class DistributedClusterStore
56 57
57 rawNodes = theInstance.getMap("nodes"); 58 rawNodes = theInstance.getMap("nodes");
58 OptionalCacheLoader<NodeId, DefaultControllerNode> nodeLoader 59 OptionalCacheLoader<NodeId, DefaultControllerNode> nodeLoader
59 - = new OptionalCacheLoader<>(storeService, rawNodes); 60 + = new OptionalCacheLoader<>(kryoSerializationService, rawNodes);
60 nodes = new AbsentInvalidatingLoadingCache<>(newBuilder().build(nodeLoader)); 61 nodes = new AbsentInvalidatingLoadingCache<>(newBuilder().build(nodeLoader));
61 rawNodes.addEntryListener(new RemoteCacheEventHandler<>(nodes), true); 62 rawNodes.addEntryListener(new RemoteCacheEventHandler<>(nodes), true);
62 63
......
...@@ -52,7 +52,7 @@ implements MastershipStore { ...@@ -52,7 +52,7 @@ implements MastershipStore {
52 52
53 rawMasters = theInstance.getMap("masters"); 53 rawMasters = theInstance.getMap("masters");
54 OptionalCacheLoader<DeviceId, NodeId> nodeLoader 54 OptionalCacheLoader<DeviceId, NodeId> nodeLoader
55 - = new OptionalCacheLoader<>(storeService, rawMasters); 55 + = new OptionalCacheLoader<>(kryoSerializationService, rawMasters);
56 masters = new AbsentInvalidatingLoadingCache<>(newBuilder().build(nodeLoader)); 56 masters = new AbsentInvalidatingLoadingCache<>(newBuilder().build(nodeLoader));
57 rawMasters.addEntryListener(new RemoteMasterShipEventHandler(masters), true); 57 rawMasters.addEntryListener(new RemoteMasterShipEventHandler(masters), true);
58 58
......
...@@ -15,6 +15,7 @@ import org.apache.felix.scr.annotations.ReferenceCardinality; ...@@ -15,6 +15,7 @@ import org.apache.felix.scr.annotations.ReferenceCardinality;
15 import org.onlab.onos.event.Event; 15 import org.onlab.onos.event.Event;
16 import org.onlab.onos.store.AbstractStore; 16 import org.onlab.onos.store.AbstractStore;
17 import org.onlab.onos.store.StoreDelegate; 17 import org.onlab.onos.store.StoreDelegate;
18 +import org.onlab.onos.store.serializers.KryoSerializationService;
18 import org.slf4j.Logger; 19 import org.slf4j.Logger;
19 20
20 import static com.google.common.base.Preconditions.checkNotNull; 21 import static com.google.common.base.Preconditions.checkNotNull;
...@@ -32,6 +33,9 @@ public abstract class AbstractHazelcastStore<E extends Event, D extends StoreDel ...@@ -32,6 +33,9 @@ public abstract class AbstractHazelcastStore<E extends Event, D extends StoreDel
32 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) 33 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
33 protected StoreService storeService; 34 protected StoreService storeService;
34 35
36 + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
37 + protected KryoSerializationService kryoSerializationService;
38 +
35 protected HazelcastInstance theInstance; 39 protected HazelcastInstance theInstance;
36 40
37 @Activate 41 @Activate
...@@ -46,7 +50,7 @@ public abstract class AbstractHazelcastStore<E extends Event, D extends StoreDel ...@@ -46,7 +50,7 @@ public abstract class AbstractHazelcastStore<E extends Event, D extends StoreDel
46 * @return serialized object 50 * @return serialized object
47 */ 51 */
48 protected byte[] serialize(Object obj) { 52 protected byte[] serialize(Object obj) {
49 - return storeService.serialize(obj); 53 + return kryoSerializationService.serialize(obj);
50 } 54 }
51 55
52 /** 56 /**
...@@ -57,7 +61,7 @@ public abstract class AbstractHazelcastStore<E extends Event, D extends StoreDel ...@@ -57,7 +61,7 @@ public abstract class AbstractHazelcastStore<E extends Event, D extends StoreDel
57 * @return deserialized object 61 * @return deserialized object
58 */ 62 */
59 protected <T> T deserialize(byte[] bytes) { 63 protected <T> T deserialize(byte[] bytes) {
60 - return storeService.deserialize(bytes); 64 + return kryoSerializationService.deserialize(bytes);
61 } 65 }
62 66
63 67
......
...@@ -2,6 +2,8 @@ package org.onlab.onos.store.common; ...@@ -2,6 +2,8 @@ package org.onlab.onos.store.common;
2 2
3 import static com.google.common.base.Preconditions.checkNotNull; 3 import static com.google.common.base.Preconditions.checkNotNull;
4 4
5 +import org.onlab.onos.store.serializers.KryoSerializationService;
6 +
5 import com.google.common.base.Optional; 7 import com.google.common.base.Optional;
6 import com.google.common.cache.CacheLoader; 8 import com.google.common.cache.CacheLoader;
7 import com.hazelcast.core.IMap; 9 import com.hazelcast.core.IMap;
...@@ -16,28 +18,28 @@ import com.hazelcast.core.IMap; ...@@ -16,28 +18,28 @@ import com.hazelcast.core.IMap;
16 public final class OptionalCacheLoader<K, V> extends 18 public final class OptionalCacheLoader<K, V> extends
17 CacheLoader<K, Optional<V>> { 19 CacheLoader<K, Optional<V>> {
18 20
19 - private final StoreService storeService; 21 + private final KryoSerializationService kryoSerializationService;
20 private IMap<byte[], byte[]> rawMap; 22 private IMap<byte[], byte[]> rawMap;
21 23
22 /** 24 /**
23 * Constructor. 25 * Constructor.
24 * 26 *
25 - * @param storeService to use for serialization 27 + * @param kryoSerializationService to use for serialization
26 * @param rawMap underlying IMap 28 * @param rawMap underlying IMap
27 */ 29 */
28 - public OptionalCacheLoader(StoreService storeService, IMap<byte[], byte[]> rawMap) { 30 + public OptionalCacheLoader(KryoSerializationService kryoSerializationService, IMap<byte[], byte[]> rawMap) {
29 - this.storeService = checkNotNull(storeService); 31 + this.kryoSerializationService = checkNotNull(kryoSerializationService);
30 this.rawMap = checkNotNull(rawMap); 32 this.rawMap = checkNotNull(rawMap);
31 } 33 }
32 34
33 @Override 35 @Override
34 public Optional<V> load(K key) throws Exception { 36 public Optional<V> load(K key) throws Exception {
35 - byte[] keyBytes = storeService.serialize(key); 37 + byte[] keyBytes = kryoSerializationService.serialize(key);
36 byte[] valBytes = rawMap.get(keyBytes); 38 byte[] valBytes = rawMap.get(keyBytes);
37 if (valBytes == null) { 39 if (valBytes == null) {
38 return Optional.absent(); 40 return Optional.absent();
39 } 41 }
40 - V dev = storeService.deserialize(valBytes); 42 + V dev = kryoSerializationService.deserialize(valBytes);
41 return Optional.of(dev); 43 return Optional.of(dev);
42 } 44 }
43 } 45 }
......
...@@ -5,46 +5,14 @@ import com.hazelcast.config.FileSystemXmlConfig; ...@@ -5,46 +5,14 @@ import com.hazelcast.config.FileSystemXmlConfig;
5 import com.hazelcast.core.Hazelcast; 5 import com.hazelcast.core.Hazelcast;
6 import com.hazelcast.core.HazelcastInstance; 6 import com.hazelcast.core.HazelcastInstance;
7 7
8 -import de.javakaffee.kryoserializers.URISerializer;
9 -
10 import org.apache.felix.scr.annotations.Activate; 8 import org.apache.felix.scr.annotations.Activate;
11 import org.apache.felix.scr.annotations.Component; 9 import org.apache.felix.scr.annotations.Component;
12 import org.apache.felix.scr.annotations.Deactivate; 10 import org.apache.felix.scr.annotations.Deactivate;
13 import org.apache.felix.scr.annotations.Service; 11 import org.apache.felix.scr.annotations.Service;
14 -import org.onlab.onos.cluster.ControllerNode;
15 -import org.onlab.onos.cluster.DefaultControllerNode;
16 -import org.onlab.onos.cluster.NodeId;
17 -import org.onlab.onos.net.ConnectPoint;
18 -import org.onlab.onos.net.DefaultDevice;
19 -import org.onlab.onos.net.DefaultLink;
20 -import org.onlab.onos.net.DefaultPort;
21 -import org.onlab.onos.net.Device;
22 -import org.onlab.onos.net.DeviceId;
23 -import org.onlab.onos.net.Element;
24 -import org.onlab.onos.net.Link;
25 -import org.onlab.onos.net.LinkKey;
26 -import org.onlab.onos.net.MastershipRole;
27 -import org.onlab.onos.net.Port;
28 -import org.onlab.onos.net.PortNumber;
29 -import org.onlab.onos.net.provider.ProviderId;
30 -import org.onlab.onos.store.serializers.ConnectPointSerializer;
31 -import org.onlab.onos.store.serializers.DefaultLinkSerializer;
32 -import org.onlab.onos.store.serializers.DefaultPortSerializer;
33 -import org.onlab.onos.store.serializers.DeviceIdSerializer;
34 -import org.onlab.onos.store.serializers.IpPrefixSerializer;
35 -import org.onlab.onos.store.serializers.LinkKeySerializer;
36 -import org.onlab.onos.store.serializers.NodeIdSerializer;
37 -import org.onlab.onos.store.serializers.PortNumberSerializer;
38 -import org.onlab.onos.store.serializers.ProviderIdSerializer;
39 -import org.onlab.packet.IpPrefix;
40 -import org.onlab.util.KryoPool;
41 import org.slf4j.Logger; 12 import org.slf4j.Logger;
42 import org.slf4j.LoggerFactory; 13 import org.slf4j.LoggerFactory;
43 14
44 import java.io.FileNotFoundException; 15 import java.io.FileNotFoundException;
45 -import java.net.URI;
46 -import java.util.ArrayList;
47 -import java.util.HashMap;
48 16
49 /** 17 /**
50 * Auxiliary bootstrap of distributed store. 18 * Auxiliary bootstrap of distributed store.
...@@ -58,55 +26,18 @@ public class StoreManager implements StoreService { ...@@ -58,55 +26,18 @@ public class StoreManager implements StoreService {
58 private final Logger log = LoggerFactory.getLogger(getClass()); 26 private final Logger log = LoggerFactory.getLogger(getClass());
59 27
60 protected HazelcastInstance instance; 28 protected HazelcastInstance instance;
61 - private KryoPool serializerPool;
62 -
63 29
64 @Activate 30 @Activate
65 public void activate() { 31 public void activate() {
66 try { 32 try {
67 Config config = new FileSystemXmlConfig(HAZELCAST_XML_FILE); 33 Config config = new FileSystemXmlConfig(HAZELCAST_XML_FILE);
68 instance = Hazelcast.newHazelcastInstance(config); 34 instance = Hazelcast.newHazelcastInstance(config);
69 - setupKryoPool();
70 log.info("Started"); 35 log.info("Started");
71 } catch (FileNotFoundException e) { 36 } catch (FileNotFoundException e) {
72 log.error("Unable to configure Hazelcast", e); 37 log.error("Unable to configure Hazelcast", e);
73 } 38 }
74 } 39 }
75 40
76 - /**
77 - * Sets up the common serialzers pool.
78 - */
79 - protected void setupKryoPool() {
80 - // FIXME Slice out types used in common to separate pool/namespace.
81 - serializerPool = KryoPool.newBuilder()
82 - .register(ArrayList.class,
83 - HashMap.class,
84 -
85 - ControllerNode.State.class,
86 - Device.Type.class,
87 -
88 - DefaultControllerNode.class,
89 - DefaultDevice.class,
90 - MastershipRole.class,
91 - Port.class,
92 - Element.class,
93 -
94 - Link.Type.class
95 - )
96 - .register(IpPrefix.class, new IpPrefixSerializer())
97 - .register(URI.class, new URISerializer())
98 - .register(NodeId.class, new NodeIdSerializer())
99 - .register(ProviderId.class, new ProviderIdSerializer())
100 - .register(DeviceId.class, new DeviceIdSerializer())
101 - .register(PortNumber.class, new PortNumberSerializer())
102 - .register(DefaultPort.class, new DefaultPortSerializer())
103 - .register(LinkKey.class, new LinkKeySerializer())
104 - .register(ConnectPoint.class, new ConnectPointSerializer())
105 - .register(DefaultLink.class, new DefaultLinkSerializer())
106 - .build()
107 - .populate(10);
108 - }
109 -
110 @Deactivate 41 @Deactivate
111 public void deactivate() { 42 public void deactivate() {
112 instance.shutdown(); 43 instance.shutdown();
...@@ -118,18 +49,4 @@ public class StoreManager implements StoreService { ...@@ -118,18 +49,4 @@ public class StoreManager implements StoreService {
118 return instance; 49 return instance;
119 } 50 }
120 51
121 -
122 - @Override
123 - public byte[] serialize(final Object obj) {
124 - return serializerPool.serialize(obj);
125 - }
126 -
127 - @Override
128 - public <T> T deserialize(final byte[] bytes) {
129 - if (bytes == null) {
130 - return null;
131 - }
132 - return serializerPool.deserialize(bytes);
133 - }
134 -
135 } 52 }
......
...@@ -15,22 +15,4 @@ public interface StoreService { ...@@ -15,22 +15,4 @@ public interface StoreService {
15 */ 15 */
16 HazelcastInstance getHazelcastInstance(); 16 HazelcastInstance getHazelcastInstance();
17 17
18 - /**
19 - * Serializes the specified object into bytes using one of the
20 - * pre-registered serializers.
21 - *
22 - * @param obj object to be serialized
23 - * @return serialized bytes
24 - */
25 - public byte[] serialize(final Object obj);
26 -
27 - /**
28 - * Deserializes the specified bytes into an object using one of the
29 - * pre-registered serializers.
30 - *
31 - * @param bytes bytes to be deserialized
32 - * @return deserialized object
33 - */
34 - public <T> T deserialize(final byte[] bytes);
35 -
36 } 18 }
......
...@@ -46,9 +46,8 @@ public class TestStoreManager extends StoreManager { ...@@ -46,9 +46,8 @@ public class TestStoreManager extends StoreManager {
46 this.instance = instance; 46 this.instance = instance;
47 } 47 }
48 48
49 - // Hazelcast setup removed from original code.
50 @Override 49 @Override
51 public void activate() { 50 public void activate() {
52 - setupKryoPool(); 51 + // Hazelcast setup removed from original code.
53 } 52 }
54 } 53 }
......
...@@ -87,7 +87,7 @@ public class DistributedDeviceStore ...@@ -87,7 +87,7 @@ public class DistributedDeviceStore
87 // TODO decide on Map name scheme to avoid collision 87 // TODO decide on Map name scheme to avoid collision
88 rawDevices = theInstance.getMap("devices"); 88 rawDevices = theInstance.getMap("devices");
89 final OptionalCacheLoader<DeviceId, DefaultDevice> deviceLoader 89 final OptionalCacheLoader<DeviceId, DefaultDevice> deviceLoader
90 - = new OptionalCacheLoader<>(storeService, rawDevices); 90 + = new OptionalCacheLoader<>(kryoSerializationService, rawDevices);
91 devices = new AbsentInvalidatingLoadingCache<>(newBuilder().build(deviceLoader)); 91 devices = new AbsentInvalidatingLoadingCache<>(newBuilder().build(deviceLoader));
92 // refresh/populate cache based on notification from other instance 92 // refresh/populate cache based on notification from other instance
93 devicesListener = rawDevices.addEntryListener(new RemoteDeviceEventHandler(devices), includeValue); 93 devicesListener = rawDevices.addEntryListener(new RemoteDeviceEventHandler(devices), includeValue);
...@@ -97,7 +97,7 @@ public class DistributedDeviceStore ...@@ -97,7 +97,7 @@ public class DistributedDeviceStore
97 97
98 rawDevicePorts = theInstance.getMap("devicePorts"); 98 rawDevicePorts = theInstance.getMap("devicePorts");
99 final OptionalCacheLoader<DeviceId, Map<PortNumber, Port>> devicePortLoader 99 final OptionalCacheLoader<DeviceId, Map<PortNumber, Port>> devicePortLoader
100 - = new OptionalCacheLoader<>(storeService, rawDevicePorts); 100 + = new OptionalCacheLoader<>(kryoSerializationService, rawDevicePorts);
101 devicePorts = new AbsentInvalidatingLoadingCache<>(newBuilder().build(devicePortLoader)); 101 devicePorts = new AbsentInvalidatingLoadingCache<>(newBuilder().build(devicePortLoader));
102 // refresh/populate cache based on notification from other instance 102 // refresh/populate cache based on notification from other instance
103 portsListener = rawDevicePorts.addEntryListener(new RemotePortEventHandler(devicePorts), includeValue); 103 portsListener = rawDevicePorts.addEntryListener(new RemotePortEventHandler(devicePorts), includeValue);
......
...@@ -70,7 +70,7 @@ public class DistributedLinkStore ...@@ -70,7 +70,7 @@ public class DistributedLinkStore
70 // TODO decide on Map name scheme to avoid collision 70 // TODO decide on Map name scheme to avoid collision
71 rawLinks = theInstance.getMap("links"); 71 rawLinks = theInstance.getMap("links");
72 final OptionalCacheLoader<LinkKey, DefaultLink> linkLoader 72 final OptionalCacheLoader<LinkKey, DefaultLink> linkLoader
73 - = new OptionalCacheLoader<>(storeService, rawLinks); 73 + = new OptionalCacheLoader<>(kryoSerializationService, rawLinks);
74 links = new AbsentInvalidatingLoadingCache<>(newBuilder().build(linkLoader)); 74 links = new AbsentInvalidatingLoadingCache<>(newBuilder().build(linkLoader));
75 // refresh/populate cache based on notification from other instance 75 // refresh/populate cache based on notification from other instance
76 linksListener = rawLinks.addEntryListener(new RemoteLinkEventHandler(links), includeValue); 76 linksListener = rawLinks.addEntryListener(new RemoteLinkEventHandler(links), includeValue);
......
...@@ -36,6 +36,8 @@ import org.onlab.onos.net.provider.ProviderId; ...@@ -36,6 +36,8 @@ import org.onlab.onos.net.provider.ProviderId;
36 import org.onlab.onos.store.common.StoreManager; 36 import org.onlab.onos.store.common.StoreManager;
37 import org.onlab.onos.store.common.StoreService; 37 import org.onlab.onos.store.common.StoreService;
38 import org.onlab.onos.store.common.TestStoreManager; 38 import org.onlab.onos.store.common.TestStoreManager;
39 +import org.onlab.onos.store.serializers.KryoSerializationManager;
40 +import org.onlab.onos.store.serializers.KryoSerializationService;
39 41
40 import com.google.common.collect.Iterables; 42 import com.google.common.collect.Iterables;
41 import com.google.common.collect.Sets; 43 import com.google.common.collect.Sets;
...@@ -61,6 +63,7 @@ public class DistributedDeviceStoreTest { ...@@ -61,6 +63,7 @@ public class DistributedDeviceStoreTest {
61 private static final PortNumber P3 = PortNumber.portNumber(3); 63 private static final PortNumber P3 = PortNumber.portNumber(3);
62 64
63 private DistributedDeviceStore deviceStore; 65 private DistributedDeviceStore deviceStore;
66 + private KryoSerializationManager serializationMgr;
64 67
65 private StoreManager storeManager; 68 private StoreManager storeManager;
66 69
...@@ -82,7 +85,10 @@ public class DistributedDeviceStoreTest { ...@@ -82,7 +85,10 @@ public class DistributedDeviceStoreTest {
82 storeManager = new TestStoreManager(Hazelcast.newHazelcastInstance(config)); 85 storeManager = new TestStoreManager(Hazelcast.newHazelcastInstance(config));
83 storeManager.activate(); 86 storeManager.activate();
84 87
85 - deviceStore = new TestDistributedDeviceStore(storeManager); 88 + serializationMgr = new KryoSerializationManager();
89 + serializationMgr.activate();
90 +
91 + deviceStore = new TestDistributedDeviceStore(storeManager, serializationMgr);
86 deviceStore.activate(); 92 deviceStore.activate();
87 } 93 }
88 94
...@@ -90,6 +96,8 @@ public class DistributedDeviceStoreTest { ...@@ -90,6 +96,8 @@ public class DistributedDeviceStoreTest {
90 public void tearDown() throws Exception { 96 public void tearDown() throws Exception {
91 deviceStore.deactivate(); 97 deviceStore.deactivate();
92 98
99 + serializationMgr.deactivate();
100 +
93 storeManager.deactivate(); 101 storeManager.deactivate();
94 } 102 }
95 103
...@@ -384,8 +392,10 @@ public class DistributedDeviceStoreTest { ...@@ -384,8 +392,10 @@ public class DistributedDeviceStoreTest {
384 } 392 }
385 393
386 private class TestDistributedDeviceStore extends DistributedDeviceStore { 394 private class TestDistributedDeviceStore extends DistributedDeviceStore {
387 - public TestDistributedDeviceStore(StoreService storeService) { 395 + public TestDistributedDeviceStore(StoreService storeService,
396 + KryoSerializationService kryoSerializationService) {
388 this.storeService = storeService; 397 this.storeService = storeService;
398 + this.kryoSerializationService = kryoSerializationService;
389 } 399 }
390 } 400 }
391 } 401 }
......
...@@ -30,6 +30,8 @@ import org.onlab.onos.net.provider.ProviderId; ...@@ -30,6 +30,8 @@ import org.onlab.onos.net.provider.ProviderId;
30 import org.onlab.onos.store.common.StoreManager; 30 import org.onlab.onos.store.common.StoreManager;
31 import org.onlab.onos.store.common.StoreService; 31 import org.onlab.onos.store.common.StoreService;
32 import org.onlab.onos.store.common.TestStoreManager; 32 import org.onlab.onos.store.common.TestStoreManager;
33 +import org.onlab.onos.store.serializers.KryoSerializationManager;
34 +import org.onlab.onos.store.serializers.KryoSerializationService;
33 35
34 import com.google.common.collect.Iterables; 36 import com.google.common.collect.Iterables;
35 import com.hazelcast.config.Config; 37 import com.hazelcast.config.Config;
...@@ -49,6 +51,7 @@ public class DistributedLinkStoreTest { ...@@ -49,6 +51,7 @@ public class DistributedLinkStoreTest {
49 private static final PortNumber P3 = PortNumber.portNumber(3); 51 private static final PortNumber P3 = PortNumber.portNumber(3);
50 52
51 private StoreManager storeManager; 53 private StoreManager storeManager;
54 + private KryoSerializationManager serializationMgr;
52 55
53 private DistributedLinkStore linkStore; 56 private DistributedLinkStore linkStore;
54 57
...@@ -68,13 +71,17 @@ public class DistributedLinkStoreTest { ...@@ -68,13 +71,17 @@ public class DistributedLinkStoreTest {
68 storeManager = new TestStoreManager(Hazelcast.newHazelcastInstance(config)); 71 storeManager = new TestStoreManager(Hazelcast.newHazelcastInstance(config));
69 storeManager.activate(); 72 storeManager.activate();
70 73
71 - linkStore = new TestDistributedLinkStore(storeManager); 74 + serializationMgr = new KryoSerializationManager();
75 + serializationMgr.activate();
76 +
77 + linkStore = new TestDistributedLinkStore(storeManager, serializationMgr);
72 linkStore.activate(); 78 linkStore.activate();
73 } 79 }
74 80
75 @After 81 @After
76 public void tearDown() throws Exception { 82 public void tearDown() throws Exception {
77 linkStore.deactivate(); 83 linkStore.deactivate();
84 + serializationMgr.deactivate();
78 storeManager.deactivate(); 85 storeManager.deactivate();
79 } 86 }
80 87
...@@ -354,8 +361,10 @@ public class DistributedLinkStoreTest { ...@@ -354,8 +361,10 @@ public class DistributedLinkStoreTest {
354 361
355 362
356 class TestDistributedLinkStore extends DistributedLinkStore { 363 class TestDistributedLinkStore extends DistributedLinkStore {
357 - TestDistributedLinkStore(StoreService storeService) { 364 + TestDistributedLinkStore(StoreService storeService,
365 + KryoSerializationService kryoSerializationService) {
358 this.storeService = storeService; 366 this.storeService = storeService;
367 + this.kryoSerializationService = kryoSerializationService;
359 } 368 }
360 } 369 }
361 } 370 }
......
1 +package org.onlab.onos.store.serializers;
2 +
3 +import java.net.URI;
4 +import java.util.ArrayList;
5 +import java.util.HashMap;
6 +
7 +import org.apache.felix.scr.annotations.Activate;
8 +import org.apache.felix.scr.annotations.Component;
9 +import org.apache.felix.scr.annotations.Deactivate;
10 +import org.apache.felix.scr.annotations.Service;
11 +import org.onlab.onos.cluster.ControllerNode;
12 +import org.onlab.onos.cluster.DefaultControllerNode;
13 +import org.onlab.onos.cluster.NodeId;
14 +import org.onlab.onos.net.ConnectPoint;
15 +import org.onlab.onos.net.DefaultDevice;
16 +import org.onlab.onos.net.DefaultLink;
17 +import org.onlab.onos.net.DefaultPort;
18 +import org.onlab.onos.net.Device;
19 +import org.onlab.onos.net.DeviceId;
20 +import org.onlab.onos.net.Element;
21 +import org.onlab.onos.net.Link;
22 +import org.onlab.onos.net.LinkKey;
23 +import org.onlab.onos.net.MastershipRole;
24 +import org.onlab.onos.net.Port;
25 +import org.onlab.onos.net.PortNumber;
26 +import org.onlab.onos.net.provider.ProviderId;
27 +import org.onlab.packet.IpPrefix;
28 +import org.onlab.util.KryoPool;
29 +import org.slf4j.Logger;
30 +import org.slf4j.LoggerFactory;
31 +
32 +import de.javakaffee.kryoserializers.URISerializer;
33 +
34 +/**
35 + * Serialization service using Kryo.
36 + */
37 +@Component(immediate = true)
38 +@Service
39 +public class KryoSerializationManager implements KryoSerializationService {
40 +
41 + private final Logger log = LoggerFactory.getLogger(getClass());
42 + private KryoPool serializerPool;
43 +
44 +
45 + @Activate
46 + public void activate() {
47 + setupKryoPool();
48 + log.info("Started");
49 + }
50 +
51 + @Deactivate
52 + public void deactivate() {
53 + log.info("Stopped");
54 + }
55 +
56 + /**
57 + * Sets up the common serialzers pool.
58 + */
59 + protected void setupKryoPool() {
60 + // FIXME Slice out types used in common to separate pool/namespace.
61 + serializerPool = KryoPool.newBuilder()
62 + .register(ArrayList.class,
63 + HashMap.class,
64 +
65 + ControllerNode.State.class,
66 + Device.Type.class,
67 +
68 + DefaultControllerNode.class,
69 + DefaultDevice.class,
70 + MastershipRole.class,
71 + Port.class,
72 + Element.class,
73 +
74 + Link.Type.class
75 + )
76 + .register(IpPrefix.class, new IpPrefixSerializer())
77 + .register(URI.class, new URISerializer())
78 + .register(NodeId.class, new NodeIdSerializer())
79 + .register(ProviderId.class, new ProviderIdSerializer())
80 + .register(DeviceId.class, new DeviceIdSerializer())
81 + .register(PortNumber.class, new PortNumberSerializer())
82 + .register(DefaultPort.class, new DefaultPortSerializer())
83 + .register(LinkKey.class, new LinkKeySerializer())
84 + .register(ConnectPoint.class, new ConnectPointSerializer())
85 + .register(DefaultLink.class, new DefaultLinkSerializer())
86 + .build()
87 + .populate(1);
88 + }
89 +
90 + @Override
91 + public byte[] serialize(final Object obj) {
92 + return serializerPool.serialize(obj);
93 + }
94 +
95 + @Override
96 + public <T> T deserialize(final byte[] bytes) {
97 + if (bytes == null) {
98 + return null;
99 + }
100 + return serializerPool.deserialize(bytes);
101 + }
102 +
103 +}
1 +package org.onlab.onos.store.serializers;
2 +
3 +// TODO: To be replaced with SerializationService from IOLoop activity
4 +/**
5 + * Service to serialize Objects into byte array.
6 + */
7 +public interface KryoSerializationService {
8 +
9 + /**
10 + * Serializes the specified object into bytes using one of the
11 + * pre-registered serializers.
12 + *
13 + * @param obj object to be serialized
14 + * @return serialized bytes
15 + */
16 + public byte[] serialize(final Object obj);
17 +
18 + /**
19 + * Deserializes the specified bytes into an object using one of the
20 + * pre-registered serializers.
21 + *
22 + * @param bytes bytes to be deserialized
23 + * @return deserialized object
24 + */
25 + public <T> T deserialize(final byte[] bytes);
26 +
27 +}
...@@ -19,6 +19,9 @@ ...@@ -19,6 +19,9 @@
19 <bundle>mvn:de.javakaffee/kryo-serializers/0.27</bundle> 19 <bundle>mvn:de.javakaffee/kryo-serializers/0.27</bundle>
20 20
21 <bundle>mvn:org.onlab.onos/onlab-nio/1.0.0-SNAPSHOT</bundle> 21 <bundle>mvn:org.onlab.onos/onlab-nio/1.0.0-SNAPSHOT</bundle>
22 +
23 + <bundle>mvn:org.codehaus.jackson/jackson-core-asl/1.9.13</bundle>
24 + <bundle>mvn:org.codehaus.jackson/jackson-mapper-asl/1.9.13</bundle>
22 </feature> 25 </feature>
23 26
24 <feature name="onos-thirdparty-web" version="1.0.0" 27 <feature name="onos-thirdparty-web" version="1.0.0"
...@@ -122,4 +125,10 @@ ...@@ -122,4 +125,10 @@
122 <bundle>mvn:org.onlab.onos/onos-app-foo/1.0.0-SNAPSHOT</bundle> 125 <bundle>mvn:org.onlab.onos/onos-app-foo/1.0.0-SNAPSHOT</bundle>
123 </feature> 126 </feature>
124 127
128 + <feature name="onos-app-config" version="1.0.0"
129 + description="ONOS network config reader">
130 + <feature>onos-api</feature>
131 + <bundle>mvn:org.onlab.onos/onos-app-config/1.0.0-SNAPSHOT</bundle>
132 + </feature>
133 +
125 </features> 134 </features>
......
...@@ -92,6 +92,17 @@ ...@@ -92,6 +92,17 @@
92 <version>3.3.2</version> 92 <version>3.3.2</version>
93 </dependency> 93 </dependency>
94 94
95 + <dependency>
96 + <groupId>org.codehaus.jackson</groupId>
97 + <artifactId>jackson-core-asl</artifactId>
98 + <version>1.9.13</version>
99 + </dependency>
100 + <dependency>
101 + <groupId>org.codehaus.jackson</groupId>
102 + <artifactId>jackson-mapper-asl</artifactId>
103 + <version>1.9.13</version>
104 + </dependency>
105 +
95 106
96 <!-- Web related --> 107 <!-- Web related -->
97 <dependency> 108 <dependency>
......
...@@ -106,10 +106,10 @@ public class OpenFlowPacketProvider extends AbstractProvider implements PacketPr ...@@ -106,10 +106,10 @@ public class OpenFlowPacketProvider extends AbstractProvider implements PacketPr
106 for (Instruction inst : packet.treatment().instructions()) { 106 for (Instruction inst : packet.treatment().instructions()) {
107 if (inst.type().equals(Instruction.Type.OUTPUT)) { 107 if (inst.type().equals(Instruction.Type.OUTPUT)) {
108 p = portDesc(((OutputInstruction) inst).port()); 108 p = portDesc(((OutputInstruction) inst).port());
109 - if (!sw.getPorts().contains(p)) { 109 + /*if (!sw.getPorts().contains(p)) {
110 - log.warn("Tried to write out non-existint port {}", p.getPortNo()); 110 + log.warn("Tried to write out non-existent port {}", p.getPortNo());
111 continue; 111 continue;
112 - } 112 + }*/
113 OFPacketOut po = packetOut(sw, eth, p.getPortNo()); 113 OFPacketOut po = packetOut(sw, eth, p.getPortNo());
114 sw.sendMsg(po); 114 sw.sendMsg(po);
115 } 115 }
......
...@@ -154,9 +154,9 @@ public class OpenFlowPacketProviderTest { ...@@ -154,9 +154,9 @@ public class OpenFlowPacketProviderTest {
154 assertEquals("message sent incorrectly", 0, sw.sent.size()); 154 assertEquals("message sent incorrectly", 0, sw.sent.size());
155 155
156 //to missing port 156 //to missing port
157 - OutboundPacket portFailPkt = outPacket(DID, TR_MISSING, eth); 157 + //OutboundPacket portFailPkt = outPacket(DID, TR_MISSING, eth);
158 - provider.emit(portFailPkt); 158 + //provider.emit(portFailPkt);
159 - assertEquals("extra message sent", 1, sw.sent.size()); 159 + //assertEquals("extra message sent", 1, sw.sent.size());
160 160
161 } 161 }
162 162
......
...@@ -9,5 +9,5 @@ ...@@ -9,5 +9,5 @@
9 nodes=$(env | sort | egrep "OC[0-9]+" | cut -d= -f2) 9 nodes=$(env | sort | egrep "OC[0-9]+" | cut -d= -f2)
10 10
11 onos-package 11 onos-package
12 -for node in $nodes; do printf "%s: " $node; onos-install -f $node; done 12 +for node in $nodes; do (printf "%s: %s\n" "$node" "`onos-install -f $node`")& done
13 for node in $nodes; do onos-wait-for-start $node; done 13 for node in $nodes; do onos-wait-for-start $node; done
......