Merge remote-tracking branch 'origin/master'
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/OnosDistributedDeviceStore.java b/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/OnosDistributedDeviceStore.java
index 30374c3..bd5f2fd 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/OnosDistributedDeviceStore.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/OnosDistributedDeviceStore.java
@@ -86,46 +86,48 @@
@Override
public Iterable<Device> getDevices() {
- // TODO builder v.s. copyOf. Guava semms to be using copyOf?
- // FIXME: synchronize.
Builder<Device> builder = ImmutableSet.builder();
- for (VersionedValue<? extends Device> device : devices.values()) {
- builder.add(device.entity());
+ synchronized (this) {
+ for (VersionedValue<Device> device : devices.values()) {
+ builder.add(device.entity());
+ }
+ return builder.build();
}
- return builder.build();
}
@Override
public Device getDevice(DeviceId deviceId) {
- return devices.get(deviceId).entity();
+ VersionedValue<Device> device = devices.get(deviceId);
+ checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
+ return device.entity();
}
@Override
public DeviceEvent createOrUpdateDevice(ProviderId providerId, DeviceId deviceId,
DeviceDescription deviceDescription) {
- Timestamp now = clockService.getTimestamp(deviceId);
+ Timestamp newTimestamp = clockService.getTimestamp(deviceId);
VersionedValue<Device> device = devices.get(deviceId);
if (device == null) {
- return createDevice(providerId, deviceId, deviceDescription, now);
+ return createDevice(providerId, deviceId, deviceDescription, newTimestamp);
}
- checkState(now.compareTo(device.timestamp()) > 0,
+ checkState(newTimestamp.compareTo(device.timestamp()) > 0,
"Existing device has a timestamp in the future!");
- return updateDevice(providerId, device.entity(), deviceDescription, now);
+ return updateDevice(providerId, device.entity(), deviceDescription, newTimestamp);
}
// Creates the device and returns the appropriate event if necessary.
private DeviceEvent createDevice(ProviderId providerId, DeviceId deviceId,
DeviceDescription desc, Timestamp timestamp) {
- DefaultDevice device = new DefaultDevice(providerId, deviceId, desc.type(),
+ Device device = new DefaultDevice(providerId, deviceId, desc.type(),
desc.manufacturer(),
desc.hwVersion(), desc.swVersion(),
desc.serialNumber());
- devices.put(deviceId, new VersionedValue<Device>(device, true, timestamp));
- // FIXME: broadcast a message telling peers of a device event.
+ devices.put(deviceId, new VersionedValue<>(device, true, timestamp));
+ // TODO,FIXME: broadcast a message telling peers of a device event.
return new DeviceEvent(DEVICE_ADDED, device, null);
}
@@ -148,7 +150,7 @@
}
// Otherwise merely attempt to change availability
- DefaultDevice updated = new DefaultDevice(providerId, device.id(),
+ Device updated = new DefaultDevice(providerId, device.id(),
desc.type(),
desc.manufacturer(),
desc.hwVersion(),
@@ -196,18 +198,18 @@
VersionedValue<Device> device = devices.get(deviceId);
checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
Map<PortNumber, VersionedValue<Port>> ports = getPortMap(deviceId);
- Timestamp timestamp = clockService.getTimestamp(deviceId);
+ Timestamp newTimestamp = clockService.getTimestamp(deviceId);
// Add new ports
Set<PortNumber> processed = new HashSet<>();
for (PortDescription portDescription : portDescriptions) {
VersionedValue<Port> port = ports.get(portDescription.portNumber());
if (port == null) {
- events.add(createPort(device, portDescription, ports, timestamp));
+ events.add(createPort(device, portDescription, ports, newTimestamp));
}
- checkState(timestamp.compareTo(port.timestamp()) > 0,
+ checkState(newTimestamp.compareTo(port.timestamp()) > 0,
"Existing port state has a timestamp in the future!");
- events.add(updatePort(device, port, portDescription, ports, timestamp));
+ events.add(updatePort(device.entity(), port.entity(), portDescription, ports, newTimestamp));
processed.add(portDescription.portNumber());
}
@@ -233,19 +235,19 @@
// Checks if the specified port requires update and if so, it replaces the
// existing entry in the map and returns corresponding event.
//@GuardedBy("this")
- private DeviceEvent updatePort(VersionedValue<Device> device, VersionedValue<Port> port,
+ private DeviceEvent updatePort(Device device, Port port,
PortDescription portDescription,
Map<PortNumber, VersionedValue<Port>> ports,
Timestamp timestamp) {
- if (port.entity().isEnabled() != portDescription.isEnabled()) {
+ if (port.isEnabled() != portDescription.isEnabled()) {
VersionedValue<Port> updatedPort = new VersionedValue<Port>(
- new DefaultPort(device.entity(), portDescription.portNumber(),
+ new DefaultPort(device, portDescription.portNumber(),
portDescription.isEnabled()),
portDescription.isEnabled(),
timestamp);
- ports.put(port.entity().number(), updatedPort);
- updatePortMap(device.entity().id(), ports);
- return new DeviceEvent(PORT_UPDATED, device.entity(), updatedPort.entity());
+ ports.put(port.number(), updatedPort);
+ updatePortMap(device.id(), ports);
+ return new DeviceEvent(PORT_UPDATED, device, updatedPort.entity());
}
return null;
}
@@ -300,7 +302,7 @@
Map<PortNumber, VersionedValue<Port>> ports = getPortMap(deviceId);
VersionedValue<Port> port = ports.get(portDescription.portNumber());
Timestamp timestamp = clockService.getTimestamp(deviceId);
- return updatePort(device, port, portDescription, ports, timestamp);
+ return updatePort(device.entity(), port.entity(), portDescription, ports, timestamp);
}
@Override
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/OnosDistributedLinkStore.java b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/OnosDistributedLinkStore.java
new file mode 100644
index 0000000..5df25b4
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/link/impl/OnosDistributedLinkStore.java
@@ -0,0 +1,246 @@
+package org.onlab.onos.store.link.impl;
+
+import static org.onlab.onos.net.Link.Type.DIRECT;
+import static org.onlab.onos.net.Link.Type.INDIRECT;
+import static org.onlab.onos.net.link.LinkEvent.Type.LINK_ADDED;
+import static org.onlab.onos.net.link.LinkEvent.Type.LINK_REMOVED;
+import static org.onlab.onos.net.link.LinkEvent.Type.LINK_UPDATED;
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.net.ConnectPoint;
+import org.onlab.onos.net.DefaultLink;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.Link;
+import org.onlab.onos.net.LinkKey;
+import org.onlab.onos.net.link.LinkDescription;
+import org.onlab.onos.net.link.LinkEvent;
+import org.onlab.onos.net.link.LinkStore;
+import org.onlab.onos.net.link.LinkStoreDelegate;
+import org.onlab.onos.net.provider.ProviderId;
+import org.onlab.onos.store.AbstractStore;
+import org.onlab.onos.store.ClockService;
+import org.onlab.onos.store.Timestamp;
+import org.onlab.onos.store.device.impl.VersionedValue;
+import org.slf4j.Logger;
+
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Multimap;
+import com.google.common.collect.ImmutableSet.Builder;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Manages inventory of infrastructure links using a protocol that takes into consideration
+ * the order in which events occur.
+ */
+// FIXME: This does not yet implement the full protocol.
+// The full protocol requires the sender of LLDP message to include the
+// version information of src device/port and the receiver to
+// take that into account when figuring out if a more recent src
+// device/port down event renders the link discovery obsolete.
+@Component(immediate = true)
+@Service
+public class OnosDistributedLinkStore
+ extends AbstractStore<LinkEvent, LinkStoreDelegate>
+ implements LinkStore {
+
+ private final Logger log = getLogger(getClass());
+
+ // Link inventory
+ private ConcurrentMap<LinkKey, VersionedValue<Link>> links;
+
+ public static final String LINK_NOT_FOUND = "Link between %s and %s not found";
+
+ // TODO synchronize?
+ // Egress and ingress link sets
+ private final Multimap<DeviceId, VersionedValue<Link>> srcLinks = HashMultimap.create();
+ private final Multimap<DeviceId, VersionedValue<Link>> dstLinks = HashMultimap.create();
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClockService clockService;
+
+ @Activate
+ public void activate() {
+
+ links = new ConcurrentHashMap<>();
+
+ log.info("Started");
+ }
+
+ @Deactivate
+ public void deactivate() {
+ log.info("Stopped");
+ }
+
+ @Override
+ public int getLinkCount() {
+ return links.size();
+ }
+
+ @Override
+ public Iterable<Link> getLinks() {
+ Builder<Link> builder = ImmutableSet.builder();
+ synchronized (this) {
+ for (VersionedValue<Link> link : links.values()) {
+ builder.add(link.entity());
+ }
+ return builder.build();
+ }
+ }
+
+ @Override
+ public Set<Link> getDeviceEgressLinks(DeviceId deviceId) {
+ Set<VersionedValue<Link>> egressLinks = ImmutableSet.copyOf(srcLinks.get(deviceId));
+ Set<Link> rawEgressLinks = new HashSet<>();
+ for (VersionedValue<Link> link : egressLinks) {
+ rawEgressLinks.add(link.entity());
+ }
+ return rawEgressLinks;
+ }
+
+ @Override
+ public Set<Link> getDeviceIngressLinks(DeviceId deviceId) {
+ Set<VersionedValue<Link>> ingressLinks = ImmutableSet.copyOf(dstLinks.get(deviceId));
+ Set<Link> rawIngressLinks = new HashSet<>();
+ for (VersionedValue<Link> link : ingressLinks) {
+ rawIngressLinks.add(link.entity());
+ }
+ return rawIngressLinks;
+ }
+
+ @Override
+ public Link getLink(ConnectPoint src, ConnectPoint dst) {
+ VersionedValue<Link> link = links.get(new LinkKey(src, dst));
+ checkArgument(link != null, "LINK_NOT_FOUND", src, dst);
+ return link.entity();
+ }
+
+ @Override
+ public Set<Link> getEgressLinks(ConnectPoint src) {
+ Set<Link> egressLinks = new HashSet<>();
+ for (VersionedValue<Link> link : srcLinks.get(src.deviceId())) {
+ if (link.entity().src().equals(src)) {
+ egressLinks.add(link.entity());
+ }
+ }
+ return egressLinks;
+ }
+
+ @Override
+ public Set<Link> getIngressLinks(ConnectPoint dst) {
+ Set<Link> ingressLinks = new HashSet<>();
+ for (VersionedValue<Link> link : dstLinks.get(dst.deviceId())) {
+ if (link.entity().dst().equals(dst)) {
+ ingressLinks.add(link.entity());
+ }
+ }
+ return ingressLinks;
+ }
+
+ @Override
+ public LinkEvent createOrUpdateLink(ProviderId providerId,
+ LinkDescription linkDescription) {
+
+ final DeviceId destinationDeviceId = linkDescription.dst().deviceId();
+ final Timestamp newTimestamp = clockService.getTimestamp(destinationDeviceId);
+
+ LinkKey key = new LinkKey(linkDescription.src(), linkDescription.dst());
+ VersionedValue<Link> link = links.get(key);
+ if (link == null) {
+ return createLink(providerId, key, linkDescription, newTimestamp);
+ }
+
+ checkState(newTimestamp.compareTo(link.timestamp()) > 0,
+ "Existing Link has a timestamp in the future!");
+
+ return updateLink(providerId, link, key, linkDescription, newTimestamp);
+ }
+
+ // Creates and stores the link and returns the appropriate event.
+ private LinkEvent createLink(ProviderId providerId, LinkKey key,
+ LinkDescription linkDescription, Timestamp timestamp) {
+ VersionedValue<Link> link = new VersionedValue<Link>(new DefaultLink(providerId, key.src(), key.dst(),
+ linkDescription.type()), true, timestamp);
+ synchronized (this) {
+ links.put(key, link);
+ addNewLink(link, timestamp);
+ }
+ // FIXME: notify peers.
+ return new LinkEvent(LINK_ADDED, link.entity());
+ }
+
+ // update Egress and ingress link sets
+ private void addNewLink(VersionedValue<Link> link, Timestamp timestamp) {
+ Link rawLink = link.entity();
+ synchronized (this) {
+ srcLinks.put(rawLink.src().deviceId(), link);
+ dstLinks.put(rawLink.dst().deviceId(), link);
+ }
+ }
+
+ // Updates, if necessary the specified link and returns the appropriate event.
+ private LinkEvent updateLink(ProviderId providerId, VersionedValue<Link> existingLink,
+ LinkKey key, LinkDescription linkDescription, Timestamp timestamp) {
+ // FIXME confirm Link update condition is OK
+ if (existingLink.entity().type() == INDIRECT && linkDescription.type() == DIRECT) {
+ synchronized (this) {
+
+ VersionedValue<Link> updatedLink = new VersionedValue<Link>(
+ new DefaultLink(providerId, existingLink.entity().src(), existingLink.entity().dst(),
+ linkDescription.type()), true, timestamp);
+ links.replace(key, existingLink, updatedLink);
+
+ replaceLink(existingLink, updatedLink);
+ // FIXME: notify peers.
+ return new LinkEvent(LINK_UPDATED, updatedLink.entity());
+ }
+ }
+ return null;
+ }
+
+ // update Egress and ingress link sets
+ private void replaceLink(VersionedValue<Link> current, VersionedValue<Link> updated) {
+ synchronized (this) {
+ srcLinks.remove(current.entity().src().deviceId(), current);
+ dstLinks.remove(current.entity().dst().deviceId(), current);
+
+ srcLinks.put(current.entity().src().deviceId(), updated);
+ dstLinks.put(current.entity().dst().deviceId(), updated);
+ }
+ }
+
+ @Override
+ public LinkEvent removeLink(ConnectPoint src, ConnectPoint dst) {
+ synchronized (this) {
+ LinkKey key = new LinkKey(src, dst);
+ VersionedValue<Link> link = links.remove(key);
+ if (link != null) {
+ removeLink(link);
+ // notify peers
+ return new LinkEvent(LINK_REMOVED, link.entity());
+ }
+ return null;
+ }
+ }
+
+ // update Egress and ingress link sets
+ private void removeLink(VersionedValue<Link> link) {
+ synchronized (this) {
+ srcLinks.remove(link.entity().src().deviceId(), link);
+ dstLinks.remove(link.entity().dst().deviceId(), link);
+ }
+ }
+}