* Refactor some of of the internals of the FlowEventHandler, so
now the processing logic is simpler.
* Change the logic of writing Flow state to Titan: now we write the
complete FlowPath, so the ordering of the corresponding FlowEntries
is preserved.
This makes debugging easier, and fixes an issue with the GUI where
the flows didn't show-up nicely.
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java
index 58a590f..71479a1 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java
@@ -68,6 +68,21 @@
}
//
+ // Remove the old Flow Entries
+ //
+ if (found) {
+ Iterable<IFlowEntry> flowEntries = flowObj.getFlowEntries();
+ LinkedList<IFlowEntry> deleteFlowEntries =
+ new LinkedList<IFlowEntry>();
+ for (IFlowEntry flowEntryObj : flowEntries)
+ deleteFlowEntries.add(flowEntryObj);
+ for (IFlowEntry flowEntryObj : deleteFlowEntries) {
+ flowObj.removeFlowEntry(flowEntryObj);
+ dbHandler.removeFlowEntry(flowEntryObj);
+ }
+ }
+
+ //
// Set the Flow key:
// - flowId
//
@@ -155,6 +170,9 @@
// flowPath.dataPath().flowEntries()
//
for (FlowEntry flowEntry : flowPath.dataPath().flowEntries()) {
+ if (flowEntry.flowEntryUserState() == FlowEntryUserState.FE_USER_DELETE)
+ continue; // Skip: all Flow Entries were deleted earlier
+
if (addFlowEntry(flowManager, dbHandler, flowObj, flowEntry) == null) {
dbHandler.rollback();
return false;
@@ -186,14 +204,6 @@
// Flow edges
// HeadFE (TODO)
- //
- // Assign the FlowEntry ID.
- //
- if (! flowEntry.isValidFlowEntryId()) {
- long id = flowManager.getNextFlowEntryId();
- flowEntry.setFlowEntryId(new FlowEntryId(id));
- }
-
IFlowEntry flowEntryObj = null;
boolean found = false;
try {
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 eccf40b..feb80e1 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
@@ -58,9 +58,6 @@
private FlowManager flowManager; // The Flow Manager to use
private IDatagridService datagridService; // The Datagrid Service to use
private Topology topology; // The network topology
- private Map<Long, FlowPath> allFlowPaths = new HashMap<Long, FlowPath>();
- private Map<Long, FlowEntry> unmatchedFlowEntryAdd =
- new HashMap<Long, FlowEntry>();
// The queue with Flow Path and Topology Element updates
private BlockingQueue<EventEntry<?>> networkEvents =
@@ -74,21 +71,22 @@
private List<EventEntry<FlowEntry>> flowEntryEvents =
new LinkedList<EventEntry<FlowEntry>>();
+ // All internally computed Flow Paths
+ private Map<Long, FlowPath> allFlowPaths = new HashMap<Long, FlowPath>();
+
+ // The Flow Entries received as notifications with unmatched Flow Paths
+ private Map<Long, FlowEntry> unmatchedFlowEntryAdd =
+ new HashMap<Long, FlowEntry>();
+
//
// Transient state for processing the Flow Paths:
- // - The new Flow Paths
// - The Flow Paths that should be recomputed
// - The Flow Paths with modified Flow Entries
- // - The Flow Entries that were updated
//
- private List<FlowPath> newFlowPaths = new LinkedList<FlowPath>();
- private List<FlowPath> recomputeFlowPaths = new LinkedList<FlowPath>();
- private List<FlowPath> modifiedFlowPaths = new LinkedList<FlowPath>();
- private List<FlowPathEntryPair> updatedFlowEntries =
- new LinkedList<FlowPathEntryPair>();
- private List<FlowPathEntryPair> unmatchedDeleteFlowEntries =
- new LinkedList<FlowPathEntryPair>();
-
+ private Map<Long, FlowPath> shouldRecomputeFlowPaths =
+ new HashMap<Long, FlowPath>();
+ private Map<Long, FlowPath> modifiedFlowPaths =
+ new HashMap<Long, FlowPath>();
/**
* Constructor for a given Flow Manager and Datagrid Service.
@@ -218,22 +216,15 @@
processFlowPathEvents();
processTopologyEvents();
- //
- // Add all new Flows: should be done after processing the Flow Path
- // and Topology events.
- //
- for (FlowPath flowPath : newFlowPaths) {
- allFlowPaths.put(flowPath.flowId().value(), flowPath);
- }
-
processFlowEntryEvents();
// Recompute all affected Flow Paths and keep only the modified
- for (FlowPath flowPath : recomputeFlowPaths) {
+ for (FlowPath flowPath : shouldRecomputeFlowPaths.values()) {
if (recomputeFlowPath(flowPath))
- modifiedFlowPaths.add(flowPath);
+ modifiedFlowPaths.put(flowPath.flowId().value(), flowPath);
}
+ // Extract the modified Flow Entries
modifiedFlowEntries = extractModifiedFlowEntries(modifiedFlowPaths);
// Assign missing Flow Entry IDs
@@ -244,14 +235,12 @@
//
flowManager.pushModifiedFlowEntriesToSwitches(modifiedFlowEntries);
flowManager.pushModifiedFlowEntriesToDatagrid(modifiedFlowEntries);
- flowManager.pushModifiedFlowEntriesToDatabase(modifiedFlowEntries);
- flowManager.pushModifiedFlowEntriesToDatabase(updatedFlowEntries);
- flowManager.pushModifiedFlowEntriesToDatabase(unmatchedDeleteFlowEntries);
+ flowManager.pushModifiedFlowPathsToDatabase(modifiedFlowPaths.values());
//
// Remove Flow Entries that were deleted
//
- for (FlowPath flowPath : modifiedFlowPaths)
+ for (FlowPath flowPath : modifiedFlowPaths.values())
flowPath.dataPath().removeDeletedFlowEntries();
// Cleanup
@@ -259,23 +248,20 @@
flowPathEvents.clear();
flowEntryEvents.clear();
//
- newFlowPaths.clear();
- recomputeFlowPaths.clear();
+ shouldRecomputeFlowPaths.clear();
modifiedFlowPaths.clear();
- updatedFlowEntries.clear();
- unmatchedDeleteFlowEntries.clear();
}
/**
* Extract the modified Flow Entries.
*/
private List<FlowPathEntryPair> extractModifiedFlowEntries(
- List<FlowPath> modifiedFlowPaths) {
+ Map<Long, FlowPath> modifiedFlowPaths) {
List<FlowPathEntryPair> modifiedFlowEntries =
new LinkedList<FlowPathEntryPair>();
// Extract only the modified Flow Entries
- for (FlowPath flowPath : modifiedFlowPaths) {
+ for (FlowPath flowPath : modifiedFlowPaths.values()) {
for (FlowEntry flowEntry : flowPath.flowEntries()) {
if (flowEntry.flowEntrySwitchState() ==
FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED) {
@@ -334,10 +320,8 @@
if (allFlowPaths.get(flowPath.flowId().value()) != null) {
//
// TODO: What to do if the Flow Path already exists?
- // Remove and then re-add it, or merge the info?
- // For now, we don't have to do anything.
+ // Fow now, we just re-add it.
//
- break;
}
switch (flowPath.flowPathType()) {
@@ -347,7 +331,8 @@
// we are going to recompute it anyway.
//
flowPath.flowEntries().clear();
- recomputeFlowPaths.add(flowPath);
+ shouldRecomputeFlowPaths.put(flowPath.flowId().value(),
+ flowPath);
break;
case FP_TYPE_EXPLICIT_PATH:
//
@@ -356,10 +341,10 @@
for (FlowEntry flowEntry : flowPath.flowEntries()) {
flowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
}
- modifiedFlowPaths.add(flowPath);
+ modifiedFlowPaths.put(flowPath.flowId().value(), flowPath);
break;
}
- newFlowPaths.add(flowPath);
+ allFlowPaths.put(flowPath.flowId().value(), flowPath);
break;
}
@@ -382,8 +367,11 @@
flowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
}
- allFlowPaths.remove(existingFlowPath.flowId().value());
- modifiedFlowPaths.add(existingFlowPath);
+ // Remove the Flow Path from the internal state
+ Long key = existingFlowPath.flowId().value();
+ allFlowPaths.remove(key);
+ shouldRecomputeFlowPaths.remove(key);
+ modifiedFlowPaths.put(key, existingFlowPath);
break;
}
@@ -416,7 +404,7 @@
}
if (isTopologyModified) {
// TODO: For now, if the topology changes, we recompute all Flows
- recomputeFlowPaths.addAll(allFlowPaths.values());
+ shouldRecomputeFlowPaths.putAll(allFlowPaths);
}
}
@@ -444,7 +432,7 @@
continue;
}
flowPair = new FlowPathEntryPair(flowPath, updatedFlowEntry);
- updatedFlowEntries.add(flowPair);
+ modifiedFlowPaths.put(flowPath.flowId().value(), flowPath);
}
unmatchedFlowEntryAdd = remainingUpdates;
}
@@ -481,15 +469,15 @@
}
// Add the updated entry to the list of updated Flow Entries
flowPair = new FlowPathEntryPair(flowPath, updatedFlowEntry);
- updatedFlowEntries.add(flowPair);
+ modifiedFlowPaths.put(flowPath.flowId().value(), flowPath);
break;
case ENTRY_REMOVE:
flowEntry.setFlowEntryUserState(FlowEntryUserState.FE_USER_DELETE);
if (unmatchedFlowEntryAdd.remove(flowEntry.flowEntryId().value()) != null) {
- continue; // Match found
+ continue; // Removed previously unmatched entry
}
-
+
flowPath = allFlowPaths.get(flowEntry.flowId().value());
if (flowPath == null) {
// Flow Path not found: ignore the update
@@ -497,13 +485,11 @@
}
updatedFlowEntry = updateFlowEntryRemove(flowPath, flowEntry);
if (updatedFlowEntry == null) {
- // Flow Entry not found: add to list of deleted entries
- flowPair = new FlowPathEntryPair(flowPath, flowEntry);
- unmatchedDeleteFlowEntries.add(flowPair);
+ // Flow Entry not found: ignore it
break;
}
flowPair = new FlowPathEntryPair(flowPath, updatedFlowEntry);
- updatedFlowEntries.add(flowPair);
+ modifiedFlowPaths.put(flowPath.flowId().value(), flowPath);
break;
}
}
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 bdd0355..7f43fe9 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
@@ -57,8 +57,8 @@
private final static Logger log = LoggerFactory.getLogger(FlowManager.class);
// The queue to write Flow Entries to the database
- private BlockingQueue<FlowPathEntryPair> flowEntriesToDatabaseQueue =
- new LinkedBlockingQueue<FlowPathEntryPair>();
+ private BlockingQueue<FlowPath> flowPathsToDatabaseQueue =
+ new LinkedBlockingQueue<FlowPath>();
FlowDatabaseWriter flowDatabaseWriter;
/**
@@ -192,7 +192,7 @@
// The thread to write to the database
//
flowDatabaseWriter = new FlowDatabaseWriter(this,
- flowEntriesToDatabaseQueue);
+ flowPathsToDatabaseQueue);
flowDatabaseWriter.start();
//
@@ -510,7 +510,7 @@
*/
class FlowDatabaseWriter extends Thread {
private FlowManager flowManager;
- private BlockingQueue<FlowPathEntryPair> blockingQueue;
+ private BlockingQueue<FlowPath> blockingQueue;
/**
* Constructor.
@@ -519,7 +519,7 @@
* @param blockingQueue the blocking queue to use.
*/
FlowDatabaseWriter(FlowManager flowManager,
- BlockingQueue<FlowPathEntryPair> blockingQueue) {
+ BlockingQueue<FlowPath> blockingQueue) {
this.flowManager = flowManager;
this.blockingQueue = blockingQueue;
}
@@ -532,14 +532,13 @@
//
// The main loop
//
- Collection<FlowPathEntryPair> collection =
- new LinkedList<FlowPathEntryPair>();
+ Collection<FlowPath> collection = new LinkedList<FlowPath>();
try {
while (true) {
- FlowPathEntryPair entryPair = blockingQueue.take();
- collection.add(entryPair);
+ FlowPath flowPath = blockingQueue.take();
+ collection.add(flowPath);
blockingQueue.drainTo(collection);
- flowManager.writeModifiedFlowEntriesToDatabase(collection);
+ flowManager.writeModifiedFlowPathsToDatabase(collection);
collection.clear();
}
} catch (Exception exception) {
@@ -549,46 +548,42 @@
}
/**
- * Push Flow Entries to the Network MAP.
+ * Push Flow Paths to the Network MAP.
*
- * NOTE: The Flow Entries are pushed only on the instance responsible
- * for the first switch. This is to avoid database errors when multiple
- * instances are writing Flow Entries for the same Flow Path.
+ * NOTE: The complete Flow Paths are pushed only on the instance
+ * responsible for the first switch. This is to avoid database errors
+ * when multiple instances are writing Flow Entries for the same Flow Path.
*
- * @param modifiedFlowEntries the collection of Flow Entries to push.
+ * @param modifiedFlowPaths the collection of Flow Paths to push.
*/
- void pushModifiedFlowEntriesToDatabase(
- Collection<FlowPathEntryPair> modifiedFlowEntries) {
+ void pushModifiedFlowPathsToDatabase(
+ Collection<FlowPath> modifiedFlowPaths) {
//
- // We only add the Flow Entries to the Database Queue.
+ // We only add the Flow Paths to the Database Queue.
// The FlowDatabaseWriter thread is responsible for the actual writing.
//
- flowEntriesToDatabaseQueue.addAll(modifiedFlowEntries);
+ flowPathsToDatabaseQueue.addAll(modifiedFlowPaths);
}
/**
- * Write Flow Entries to the Network MAP.
+ * Write Flow Paths to the Network MAP.
*
- * NOTE: The Flow Entries are written only on the instance responsible
- * for the first switch. This is to avoid database errors when multiple
- * instances are writing Flow Entries for the same Flow Path.
+ * NOTE: The complete Flow Paths are pushed only on the instance
+ * responsible for the first switch. This is to avoid database errors
+ * when multiple instances are writing Flow Entries for the same Flow Path.
*
- * @param modifiedFlowEntries the collection of Flow Entries to write.
+ * @param modifiedFlowPaths the collection of Flow Paths to write.
*/
- private void writeModifiedFlowEntriesToDatabase(
- Collection<FlowPathEntryPair> modifiedFlowEntries) {
- if (modifiedFlowEntries.isEmpty())
+ private void writeModifiedFlowPathsToDatabase(
+ Collection<FlowPath> modifiedFlowPaths) {
+ if (modifiedFlowPaths.isEmpty())
return;
+ FlowId dummyFlowId = new FlowId();
+
Map<Long, IOFSwitch> mySwitches = getMySwitches();
- for (FlowPathEntryPair flowPair : modifiedFlowEntries) {
- FlowPath flowPath = flowPair.flowPath;
- FlowEntry flowEntry = flowPair.flowEntry;
-
- if (! flowEntry.isValidFlowEntryId())
- continue;
-
+ for (FlowPath flowPath : modifiedFlowPaths) {
//
// Push the changes only on the instance responsible for the
// first switch.
@@ -598,69 +593,37 @@
if (mySrcSwitch == null)
continue;
- log.debug("Pushing Flow Entry To Database: {}", flowEntry.toString());
//
- // Write the Flow Entry to the Network Map
+ // Test whether all Flow Entries are valid
//
- // NOTE: We try a number of times, in case somehow some other
- // instances are writing at the same time.
- // Apparently, if other instances are writing at the same time
- // this will trigger an error.
- //
- for (int i = 0; i < 6; i++) {
- try {
- //
- // Find the Flow Path in the Network MAP.
- //
- // NOTE: The Flow Path might not be found if the Flow was
- // just removed by some other controller instance.
- //
- IFlowPath flowObj =
- dbHandlerInner.searchFlowPath(flowEntry.flowId());
- if (flowObj == null) {
- String logMsg = "Cannot find Network MAP entry for Flow Path " + flowEntry.flowId();
- log.error(logMsg);
- break;
- }
-
- // Write the Flow Entry
- switch (flowEntry.flowEntryUserState()) {
- case FE_USER_ADD:
- // FALLTHROUGH
- case FE_USER_MODIFY:
- if (addFlowEntry(flowObj, flowEntry) == null) {
- String logMsg = "Cannot write to Network MAP Flow Entry " +
- flowEntry.flowEntryId() +
- " from Flow Path " + flowEntry.flowId() +
- " on switch " + flowEntry.dpid();
- log.error(logMsg);
- }
- break;
- case FE_USER_DELETE:
- if (deleteFlowEntry(flowObj, flowEntry) == false) {
- String logMsg = "Cannot remove from Network MAP Flow Entry " +
- flowEntry.flowEntryId() +
- " from Flow Path " + flowEntry.flowId() +
- " on switch " + flowEntry.dpid();
- log.error(logMsg);
- }
- break;
- }
-
- // Commit to the database
- dbHandlerInner.commit();
- break; // Success
-
- } catch (Exception e) {
- log.debug("Exception writing Flow Entry to Network MAP: ", e);
- dbHandlerInner.rollback();
- // Wait a bit (random value [1ms, 20ms] and try again
- int delay = 1 + randomGenerator.nextInt() % 20;
- try {
- Thread.sleep(delay);
- } catch (Exception e0) {
- }
+ boolean allValid = true;
+ for (FlowEntry flowEntry : flowPath.flowEntries()) {
+ if (flowEntry.flowEntryUserState() ==
+ FlowEntryUserState.FE_USER_DELETE) {
+ continue;
}
+ if (! flowEntry.isValidFlowEntryId()) {
+ allValid = false;
+ break;
+ }
+ }
+ if (! allValid)
+ continue;
+
+ log.debug("Pushing Flow Path To Database: {}", flowPath.toString());
+
+ //
+ // Write the Flow Path to the Network Map
+ //
+ try {
+ if (! FlowDatabaseOperation.addFlow(this, dbHandlerInner,
+ flowPath, dummyFlowId)) {
+ String logMsg = "Cannot write to Network Map Flow Path " +
+ flowPath.flowId();
+ log.error(logMsg);
+ }
+ } catch (Exception e) {
+ log.error("Exception writing Flow Path to Network MAP: ", e);
}
}
}