Merge "Moved old REST APIs to point to the new network graph/topology implementation."
diff --git a/src/main/java/net/onrc/onos/datagrid/HazelcastDatagrid.java b/src/main/java/net/onrc/onos/datagrid/HazelcastDatagrid.java
index 33fa54d..e7ad821 100755
--- a/src/main/java/net/onrc/onos/datagrid/HazelcastDatagrid.java
+++ b/src/main/java/net/onrc/onos/datagrid/HazelcastDatagrid.java
@@ -31,6 +31,7 @@
import net.onrc.onos.ofcontroller.util.FlowId;
import net.onrc.onos.ofcontroller.util.FlowPath;
import net.onrc.onos.ofcontroller.util.Pair;
+import net.onrc.onos.ofcontroller.util.PerformanceMonitor;
import net.onrc.onos.ofcontroller.util.serializers.KryoFactory;
import org.slf4j.Logger;
@@ -50,8 +51,6 @@
import com.hazelcast.instance.GroupProperties;
import net.onrc.onos.intent.Intent;
-import net.onrc.onos.ofcontroller.flowmanager.PerformanceMonitor;
-
/**
* A datagrid service that uses Hazelcast as a datagrid.
* The relevant data is stored in the Hazelcast datagrid and shared as
diff --git a/src/main/java/net/onrc/onos/ofcontroller/bgproute/BgpRoute.java b/src/main/java/net/onrc/onos/ofcontroller/bgproute/BgpRoute.java
index 1363c19..a3cb17b 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/bgproute/BgpRoute.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/bgproute/BgpRoute.java
@@ -3,7 +3,6 @@
import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
-import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@@ -90,10 +89,10 @@
import com.google.common.net.InetAddresses;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
-public class BgpRoute implements IFloodlightModule, IBgpRouteService,
+public class BgpRoute implements IFloodlightModule, IBgpRouteService,
ITopologyListener, IArpRequester,
IOFSwitchListener, IConfigInfoService {
-
+
private final static Logger log = LoggerFactory.getLogger(BgpRoute.class);
private IFloodlightProviderService floodlightProvider;
@@ -105,15 +104,15 @@
protected volatile IFlowService flowManagerService;
private IDeviceStorage deviceStorage;
private ITopoSwitchService topoSwitchService;
-
+
private IPatriciaTrie<RibEntry> ptree;
private IPatriciaTrie<Interface> interfacePtrie;
private BlockingQueue<RibUpdate> ribUpdates;
-
+
private String bgpdRestIp;
private String routerId;
private String configFilename = "config.json";
-
+
//We need to identify our flows somehow. But like it says in LearningSwitch.java,
//the controller/OS should hand out cookie IDs to prevent conflicts.
private final long APP_COOKIE = 0xa0000000000000L;
@@ -127,11 +126,11 @@
//need to be higher priority than this otherwise the rewrite may not get done
private final short SDNIP_PRIORITY = 10;
private final short ARP_PRIORITY = 20;
-
+
private final short BGP_PORT = 179;
-
+
private final int TOPO_DETECTION_WAIT = 2; //seconds
-
+
//Configuration stuff
private List<String> switches;
private Map<String, Interface> interfaces;
@@ -139,7 +138,7 @@
private SwitchPort bgpdAttachmentPoint;
private MACAddress bgpdMacAddress;
private short vlan;
-
+
//True when all switches have connected
private volatile boolean switchesConnected = false;
//True when we have a full mesh of shortest paths between gateways
@@ -147,22 +146,22 @@
private ArrayList<LDUpdate> linkUpdates;
private SingletonTask topologyChangeDetectorTask;
-
+
private SetMultimap<InetAddress, RibUpdate> prefixesWaitingOnArp;
-
+
private Map<InetAddress, Path> pathsWaitingOnArp;
-
+
private ExecutorService bgpUpdatesExecutor;
-
+
private Map<InetAddress, Path> pushedPaths;
private Map<Prefix, Path> prefixToPath;
// private Multimap<Prefix, PushedFlowMod> pushedFlows;
private Multimap<Prefix, FlowId> pushedFlowIds;
-
+
private FlowCache flowCache;
-
+
private volatile Topology topology = null;
-
+
private class TopologyChangeDetector implements Runnable {
@Override
public void run() {
@@ -170,21 +169,21 @@
synchronized (linkUpdates) {
//This is the model the REST API uses to retrieve network graph info
ITopoLinkService topoLinkService = new TopoLinkServiceImpl();
-
+
List<Link> activeLinks = topoLinkService.getActiveLinks();
-
+
Iterator<LDUpdate> it = linkUpdates.iterator();
while (it.hasNext()){
LDUpdate ldu = it.next();
- Link l = new Link(ldu.getSrc(), ldu.getSrcPort(),
+ Link l = new Link(ldu.getSrc(), ldu.getSrcPort(),
ldu.getDst(), ldu.getDstPort());
-
+
if (activeLinks.contains(l)){
it.remove();
}
}
}
-
+
if (!topologyReady) {
if (linkUpdates.isEmpty()){
//All updates have been seen in network map.
@@ -200,14 +199,14 @@
}
}
}
-
+
private void readConfiguration(String configFilename){
File gatewaysFile = new File(configFilename);
ObjectMapper mapper = new ObjectMapper();
-
+
try {
Configuration config = mapper.readValue(gatewaysFile, Configuration.class);
-
+
switches = config.getSwitches();
interfaces = new HashMap<String, Interface>();
for (Interface intf : config.getInterfaces()){
@@ -217,11 +216,11 @@
for (BgpPeer peer : config.getPeers()){
bgpPeers.put(peer.getIpAddress(), peer);
}
-
+
bgpdAttachmentPoint = new SwitchPort(
new Dpid(config.getBgpdAttachmentDpid()),
new Port(config.getBgpdAttachmentPort()));
-
+
bgpdMacAddress = config.getBgpdMacAddress();
vlan = config.getVlan();
} catch (JsonParseException e) {
@@ -234,26 +233,26 @@
log.error("Error reading JSON file", e);
System.exit(1);
}
-
+
//Populate the interface Patricia Trie
for (Interface intf : interfaces.values()) {
Prefix prefix = new Prefix(intf.getIpAddress().getAddress(), intf.getPrefixLength());
interfacePtrie.put(prefix, intf);
}
}
-
+
@Override
public Collection<Class<? extends IFloodlightService>> getModuleServices() {
- Collection<Class<? extends IFloodlightService>> l
+ Collection<Class<? extends IFloodlightService>> l
= new ArrayList<Class<? extends IFloodlightService>>();
l.add(IBgpRouteService.class);
l.add(IConfigInfoService.class);
return l;
}
-
+
@Override
public Map<Class<? extends IFloodlightService>, IFloodlightService> getServiceImpls() {
- Map<Class<? extends IFloodlightService>, IFloodlightService> m
+ Map<Class<? extends IFloodlightService>, IFloodlightService> m
= new HashMap<Class<? extends IFloodlightService>, IFloodlightService>();
m.put(IBgpRouteService.class, this);
m.put(IConfigInfoService.class, this);
@@ -262,23 +261,23 @@
@Override
public Collection<Class<? extends IFloodlightService>> getModuleDependencies() {
- Collection<Class<? extends IFloodlightService>> l
+ Collection<Class<? extends IFloodlightService>> l
= new ArrayList<Class<? extends IFloodlightService>>();
l.add(IFloodlightProviderService.class);
l.add(ITopologyService.class);
l.add(IRestApiService.class);
return l;
}
-
+
@Override
public void init(FloodlightModuleContext context)
throws FloodlightModuleException {
-
+
ptree = new PatriciaTrie<RibEntry>(32);
interfacePtrie = new PatriciaTrie<Interface>(32);
-
+
ribUpdates = new LinkedBlockingQueue<RibUpdate>();
-
+
// Register floodlight provider and REST handler.
floodlightProvider = context.getServiceImpl(IFloodlightProviderService.class);
topologyService = context.getServiceImpl(ITopologyService.class);
@@ -300,17 +299,17 @@
pathsWaitingOnArp = new HashMap<InetAddress, Path>();
prefixesWaitingOnArp = Multimaps.synchronizedSetMultimap(
HashMultimap.<InetAddress, RibUpdate>create());
-
+
pushedPaths = new HashMap<InetAddress, Path>();
prefixToPath = new HashMap<Prefix, Path>();
// pushedFlows = HashMultimap.<Prefix, PushedFlowMod>create();
pushedFlowIds = HashMultimap.<Prefix, FlowId>create();
-
+
flowCache = new FlowCache(floodlightProvider);
-
+
bgpUpdatesExecutor = Executors.newSingleThreadExecutor(
new ThreadFactoryBuilder().setNameFormat("bgp-updates-%d").build());
-
+
//Read in config values
bgpdRestIp = context.getConfigParams(this).get("BgpdRestIp");
if (bgpdRestIp == null){
@@ -320,7 +319,7 @@
else {
log.info("BgpdRestIp set to {}", bgpdRestIp);
}
-
+
routerId = context.getConfigParams(this).get("RouterId");
if (routerId == null){
log.error("RouterId property not found in config file");
@@ -335,10 +334,10 @@
configFilename = configFilenameParameter;
}
log.debug("Config file set to {}", configFilename);
-
+
readConfiguration(configFilename);
}
-
+
@Override
public void startUp(FloodlightModuleContext context) {
restApi.addRestletRoutable(new BgpRouteWebRoutable());
@@ -349,39 +348,43 @@
retrieveRib();
}
+ @Override
public IPatriciaTrie<RibEntry> getPtree() {
return ptree;
}
-
+
+ @Override
public void clearPtree() {
ptree = new PatriciaTrie<RibEntry>(32);
}
-
+
+ @Override
public String getBGPdRestIp() {
return bgpdRestIp;
}
-
+
+ @Override
public String getRouterId() {
return routerId;
}
-
+
private void retrieveRib(){
String url = "http://" + bgpdRestIp + "/wm/bgp/" + routerId;
String response = RestClient.get(url);
-
+
if (response.equals("")){
return;
}
-
+
response = response.replaceAll("\"", "'");
- JSONObject jsonObj = (JSONObject) JSONSerializer.toJSON(response);
+ JSONObject jsonObj = (JSONObject) JSONSerializer.toJSON(response);
JSONArray rib_json_array = jsonObj.getJSONArray("rib");
String router_id = jsonObj.getString("router-id");
int size = rib_json_array.size();
log.info("Retrived RIB of {} entries from BGPd", size);
-
+
for (int j = 0; j < size; j++) {
JSONObject second_json_object = rib_json_array.getJSONObject(j);
String prefix = second_json_object.getString("prefix");
@@ -402,17 +405,17 @@
log.warn("Wrong prefix format in RIB JSON: {}", prefix1);
continue;
}
-
+
RibEntry rib = new RibEntry(router_id, nexthop);
-
+
try {
ribUpdates.put(new RibUpdate(Operation.UPDATE, p, rib));
} catch (InterruptedException e) {
log.debug("Interrupted while pushing onto update queue");
}
- }
+ }
}
-
+
@Override
public void newRibUpdate(RibUpdate update) {
try {
@@ -422,36 +425,36 @@
Thread.currentThread().interrupt();
}
}
-
+
public synchronized void processRibAdd(RibUpdate update) {
Prefix prefix = update.getPrefix();
-
+
log.debug("Processing prefix add {}", prefix);
-
+
RibEntry rib = ptree.put(prefix, update.getRibEntry());
-
+
if (rib != null && !rib.equals(update.getRibEntry())) {
//There was an existing nexthop for this prefix. This update supersedes that,
//so we need to remove the old flows for this prefix from the switches
_processDeletePrefix(prefix, rib);
}
-
+
if (update.getRibEntry().getNextHop().equals(
InetAddresses.forString("0.0.0.0"))) {
//Route originated by SDN domain
//We don't handle these at the moment
- log.debug("Own route {} to {}", prefix,
+ log.debug("Own route {} to {}", prefix,
update.getRibEntry().getNextHop().getHostAddress());
return;
}
-
+
_processRibAdd(update);
}
-
+
private void _processRibAdd(RibUpdate update) {
Prefix prefix = update.getPrefix();
RibEntry rib = update.getRibEntry();
-
+
InetAddress dstIpAddress = rib.getNextHop();
MACAddress nextHopMacAddress;
@@ -459,7 +462,7 @@
// TODO if we do not treat the next hop as a device in the future, we need to update this
IDeviceObject nextHopDevice =
deviceStorage.getDeviceByIP(InetAddresses.coerceToInteger(dstIpAddress));
-
+
if (nextHopDevice == null){
log.debug("NextHopDevice for IP: {} is null", dstIpAddress);
prefixesWaitingOnArp.put(dstIpAddress,
@@ -469,7 +472,7 @@
}
nextHopMacAddress = MACAddress.valueOf(nextHopDevice.getMACAddress());
-
+
// Find the attachment point (egress interface) of the next hop
Interface egressInterface = null;
if (bgpPeers.containsKey(dstIpAddress)) {
@@ -488,9 +491,9 @@
return;
}
}
-
+
if (nextHopMacAddress == null) {
- prefixesWaitingOnArp.put(dstIpAddress,
+ prefixesWaitingOnArp.put(dstIpAddress,
new RibUpdate(Operation.UPDATE, prefix, rib));
proxyArp.sendArpRequest(dstIpAddress, this, true);
return;
@@ -505,16 +508,16 @@
calculateAndPushPath(path, nextHopMacAddress);
pushedPaths.put(dstIpAddress, path);
}
-
+
path.incrementUsers();
prefixToPath.put(prefix, path);
}
-
+
//For all prefixes we need to add the first-hop mac-rewriting flows
addPrefixFlows(prefix, egressInterface, nextHopMacAddress);
}
}
-
+
/**
* Add a flow to match dst-IP prefix and rewrite MAC for one IP prefix
* to all other border switches
@@ -523,10 +526,10 @@
MACAddress nextHopMacAddress) {
log.debug("Adding flows for prefix {}, next hop mac {}",
prefix, nextHopMacAddress);
-
+
FlowPath flowPath = new FlowPath();
flowPath.setInstallerId(new CallerId("SDNIP"));
-
+
// Set flowPath FlowPathType and FlowPathUserState
flowPath.setFlowPathType(FlowPathType.FP_TYPE_SHORTEST_PATH);
flowPath.setFlowPathUserState(FlowPathUserState.FP_USER_ADD);
@@ -545,7 +548,7 @@
// We only need one flow mod per switch, so pick one interface on each switch
Map<Long, Interface> srcInterfaces = new HashMap<Long, Interface>();
for (Interface intf : interfaces.values()) {
- if (!srcInterfaces.containsKey(intf.getDpid())
+ if (!srcInterfaces.containsKey(intf.getDpid())
&& !intf.equals(egressInterface)) {
srcInterfaces.put(intf.getDpid(), intf);
}
@@ -557,8 +560,6 @@
}
// Create flowPath FlowId
- //FlowId returnByRefFlowId = new FlowId(flowManagerService.getNextFlowEntryId());
- //flowPath.setFlowId(returnByRefFlowId);
flowPath.setFlowId(new FlowId());
// Create DataPath object: srcSwitchPort
@@ -603,10 +604,10 @@
}
}
}
-
+
public synchronized void processRibDelete(RibUpdate update) {
Prefix prefix = update.getPrefix();
-
+
if (ptree.remove(prefix, update.getRibEntry())) {
/*
* Only delete flows if an entry was actually removed from the trie.
@@ -616,20 +617,20 @@
_processDeletePrefix(prefix, update.getRibEntry());
}
}
-
+
private void _processDeletePrefix(Prefix prefix, RibEntry ribEntry) {
deletePrefixFlows(prefix);
-
+
log.debug("Deleting {} to {}", prefix, ribEntry.getNextHop());
-
+
if (!bgpPeers.containsKey(ribEntry.getNextHop())) {
log.debug("Getting path for route with non-peer nexthop");
Path path = prefixToPath.remove(prefix);
-
+
if (path != null) {
//path could be null if we added to the Ptree but didn't push
//flows yet because we were waiting to resolve ARP
-
+
path.decrementUsers();
if (path.getUsers() <= 0 && !path.isPermanent()) {
deletePath(path);
@@ -638,11 +639,11 @@
}
}
}
-
+
// TODO have not tested this module
private void deletePrefixFlows(Prefix prefix) {
log.debug("Deleting flows for prefix {}", prefix);
-
+
Collection<FlowId> flowIds = pushedFlowIds.removeAll(prefix);
for (FlowId flowId : flowIds) {
if (log.isTraceEnabled()) {
@@ -650,7 +651,7 @@
log.trace("Pushing a DELETE flow mod to flowPath : {}",
flowManagerService.getFlow(flowId).toString());
}
-
+
if( flowManagerService.deleteFlow(flowId))
{
log.debug("Successfully deleted FlowId: {}",flowId);
@@ -661,12 +662,12 @@
}
}
}
-
+
// TODO need to record the path and then delete here
private void deletePath(Path path) {
- log.debug("Deleting flows for path to {}",
+ log.debug("Deleting flows for path to {}",
path.getDstIpAddress().getHostAddress());
-
+
// TODO need update
/*for (PushedFlowMod pfm : path.getFlowMods()) {
if (log.isTraceEnabled()) {
@@ -679,8 +680,8 @@
sendDeleteFlowMod(pfm.getFlowMod(), pfm.getDpid());
}*/
}
-
-
+
+
//TODO test next-hop changes
//TODO check delete/add synchronization
@@ -692,17 +693,17 @@
//For each border router, calculate and install a path from every other
//border switch to said border router. However, don't install the entry
//in to the first hop switch, as we need to install an entry to rewrite
- //for each prefix received. This will be done later when prefixes have
+ //for each prefix received. This will be done later when prefixes have
//actually been received.
-
+
for (BgpPeer peer : bgpPeers.values()) {
Interface peerInterface = interfaces.get(peer.getInterfaceName());
-
+
//We know there's not already a Path here pushed, because this is
//called before all other routing
Path path = new Path(peerInterface, peer.getIpAddress());
path.setPermanent();
-
+
//See if we know the MAC address of the peer. If not we can't
//do anything until we learn it
MACAddress macAddress;
@@ -716,7 +717,7 @@
proxyArp.sendArpRequest(peer.getIpAddress(), this, true);
continue;
}
-
+
macAddress = MACAddress.valueOf(nextHopDevice.getMACAddress());
if (macAddress == null) {
@@ -726,15 +727,15 @@
proxyArp.sendArpRequest(peer.getIpAddress(), this, true);
continue;
}
-
+
//If we know the MAC, lets go ahead and push the paths to this peer
calculateAndPushPath(path, macAddress);
}
}
-
+
private void calculateAndPushPath(Path path, MACAddress dstMacAddress) {
Interface dstInterface = path.getDstInterface();
-
+
log.debug("Setting up path to {}, {}", path.getDstIpAddress().getHostAddress(),
dstMacAddress);
@@ -763,8 +764,6 @@
}
// Create flowPath FlowId
- //FlowId returnByRefFlowId = new FlowId(flowManagerService.getNextFlowEntryId());
- //flowPath.setFlowId(returnByRefFlowId);
flowPath.setFlowId(new FlowId());
// Create the DataPath object: srcSwitchPort
@@ -818,11 +817,6 @@
Interface peerInterface = interfaces.get(bgpPeer.getInterfaceName());
- // set flowPath FlowId
- //FlowId returnByRefFlowId = new FlowId(flowManagerService.getNextFlowEntryId());
- //flowPath.setFlowId(returnByRefFlowId);
-
-
// Create the Flow Path Match condition(s)
FlowEntryMatch flowEntryMatch = new FlowEntryMatch();
flowEntryMatch.enableEthernetFrameType(Ethernet.TYPE_IPv4);
@@ -874,8 +868,6 @@
flowPath.setFlowEntryMatch(flowEntryMatch);
// Create a new FlowId
- //FlowId returnByRefFlowId2 = new FlowId(flowManagerService.getNextFlowEntryId());
- //flowPath.setFlowId(returnByRefFlowId2);
flowPath.setFlowId(new FlowId());
if (flowManagerService.addFlow(flowPath) == null) {
@@ -891,8 +883,6 @@
* Create the DataPath: BGP <-BGP peer
*/
// Reversed BGP flow path for src-TCP-port
- //FlowId returnByRefFlowId3 = new FlowId(flowManagerService.getNextFlowEntryId());
- //flowPath.setFlowId(returnByRefFlowId3);
flowPath.setFlowId(new FlowId());
DataPath reverse_dataPath = new DataPath();
@@ -926,8 +916,6 @@
}
// Reversed BGP flow path for dst-TCP-port
- //FlowId returnByRefFlowId4 = new FlowId(flowManagerService.getNextFlowEntryId());
- //flowPath.setFlowId(returnByRefFlowId4);
flowPath.setFlowId(new FlowId());
// Disable src-TCP-port, and set the dst-TCP-port
@@ -936,7 +924,7 @@
flowPath.setFlowEntryMatch(flowEntryMatch);
log.debug("Reversed BGP FlowPath: {}", flowPath.toString());
-
+
if (flowManagerService.addFlow(flowPath) == null) {
log.error("Failed to setting up path BGP <- Peer {}" + "; dst-TCP-port:179",
bgpPeer.getIpAddress().getHostAddress());
@@ -950,8 +938,6 @@
* ICMP paths between BGPd and its peers
*/
//match ICMP protocol BGP <- Peer
- //FlowId returnByRefFlowId5 = new FlowId(flowManagerService.getNextFlowEntryId());
- //flowPath.setFlowId(returnByRefFlowId5);
flowPath.setFlowId(new FlowId());
flowEntryMatch.enableIpProto(IPv4.PROTOCOL_ICMP);
@@ -975,8 +961,6 @@
}
//match ICMP protocol BGP -> Peer
- //FlowId returnByRefFlowId6 = new FlowId(flowManagerService.getNextFlowEntryId());
- //flowPath.setFlowId(returnByRefFlowId6);
flowPath.setFlowId(new FlowId());
flowEntryMatch.enableDstIPv4Net(dstIPv4Net);
@@ -999,31 +983,31 @@
}
}
}
-
+
@Override
public void arpResponse(InetAddress ipAddress, MACAddress macAddress) {
- log.debug("Received ARP response: {} => {}",
+ log.debug("Received ARP response: {} => {}",
ipAddress.getHostAddress(), macAddress);
-
+
/*
* We synchronize on this to prevent changes to the ptree while we're pushing
* flows to the switches. If the ptree changes, the ptree and switches
- * could get out of sync.
+ * could get out of sync.
*/
synchronized (this) {
Path path = pathsWaitingOnArp.remove(ipAddress);
-
+
if (path != null) {
log.debug("Pushing path to {} at {} on {}", new Object[] {
path.getDstIpAddress().getHostAddress(), macAddress,
path.getDstInterface().getSwitchPort()});
//These paths should always be to BGP peers. Paths to non-peers are
//handled once the first prefix is ready to push
- if (pushedPaths.containsKey(path.getDstInterface())) {
+ if (pushedPaths.containsKey(path.getDstIpAddress())) {
//A path already got pushed to this endpoint while we were waiting
//for ARP. We'll copy over the permanent attribute if it is set on this path.
if (path.isPermanent()) {
- pushedPaths.get(path.getDstInterface()).setPermanent();
+ pushedPaths.get(path.getDstIpAddress()).setPermanent();
}
}
else {
@@ -1031,45 +1015,45 @@
pushedPaths.put(path.getDstIpAddress(), path);
}
}
-
+
Set<RibUpdate> prefixesToPush = prefixesWaitingOnArp.removeAll(ipAddress);
-
+
for (RibUpdate update : prefixesToPush) {
//These will always be adds
-
- RibEntry rib = ptree.lookup(update.getPrefix());
+
+ RibEntry rib = ptree.lookup(update.getPrefix());
if (rib != null && rib.equals(update.getRibEntry())) {
- log.debug("Pushing prefix {} next hop {}", update.getPrefix(),
+ log.debug("Pushing prefix {} next hop {}", update.getPrefix(),
rib.getNextHop().getHostAddress());
//We only push prefix flows if the prefix is still in the ptree
- //and the next hop is the same as our update. The prefix could
- //have been removed while we were waiting for the ARP, or the
+ //and the next hop is the same as our update. The prefix could
+ //have been removed while we were waiting for the ARP, or the
//next hop could have changed.
_processRibAdd(update);
} else {
- log.debug("Received ARP response, but {},{} is no longer in ptree",
+ log.debug("Received ARP response, but {},{} is no longer in ptree",
update.getPrefix(), update.getRibEntry());
}
}
}
}
-
+
//TODO wait the priority module of the flow Manager
private void setupArpFlows() {
OFMatch match = new OFMatch();
match.setDataLayerType(Ethernet.TYPE_ARP);
match.setWildcards(match.getWildcards() & ~OFMatch.OFPFW_DL_TYPE);
-
+
OFFlowMod fm = new OFFlowMod();
fm.setMatch(match);
-
+
OFActionOutput action = new OFActionOutput();
action.setPort(OFPort.OFPP_CONTROLLER.getValue());
action.setMaxLength((short)0xffff);
List<OFAction> actions = new ArrayList<OFAction>(1);
actions.add(action);
fm.setActions(actions);
-
+
fm.setIdleTimeout((short)0)
.setHardTimeout((short)0)
.setBufferId(OFPacketOut.BUFFER_ID_NONE)
@@ -1077,7 +1061,7 @@
.setCommand(OFFlowMod.OFPFC_ADD)
.setPriority(ARP_PRIORITY)
.setLengthU(OFFlowMod.MINIMUM_LENGTH + OFActionOutput.MINIMUM_LENGTH);
-
+
for (String strdpid : switches){
flowCache.write(HexString.toLong(strdpid), fm);
}
@@ -1087,7 +1071,7 @@
OFFlowMod fm = new OFFlowMod();
fm.setMatch(new OFMatch());
fm.setActions(new ArrayList<OFAction>()); //No action means drop
-
+
fm.setIdleTimeout((short)0)
.setHardTimeout((short)0)
.setBufferId(OFPacketOut.BUFFER_ID_NONE)
@@ -1095,7 +1079,7 @@
.setCommand(OFFlowMod.OFPFC_ADD)
.setPriority((short)0)
.setLengthU(OFFlowMod.MINIMUM_LENGTH);
-
+
OFFlowMod fmLLDP;
OFFlowMod fmBDDP;
try {
@@ -1105,57 +1089,57 @@
log.error("Error cloning flow mod", e1);
return;
}
-
+
OFMatch matchLLDP = new OFMatch();
matchLLDP.setDataLayerType((short)0x88cc);
matchLLDP.setWildcards(matchLLDP.getWildcards() & ~ OFMatch.OFPFW_DL_TYPE);
fmLLDP.setMatch(matchLLDP);
-
+
OFMatch matchBDDP = new OFMatch();
matchBDDP.setDataLayerType((short)0x8942);
matchBDDP.setWildcards(matchBDDP.getWildcards() & ~ OFMatch.OFPFW_DL_TYPE);
fmBDDP.setMatch(matchBDDP);
-
+
OFActionOutput action = new OFActionOutput();
action.setPort(OFPort.OFPP_CONTROLLER.getValue());
action.setMaxLength((short)0xffff);
List<OFAction> actions = new ArrayList<OFAction>(1);
actions.add(action);
-
+
fmLLDP.setActions(actions);
fmBDDP.setActions(actions);
-
+
fmLLDP.setPriority(ARP_PRIORITY);
fmLLDP.setLengthU(OFFlowMod.MINIMUM_LENGTH + OFActionOutput.MINIMUM_LENGTH);
fmBDDP.setPriority(ARP_PRIORITY);
fmBDDP.setLengthU(OFFlowMod.MINIMUM_LENGTH + OFActionOutput.MINIMUM_LENGTH);
-
- List<OFFlowMod> flowModList = new ArrayList<OFFlowMod>(3);
+
+ List<OFFlowMod> flowModList = new ArrayList<OFFlowMod>(3);
flowModList.add(fm);
flowModList.add(fmLLDP);
flowModList.add(fmBDDP);
-
+
for (String strdpid : switches){
flowCache.write(HexString.toLong(strdpid), flowModList);
}
}
-
+
private void beginRouting(){
log.debug("Topology is now ready, beginning routing function");
topology = topologyNetService.newDatabaseTopology();
-
+
// Wait Pavlin's API. We need the following functions.
/*setupArpFlows();
setupDefaultDropFlows();*/
-
+
setupBgpPaths();
setupFullMesh();
-
+
//Suppress link discovery on external-facing router ports
for (Interface intf : interfaces.values()) {
linkDiscoveryService.AddToSuppressLLDPs(intf.getDpid(), intf.getPort());
}
-
+
bgpUpdatesExecutor.execute(new Runnable() {
@Override
public void run() {
@@ -1163,7 +1147,7 @@
}
});
}
-
+
// Before inserting the paths for BGP traffic, we should check
// whether all the switches in the configure file are discovered by onos.
private void checkSwitchesConnected(){
@@ -1184,19 +1168,19 @@
}
switchesConnected = true;
}
-
+
//Actually we only need to go half way round to verify full mesh connectivity
//(n^2)/2
private void checkTopologyReady(){
for (Interface dstInterface : interfaces.values()) {
- for (Interface srcInterface : interfaces.values()) {
+ for (Interface srcInterface : interfaces.values()) {
if (dstInterface.equals(srcInterface)) {
continue;
}
-
+
DataPath shortestPath = topologyNetService.getDatabaseShortestPath(
srcInterface.getSwitchPort(), dstInterface.getSwitchPort());
-
+
if (shortestPath == null){
log.debug("Shortest path between {} and {} not found",
srcInterface.getSwitchPort(), dstInterface.getSwitchPort());
@@ -1206,7 +1190,7 @@
}
topologyReady = true;
}
-
+
private void checkStatus(){
if (!switchesConnected){
checkSwitchesConnected();
@@ -1259,11 +1243,11 @@
}
}
}
-
+
private boolean validateUpdate(RibUpdate update) {
RibEntry newEntry = update.getRibEntry();
RibEntry oldEntry = ptree.lookup(update.getPrefix());
-
+
//If there is no existing entry we must assume this is the most recent
//update. However this might not always be the case as we might have a
//POST then DELETE reordering.
@@ -1271,13 +1255,13 @@
if (oldEntry == null) {
return true;
}
-
+
// This handles the case where routes are gathered in the initial
// request because they don't have sequence number info
if (newEntry.getSysUpTime() == -1 && newEntry.getSequenceNum() == -1) {
return true;
}
-
+
if (newEntry.getSysUpTime() > oldEntry.getSysUpTime()) {
return true;
}
@@ -1299,7 +1283,7 @@
if (topologyReady) {
return;
}
-
+
boolean refreshNeeded = false;
for (LDUpdate ldu : topologyService.getLastLinkUpdates()){
if (!ldu.getOperation().equals(ILinkDiscovery.UpdateOperation.LINK_UPDATED)){
@@ -1307,16 +1291,16 @@
//They happen very frequently
refreshNeeded = true;
}
-
+
log.debug("Topo change {}", ldu.getOperation());
-
+
if (ldu.getOperation().equals(ILinkDiscovery.UpdateOperation.LINK_ADDED)){
synchronized (linkUpdates) {
linkUpdates.add(ldu);
}
}
}
-
+
if (refreshNeeded && !topologyReady){
topologyChangeDetectorTask.reschedule(TOPO_DETECTION_WAIT, TimeUnit.SECONDS);
}
@@ -1327,7 +1311,7 @@
if (!topologyReady) {
sw.clearAllFlowMods();
}
-
+
flowCache.switchConnected(sw);
}
@@ -1341,23 +1325,23 @@
public String getName() {
return "BgpRoute";
}
-
+
/*
* IConfigInfoService methods
*/
-
+
@Override
public boolean isInterfaceAddress(InetAddress address) {
Interface intf = interfacePtrie.match(new Prefix(address.getAddress(), 32));
return (intf != null && intf.getIpAddress().equals(address));
}
-
+
@Override
public boolean inConnectedNetwork(InetAddress address) {
Interface intf = interfacePtrie.match(new Prefix(address.getAddress(), 32));
return (intf != null && !intf.getIpAddress().equals(address));
}
-
+
@Override
public boolean fromExternalNetwork(long inDpid, short inPort) {
for (Interface intf : interfaces.values()) {
@@ -1367,22 +1351,22 @@
}
return false;
}
-
+
@Override
public Interface getOutgoingInterface(InetAddress dstIpAddress) {
return interfacePtrie.match(new Prefix(dstIpAddress.getAddress(), 32));
}
-
+
@Override
public boolean hasLayer3Configuration() {
return !interfaces.isEmpty();
}
-
+
@Override
public MACAddress getRouterMacAddress() {
return bgpdMacAddress;
}
-
+
@Override
public short getVlan() {
return vlan;
diff --git a/src/main/java/net/onrc/onos/ofcontroller/devicemanager/OnosDeviceManager.java b/src/main/java/net/onrc/onos/ofcontroller/devicemanager/OnosDeviceManager.java
index e9ddfa7..7b2b29b 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/devicemanager/OnosDeviceManager.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/devicemanager/OnosDeviceManager.java
@@ -45,37 +45,37 @@
protected final static Logger log = LoggerFactory.getLogger(OnosDeviceManager.class);
private static final int CLEANUP_SECOND = 60*60;
private static final int AGEING_MILLSEC = 60*60*1000;
-
+
private IDeviceStorage deviceStorage;
private IFloodlightProviderService floodlightProvider;
private final static ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
-
+
private IDatagridService datagrid;
private Map<Long, OnosDevice> mapDevice = new ConcurrentHashMap<Long, OnosDevice>();
-
+
public enum OnosDeviceUpdateType {
ADD, DELETE, UPDATE;
}
-
+
private class OnosDeviceUpdate implements IUpdate {
private OnosDevice device;
private OnosDeviceUpdateType type;
-
+
public OnosDeviceUpdate(OnosDevice device, OnosDeviceUpdateType type) {
this.device = device;
this.type = type;
}
-
+
@Override
public void dispatch() {
if(type == OnosDeviceUpdateType.ADD) {
- deviceStorage.addOnosDevice(device);
+ deviceStorage.addOnosDevice(device);
} else if (type == OnosDeviceUpdateType.DELETE){
- deviceStorage.deleteOnosDevice(device);
+ deviceStorage.deleteOnosDevice(device);
}
}
}
-
+
@Override
public String getName() {
return "onosdevicemanager";
@@ -90,7 +90,7 @@
@Override
public boolean isCallbackOrderingPostreq(OFType type, String name) {
- return type == OFType.PACKET_IN &&
+ return type == OFType.PACKET_IN &&
("proxyarpmanager".equals(name) || "onosforwarding".equals(name));
}
@@ -98,16 +98,16 @@
public Command receive(IOFSwitch sw, OFMessage msg, FloodlightContext cntx) {
if (msg.getType().equals(OFType.PACKET_IN)) {
OFPacketIn pi = (OFPacketIn) msg;
-
+
Ethernet eth = IFloodlightProviderService.bcStore.
get(cntx, IFloodlightProviderService.CONTEXT_PI_PAYLOAD);
-
+
return processPacketIn(sw, pi, eth);
}
-
+
return Command.CONTINUE;
}
-
+
private Command processPacketIn(IOFSwitch sw, OFPacketIn pi, Ethernet eth) {
long dpid =sw.getId();
short portId = pi.getInPort();
@@ -119,7 +119,7 @@
if (srcDevice == null){
return Command.STOP;
}
-
+
//We check if it is the same device in datagrid to suppress the device update
OnosDevice exDev = null;
if((exDev = mapDevice.get(mac)) != null ){
@@ -132,11 +132,11 @@
dpid, portId, srcDevice.getMacAddress(), srcDevice.getIpv4Address(), srcDevice.getLastSeenTimestamp().getTime());
}
return Command.CONTINUE;
- } else if (srcDevice.getIpv4Address() == null &&
- exDev.getSwitchDPID() == srcDevice.getSwitchDPID() &&
+ } else if (srcDevice.getIpv4Address() == null &&
+ exDev.getSwitchDPID().equals(srcDevice.getSwitchDPID()) &&
exDev.getSwitchPort() == srcDevice.getSwitchPort() &&
- exDev.getVlan() == srcDevice.getVlan()) {
- //Device attachment point and mac address are the same
+ exDev.getVlan().equals(srcDevice.getVlan())) {
+ //Device attachment point and mac address are the same
//but the packet does not have an ip address.
exDev.setLastSeenTimestamp(new Date());
if(log.isTraceEnabled()) {
@@ -147,7 +147,7 @@
return Command.CONTINUE;
}
}
-
+
//If the switch port we try to attach a new device already has a link, then stop adding device
Collection<TopologyElement> list = datagrid.getAllTopologyElements();
for(TopologyElement elem: list) {
@@ -162,17 +162,17 @@
}
}
}
-
+
addOnosDevice(mac, srcDevice);
-
+
if(log.isTraceEnabled()) {
log.debug("Add device info in the set. dpid {}, port {}, mac {}, ip {}, lastSeenTime {}",
dpid, portId, srcDevice.getMacAddress(), srcDevice.getIpv4Address(), srcDevice.getLastSeenTimestamp().getTime());
}
return Command.CONTINUE;
}
-
- //Thread to delete devices periodically.
+
+ //Thread to delete devices periodically.
//Remove all devices from the map first and then finally delete devices from the DB.
private class CleanDevice implements Runnable {
@Override
@@ -191,9 +191,9 @@
deleteSet.add(dev);
}
}
-
+
for(OnosDevice dev : deleteSet) {
- deleteOnosDevice(dev);
+ deleteOnosDevice(dev);
}
} catch(Exception e) {
log.error("Error:", e);
@@ -202,7 +202,7 @@
}
/**
- * Get IP address from packet if the packet is either an ARP
+ * Get IP address from packet if the packet is either an ARP
* or a DHCP packet
* @param eth
* @param dlAddr
@@ -259,7 +259,7 @@
@Override
public Collection<Class<? extends IFloodlightService>> getModuleServices() {
- List<Class<? extends IFloodlightService>> services =
+ List<Class<? extends IFloodlightService>> services =
new ArrayList<Class<? extends IFloodlightService>>();
services.add(IOnosDeviceService.class);
return services;
@@ -267,7 +267,7 @@
@Override
public Map<Class<? extends IFloodlightService>, IFloodlightService> getServiceImpls() {
- Map<Class<? extends IFloodlightService>, IFloodlightService> impls =
+ Map<Class<? extends IFloodlightService>, IFloodlightService> impls =
new HashMap<Class<? extends IFloodlightService>, IFloodlightService>();
impls.put(IOnosDeviceService.class, this);
return impls;
@@ -275,7 +275,7 @@
@Override
public Collection<Class<? extends IFloodlightService>> getModuleDependencies() {
- List<Class<? extends IFloodlightService>> dependencies =
+ List<Class<? extends IFloodlightService>> dependencies =
new ArrayList<Class<? extends IFloodlightService>>();
dependencies.add(IFloodlightProviderService.class);
return dependencies;
@@ -288,7 +288,7 @@
executor.scheduleAtFixedRate(new CleanDevice(), 30 ,CLEANUP_SECOND, TimeUnit.SECONDS);
deviceStorage = new DeviceStorageImpl();
deviceStorage.init("","");
-
+
datagrid = context.getServiceImpl(IDatagridService.class);
}
@@ -303,30 +303,30 @@
datagrid.sendNotificationDeviceDeleted(dev);
floodlightProvider.publishUpdate(new OnosDeviceUpdate(dev, OnosDeviceUpdateType.DELETE));
}
-
+
@Override
public void addOnosDevice(Long mac, OnosDevice dev) {
datagrid.sendNotificationDeviceAdded(mac, dev);
floodlightProvider.publishUpdate(new OnosDeviceUpdate(dev, OnosDeviceUpdateType.ADD));
}
-
+
//This is listener for datagrid mapDevice change.
class MapDevListener implements IDeviceEventHandler {
@Override
- public void addDeviceEvent(Long mac, OnosDevice dev) {
+ public void addDeviceEvent(Long mac, OnosDevice dev) {
mapDevice.put(mac, dev);
log.debug("addDeviceMap: device mac {}", mac);
}
@Override
- public void deleteDeviceEvent(Long mac, OnosDevice dev) {
+ public void deleteDeviceEvent(Long mac, OnosDevice dev) {
mapDevice.remove(mac);
- log.debug("deleteDeviceMap: device mac {}", mac);
+ log.debug("deleteDeviceMap: device mac {}", mac);
}
@Override
- public void updateDeviceEvent(Long mac, OnosDevice dev) {
+ public void updateDeviceEvent(Long mac, OnosDevice dev) {
mapDevice.put(mac, dev);
log.debug("updateDeviceMap: device mac {}", mac);
}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/floodlightlistener/NetworkGraphPublisher.java b/src/main/java/net/onrc/onos/ofcontroller/floodlightlistener/NetworkGraphPublisher.java
index f77a62c..5495ca0 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/floodlightlistener/NetworkGraphPublisher.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/floodlightlistener/NetworkGraphPublisher.java
@@ -41,12 +41,11 @@
import net.onrc.onos.ofcontroller.linkdiscovery.LinkInfo;
import net.onrc.onos.ofcontroller.proxyarp.ArpReplyNotification;
import net.onrc.onos.ofcontroller.topology.TopologyElement;
+import net.onrc.onos.ofcontroller.util.PerformanceMonitor;
import net.onrc.onos.registry.controller.IControllerRegistryService;
import net.onrc.onos.registry.controller.IControllerRegistryService.ControlChangeCallback;
import net.onrc.onos.registry.controller.RegistryException;
-import net.onrc.onos.ofcontroller.flowmanager.PerformanceMonitor;
-
import org.openflow.protocol.OFPhysicalPort;
import org.openflow.util.HexString;
import org.slf4j.Logger;
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
index 116af1f..4c2bdf8 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
@@ -34,8 +34,10 @@
import net.onrc.onos.ofcontroller.util.FlowPath;
import net.onrc.onos.ofcontroller.util.FlowPathUserState;
import net.onrc.onos.ofcontroller.util.Pair;
+import net.onrc.onos.ofcontroller.util.PerformanceMonitor;
import net.onrc.onos.ofcontroller.util.Port;
import net.onrc.onos.ofcontroller.util.serializers.KryoFactory;
+import net.onrc.onos.registry.controller.IControllerRegistryService;
import com.esotericsoftware.kryo.Kryo;
import org.slf4j.Logger;
@@ -63,6 +65,7 @@
private FlowManager flowManager; // The Flow Manager to use
private IDatagridService datagridService; // The Datagrid Service to use
+ private IControllerRegistryService registryService; // The Registry Service
private Topology topology; // The network topology
private KryoFactory kryoFactory = new KryoFactory();
@@ -109,11 +112,14 @@
*
* @param flowManager the Flow Manager to use.
* @param datagridService the Datagrid Service to use.
+ * @param registryService the Registry Service to use.
*/
FlowEventHandler(FlowManager flowManager,
- IDatagridService datagridService) {
+ IDatagridService datagridService,
+ IControllerRegistryService registryService) {
this.flowManager = flowManager;
this.datagridService = datagridService;
+ this.registryService = registryService;
this.topology = new Topology();
}
@@ -323,7 +329,7 @@
for (FlowPath flowPath : modifiedFlowPaths.values()) {
for (FlowEntry flowEntry : flowPath.flowEntries()) {
if (! flowEntry.isValidFlowEntryId()) {
- long id = flowManager.getNextFlowEntryId();
+ long id = registryService.getNextUniqueId();
flowEntry.setFlowEntryId(new FlowEntryId(id));
}
}
@@ -543,7 +549,7 @@
if (mySwitch == null)
continue;
if (! flowEntry.isValidFlowEntryId()) {
- long id = flowManager.getNextFlowEntryId();
+ long id = registryService.getNextUniqueId();
flowEntry.setFlowEntryId(new FlowEntryId(id));
}
}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
index 6c09977..55266b3 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
@@ -6,7 +6,6 @@
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Random;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
@@ -38,6 +37,7 @@
import net.onrc.onos.ofcontroller.util.FlowPath;
import net.onrc.onos.ofcontroller.util.FlowPathUserState;
import net.onrc.onos.ofcontroller.util.Pair;
+import net.onrc.onos.ofcontroller.util.PerformanceMonitor;
import net.onrc.onos.ofcontroller.util.serializers.KryoFactory;
import net.onrc.onos.registry.controller.IControllerRegistryService;
@@ -69,11 +69,6 @@
private KryoFactory kryoFactory = new KryoFactory();
- // Flow Entry ID generation state
- private static Random randomGenerator = new Random();
- private static int nextFlowEntryIdPrefix = 0;
- private static int nextFlowEntryIdSuffix = 0;
-
/** The logger. */
private final static Logger log = LoggerFactory.getLogger(FlowManager.class);
@@ -185,30 +180,6 @@
}
/**
- * Get the next Flow Entry ID to use.
- *
- * @return the next Flow Entry ID to use.
- */
- @Override
- public synchronized long getNextFlowEntryId() {
- //
- // Generate the next Flow Entry ID.
- // NOTE: For now, the higher 32 bits are random, and
- // the lower 32 bits are sequential.
- // In the future, we need a better allocation mechanism.
- //
- if ((nextFlowEntryIdSuffix & 0xffffffffL) == 0xffffffffL) {
- nextFlowEntryIdPrefix = randomGenerator.nextInt();
- nextFlowEntryIdSuffix = 0;
- } else {
- nextFlowEntryIdSuffix++;
- }
- long result = (long)nextFlowEntryIdPrefix << 32;
- result = result | (0xffffffffL & nextFlowEntryIdSuffix);
- return result;
- }
-
- /**
* Startup module operation.
*
* @param context the module context to use for the startup.
@@ -217,9 +188,6 @@
public void startUp(FloodlightModuleContext context) {
restApi.addRestletRoutable(new FlowWebRoutable());
- // Initialize the Flow Entry ID generator
- nextFlowEntryIdPrefix = randomGenerator.nextInt();
-
//
// The thread to write to the database
//
@@ -233,7 +201,8 @@
// - register with the Datagrid Service
// - startup
//
- flowEventHandler = new FlowEventHandler(this, datagridService);
+ flowEventHandler = new FlowEventHandler(this, datagridService,
+ registryService);
floodlightProvider.addOFSwitchListener(flowEventHandler);
datagridService.registerFlowEventHandlerService(flowEventHandler);
flowEventHandler.start();
@@ -293,7 +262,7 @@
// Allocate the Flow ID if necessary
if (! flowPath.isValidFlowId()) {
- long id = getNextFlowEntryId();
+ long id = registryService.getNextUniqueId();
flowPath.setFlowId(new FlowId(id));
}
@@ -309,7 +278,7 @@
}
// The Flow Entry ID
if (! flowEntry.isValidFlowEntryId()) {
- long id = getNextFlowEntryId();
+ long id = registryService.getNextUniqueId();
flowEntry.setFlowEntryId(new FlowEntryId(id));
}
// The Flow ID
@@ -674,7 +643,7 @@
// assignments by the caller).
//
if (! flowEntry.isValidFlowEntryId()) {
- long id = getNextFlowEntryId();
+ long id = registryService.getNextUniqueId();
flowEntry.setFlowEntryId(new FlowEntryId(id));
}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowService.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowService.java
index 549a0fc..d6a5db1 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowService.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowService.java
@@ -71,14 +71,6 @@
Topology getTopology();
/**
- * Get a globally unique flow ID from the flow service.
- * NOTE: Not currently guaranteed to be globally unique.
- *
- * @return unique flow ID
- */
- public long getNextFlowEntryId();
-
- /**
* Inform the Flow Manager that a Flow Entry on switch expired.
*
* @param sw the switch the Flow Entry expired on.
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/ParallelFlowDatabaseOperation.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/ParallelFlowDatabaseOperation.java
index e9cafee..b5fd95d 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/ParallelFlowDatabaseOperation.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/ParallelFlowDatabaseOperation.java
@@ -23,6 +23,7 @@
import net.onrc.onos.ofcontroller.util.FlowEntrySwitchState;
import net.onrc.onos.ofcontroller.util.FlowId;
import net.onrc.onos.ofcontroller.util.FlowPath;
+import net.onrc.onos.ofcontroller.util.PerformanceMonitor;
import net.onrc.onos.ofcontroller.util.serializers.KryoFactory;
import com.esotericsoftware.kryo.Kryo;
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/web/GetFlowByIdResource.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/web/GetFlowByIdResource.java
index de625d7..1cbeece 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/web/GetFlowByIdResource.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/web/GetFlowByIdResource.java
@@ -1,7 +1,6 @@
package net.onrc.onos.ofcontroller.flowmanager.web;
import net.onrc.onos.ofcontroller.flowmanager.IFlowService;
-import net.onrc.onos.ofcontroller.flowmanager.PerformanceMonitor;
import net.onrc.onos.ofcontroller.util.FlowId;
import net.onrc.onos.ofcontroller.util.FlowPath;
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowPusher.java b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowPusher.java
index d3476c7..3a0407f 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowPusher.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowPusher.java
@@ -14,7 +14,9 @@
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Semaphore;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import org.openflow.protocol.*;
import org.openflow.protocol.action.*;
@@ -271,33 +273,39 @@
*
*/
private class FlowPusherThread extends Thread {
- private Map<IOFSwitch,SwitchQueue> queues
+ private Map<IOFSwitch,SwitchQueue> assignedQueues
= new ConcurrentHashMap<IOFSwitch,SwitchQueue>();
- // reusable latch used for waiting for arrival of message
- private Semaphore mutex = new Semaphore(0);
+ final Lock queuingLock = new ReentrantLock();
+ final Condition messagePushed = queuingLock.newCondition();
@Override
public void run() {
this.setName("FlowPusherThread " + this.getId() );
while (true) {
- try {
- // wait for message pushed to queue
- mutex.acquire();
- } catch (InterruptedException e) {
- // Interrupted to be shut down (not an error)
- log.debug("FlowPusherThread is interrupted");
- return;
+ while (! queuesHasMessageToSend()) {
+ queuingLock.lock();
+
+ try {
+ // wait for message pushed to queue
+ messagePushed.await();
+ } catch (InterruptedException e) {
+ // Interrupted to be shut down (not an error)
+ log.debug("FlowPusherThread is interrupted");
+ return;
+ } finally {
+ queuingLock.unlock();
+ }
}
// for safety of concurrent access, copy set of key objects
- Set<IOFSwitch> keys = new HashSet<IOFSwitch>(queues.size());
- for (IOFSwitch sw : queues.keySet()) {
+ Set<IOFSwitch> keys = new HashSet<IOFSwitch>(assignedQueues.size());
+ for (IOFSwitch sw : assignedQueues.keySet()) {
keys.add(sw);
}
for (IOFSwitch sw : keys) {
- SwitchQueue queue = queues.get(sw);
+ SwitchQueue queue = assignedQueues.get(sw);
if (sw == null || queue == null) {
continue;
@@ -305,16 +313,9 @@
synchronized (queue) {
processQueue(sw, queue, MAX_MESSAGE_SEND);
- if (! queue.hasMessageToSend()) {
+ if (queue.toBeDeleted && ! queue.hasMessageToSend()) {
// remove queue if flagged to be.
- if (queue.toBeDeleted) {
- queues.remove(sw);
- }
- } else {
- // Free the latch if message remains in queue
- if (mutex.availablePermits() == 0) {
- mutex.release();
- }
+ assignedQueues.remove(sw);
}
}
}
@@ -365,6 +366,25 @@
queue.logSentData(current_time, size);
}
}
+
+ private boolean queuesHasMessageToSend() {
+ for (SwitchQueue queue : assignedQueues.values()) {
+ if (queue.hasMessageToSend()) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ private void notifyMessagePushed() {
+ queuingLock.lock();
+ try {
+ messagePushed.signal();
+ } finally {
+ queuingLock.unlock();
+ }
+ }
}
/**
@@ -403,7 +423,8 @@
this.context = context;
this.factory = factory;
this.threadPool = modContext.getServiceImpl(IThreadPoolService.class);
- IFloodlightProviderService flservice = modContext.getServiceImpl(IFloodlightProviderService.class);
+ IFloodlightProviderService flservice
+ = modContext.getServiceImpl(IFloodlightProviderService.class);
flservice.addOFMessageListener(OFType.BARRIER_REPLY, this);
if (damper != null) {
@@ -467,9 +488,8 @@
// Free the latch if queue has any messages
FlowPusherThread thread = getProcessingThread(sw);
- if (queue.hasMessageToSend() &&
- thread.mutex.availablePermits() == 0) {
- thread.mutex.release();
+ if (queue.hasMessageToSend()) {
+ thread.notifyMessagePushed();
}
return true;
}
@@ -532,7 +552,7 @@
FlowPusherThread proc = getProcessingThread(sw);
queue = new SwitchQueue();
queue.state = QueueState.READY;
- proc.queues.put(sw, queue);
+ proc.assignedQueues.put(sw, queue);
return queue;
}
@@ -547,7 +567,7 @@
FlowPusherThread proc = getProcessingThread(sw);
if (forceStop) {
- SwitchQueue queue = proc.queues.remove(sw);
+ SwitchQueue queue = proc.assignedQueues.remove(sw);
if (queue == null) {
return false;
}
@@ -889,7 +909,8 @@
* @return
*/
protected boolean addMessageImpl(IOFSwitch sw, OFMessage msg, MsgPriority priority) {
- FlowPusherThread proc = getProcessingThread(sw);
+ FlowPusherThread thread = getProcessingThread(sw);
+
SwitchQueue queue = getQueue(sw);
// create queue at first addition of message
@@ -898,6 +919,7 @@
}
SwitchQueueEntry entry = new SwitchQueueEntry(msg);
+
synchronized (queue) {
queue.add(entry,priority);
if (log.isTraceEnabled()) {
@@ -905,9 +927,7 @@
}
}
- if (proc.mutex.availablePermits() == 0) {
- proc.mutex.release();
- }
+ thread.notifyMessagePushed();
return true;
}
@@ -972,7 +992,7 @@
return null;
}
- return th.queues.get(sw);
+ return th.assignedQueues.get(sw);
}
/**
diff --git a/src/main/java/net/onrc/onos/ofcontroller/forwarding/Forwarding.java b/src/main/java/net/onrc/onos/ofcontroller/forwarding/Forwarding.java
index 9c3b5ee..5ddee15 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/forwarding/Forwarding.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/forwarding/Forwarding.java
@@ -42,6 +42,7 @@
import net.onrc.onos.ofcontroller.util.FlowPathUserState;
import net.onrc.onos.ofcontroller.util.Port;
import net.onrc.onos.ofcontroller.util.SwitchPort;
+import net.onrc.onos.registry.controller.IControllerRegistryService;
import org.openflow.protocol.OFMessage;
import org.openflow.protocol.OFPacketIn;
@@ -74,6 +75,7 @@
private IFlowService flowService;
private IFlowPusherService flowPusher;
private IDatagridService datagrid;
+ private IControllerRegistryService controllerRegistryService;
private IDeviceStorage deviceStorage;
private TopologyManager topologyService;
@@ -163,6 +165,7 @@
dependencies.add(IFlowService.class);
dependencies.add(IFlowPusherService.class);
dependencies.add(IOnosDeviceService.class);
+ dependencies.add(IControllerRegistryService.class);
// We don't use the IProxyArpService directly, but reactive forwarding
// requires it to be loaded and answering ARP requests
dependencies.add(IProxyArpService.class);
@@ -176,6 +179,7 @@
flowService = context.getServiceImpl(IFlowService.class);
flowPusher = context.getServiceImpl(IFlowPusherService.class);
datagrid = context.getServiceImpl(IDatagridService.class);
+ controllerRegistryService = context.getServiceImpl(IControllerRegistryService.class);
floodlightProvider.addOFMessageListener(OFType.PACKET_IN, this);
@@ -411,7 +415,7 @@
flowPath.flowEntryMatch().enableEthernetFrameType(Ethernet.TYPE_IPv4);
flowPath.setDataPath(datapath);
- FlowId flowId = new FlowId(flowService.getNextFlowEntryId());
+ FlowId flowId = new FlowId(controllerRegistryService.getNextUniqueId());
flowPath.setFlowId(flowId);
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/PerformanceMonitor.java b/src/main/java/net/onrc/onos/ofcontroller/util/PerformanceMonitor.java
similarity index 99%
rename from src/main/java/net/onrc/onos/ofcontroller/flowmanager/PerformanceMonitor.java
rename to src/main/java/net/onrc/onos/ofcontroller/util/PerformanceMonitor.java
index 15e5cdc..189ea6c 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/PerformanceMonitor.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/util/PerformanceMonitor.java
@@ -1,4 +1,4 @@
-package net.onrc.onos.ofcontroller.flowmanager;
+package net.onrc.onos.ofcontroller.util;
import java.util.Map.Entry;
import java.util.Queue;
diff --git a/src/main/java/net/onrc/onos/registry/controller/IControllerRegistryService.java b/src/main/java/net/onrc/onos/registry/controller/IControllerRegistryService.java
index eb27cf9..41f7d55 100755
--- a/src/main/java/net/onrc/onos/registry/controller/IControllerRegistryService.java
+++ b/src/main/java/net/onrc/onos/registry/controller/IControllerRegistryService.java
@@ -135,5 +135,12 @@
* Get next unique id and retrieve a new range of ids if needed.
*/
public IdBlock allocateUniqueIdBlock(long range);
-
+
+
+ /**
+ * Get a globally unique ID.
+ *
+ * @return a globally unique ID.
+ */
+ public long getNextUniqueId();
}
diff --git a/src/main/java/net/onrc/onos/registry/controller/StandaloneRegistry.java b/src/main/java/net/onrc/onos/registry/controller/StandaloneRegistry.java
index 1e2934b..d581590 100755
--- a/src/main/java/net/onrc/onos/registry/controller/StandaloneRegistry.java
+++ b/src/main/java/net/onrc/onos/registry/controller/StandaloneRegistry.java
@@ -5,6 +5,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
import net.floodlightcontroller.core.IFloodlightProviderService;
import net.floodlightcontroller.core.module.FloodlightModuleContext;
@@ -33,7 +34,11 @@
protected String controllerId = null;
protected Map<String, ControlChangeCallback> switchCallbacks;
-
+
+ //
+ // Unique ID generation state
+ //
+ private static AtomicLong nextUniqueId = new AtomicLong(0);
@Override
public void requestControl(long dpid, ControlChangeCallback cb)
@@ -141,6 +146,16 @@
return block;
}
+ /**
+ * Get a globally unique ID.
+ *
+ * @return a globally unique ID.
+ */
+ @Override
+ public long getNextUniqueId() {
+ return nextUniqueId.incrementAndGet();
+ }
+
@Override
public Collection<Class<? extends IFloodlightService>> getModuleServices() {
Collection<Class<? extends IFloodlightService>> l =
diff --git a/src/main/java/net/onrc/onos/registry/controller/ZookeeperRegistry.java b/src/main/java/net/onrc/onos/registry/controller/ZookeeperRegistry.java
index b8a1021..ef45c64 100755
--- a/src/main/java/net/onrc/onos/registry/controller/ZookeeperRegistry.java
+++ b/src/main/java/net/onrc/onos/registry/controller/ZookeeperRegistry.java
@@ -8,6 +8,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
@@ -88,7 +89,16 @@
//Zookeeper performance-related configuration
protected static final int sessionTimeout = 5000;
protected static final int connectionTimeout = 7000;
-
+
+ //
+ // Unique ID generation state
+ // TODO: The implementation must be updated to use the Zookeeper
+ // instead of a ramdon generator.
+ //
+ private static Random randomGenerator = new Random();
+ private static int nextUniqueIdPrefix = 0;
+ private static int nextUniqueIdSuffix = 0;
+
private final BlockingQueue<SwitchLeaderEvent> switchLeadershipEvents =
new LinkedBlockingQueue<SwitchLeaderEvent>();
@@ -474,7 +484,33 @@
public IdBlock allocateUniqueIdBlock(){
return allocateUniqueIdBlock(ID_BLOCK_SIZE);
}
-
+
+ /**
+ * Get a globally unique ID.
+ *
+ * @return a globally unique ID.
+ */
+ @Override
+ public synchronized long getNextUniqueId() {
+ //
+ // Generate the next Unique ID.
+ //
+ // TODO: For now, the higher 32 bits are random, and
+ // the lower 32 bits are sequential.
+ // The implementation must be updated to use the Zookeeper
+ // to allocate the higher 32 bits (globally unique).
+ //
+ if ((nextUniqueIdSuffix & 0xffffffffL) == 0xffffffffL) {
+ nextUniqueIdPrefix = randomGenerator.nextInt();
+ nextUniqueIdSuffix = 0;
+ } else {
+ nextUniqueIdSuffix++;
+ }
+ long result = (long)nextUniqueIdPrefix << 32;
+ result = result | (0xffffffffL & nextUniqueIdSuffix);
+ return result;
+ }
+
/*
* IFloodlightModule
*/
@@ -517,7 +553,13 @@
this.connectionString = connectionString;
}
log.info("Setting Zookeeper connection string to {}", this.connectionString);
-
+
+ //
+ // Initialize the Unique ID generator
+ // TODO: This must be replaced by Zookeeper-based allocation
+ //
+ nextUniqueIdPrefix = randomGenerator.nextInt();
+
restApi = context.getServiceImpl(IRestApiService.class);
switches = new ConcurrentHashMap<String, SwitchLeadershipData>();