홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Madan Jampani
bbc98cfc 2 parents a185d861 770218a0

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

Expand all
Inline Side-by-side
Showing 72 changed files with 1639 additions and 387 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 +}
...@@ -233,7 +233,7 @@ public class IOLoopTestClient { ...@@ -233,7 +233,7 @@ public class IOLoopTestClient {
233 } 233 }
234 234
235 @Override 235 @Override
236 - protected void connect(SelectionKey key) { 236 + protected void connect(SelectionKey key) throws IOException {
237 super.connect(key); 237 super.connect(key);
238 TestMessageStream b = (TestMessageStream) key.attachment(); 238 TestMessageStream b = (TestMessageStream) key.attachment();
239 Worker w = ((CustomIOLoop) b.loop()).worker; 239 Worker w = ((CustomIOLoop) b.loop()).worker;
......
...@@ -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.
......
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-mobility</artifactId>
15 + <packaging>bundle</packaging>
16 +
17 + <description>ONOS simple Mobility app</description>
18 +
19 +</project>
1 +package org.onlab.onos.mobility;
2 +import static org.slf4j.LoggerFactory.getLogger;
3 +
4 +import java.util.Collection;
5 +import java.util.List;
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.Reference;
11 +import org.apache.felix.scr.annotations.ReferenceCardinality;
12 +import org.onlab.onos.ApplicationId;
13 +import org.onlab.onos.net.Device;
14 +import org.onlab.onos.net.Host;
15 +import org.onlab.onos.net.device.DeviceService;
16 +import org.onlab.onos.net.flow.FlowRule;
17 +import org.onlab.onos.net.flow.FlowRuleService;
18 +import org.onlab.onos.net.flow.criteria.Criteria.EthCriterion;
19 +import org.onlab.onos.net.flow.criteria.Criterion;
20 +import org.onlab.onos.net.flow.criteria.Criterion.Type;
21 +import org.onlab.onos.net.host.HostEvent;
22 +import org.onlab.onos.net.host.HostListener;
23 +import org.onlab.onos.net.host.HostService;
24 +import org.onlab.packet.MacAddress;
25 +import org.slf4j.Logger;
26 +
27 +import com.google.common.collect.Lists;
28 +
29 +
30 +/**
31 + * Sample mobility application. Cleans up flowmods when a host moves.
32 + */
33 +@Component(immediate = true)
34 +public class HostMobility {
35 +
36 + private final Logger log = getLogger(getClass());
37 +
38 + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
39 + protected HostService hostService;
40 +
41 + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
42 + protected FlowRuleService flowRuleService;
43 +
44 + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
45 + protected DeviceService deviceService;
46 +
47 + private ApplicationId appId;
48 +
49 + @Activate
50 + public void activate() {
51 + appId = ApplicationId.getAppId();
52 + hostService.addListener(new InternalHostListener());
53 + log.info("Started with Application ID {}", appId.id());
54 + }
55 +
56 + @Deactivate
57 + public void deactivate() {
58 + flowRuleService.removeFlowRulesById(appId);
59 + log.info("Stopped");
60 + }
61 +
62 + public class InternalHostListener
63 + implements HostListener {
64 +
65 + @Override
66 + public void event(HostEvent event) {
67 + switch (event.type()) {
68 + case HOST_ADDED:
69 + case HOST_REMOVED:
70 + case HOST_UPDATED:
71 + // don't care if a host has been added, removed.
72 + break;
73 + case HOST_MOVED:
74 + log.info("Host {} has moved; cleaning up.", event.subject());
75 + cleanup(event.subject());
76 + break;
77 +
78 + default:
79 + break;
80 +
81 + }
82 +
83 + }
84 +
85 + /**
86 + * For a given host, remove any flow rule which references it's addresses.
87 + * @param host the host to clean up for
88 + */
89 + private void cleanup(Host host) {
90 + Iterable<Device> devices = deviceService.getDevices();
91 + List<FlowRule> flowRules = Lists.newLinkedList();
92 + for (Device device : devices) {
93 + flowRules.addAll(cleanupDevice(device, host));
94 + }
95 + FlowRule[] flows = new FlowRule[flowRules.size()];
96 + flows = flowRules.toArray(flows);
97 + flowRuleService.removeFlowRules(flows);
98 + }
99 +
100 + private Collection<? extends FlowRule> cleanupDevice(Device device, Host host) {
101 + List<FlowRule> flowRules = Lists.newLinkedList();
102 + MacAddress mac = host.mac();
103 + for (FlowRule rule : flowRuleService.getFlowEntries(device.id())) {
104 + for (Criterion c : rule.selector().criteria()) {
105 + if (c.type() == Type.ETH_DST || c.type() == Type.ETH_SRC) {
106 + EthCriterion eth = (EthCriterion) c;
107 + if (eth.mac().equals(mac)) {
108 + flowRules.add(rule);
109 + }
110 + }
111 + }
112 + }
113 + //TODO: handle ip cleanup
114 + return flowRules;
115 + }
116 +
117 + }
118 +
119 +}
120 +
121 +
1 +/**
2 + * Trivial application that provides simple form of reactive forwarding.
3 + */
4 +package org.onlab.onos.mobility;
...@@ -20,6 +20,8 @@ ...@@ -20,6 +20,8 @@
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>mobility</module>
24 + <module>config</module>
23 </modules> 25 </modules>
24 26
25 <properties> 27 <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 }
......
...@@ -161,7 +161,7 @@ implements FlowRuleService, FlowRuleProviderRegistry { ...@@ -161,7 +161,7 @@ implements FlowRuleService, FlowRuleProviderRegistry {
161 switch (stored.state()) { 161 switch (stored.state()) {
162 case ADDED: 162 case ADDED:
163 case PENDING_ADD: 163 case PENDING_ADD:
164 - frp.applyFlowRule(flowRule); 164 + frp.applyFlowRule(stored);
165 break; 165 break;
166 case PENDING_REMOVE: 166 case PENDING_REMOVE:
167 case REMOVED: 167 case REMOVED:
......
...@@ -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.onos.cluster.DefaultControllerNode;
4 +
5 +/**
6 + * Service for administering communications manager.
7 + */
8 +public interface ClusterCommunicationAdminService {
9 +
10 + /**
11 + * Adds the node to the list of monitored nodes.
12 + *
13 + * @param node node to be added
14 + */
15 + void addNode(DefaultControllerNode node);
16 +
17 + /**
18 + * Removes the node from the list of monitored nodes.
19 + *
20 + * @param node node to be removed
21 + */
22 + void removeNode(DefaultControllerNode node);
23 +
24 + /**
25 + * Starts-up the communications engine.
26 + *
27 + * @param localNode local controller node
28 + * @param delegate nodes delegate
29 + */
30 + void startUp(DefaultControllerNode localNode, ClusterNodesDelegate delegate);
31 +
32 + /**
33 + * Clears all nodes and streams as part of leaving the cluster.
34 + */
35 + void clearAllNodesAndStreams();
36 +}
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 ClusterCommunicationManager manager;
27 +
28 + ClusterConnectionListener(ClusterCommunicationManager manager,
29 + IpPrefix ip, int tcpPort) throws IOException {
30 + super(SELECT_TIMEOUT, new InetSocketAddress(getByAddress(ip.toOctets()), tcpPort));
31 + this.manager = manager;
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 + manager.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.cluster.NodeId;
7 +import org.onlab.onos.store.cluster.messaging.ClusterMessage;
8 +import org.onlab.onos.store.cluster.messaging.HelloMessage;
9 +import org.onlab.onos.store.cluster.messaging.SerializationService;
10 +import org.slf4j.Logger;
11 +import org.slf4j.LoggerFactory;
12 +
13 +import java.io.IOException;
14 +import java.net.InetSocketAddress;
15 +import java.nio.channels.ByteChannel;
16 +import java.nio.channels.SelectionKey;
17 +import java.nio.channels.SocketChannel;
18 +import java.util.List;
19 +import java.util.Objects;
20 +
21 +import static org.onlab.packet.IpPrefix.valueOf;
22 +
23 +/**
24 + * Performs the IO operations related to a cluster-wide communications.
25 + */
26 +public class ClusterIOWorker extends
27 + IOLoop<ClusterMessage, ClusterMessageStream> {
28 +
29 + private final Logger log = LoggerFactory.getLogger(getClass());
30 +
31 + private static final long SELECT_TIMEOUT = 50;
32 +
33 + private final ClusterCommunicationManager manager;
34 + private final SerializationService serializationService;
35 + private final ClusterMessage helloMessage;
36 +
37 + /**
38 + * Creates a new cluster IO worker.
39 + *
40 + * @param manager parent comms manager
41 + * @param serializationService serialization service for encode/decode
42 + * @param helloMessage hello message for greeting peers
43 + * @throws IOException if errors occur during IO loop ignition
44 + */
45 + ClusterIOWorker(ClusterCommunicationManager manager,
46 + SerializationService serializationService,
47 + ClusterMessage helloMessage) throws IOException {
48 + super(SELECT_TIMEOUT);
49 + this.manager = manager;
50 + this.serializationService = serializationService;
51 + this.helloMessage = helloMessage;
52 + }
53 +
54 + @Override
55 + protected ClusterMessageStream createStream(ByteChannel byteChannel) {
56 + return new ClusterMessageStream(serializationService, this, byteChannel);
57 + }
58 +
59 + @Override
60 + protected void processMessages(List<ClusterMessage> messages, MessageStream<ClusterMessage> stream) {
61 + NodeId nodeId = getNodeId(messages, (ClusterMessageStream) stream);
62 + for (ClusterMessage message : messages) {
63 + manager.dispatch(message, nodeId);
64 + }
65 + }
66 +
67 + // Retrieves the node from the stream. If one is not bound, it attempts
68 + // to bind it using the knowledge that the first message must be a hello.
69 + private NodeId getNodeId(List<ClusterMessage> messages, ClusterMessageStream stream) {
70 + DefaultControllerNode node = stream.node();
71 + if (node == null && !messages.isEmpty()) {
72 + ClusterMessage firstMessage = messages.get(0);
73 + if (firstMessage instanceof HelloMessage) {
74 + HelloMessage hello = (HelloMessage) firstMessage;
75 + node = manager.addNodeStream(hello.nodeId(), hello.ipAddress(),
76 + hello.tcpPort(), stream);
77 + }
78 + }
79 + return node != null ? node.id() : null;
80 + }
81 +
82 + @Override
83 + public ClusterMessageStream acceptStream(SocketChannel channel) {
84 + ClusterMessageStream stream = super.acceptStream(channel);
85 + try {
86 + InetSocketAddress sa = (InetSocketAddress) channel.getRemoteAddress();
87 + log.info("Accepted connection from node {}", valueOf(sa.getAddress().getAddress()));
88 + stream.write(helloMessage);
89 +
90 + } catch (IOException e) {
91 + log.warn("Unable to accept connection from an unknown end-point", e);
92 + }
93 + return stream;
94 + }
95 +
96 + @Override
97 + protected void connect(SelectionKey key) throws IOException {
98 + try {
99 + super.connect(key);
100 + ClusterMessageStream stream = (ClusterMessageStream) key.attachment();
101 + stream.write(helloMessage);
102 +
103 + } catch (IOException e) {
104 + if (!Objects.equals(e.getMessage(), "Connection refused")) {
105 + throw e;
106 + }
107 + }
108 + }
109 +
110 + @Override
111 + protected void removeStream(MessageStream<ClusterMessage> stream) {
112 + DefaultControllerNode node = ((ClusterMessageStream) stream).node();
113 + if (node != null) {
114 + log.info("Closed connection to node {}", node.id());
115 + manager.removeNodeStream(node);
116 + }
117 + super.removeStream(stream);
118 + }
119 +
120 +}
...@@ -3,6 +3,8 @@ package org.onlab.onos.store.cluster.impl; ...@@ -3,6 +3,8 @@ package org.onlab.onos.store.cluster.impl;
3 import org.onlab.nio.IOLoop; 3 import org.onlab.nio.IOLoop;
4 import org.onlab.nio.MessageStream; 4 import org.onlab.nio.MessageStream;
5 import org.onlab.onos.cluster.DefaultControllerNode; 5 import org.onlab.onos.cluster.DefaultControllerNode;
6 +import org.onlab.onos.store.cluster.messaging.ClusterMessage;
7 +import org.onlab.onos.store.cluster.messaging.SerializationService;
6 8
7 import java.nio.ByteBuffer; 9 import java.nio.ByteBuffer;
8 import java.nio.channels.ByteChannel; 10 import java.nio.channels.ByteChannel;
...@@ -10,29 +12,29 @@ import java.nio.channels.ByteChannel; ...@@ -10,29 +12,29 @@ import java.nio.channels.ByteChannel;
10 import static com.google.common.base.Preconditions.checkState; 12 import static com.google.common.base.Preconditions.checkState;
11 13
12 /** 14 /**
13 - * Stream for transferring TLV messages between cluster members. 15 + * Stream for transferring messages between two cluster members.
14 */ 16 */
15 -public class TLVMessageStream extends MessageStream<TLVMessage> { 17 +public class ClusterMessageStream extends MessageStream<ClusterMessage> {
16 18
17 - public static final int METADATA_LENGTH = 16; // 8 + 4 + 4 19 + private static final int COMM_BUFFER_SIZE = 32 * 1024;
18 - 20 + private static final int COMM_IDLE_TIME = 500;
19 - private static final int LENGTH_OFFSET = 12;
20 - private static final long MARKER = 0xfeedcafecafefeedL;
21 21
22 private DefaultControllerNode node; 22 private DefaultControllerNode node;
23 + private SerializationService serializationService;
23 24
24 /** 25 /**
25 * Creates a message stream associated with the specified IO loop and 26 * Creates a message stream associated with the specified IO loop and
26 * backed by the given byte channel. 27 * backed by the given byte channel.
27 * 28 *
28 - * @param loop IO loop 29 + * @param serializationService service for encoding/decoding messages
29 - * @param byteChannel backing byte channel 30 + * @param loop IO loop
30 - * @param bufferSize size of the backing byte buffers 31 + * @param byteChannel backing byte channel
31 - * @param maxIdleMillis maximum number of millis the stream can be idle
32 */ 32 */
33 - protected TLVMessageStream(IOLoop<TLVMessage, ?> loop, ByteChannel byteChannel, 33 + public ClusterMessageStream(SerializationService serializationService,
34 - int bufferSize, int maxIdleMillis) { 34 + IOLoop<ClusterMessage, ?> loop,
35 - super(loop, byteChannel, bufferSize, maxIdleMillis); 35 + ByteChannel byteChannel) {
36 + super(loop, byteChannel, COMM_BUFFER_SIZE, COMM_IDLE_TIME);
37 + this.serializationService = serializationService;
36 } 38 }
37 39
38 /** 40 /**
...@@ -40,7 +42,7 @@ public class TLVMessageStream extends MessageStream<TLVMessage> { ...@@ -40,7 +42,7 @@ public class TLVMessageStream extends MessageStream<TLVMessage> {
40 * 42 *
41 * @return controller node 43 * @return controller node
42 */ 44 */
43 - DefaultControllerNode node() { 45 + public DefaultControllerNode node() {
44 return node; 46 return node;
45 } 47 }
46 48
...@@ -49,47 +51,19 @@ public class TLVMessageStream extends MessageStream<TLVMessage> { ...@@ -49,47 +51,19 @@ public class TLVMessageStream extends MessageStream<TLVMessage> {
49 * 51 *
50 * @param node controller node 52 * @param node controller node
51 */ 53 */
52 - void setNode(DefaultControllerNode node) { 54 + public void setNode(DefaultControllerNode node) {
53 checkState(this.node == null, "Stream is already bound to a node"); 55 checkState(this.node == null, "Stream is already bound to a node");
54 this.node = node; 56 this.node = node;
55 } 57 }
56 58
57 @Override 59 @Override
58 - protected TLVMessage read(ByteBuffer buffer) { 60 + protected ClusterMessage read(ByteBuffer buffer) {
59 - // Do we have enough bytes to read the header? If not, bail. 61 + 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 } 62 }
84 63
85 @Override 64 @Override
86 - protected void write(TLVMessage message, ByteBuffer buffer) { 65 + protected void write(ClusterMessage message, ByteBuffer buffer) {
87 - buffer.putLong(MARKER); 66 + 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 } 67 }
94 68
95 } 69 }
......
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.packet.IpPrefix;
6 +
7 +/**
8 + * Simple back interface through which connection manager can interact with
9 + * the cluster store.
10 + */
11 +public interface ClusterNodesDelegate {
12 +
13 + /**
14 + * Notifies about cluster node coming online.
15 + *
16 + * @param nodeId newly detected cluster node id
17 + * @param ip node IP listen address
18 + * @param tcpPort node TCP listen port
19 + * @return the controller node
20 + */
21 + DefaultControllerNode nodeDetected(NodeId nodeId, IpPrefix ip, int tcpPort);
22 +
23 + /**
24 + * Notifies about cluster node going offline.
25 + *
26 + * @param nodeId identifier of the cluster node that vanished
27 + */
28 + void nodeVanished(NodeId nodeId);
29 +
30 + /**
31 + * Notifies about remote request to remove node from cluster.
32 + *
33 + * @param nodeId identifier of the cluster node that was removed
34 + */
35 + void nodeRemoved(NodeId nodeId);
36 +
37 +}
1 +package org.onlab.onos.store.cluster.impl;
2 +
3 +import de.javakaffee.kryoserializers.URISerializer;
4 +import org.apache.felix.scr.annotations.Activate;
5 +import org.apache.felix.scr.annotations.Component;
6 +import org.apache.felix.scr.annotations.Deactivate;
7 +import org.apache.felix.scr.annotations.Service;
8 +import org.onlab.onos.cluster.ControllerNode;
9 +import org.onlab.onos.cluster.DefaultControllerNode;
10 +import org.onlab.onos.cluster.NodeId;
11 +import org.onlab.onos.net.ConnectPoint;
12 +import org.onlab.onos.net.DefaultDevice;
13 +import org.onlab.onos.net.DefaultLink;
14 +import org.onlab.onos.net.DefaultPort;
15 +import org.onlab.onos.net.Device;
16 +import org.onlab.onos.net.DeviceId;
17 +import org.onlab.onos.net.Element;
18 +import org.onlab.onos.net.Link;
19 +import org.onlab.onos.net.LinkKey;
20 +import org.onlab.onos.net.MastershipRole;
21 +import org.onlab.onos.net.Port;
22 +import org.onlab.onos.net.PortNumber;
23 +import org.onlab.onos.net.provider.ProviderId;
24 +import org.onlab.onos.store.cluster.messaging.ClusterMessage;
25 +import org.onlab.onos.store.cluster.messaging.EchoMessage;
26 +import org.onlab.onos.store.cluster.messaging.LeavingMemberMessage;
27 +import org.onlab.onos.store.cluster.messaging.HelloMessage;
28 +import org.onlab.onos.store.cluster.messaging.MessageSubject;
29 +import org.onlab.onos.store.cluster.messaging.NewMemberMessage;
30 +import org.onlab.onos.store.cluster.messaging.SerializationService;
31 +import org.onlab.onos.store.serializers.ConnectPointSerializer;
32 +import org.onlab.onos.store.serializers.DefaultLinkSerializer;
33 +import org.onlab.onos.store.serializers.DefaultPortSerializer;
34 +import org.onlab.onos.store.serializers.DeviceIdSerializer;
35 +import org.onlab.onos.store.serializers.IpPrefixSerializer;
36 +import org.onlab.onos.store.serializers.LinkKeySerializer;
37 +import org.onlab.onos.store.serializers.NodeIdSerializer;
38 +import org.onlab.onos.store.serializers.PortNumberSerializer;
39 +import org.onlab.onos.store.serializers.ProviderIdSerializer;
40 +import org.onlab.packet.IpPrefix;
41 +import org.onlab.util.KryoPool;
42 +import org.slf4j.Logger;
43 +import org.slf4j.LoggerFactory;
44 +
45 +import java.net.URI;
46 +import java.nio.ByteBuffer;
47 +import java.util.ArrayList;
48 +import java.util.HashMap;
49 +
50 +import static com.google.common.base.Preconditions.checkState;
51 +
52 +/**
53 + * Factory for parsing messages sent between cluster members.
54 + */
55 +@Component(immediate = true)
56 +@Service
57 +public class MessageSerializer implements SerializationService {
58 +
59 + private final Logger log = LoggerFactory.getLogger(getClass());
60 +
61 + private static final int METADATA_LENGTH = 12; // 8 + 4
62 + private static final int LENGTH_OFFSET = 8;
63 +
64 + private static final long MARKER = 0xfeedcafebeaddeadL;
65 +
66 + private KryoPool serializerPool;
67 +
68 + @Activate
69 + public void activate() {
70 + setupKryoPool();
71 + log.info("Started");
72 + }
73 +
74 + @Deactivate
75 + public void deactivate() {
76 + log.info("Stopped");
77 + }
78 +
79 + /**
80 + * Sets up the common serialzers pool.
81 + */
82 + protected void setupKryoPool() {
83 + // FIXME Slice out types used in common to separate pool/namespace.
84 + serializerPool = KryoPool.newBuilder()
85 + .register(ArrayList.class,
86 + HashMap.class,
87 +
88 + ControllerNode.State.class,
89 + Device.Type.class,
90 +
91 + DefaultControllerNode.class,
92 + DefaultDevice.class,
93 + MastershipRole.class,
94 + Port.class,
95 + Element.class,
96 +
97 + Link.Type.class,
98 +
99 + MessageSubject.class,
100 + HelloMessage.class,
101 + NewMemberMessage.class,
102 + LeavingMemberMessage.class,
103 + EchoMessage.class
104 + )
105 + .register(IpPrefix.class, new IpPrefixSerializer())
106 + .register(URI.class, new URISerializer())
107 + .register(NodeId.class, new NodeIdSerializer())
108 + .register(ProviderId.class, new ProviderIdSerializer())
109 + .register(DeviceId.class, new DeviceIdSerializer())
110 + .register(PortNumber.class, new PortNumberSerializer())
111 + .register(DefaultPort.class, new DefaultPortSerializer())
112 + .register(LinkKey.class, new LinkKeySerializer())
113 + .register(ConnectPoint.class, new ConnectPointSerializer())
114 + .register(DefaultLink.class, new DefaultLinkSerializer())
115 + .build()
116 + .populate(1);
117 + }
118 +
119 +
120 + @Override
121 + public ClusterMessage decode(ByteBuffer buffer) {
122 + try {
123 + // Do we have enough bytes to read the header? If not, bail.
124 + if (buffer.remaining() < METADATA_LENGTH) {
125 + return null;
126 + }
127 +
128 + // Peek at the length and if we have enough to read the entire message
129 + // go ahead, otherwise bail.
130 + int length = buffer.getInt(buffer.position() + LENGTH_OFFSET);
131 + if (buffer.remaining() < length) {
132 + return null;
133 + }
134 +
135 + // At this point, we have enough data to read a complete message.
136 + long marker = buffer.getLong();
137 + checkState(marker == MARKER, "Incorrect message marker");
138 + length = buffer.getInt();
139 +
140 + // TODO: sanity checking for length
141 + byte[] data = new byte[length - METADATA_LENGTH];
142 + buffer.get(data);
143 + return (ClusterMessage) serializerPool.deserialize(data);
144 +
145 + } catch (Exception e) {
146 + // TODO: recover from exceptions by forwarding stream to next marker
147 + log.warn("Unable to decode message due to: " + e);
148 + }
149 + return null;
150 + }
151 +
152 + @Override
153 + public void encode(ClusterMessage message, ByteBuffer buffer) {
154 + try {
155 + byte[] data = serializerPool.serialize(message);
156 + buffer.putLong(MARKER);
157 + buffer.putInt(data.length + METADATA_LENGTH);
158 + buffer.put(data);
159 +
160 + } catch (Exception e) {
161 + // TODO: recover from exceptions by forwarding stream to next marker
162 + log.warn("Unable to encode message due to: " + e);
163 + }
164 + }
165 +
166 +}
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 +/**
2 + * Distributed cluster store and messaging subsystem implementation.
3 + */
4 +package org.onlab.onos.store.cluster.impl;
...\ No newline at end of file ...\ No newline at end of file
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 all controller nodes.
14 + *
15 + * @param message message to send
16 + * @return true if the message was sent sucessfully to all nodes; false
17 + * if there is no stream or if there was an error for some node
18 + */
19 + boolean send(ClusterMessage message);
20 +
21 + /**
22 + * Sends a message to the specified controller node.
23 + *
24 + * @param message message to send
25 + * @param toNodeId node identifier
26 + * @return true if the message was sent sucessfully; false if there is
27 + * no stream or if there was an error
28 + */
29 + boolean send(ClusterMessage message, NodeId toNodeId);
30 +
31 + /**
32 + * Adds a new subscriber for the specified message subject.
33 + *
34 + * @param subject message subject
35 + * @param subscriber message subscriber
36 + */
37 + void addSubscriber(MessageSubject subject, MessageSubscriber subscriber);
38 +
39 + /**
40 + * Removes the specified subscriber from the given message subject.
41 + *
42 + * @param subject message subject
43 + * @param subscriber message subscriber
44 + */
45 + void removeSubscriber(MessageSubject subject, MessageSubscriber subscriber);
46 +
47 + /**
48 + * Returns the set of subscribers for the specified message subject.
49 + *
50 + * @param subject message subject
51 + * @return set of message subscribers
52 + */
53 + Set<MessageSubscriber> getSubscribers(MessageSubject subject);
54 +
55 +}
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 + * Base for cluster membership messages.
8 + */
9 +public abstract class ClusterMembershipMessage extends ClusterMessage {
10 +
11 + private NodeId nodeId;
12 + private IpPrefix ipAddress;
13 + private int tcpPort;
14 +
15 + // For serialization
16 + protected ClusterMembershipMessage() {
17 + super(MessageSubject.HELLO);
18 + nodeId = null;
19 + ipAddress = null;
20 + tcpPort = 0;
21 + }
22 +
23 + /**
24 + * Creates a new membership message for the specified end-point data.
25 + *
26 + * @param subject message subject
27 + * @param nodeId sending node identification
28 + * @param ipAddress sending node IP address
29 + * @param tcpPort sending node TCP port
30 + */
31 + protected ClusterMembershipMessage(MessageSubject subject, NodeId nodeId,
32 + IpPrefix ipAddress, int tcpPort) {
33 + super(subject);
34 + this.nodeId = nodeId;
35 + this.ipAddress = ipAddress;
36 + this.tcpPort = tcpPort;
37 + }
38 +
39 + /**
40 + * Returns the sending node identifer.
41 + *
42 + * @return node identifier
43 + */
44 + public NodeId nodeId() {
45 + return nodeId;
46 + }
47 +
48 + /**
49 + * Returns the sending node IP address.
50 + *
51 + * @return node IP address
52 + */
53 + public IpPrefix ipAddress() {
54 + return ipAddress;
55 + }
56 +
57 + /**
58 + * Returns the sending node TCP listen port.
59 + *
60 + * @return TCP listen port
61 + */
62 + public int tcpPort() {
63 + return tcpPort;
64 + }
65 +
66 +}
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.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 ClusterMembershipMessage {
10 +
11 + // For serialization
12 + private HelloMessage() {
13 + }
14 +
15 + /**
16 + * Creates a new hello message for the specified end-point data.
17 + *
18 + * @param nodeId sending node identification
19 + * @param ipAddress sending node IP address
20 + * @param tcpPort sending node TCP port
21 + */
22 + public HelloMessage(NodeId nodeId, IpPrefix ipAddress, int tcpPort) {
23 + super(MessageSubject.HELLO, nodeId, ipAddress, tcpPort);
24 + }
25 +
26 +}
1 +package org.onlab.onos.store.cluster.messaging;
2 +
3 +import org.onlab.onos.cluster.NodeId;
4 +
5 +/**
6 + * Announcement message that nodes use to gossip about team departures.
7 + */
8 +public class LeavingMemberMessage extends ClusterMembershipMessage {
9 +
10 + // For serialization
11 + private LeavingMemberMessage() {
12 + super();
13 + }
14 +
15 + /**
16 + * Creates a new goodbye message.
17 + *
18 + * @param nodeId sending node identification
19 + */
20 + public LeavingMemberMessage(NodeId nodeId) {
21 + super(MessageSubject.LEAVING_MEMBER, nodeId, null, 0);
22 + }
23 +
24 +}
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 announcement about new member. */
12 + NEW_MEMBER,
13 +
14 + /** Signifies announcement about leaving member. */
15 + LEAVING_MEMBER,
16 +
17 + /** Signifies a heart-beat message. */
18 + ECHO
19 +
20 +}
1 +package org.onlab.onos.store.cluster.messaging;
2 +
3 +import org.onlab.onos.cluster.NodeId;
4 +
5 +/**
6 + * Represents a message consumer.
7 + */
8 +public interface MessageSubscriber {
9 +
10 + /**
11 + * Receives the specified cluster message.
12 + *
13 + * @param message message to be received
14 + * @param fromNodeId node from which the message was received
15 + */
16 + void receive(ClusterMessage message, NodeId fromNodeId);
17 +
18 +}
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 + * Announcement message that nodes use to gossip about new arrivals.
8 + */
9 +public class NewMemberMessage extends ClusterMembershipMessage {
10 +
11 + // For serialization
12 + private NewMemberMessage() {
13 + }
14 +
15 + /**
16 + * Creates a new gossip message for the specified end-point data.
17 + *
18 + * @param nodeId sending node identification
19 + * @param ipAddress sending node IP address
20 + * @param tcpPort sending node TCP port
21 + */
22 + public NewMemberMessage(NodeId nodeId, IpPrefix ipAddress, int tcpPort) {
23 + super(MessageSubject.NEW_MEMBER, nodeId, ipAddress, tcpPort);
24 + }
25 +
26 +}
1 +package org.onlab.onos.store.cluster.messaging;
2 +
3 +import java.nio.ByteBuffer;
4 +
5 +/**
6 + * Service for encoding &amp; decoding intra-cluster messages.
7 + */
8 +public interface SerializationService {
9 +
10 + /**
11 + * Decodes the specified byte buffer to obtain the 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 +/**
2 + * Cluster messaging APIs for the use by the various distributed stores.
3 + */
4 +package org.onlab.onos.store.cluster.messaging;
...\ No newline at end of file ...\ No newline at end of file
1 +package org.onlab.onos.store.cluster.impl;
2 +
3 +import org.junit.After;
4 +import org.junit.Before;
5 +import org.junit.Test;
6 +import org.onlab.onos.cluster.DefaultControllerNode;
7 +import org.onlab.onos.cluster.NodeId;
8 +import org.onlab.packet.IpPrefix;
9 +
10 +import java.util.concurrent.CountDownLatch;
11 +import java.util.concurrent.TimeUnit;
12 +
13 +import static org.junit.Assert.assertEquals;
14 +import static org.junit.Assert.assertTrue;
15 +
16 +/**
17 + * Tests of the cluster communication manager.
18 + */
19 +public class ClusterCommunicationManagerTest {
20 +
21 + private static final NodeId N1 = new NodeId("n1");
22 + private static final NodeId N2 = new NodeId("n2");
23 +
24 + private static final int P1 = 9881;
25 + private static final int P2 = 9882;
26 +
27 + private static final IpPrefix IP = IpPrefix.valueOf("127.0.0.1");
28 +
29 + private ClusterCommunicationManager ccm1;
30 + private ClusterCommunicationManager ccm2;
31 +
32 + private TestDelegate cnd1 = new TestDelegate();
33 + private TestDelegate cnd2 = new TestDelegate();
34 +
35 + private DefaultControllerNode node1 = new DefaultControllerNode(N1, IP, P1);
36 + private DefaultControllerNode node2 = new DefaultControllerNode(N2, IP, P2);
37 +
38 + @Before
39 + public void setUp() {
40 + MessageSerializer messageSerializer = new MessageSerializer();
41 + messageSerializer.activate();
42 +
43 + ccm1 = new ClusterCommunicationManager();
44 + ccm1.serializationService = messageSerializer;
45 + ccm1.activate();
46 +
47 + ccm2 = new ClusterCommunicationManager();
48 + ccm2.serializationService = messageSerializer;
49 + ccm2.activate();
50 +
51 + ccm1.startUp(node1, cnd1);
52 + ccm2.startUp(node2, cnd2);
53 + }
54 +
55 + @After
56 + public void tearDown() {
57 + ccm1.deactivate();
58 + ccm2.deactivate();
59 + }
60 +
61 + @Test
62 + public void connect() throws Exception {
63 + cnd1.latch = new CountDownLatch(1);
64 + cnd2.latch = new CountDownLatch(1);
65 +
66 + ccm1.addNode(node2);
67 + validateDelegateEvent(cnd1, Op.DETECTED, node2.id());
68 + validateDelegateEvent(cnd2, Op.DETECTED, node1.id());
69 + }
70 +
71 + @Test
72 + public void disconnect() throws Exception {
73 + cnd1.latch = new CountDownLatch(1);
74 + cnd2.latch = new CountDownLatch(1);
75 +
76 + ccm1.addNode(node2);
77 + validateDelegateEvent(cnd1, Op.DETECTED, node2.id());
78 + validateDelegateEvent(cnd2, Op.DETECTED, node1.id());
79 +
80 + cnd1.latch = new CountDownLatch(1);
81 + cnd2.latch = new CountDownLatch(1);
82 + ccm1.deactivate();
83 +//
84 +// validateDelegateEvent(cnd2, Op.VANISHED, node1.id());
85 + }
86 +
87 + private void validateDelegateEvent(TestDelegate delegate, Op op, NodeId nodeId)
88 + throws InterruptedException {
89 + assertTrue("did not connect in time", delegate.latch.await(2500, TimeUnit.MILLISECONDS));
90 + assertEquals("incorrect event", op, delegate.op);
91 + assertEquals("incorrect event node", nodeId, delegate.nodeId);
92 + }
93 +
94 + enum Op { DETECTED, VANISHED, REMOVED };
95 +
96 + private class TestDelegate implements ClusterNodesDelegate {
97 +
98 + Op op;
99 + CountDownLatch latch;
100 + NodeId nodeId;
101 +
102 + @Override
103 + public DefaultControllerNode nodeDetected(NodeId nodeId, IpPrefix ip, int tcpPort) {
104 + latch(nodeId, Op.DETECTED);
105 + return new DefaultControllerNode(nodeId, ip, tcpPort);
106 + }
107 +
108 + @Override
109 + public void nodeVanished(NodeId nodeId) {
110 + latch(nodeId, Op.VANISHED);
111 + }
112 +
113 + @Override
114 + public void nodeRemoved(NodeId nodeId) {
115 + latch(nodeId, Op.REMOVED);
116 + }
117 +
118 + private void latch(NodeId nodeId, Op op) {
119 + this.op = op;
120 + this.nodeId = nodeId;
121 + latch.countDown();
122 + }
123 + }
124 +}
...\ No newline at end of file ...\ No newline at end of file
...@@ -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 +}
...@@ -105,11 +105,9 @@ public class SimpleFlowRuleStore ...@@ -105,11 +105,9 @@ public class SimpleFlowRuleStore
105 */ 105 */
106 106
107 if (flowEntries.containsEntry(did, f)) { 107 if (flowEntries.containsEntry(did, f)) {
108 - //synchronized (flowEntries) {
109 flowEntries.remove(did, f); 108 flowEntries.remove(did, f);
110 flowEntries.put(did, f); 109 flowEntries.put(did, f);
111 flowEntriesById.remove(rule.appId(), rule); 110 flowEntriesById.remove(rule.appId(), rule);
112 - //}
113 } 111 }
114 } 112 }
115 113
......
...@@ -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"
...@@ -58,7 +61,7 @@ ...@@ -58,7 +61,7 @@
58 <bundle>mvn:org.onlab.onos/onos-core-hz-common/1.0.0-SNAPSHOT</bundle> 61 <bundle>mvn:org.onlab.onos/onos-core-hz-common/1.0.0-SNAPSHOT</bundle>
59 <bundle>mvn:org.onlab.onos/onos-core-serializers/1.0.0-SNAPSHOT</bundle> 62 <bundle>mvn:org.onlab.onos/onos-core-serializers/1.0.0-SNAPSHOT</bundle>
60 <bundle>mvn:org.onlab.onos/onos-core-hz-cluster/1.0.0-SNAPSHOT</bundle> 63 <bundle>mvn:org.onlab.onos/onos-core-hz-cluster/1.0.0-SNAPSHOT</bundle>
61 - <bundle>mvn:org.onlab.onos/onos-core-dist/1.0.0-SNAPSHOT</bundle> 64 + <bundle>mvn:org.onlab.onos/onos-core-hz-net/1.0.0-SNAPSHOT</bundle>
62 </feature> 65 </feature>
63 66
64 <feature name="onos-core-trivial" version="1.0.0" 67 <feature name="onos-core-trivial" version="1.0.0"
...@@ -116,10 +119,24 @@ ...@@ -116,10 +119,24 @@
116 <bundle>mvn:org.onlab.onos/onos-app-fwd/1.0.0-SNAPSHOT</bundle> 119 <bundle>mvn:org.onlab.onos/onos-app-fwd/1.0.0-SNAPSHOT</bundle>
117 </feature> 120 </feature>
118 121
122 + <feature name="onos-app-mobility" version="1.0.0"
123 + description="ONOS sample forwarding application">
124 + <feature>onos-api</feature>
125 + <bundle>mvn:org.onlab.onos/onos-app-mobility/1.0.0-SNAPSHOT</bundle>
126 + </feature>
127 +
128 +
129 +
119 <feature name="onos-app-foo" version="1.0.0" 130 <feature name="onos-app-foo" version="1.0.0"
120 description="ONOS sample playground application"> 131 description="ONOS sample playground application">
121 <feature>onos-api</feature> 132 <feature>onos-api</feature>
122 <bundle>mvn:org.onlab.onos/onos-app-foo/1.0.0-SNAPSHOT</bundle> 133 <bundle>mvn:org.onlab.onos/onos-app-foo/1.0.0-SNAPSHOT</bundle>
123 </feature> 134 </feature>
124 135
136 + <feature name="onos-app-config" version="1.0.0"
137 + description="ONOS network config reader">
138 + <feature>onos-api</feature>
139 + <bundle>mvn:org.onlab.onos/onos-app-config/1.0.0-SNAPSHOT</bundle>
140 + </feature>
141 +
125 </features> 142 </features>
......
...@@ -98,6 +98,17 @@ ...@@ -98,6 +98,17 @@
98 <version>3.3.2</version> 98 <version>3.3.2</version>
99 </dependency> 99 </dependency>
100 100
101 + <dependency>
102 + <groupId>org.codehaus.jackson</groupId>
103 + <artifactId>jackson-core-asl</artifactId>
104 + <version>1.9.13</version>
105 + </dependency>
106 + <dependency>
107 + <groupId>org.codehaus.jackson</groupId>
108 + <artifactId>jackson-mapper-asl</artifactId>
109 + <version>1.9.13</version>
110 + </dependency>
111 +
101 112
102 <!-- Web related --> 113 <!-- Web related -->
103 <dependency> 114 <dependency>
......
...@@ -93,7 +93,7 @@ public class FlowModBuilder { ...@@ -93,7 +93,7 @@ public class FlowModBuilder {
93 OFFlowMod fm = factory.buildFlowDelete() 93 OFFlowMod fm = factory.buildFlowDelete()
94 .setCookie(U64.of(cookie.value())) 94 .setCookie(U64.of(cookie.value()))
95 .setBufferId(OFBufferId.NO_BUFFER) 95 .setBufferId(OFBufferId.NO_BUFFER)
96 - .setActions(actions) 96 + //.setActions(actions)
97 .setMatch(match) 97 .setMatch(match)
98 .setFlags(Collections.singleton(OFFlowModFlags.SEND_FLOW_REM)) 98 .setFlags(Collections.singleton(OFFlowModFlags.SEND_FLOW_REM))
99 .setPriority(priority) 99 .setPriority(priority)
...@@ -104,6 +104,9 @@ public class FlowModBuilder { ...@@ -104,6 +104,9 @@ public class FlowModBuilder {
104 104
105 private List<OFAction> buildActions() { 105 private List<OFAction> buildActions() {
106 List<OFAction> acts = new LinkedList<>(); 106 List<OFAction> acts = new LinkedList<>();
107 + if (treatment == null) {
108 + return acts;
109 + }
107 for (Instruction i : treatment.instructions()) { 110 for (Instruction i : treatment.instructions()) {
108 switch (i.type()) { 111 switch (i.type()) {
109 case DROP: 112 case DROP:
......
...@@ -31,6 +31,7 @@ import org.onlab.onos.openflow.controller.OpenFlowPacketContext; ...@@ -31,6 +31,7 @@ import org.onlab.onos.openflow.controller.OpenFlowPacketContext;
31 import org.onlab.onos.openflow.controller.PacketListener; 31 import org.onlab.onos.openflow.controller.PacketListener;
32 import org.onlab.packet.ARP; 32 import org.onlab.packet.ARP;
33 import org.onlab.packet.Ethernet; 33 import org.onlab.packet.Ethernet;
34 +import org.onlab.packet.IPv4;
34 import org.onlab.packet.IpPrefix; 35 import org.onlab.packet.IpPrefix;
35 import org.onlab.packet.VlanId; 36 import org.onlab.packet.VlanId;
36 import org.slf4j.Logger; 37 import org.slf4j.Logger;
...@@ -92,29 +93,37 @@ public class OpenFlowHostProvider extends AbstractProvider implements HostProvid ...@@ -92,29 +93,37 @@ public class OpenFlowHostProvider extends AbstractProvider implements HostProvid
92 public void handlePacket(OpenFlowPacketContext pktCtx) { 93 public void handlePacket(OpenFlowPacketContext pktCtx) {
93 Ethernet eth = pktCtx.parsed(); 94 Ethernet eth = pktCtx.parsed();
94 95
96 + VlanId vlan = VlanId.vlanId(eth.getVlanID());
97 + ConnectPoint heardOn = new ConnectPoint(deviceId(Dpid.uri(pktCtx.dpid())),
98 + portNumber(pktCtx.inPort()));
99 +
100 + // If this is not an edge port, bail out.
101 + Topology topology = topologyService.currentTopology();
102 + if (topologyService.isInfrastructure(topology, heardOn)) {
103 + return;
104 + }
105 +
106 + HostLocation hloc = new HostLocation(deviceId(Dpid.uri(pktCtx.dpid())),
107 + portNumber(pktCtx.inPort()),
108 + System.currentTimeMillis());
109 + HostId hid = HostId.hostId(eth.getSourceMAC(), vlan);
95 // Potentially a new or moved host 110 // Potentially a new or moved host
96 if (eth.getEtherType() == Ethernet.TYPE_ARP) { 111 if (eth.getEtherType() == Ethernet.TYPE_ARP) {
97 - VlanId vlan = VlanId.vlanId(eth.getVlanID());
98 - ConnectPoint heardOn = new ConnectPoint(deviceId(Dpid.uri(pktCtx.dpid())),
99 - portNumber(pktCtx.inPort()));
100 -
101 - // If this is not an edge port, bail out.
102 - Topology topology = topologyService.currentTopology();
103 - if (topologyService.isInfrastructure(topology, heardOn)) {
104 - return;
105 - }
106 112
107 - HostLocation hloc = new HostLocation(deviceId(Dpid.uri(pktCtx.dpid())),
108 - portNumber(pktCtx.inPort()),
109 - System.currentTimeMillis());
110 113
111 - HostId hid = HostId.hostId(eth.getSourceMAC(), vlan);
112 ARP arp = (ARP) eth.getPayload(); 114 ARP arp = (ARP) eth.getPayload();
113 Set<IpPrefix> ips = newHashSet(IpPrefix.valueOf(arp.getSenderProtocolAddress())); 115 Set<IpPrefix> ips = newHashSet(IpPrefix.valueOf(arp.getSenderProtocolAddress()));
114 HostDescription hdescr = 116 HostDescription hdescr =
115 new DefaultHostDescription(eth.getSourceMAC(), vlan, hloc, ips); 117 new DefaultHostDescription(eth.getSourceMAC(), vlan, hloc, ips);
116 providerService.hostDetected(hid, hdescr); 118 providerService.hostDetected(hid, hdescr);
117 119
120 + } else if (eth.getEtherType() == Ethernet.TYPE_IPV4) {
121 + IPv4 ip = (IPv4) eth.getPayload();
122 + Set<IpPrefix> ips = newHashSet(IpPrefix.valueOf(ip.getSourceAddress()));
123 + HostDescription hdescr =
124 + new DefaultHostDescription(eth.getSourceMAC(), vlan, hloc, ips);
125 + providerService.hostDetected(hid, hdescr);
126 +
118 } 127 }
119 128
120 // TODO: Use DHCP packets as well later... 129 // TODO: Use DHCP packets as well later...
......
...@@ -106,10 +106,6 @@ public class OpenFlowPacketProvider extends AbstractProvider implements PacketPr ...@@ -106,10 +106,6 @@ 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)) {
110 - log.warn("Tried to write out non-existint port {}", p.getPortNo());
111 - continue;
112 - }
113 OFPacketOut po = packetOut(sw, eth, p.getPortNo()); 109 OFPacketOut po = packetOut(sw, eth, p.getPortNo());
114 sw.sendMsg(po); 110 sw.sendMsg(po);
115 } 111 }
......
...@@ -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
......
...@@ -3,5 +3,7 @@ ...@@ -3,5 +3,7 @@
3 # ONOS remote command-line client. 3 # ONOS remote command-line client.
4 #------------------------------------------------------------------------------- 4 #-------------------------------------------------------------------------------
5 5
6 +[ "$1" = "-w" ] && shift && onos-wait-for-start $1
7 +
6 [ -n "$1" ] && OCI=$1 && shift 8 [ -n "$1" ] && OCI=$1 && shift
7 -client -h $OCI "$@" 9 +client -h $OCI "$@" 2>/dev/null
......
1 +#!/bin/bash
2 +#-------------------------------------------------------------------------------
3 +# Remotely kills the ONOS service on the specified node.
4 +#-------------------------------------------------------------------------------
5 +
6 +[ ! -d "$ONOS_ROOT" ] && echo "ONOS_ROOT is not defined" >&2 && exit 1
7 +. $ONOS_ROOT/tools/build/envDefaults
8 +
9 +ssh $ONOS_USER@${1:-$OCI} "kill -9 \$(ps -ef | grep karaf.jar | grep -v grep | cut -c10-15)"
...\ No newline at end of file ...\ No newline at end of file
1 +# Default virtual box ONOS instances 1,2 & ONOS mininet box
2 +
3 +export ONOS_NIC=192.168.56.*
4 +
5 +export OC1="192.168.56.11"
6 +export OC2="192.168.56.12"
7 +
8 +export OCN="192.168.56.7"
9 +
10 +
1 package org.onlab.util; 1 package org.onlab.util;
2 2
3 +import java.nio.ByteBuffer;
3 import java.util.ArrayList; 4 import java.util.ArrayList;
4 import java.util.List; 5 import java.util.List;
5 import java.util.concurrent.ConcurrentLinkedQueue; 6 import java.util.concurrent.ConcurrentLinkedQueue;
...@@ -8,6 +9,8 @@ import org.apache.commons.lang3.tuple.Pair; ...@@ -8,6 +9,8 @@ import org.apache.commons.lang3.tuple.Pair;
8 9
9 import com.esotericsoftware.kryo.Kryo; 10 import com.esotericsoftware.kryo.Kryo;
10 import com.esotericsoftware.kryo.Serializer; 11 import com.esotericsoftware.kryo.Serializer;
12 +import com.esotericsoftware.kryo.io.ByteBufferInput;
13 +import com.esotericsoftware.kryo.io.ByteBufferOutput;
11 import com.esotericsoftware.kryo.io.Input; 14 import com.esotericsoftware.kryo.io.Input;
12 import com.esotericsoftware.kryo.io.Output; 15 import com.esotericsoftware.kryo.io.Output;
13 import com.google.common.collect.ImmutableList; 16 import com.google.common.collect.ImmutableList;
...@@ -174,6 +177,22 @@ public final class KryoPool { ...@@ -174,6 +177,22 @@ public final class KryoPool {
174 } 177 }
175 178
176 /** 179 /**
180 + * Serializes given object to byte buffer using Kryo instance in pool.
181 + *
182 + * @param obj Object to serialize
183 + * @param buffer to write to
184 + */
185 + public void serialize(final Object obj, final ByteBuffer buffer) {
186 + ByteBufferOutput out = new ByteBufferOutput(buffer);
187 + Kryo kryo = getKryo();
188 + try {
189 + kryo.writeClassAndObject(out, obj);
190 + } finally {
191 + putKryo(kryo);
192 + }
193 + }
194 +
195 + /**
177 * Deserializes given byte array to Object using Kryo instance in pool. 196 * Deserializes given byte array to Object using Kryo instance in pool.
178 * 197 *
179 * @param bytes serialized bytes 198 * @param bytes serialized bytes
...@@ -192,6 +211,24 @@ public final class KryoPool { ...@@ -192,6 +211,24 @@ public final class KryoPool {
192 } 211 }
193 } 212 }
194 213
214 + /**
215 + * Deserializes given byte buffer to Object using Kryo instance in pool.
216 + *
217 + * @param buffer input with serialized bytes
218 + * @param <T> deserialized Object type
219 + * @return deserialized Object
220 + */
221 + public <T> T deserialize(final ByteBuffer buffer) {
222 + ByteBufferInput in = new ByteBufferInput(buffer);
223 + Kryo kryo = getKryo();
224 + try {
225 + @SuppressWarnings("unchecked")
226 + T obj = (T) kryo.readClassAndObject(in);
227 + return obj;
228 + } finally {
229 + putKryo(kryo);
230 + }
231 + }
195 232
196 /** 233 /**
197 * Creates a Kryo instance with {@link #registeredTypes} pre-registered. 234 * Creates a Kryo instance with {@link #registeredTypes} pre-registered.
......
...@@ -93,14 +93,9 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>> ...@@ -93,14 +93,9 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>>
93 * 93 *
94 * @param key selection key holding the pending connect operation. 94 * @param key selection key holding the pending connect operation.
95 */ 95 */
96 - protected void connect(SelectionKey key) { 96 + protected void connect(SelectionKey key) throws IOException {
97 - try { 97 + SocketChannel ch = (SocketChannel) key.channel();
98 - SocketChannel ch = (SocketChannel) key.channel(); 98 + ch.finishConnect();
99 - ch.finishConnect();
100 - } catch (IOException | IllegalStateException e) {
101 - log.warn("Unable to complete connection", e);
102 - }
103 -
104 if (key.isValid()) { 99 if (key.isValid()) {
105 key.interestOps(SelectionKey.OP_READ); 100 key.interestOps(SelectionKey.OP_READ);
106 } 101 }
...@@ -124,7 +119,11 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>> ...@@ -124,7 +119,11 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>>
124 119
125 // If there is a pending connect operation, complete it. 120 // If there is a pending connect operation, complete it.
126 if (key.isConnectable()) { 121 if (key.isConnectable()) {
127 - connect(key); 122 + try {
123 + connect(key);
124 + } catch (IOException | IllegalStateException e) {
125 + log.warn("Unable to complete connection", e);
126 + }
128 } 127 }
129 128
130 // If there is a read operation, slurp as much data as possible. 129 // If there is a read operation, slurp as much data as possible.
......
...@@ -10,6 +10,7 @@ import java.nio.channels.ByteChannel; ...@@ -10,6 +10,7 @@ import java.nio.channels.ByteChannel;
10 import java.nio.channels.SelectionKey; 10 import java.nio.channels.SelectionKey;
11 import java.util.ArrayList; 11 import java.util.ArrayList;
12 import java.util.List; 12 import java.util.List;
13 +import java.util.Objects;
13 14
14 import static com.google.common.base.Preconditions.checkArgument; 15 import static com.google.common.base.Preconditions.checkArgument;
15 import static com.google.common.base.Preconditions.checkNotNull; 16 import static com.google.common.base.Preconditions.checkNotNull;
...@@ -262,7 +263,7 @@ public abstract class MessageStream<M extends Message> { ...@@ -262,7 +263,7 @@ public abstract class MessageStream<M extends Message> {
262 try { 263 try {
263 channel.write(outbound); 264 channel.write(outbound);
264 } catch (IOException e) { 265 } catch (IOException e) {
265 - if (!closed && !e.getMessage().equals("Broken pipe")) { 266 + if (!closed && !Objects.equals(e.getMessage(), "Broken pipe")) {
266 log.warn("Unable to write data", e); 267 log.warn("Unable to write data", e);
267 ioError = e; 268 ioError = e;
268 } 269 }
......
...@@ -230,7 +230,7 @@ public class IOLoopTestClient { ...@@ -230,7 +230,7 @@ public class IOLoopTestClient {
230 } 230 }
231 231
232 @Override 232 @Override
233 - protected void connect(SelectionKey key) { 233 + protected void connect(SelectionKey key) throws IOException {
234 super.connect(key); 234 super.connect(key);
235 TestMessageStream b = (TestMessageStream) key.attachment(); 235 TestMessageStream b = (TestMessageStream) key.attachment();
236 Worker w = ((CustomIOLoop) b.loop()).worker; 236 Worker w = ((CustomIOLoop) b.loop()).worker;
......