Refactored packet broadcasting framework. Previously only ARP packets were being broadcasted, but as part of reactive forwarding we need to broadcast all packets that we don't know where to send. This commit refactors the Hazelcast notification system used to tell other instances to send packet-outs.
diff --git a/src/main/java/net/onrc/onos/ofcontroller/proxyarp/ProxyArpManager.java b/src/main/java/net/onrc/onos/ofcontroller/proxyarp/ProxyArpManager.java
index 32b2e9c..2ea49a9 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/proxyarp/ProxyArpManager.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/proxyarp/ProxyArpManager.java
@@ -37,6 +37,7 @@
import net.onrc.onos.ofcontroller.core.config.IConfigInfoService;
import net.onrc.onos.ofcontroller.core.internal.DeviceStorageImpl;
import net.onrc.onos.ofcontroller.core.internal.TopoSwitchServiceImpl;
+import net.onrc.onos.ofcontroller.flowprogrammer.IFlowPusherService;
import net.onrc.onos.ofcontroller.util.Dpid;
import net.onrc.onos.ofcontroller.util.Port;
import net.onrc.onos.ofcontroller.util.SwitchPort;
@@ -58,7 +59,8 @@
import com.google.common.net.InetAddresses;
public class ProxyArpManager implements IProxyArpService, IOFMessageListener,
- IArpEventHandler, IFloodlightModule {
+ IPacketOutEventHandler, IArpReplyEventHandler,
+ IFloodlightModule {
private final static Logger log = LoggerFactory.getLogger(ProxyArpManager.class);
private final long ARP_TIMER_PERIOD = 100; //ms
@@ -70,6 +72,7 @@
private IDatagridService datagrid;
private IConfigInfoService configService;
private IRestApiService restApi;
+ private IFlowPusherService flowPusher;
private IDeviceStorage deviceStorage;
private volatile ITopoSwitchService topoSwitchService;
@@ -153,6 +156,7 @@
dependencies.add(IRestApiService.class);
dependencies.add(IDatagridService.class);
dependencies.add(IConfigInfoService.class);
+ dependencies.add(IFlowPusherService.class);
return dependencies;
}
@@ -164,6 +168,7 @@
this.datagrid = context.getServiceImpl(IDatagridService.class);
this.configService = context.getServiceImpl(IConfigInfoService.class);
this.restApi = context.getServiceImpl(IRestApiService.class);
+ this.flowPusher = context.getServiceImpl(IFlowPusherService.class);
//arpCache = new ArpCache();
@@ -181,7 +186,7 @@
restApi.addRestletRoutable(new ArpWebRoutable());
floodlightProvider.addOFMessageListener(OFType.PACKET_IN, this);
- datagrid.registerArpEventHandler(this);
+ datagrid.registerPacketOutEventHandler(this);
deviceStorage = new DeviceStorageImpl();
deviceStorage.init("");
@@ -289,7 +294,7 @@
}
else if (arp.getOpCode() == ARP.OP_REPLY) {
handleArpReply(sw, pi, arp);
- sendToOtherNodesReply(eth, pi);
+ sendReplyNotification(eth, pi);
}
// Stop ARP packets here
@@ -343,7 +348,9 @@
}
// We don't know the device so broadcast the request out
- sendToOtherNodes(eth, sw.getId(), pi);
+ datagrid.sendPacketOutNotification(
+ new BroadcastPacketOutNotification(eth.serialize(),
+ sw.getId(), pi.getInPort()));
}
else {
// Even if the device exists in our database, we do not reply to
@@ -360,7 +367,6 @@
// sendArpReply(arp, sw.getId(), pi.getInPort(), macAddress);
- log.trace("Checking the device info from DB is still valid or not");
Iterable<IPortObject> outPorts = targetDevice.getAttachedPorts();
if (!outPorts.iterator().hasNext()){
@@ -369,19 +375,26 @@
" - broadcasting", macAddress);
}
- sendToOtherNodes(eth, sw.getId(), pi);
+ datagrid.sendPacketOutNotification(
+ new BroadcastPacketOutNotification(eth.serialize(),
+ sw.getId(), pi.getInPort()));
}
else {
for (IPortObject portObject : outPorts) {
- long outSwitch = 0;
- short outPort = 0;
+ //long outSwitch = 0;
+ //short outPort = 0;
+ /*
if (!portObject.getLinkedPorts().iterator().hasNext()) {
outPort = portObject.getNumber();
+ }*/
+ if (portObject.getLinkedPorts().iterator().hasNext()) {
+ continue;
}
+ short outPort = portObject.getNumber();
ISwitchObject outSwitchObject = portObject.getSwitch();
- outSwitch = HexString.toLong(outSwitchObject.getDPID());
+ long outSwitch = HexString.toLong(outSwitchObject.getDPID());
if (log.isTraceEnabled()) {
log.trace("Probing device {} on port {}/{}",
@@ -389,7 +402,9 @@
HexString.toHexString(outSwitch), outPort});
}
- sendToOtherNodes(eth, pi, outSwitch, outPort);
+ datagrid.sendPacketOutNotification(
+ new SinglePacketOutNotification(eth.serialize(),
+ outSwitch, outPort));
}
}
}
@@ -515,50 +530,7 @@
}
}
- private void sendToOtherNodes(Ethernet eth, long inSwitchId, OFPacketIn pi) {
- ARP arp = (ARP) eth.getPayload();
-
- if (log.isTraceEnabled()) {
- log.trace("Sending ARP request for {} to other ONOS instances",
- inetAddressToString(arp.getTargetProtocolAddress()));
- }
-
- InetAddress targetAddress;
- try {
- targetAddress = InetAddress.getByAddress(arp.getTargetProtocolAddress());
- } catch (UnknownHostException e) {
- log.error("Unknown host", e);
- return;
- }
-
- datagrid.sendArpRequest(ArpMessage.newRequest(targetAddress, eth.serialize(),
- -1L, (short)-1, inSwitchId, pi.getInPort()));
- }
-
- //hazelcast to other ONOS instances to send the ARP packet out on outPort of outSwitch
- private void sendToOtherNodes(Ethernet eth, OFPacketIn pi, long outSwitch, short outPort) {
- ARP arp = (ARP) eth.getPayload();
-
- if (log.isTraceEnabled()) {
- log.trace("Sending ARP request for {} to other ONOS instances with outSwitch {} ",
- inetAddressToString(arp.getTargetProtocolAddress()), String.valueOf(outSwitch));
- }
-
- InetAddress targetAddress;
- try {
- targetAddress = InetAddress.getByAddress(arp.getTargetProtocolAddress());
- } catch (UnknownHostException e) {
- log.error("Unknown host", e);
- return;
- }
-
- datagrid.sendArpRequest(ArpMessage.newRequest(targetAddress, eth.serialize(), outSwitch, outPort));
- //datagrid.sendArpRequest(ArpMessage.newRequest(targetAddress, eth.serialize()));
-
-
- }
-
- private void sendToOtherNodesReply(Ethernet eth, OFPacketIn pi) {
+ private void sendReplyNotification(Ethernet eth, OFPacketIn pi) {
ARP arp = (ARP) eth.getPayload();
if (log.isTraceEnabled()) {
@@ -575,12 +547,14 @@
log.error("Unknown host", e);
return;
}
-
- datagrid.sendArpRequest(ArpMessage.newReply(targetAddress, mac));
- //datagrid.sendArpReply(ArpMessage.newRequest(targetAddress, eth.serialize()));
-
+
+ datagrid.sendArpReplyNotification(new ArpReplyNotification(targetAddress, mac));
}
+ // This remains from the older single-instance ARP code. It used Floodlight
+ // APIs to find the edge of the network, but only worked on a single instance.
+ // We now do this using ONOS network graph APIs.
+ @Deprecated
private void broadcastArpRequestOutEdge(byte[] arpRequest, long inSwitch, short inPort) {
for (IOFSwitch sw : floodlightProvider.getSwitches().values()){
Collection<Short> enabledPorts = sw.getEnabledPortNumbers();
@@ -671,12 +645,7 @@
po.setLengthU(OFPacketOut.MINIMUM_LENGTH + actionsLength
+ arpRequest.length);
- try {
- sw.write(po, null);
- sw.flush();
- } catch (IOException e) {
- log.error("Failure writing packet out to switch", e);
- }
+ flowPusher.add(sw, po);
}
if (log.isTraceEnabled()) {
@@ -710,12 +679,7 @@
return;
}
- try {
- sw.write(po, null);
- sw.flush();
- } catch (IOException e) {
- log.error("Failure writing packet out to switch", e);
- }
+ flowPusher.add(sw, po);
}
private void sendArpReply(ARP arpRequest, long dpid, short port, MACAddress targetMac) {
@@ -738,7 +702,6 @@
.setTargetProtocolAddress(arpRequest.getSenderProtocolAddress());
-
Ethernet eth = new Ethernet();
eth.setDestinationMACAddress(arpRequest.getSenderHardwareAddress())
.setSourceMACAddress(targetMac.toBytes())
@@ -773,12 +736,7 @@
return;
}
- try {
- sw.write(msgList, null);
- sw.flush();
- } catch (IOException e) {
- log.error("Failure writing packet out to switch", e);
- }
+ flowPusher.add(sw, po);
}
private String inetAddressToString(byte[] bytes) {
@@ -818,9 +776,6 @@
}
/*
- * IArpEventHandler methods
- */
-
@Override
public void arpRequestNotification(ArpMessage arpMessage) {
log.debug("Received ARP notification from other instances");
@@ -842,6 +797,7 @@
break;
}
}
+ */
private void sendArpReplyToWaitingRequesters(InetAddress address, MACAddress mac) {
log.debug("Sending ARP reply for {} to requesters",
@@ -874,4 +830,33 @@
request.dispatchReply(address, mac);
}
}
+
+ @Override
+ public void arpReplyEvent(ArpReplyNotification arpReply) {
+ log.debug("Received ARP reply notification for {}",
+ arpReply.getTargetAddress());
+ sendArpReplyToWaitingRequesters(arpReply.getTargetAddress(),
+ arpReply.getTargetMacAddress());
+ }
+
+ @Override
+ public void packetOutNotification(
+ PacketOutNotification packetOutNotification) {
+
+ if (packetOutNotification instanceof SinglePacketOutNotification) {
+ SinglePacketOutNotification notification =
+ (SinglePacketOutNotification) packetOutNotification;
+ sendArpRequestOutPort(notification.packet, notification.getOutSwitch(),
+ notification.getOutPort());
+ }
+ else if (packetOutNotification instanceof BroadcastPacketOutNotification) {
+ BroadcastPacketOutNotification notification =
+ (BroadcastPacketOutNotification) packetOutNotification;
+ broadcastArpRequestOutMyEdge(notification.packet,
+ notification.getInSwitch(), notification.getInPort());
+ }
+ else {
+ log.warn("Unknown packet out notification received");
+ }
+ }
}