Adding ability to synchronize topology clusters' broadcast trees.
Proxy ARP now supports deferred ARP replies until instance learns of the subject host location.
Change-Id: Ib3ee97c0812858b5b4972d945e9e6d2bd397d4c5
diff --git a/core/store/dist/src/main/java/org/onosproject/store/host/impl/ECHostStore.java b/core/store/dist/src/main/java/org/onosproject/store/host/impl/ECHostStore.java
index f4071ef..17bdbbe 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/host/impl/ECHostStore.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/host/impl/ECHostStore.java
@@ -4,6 +4,7 @@
import static org.onosproject.net.DefaultAnnotations.merge;
import static org.onosproject.net.host.HostEvent.Type.HOST_ADDED;
import static org.onosproject.net.host.HostEvent.Type.HOST_REMOVED;
+import static org.onosproject.net.host.HostEvent.Type.HOST_UPDATED;
import static org.onosproject.store.service.EventuallyConsistentMapEvent.Type.PUT;
import static org.onosproject.store.service.EventuallyConsistentMapEvent.Type.REMOVE;
import static org.slf4j.LoggerFactory.getLogger;
@@ -246,14 +247,17 @@
}
private class HostLocationTracker implements EventuallyConsistentMapListener<HostId, DefaultHost> {
-
@Override
public void event(EventuallyConsistentMapEvent<HostId, DefaultHost> event) {
DefaultHost host = checkNotNull(event.value());
if (event.type() == PUT) {
- locations.put(host.location(), host);
+ boolean isNew = locations.put(host.location(), host);
+ notifyDelegate(new HostEvent(isNew ? HOST_ADDED : HOST_UPDATED, host));
} else if (event.type() == REMOVE) {
- locations.remove(host.location(), host);
+ if (locations.remove(host.location(), host)) {
+ notifyDelegate(new HostEvent(HOST_REMOVED, host));
+ }
+
}
}
}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/proxyarp/impl/DistributedProxyArpStore.java b/core/store/dist/src/main/java/org/onosproject/store/proxyarp/impl/DistributedProxyArpStore.java
new file mode 100644
index 0000000..851185b
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onosproject/store/proxyarp/impl/DistributedProxyArpStore.java
@@ -0,0 +1,174 @@
+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.store.proxyarp.impl;
+
+import com.google.common.collect.Maps;
+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.util.KryoNamespace;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.mastership.MastershipService;
+import org.onosproject.net.ConnectPoint;
+import org.onosproject.net.Host;
+import org.onosproject.net.HostId;
+import org.onosproject.net.host.HostEvent;
+import org.onosproject.net.host.HostListener;
+import org.onosproject.net.host.HostService;
+import org.onosproject.net.proxyarp.ProxyArpStore;
+import org.onosproject.net.proxyarp.ProxyArpStoreDelegate;
+import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
+import org.onosproject.store.cluster.messaging.MessageSubject;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.serializers.KryoSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+
+import static org.onlab.util.BoundedThreadPool.newFixedThreadPool;
+import static org.onlab.util.Tools.groupedThreads;
+
+/**
+ * Implementation of proxy ARP distribution mechanism.
+ */
+@Component(immediate = true)
+@Service
+public class DistributedProxyArpStore implements ProxyArpStore {
+
+ private Logger log = LoggerFactory.getLogger(getClass());
+
+ private static final MessageSubject ARP_RESPONSE_MESSAGE =
+ new MessageSubject("onos-arp-response");
+
+ protected final KryoSerializer serializer = new KryoSerializer() {
+ @Override
+ protected void setupKryoPool() {
+ serializerPool = KryoNamespace.newBuilder()
+ .register(KryoNamespaces.API)
+ .register(ArpResponseMessage.class)
+ .register(ByteBuffer.class)
+ .build();
+ }
+ };
+
+ private ProxyArpStoreDelegate delegate;
+
+ private Map<HostId, ArpResponseMessage> pendingMessages = Maps.newConcurrentMap();
+
+ private ExecutorService executor =
+ newFixedThreadPool(4, groupedThreads("onos/arp", "sender-%d"));
+
+ private NodeId localNodeId;
+
+ private HostListener hostListener = new InternalHostListener();
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected MastershipService mastershipService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClusterService clusterService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClusterCommunicationService commService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected HostService hostService;
+
+
+ @Activate
+ protected void activate() {
+ localNodeId = clusterService.getLocalNode().id();
+ hostService.addListener(hostListener);
+ commService.addSubscriber(ARP_RESPONSE_MESSAGE, serializer::decode,
+ this::processArpResponse, executor);
+ log.info("Started");
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ commService.removeSubscriber(ARP_RESPONSE_MESSAGE);
+ hostService.removeListener(hostListener);
+ log.info("Stopped");
+ }
+
+ @Override
+ public void forward(ConnectPoint outPort, Host subject, ByteBuffer packet) {
+ NodeId nodeId = mastershipService.getMasterFor(outPort.deviceId());
+ if (nodeId.equals(localNodeId)) {
+ if (delegate != null) {
+ delegate.emitResponse(outPort, packet);
+ }
+ } else {
+ log.info("Forwarding ARP response from {} to {}", subject.id(), outPort);
+ commService.unicast(new ArpResponseMessage(outPort, subject, packet.array()),
+ ARP_RESPONSE_MESSAGE, serializer::encode, nodeId);
+ }
+ }
+
+ @Override
+ public void setDelegate(ProxyArpStoreDelegate delegate) {
+ this.delegate = delegate;
+ }
+
+ // Processes the incoming ARP response message.
+ private void processArpResponse(ArpResponseMessage msg) {
+ pendingMessages.put(msg.subject.id(), msg);
+ if (hostService.getHost(msg.subject.id()) != null) {
+ checkPendingArps(msg.subject.id());
+ }
+ // FIXME: figure out pruning so stuff does not build up
+ }
+
+ // Checks for pending ARP response message for the specified host.
+ // If one exists, emit response via delegate.
+ private void checkPendingArps(HostId id) {
+ ArpResponseMessage msg = pendingMessages.remove(id);
+ if (msg != null && delegate != null) {
+ log.info("Emitting ARP response from {} to {}", id, msg.outPort);
+ delegate.emitResponse(msg.outPort, ByteBuffer.wrap(msg.packet));
+ }
+ }
+
+ // Message carrying an ARP response.
+ private static class ArpResponseMessage {
+ private ConnectPoint outPort;
+ private Host subject;
+ private byte[] packet;
+
+ public ArpResponseMessage(ConnectPoint outPort, Host subject, byte[] packet) {
+ this.outPort = outPort;
+ this.subject = subject;
+ this.packet = packet;
+ }
+
+ private ArpResponseMessage() {
+ }
+ }
+
+ private class InternalHostListener implements HostListener {
+ @Override
+ public void event(HostEvent event) {
+ checkPendingArps(event.subject().id());
+ }
+ }
+}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/topology/impl/DistributedTopologyStore.java b/core/store/dist/src/main/java/org/onosproject/store/topology/impl/DistributedTopologyStore.java
index 3a0f185..487fad9 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/topology/impl/DistributedTopologyStore.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/topology/impl/DistributedTopologyStore.java
@@ -16,19 +16,25 @@
package org.onosproject.store.topology.impl;
import static com.google.common.base.Preconditions.checkArgument;
+import static org.onlab.util.Tools.isNullOrEmpty;
import static org.onosproject.net.topology.TopologyEvent.Type.TOPOLOGY_CHANGED;
import static org.slf4j.LoggerFactory.getLogger;
import java.util.Collections;
import java.util.List;
import java.util.Set;
+import java.util.stream.Collectors;
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.util.KryoNamespace;
import org.onosproject.common.DefaultTopology;
import org.onosproject.event.Event;
+import org.onosproject.mastership.MastershipService;
import org.onosproject.net.ConnectPoint;
import org.onosproject.net.Device;
import org.onosproject.net.DeviceId;
@@ -46,6 +52,12 @@
import org.onosproject.net.topology.TopologyStore;
import org.onosproject.net.topology.TopologyStoreDelegate;
import org.onosproject.store.AbstractStore;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.EventuallyConsistentMap;
+import org.onosproject.store.service.EventuallyConsistentMapEvent;
+import org.onosproject.store.service.EventuallyConsistentMapListener;
+import org.onosproject.store.service.LogicalClockService;
+import org.onosproject.store.service.StorageService;
import org.slf4j.Logger;
/**
@@ -69,13 +81,41 @@
Collections.<Device>emptyList(),
Collections.<Link>emptyList()));
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected StorageService storageService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected LogicalClockService clockService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected MastershipService mastershipService;
+
+ // Cluster root to broadcast points bindings to allow convergence to
+ // a shared broadcast tree; node that is the master of the cluster root
+ // is the primary.
+ private EventuallyConsistentMap<DeviceId, Set<ConnectPoint>> broadcastPoints;
+
+ private EventuallyConsistentMapListener<DeviceId, Set<ConnectPoint>> listener =
+ new InternalBroadcastPointListener();
+
@Activate
public void activate() {
+ KryoNamespace.Builder hostSerializer = KryoNamespace.newBuilder()
+ .register(KryoNamespaces.API);
+
+ broadcastPoints = storageService.<DeviceId, Set<ConnectPoint>>eventuallyConsistentMapBuilder()
+ .withName("onos-broadcast-trees")
+ .withSerializer(hostSerializer)
+ .withTimestampProvider((k, v) -> clockService.getTimestamp())
+ .build();
+ broadcastPoints.addListener(listener);
log.info("Started");
}
@Deactivate
public void deactivate() {
+ broadcastPoints.removeListener(listener);
+ broadcastPoints.destroy();
log.info("Stopped");
}
@@ -136,6 +176,22 @@
return defaultTopology(topology).isBroadcastPoint(connectPoint);
}
+ private boolean isBroadcastPoint(ConnectPoint connectPoint) {
+ // Any non-infrastructure, i.e. edge points are assumed to be OK.
+ if (!current.isInfrastructure(connectPoint)) {
+ return true;
+ }
+
+ // Find the cluster to which the device belongs.
+ TopologyCluster cluster = current.getCluster(connectPoint.deviceId());
+ checkArgument(cluster != null, "No cluster found for device %s", connectPoint.deviceId());
+
+ // If the broadcast set is null or empty, or if the point explicitly
+ // belongs to it, return true;
+ Set<ConnectPoint> points = broadcastPoints.get(cluster.root().deviceId());
+ return isNullOrEmpty(points) || points.contains(connectPoint);
+ }
+
@Override
public TopologyEvent updateTopology(ProviderId providerId,
GraphDescription graphDescription,
@@ -147,7 +203,9 @@
}
// Have the default topology construct self from the description data.
- DefaultTopology newTopology = new DefaultTopology(providerId, graphDescription);
+ DefaultTopology newTopology =
+ new DefaultTopology(providerId, graphDescription, this::isBroadcastPoint);
+ updateBroadcastPoints(newTopology);
// Promote the new topology to current and return a ready-to-send event.
synchronized (this) {
@@ -156,6 +214,24 @@
}
}
+ private void updateBroadcastPoints(DefaultTopology topology) {
+ // Remove any broadcast trees rooted by devices for which we are master.
+ Set<DeviceId> toRemove = broadcastPoints.keySet().stream()
+ .filter(mastershipService::isLocalMaster)
+ .collect(Collectors.toSet());
+
+ // Update the broadcast trees rooted by devices for which we are master.
+ topology.getClusters().forEach(c -> {
+ toRemove.remove(c.root().deviceId());
+ if (mastershipService.isLocalMaster(c.root().deviceId())) {
+ broadcastPoints.put(c.root().deviceId(),
+ topology.broadcastPoints(c.id()));
+ }
+ });
+
+ toRemove.forEach(broadcastPoints::remove);
+ }
+
// Validates the specified topology and returns it as a default
private DefaultTopology defaultTopology(Topology topology) {
checkArgument(topology instanceof DefaultTopology,
@@ -163,4 +239,16 @@
return (DefaultTopology) topology;
}
+ private class InternalBroadcastPointListener
+ implements EventuallyConsistentMapListener<DeviceId, Set<ConnectPoint>> {
+ @Override
+ public void event(EventuallyConsistentMapEvent<DeviceId, Set<ConnectPoint>> event) {
+ if (event.type() == EventuallyConsistentMapEvent.Type.PUT) {
+ if (!event.value().isEmpty()) {
+ log.info("Cluster rooted at {} has {} broadcast-points; #{}",
+ event.key(), event.value().size(), event.value().hashCode());
+ }
+ }
+ }
+ }
}