Modify the logic for writing the Flow Entries to the database:
For each flow, only the controller for the first-hop switch is
writing the Flow Entries to the database.
Apparently, if multiple controllers try to write its own
Flow Entry for the same Flow Path to the database, this
triggers a Titan exception.
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 f3e47f4..0e9887a 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
@@ -32,6 +32,19 @@
import org.slf4j.LoggerFactory;
/**
+ * A class for storing a pair of Flow Path and a Flow Entry.
+ */
+class FlowPathEntryPair {
+ protected FlowPath flowPath;
+ protected FlowEntry flowEntry;
+
+ protected FlowPathEntryPair(FlowPath flowPath, FlowEntry flowEntry) {
+ this.flowPath = flowPath;
+ this.flowEntry = flowEntry;
+ }
+}
+
+/**
* Class for FlowPath Maintenance.
* This class listens for FlowEvents to:
* - Maintain a local cache of the Network Topology.
@@ -46,8 +59,8 @@
private IDatagridService datagridService; // The Datagrid Service to use
private Topology topology; // The network topology
private Map<Long, FlowPath> allFlowPaths = new HashMap<Long, FlowPath>();
- private List<FlowEntry> unmatchedFlowEntryUpdates =
- new LinkedList<FlowEntry>();
+ private Map<Long, FlowEntry> unmatchedFlowEntryAdd =
+ new HashMap<Long, FlowEntry>();
// The queue with Flow Path and Topology Element updates
private BlockingQueue<EventEntry<?>> networkEvents =
@@ -66,10 +79,15 @@
// - 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>();
/**
@@ -181,7 +199,7 @@
* Process the events (if any)
*/
private void processEvents() {
- List<FlowEntry> modifiedFlowEntries;
+ List<FlowPathEntryPair> modifiedFlowEntries;
if (topologyEvents.isEmpty() && flowPathEvents.isEmpty() &&
flowEntryEvents.isEmpty()) {
@@ -214,9 +232,11 @@
//
// Push the modified Flow Entries to switches, datagrid and database
//
- flowManager.pushModifiedFlowEntriesToSwitches(modifiedFlowPaths);
+ flowManager.pushModifiedFlowEntriesToSwitches(modifiedFlowEntries);
flowManager.pushModifiedFlowEntriesToDatagrid(modifiedFlowEntries);
flowManager.pushModifiedFlowEntriesToDatabase(modifiedFlowEntries);
+ flowManager.pushModifiedFlowEntriesToDatabase(updatedFlowEntries);
+ flowManager.pushModifiedFlowEntriesToDatabase(unmatchedDeleteFlowEntries);
//
// Remove Flow Entries that were deleted
@@ -232,21 +252,26 @@
newFlowPaths.clear();
recomputeFlowPaths.clear();
modifiedFlowPaths.clear();
+ updatedFlowEntries.clear();
+ unmatchedDeleteFlowEntries.clear();
}
/**
* Extract the modified Flow Entries.
*/
- private List<FlowEntry> extractModifiedFlowEntries(
- List<FlowPath> modifiedFlowPaths) {
- List<FlowEntry> modifiedFlowEntries = new LinkedList<FlowEntry>();
+ private List<FlowPathEntryPair> extractModifiedFlowEntries(
+ List<FlowPath> modifiedFlowPaths) {
+ List<FlowPathEntryPair> modifiedFlowEntries =
+ new LinkedList<FlowPathEntryPair>();
// Extract only the modified Flow Entries
for (FlowPath flowPath : modifiedFlowPaths) {
for (FlowEntry flowEntry : flowPath.flowEntries()) {
if (flowEntry.flowEntrySwitchState() ==
FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED) {
- modifiedFlowEntries.add(flowEntry);
+ FlowPathEntryPair flowPair =
+ new FlowPathEntryPair(flowPath, flowEntry);
+ modifiedFlowEntries.add(flowPair);
}
}
}
@@ -256,7 +281,7 @@
/**
* Assign the Flow Entry ID as needed.
*/
- private void assignFlowEntryId(List<FlowEntry> modifiedFlowEntries) {
+ private void assignFlowEntryId(List<FlowPathEntryPair> modifiedFlowEntries) {
if (modifiedFlowEntries.isEmpty())
return;
@@ -265,7 +290,8 @@
//
// Assign the Flow Entry ID only for Flow Entries for my switches
//
- for (FlowEntry flowEntry : modifiedFlowEntries) {
+ for (FlowPathEntryPair flowPair : modifiedFlowEntries) {
+ FlowEntry flowEntry = flowPair.flowEntry;
// Update the Flow Entries only for my switches
IOFSwitch mySwitch = mySwitches.get(flowEntry.dpid().value());
if (mySwitch == null)
@@ -388,17 +414,29 @@
* Process the Flow Entry events.
*/
private void processFlowEntryEvents() {
+ FlowPathEntryPair flowPair;
+ FlowPath flowPath;
+ FlowEntry updatedFlowEntry;
+
//
// Update Flow Entries with previously unmatched Flow Entry updates
//
- if ((! flowPathEvents.isEmpty()) &&
- (! unmatchedFlowEntryUpdates.isEmpty())) {
- List<FlowEntry> remainingUpdates = new LinkedList<FlowEntry>();
- for (FlowEntry flowEntry : unmatchedFlowEntryUpdates) {
- if (! updateFlowEntry(flowEntry))
- remainingUpdates.add(flowEntry);
+ if (! unmatchedFlowEntryAdd.isEmpty()) {
+ Map<Long, FlowEntry> remainingUpdates = new HashMap<Long, FlowEntry>();
+ for (FlowEntry flowEntry : unmatchedFlowEntryAdd.values()) {
+ flowPath = allFlowPaths.get(flowEntry.flowId().value());
+ if (flowPath == null)
+ continue;
+ updatedFlowEntry = updateFlowEntryAdd(flowPath, flowEntry);
+ if (updatedFlowEntry == null) {
+ remainingUpdates.put(flowEntry.flowEntryId().value(),
+ flowEntry);
+ continue;
+ }
+ flowPair = new FlowPathEntryPair(flowPath, updatedFlowEntry);
+ updatedFlowEntries.add(flowPair);
}
- unmatchedFlowEntryUpdates = remainingUpdates;
+ unmatchedFlowEntryAdd = remainingUpdates;
}
//
@@ -406,58 +444,78 @@
//
for (EventEntry<FlowEntry> eventEntry : flowEntryEvents) {
FlowEntry flowEntry = eventEntry.eventData();
+
+ log.debug("Flow Entry Event: {} {}", eventEntry.eventType(),
+ flowEntry.toString());
+
+ if ((! flowEntry.isValidFlowId()) ||
+ (! flowEntry.isValidFlowEntryId())) {
+ continue;
+ }
+
switch (eventEntry.eventType()) {
case ENTRY_ADD:
- //
- // Find the corresponding Flow Entry and update it.
- // If not found, then keep it in a local cache for
- // later matching.
- //
- if (! updateFlowEntry(flowEntry))
- unmatchedFlowEntryUpdates.add(flowEntry);
+ flowPath = allFlowPaths.get(flowEntry.flowId().value());
+ if (flowPath == null) {
+ // Flow Path not found: keep the entry for later matching
+ unmatchedFlowEntryAdd.put(flowEntry.flowEntryId().value(),
+ flowEntry);
+ break;
+ }
+ updatedFlowEntry = updateFlowEntryAdd(flowPath, flowEntry);
+ if (updatedFlowEntry == null) {
+ // Flow Entry not found: keep the entry for later matching
+ unmatchedFlowEntryAdd.put(flowEntry.flowEntryId().value(),
+ flowEntry);
+ break;
+ }
+ // Add the updated entry to the list of updated Flow Entries
+ flowPair = new FlowPathEntryPair(flowPath, updatedFlowEntry);
+ updatedFlowEntries.add(flowPair);
break;
+
case ENTRY_REMOVE:
- //
- // NOTE: For now we remove the Flow Entries based on
- // local decisions, so no need to remove them because of an
- // external event.
- //
+ flowEntry.setFlowEntryUserState(FlowEntryUserState.FE_USER_DELETE);
+ if (unmatchedFlowEntryAdd.remove(flowEntry.flowEntryId().value()) != null) {
+ continue; // Match found
+ }
+
+ flowPath = allFlowPaths.get(flowEntry.flowId().value());
+ if (flowPath == null) {
+ // Flow Path not found: ignore the update
+ break;
+ }
+ 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);
+ break;
+ }
+ flowPair = new FlowPathEntryPair(flowPath, updatedFlowEntry);
+ updatedFlowEntries.add(flowPair);
break;
}
}
}
/**
- * Update a Flow Entry because of an external event.
+ * Update a Flow Entry because of an external ENTRY_ADD event.
*
+ * @param flowPath the FlowPath for the Flow Entry to update.
* @param flowEntry the FlowEntry with the new state.
- * @return true if the Flow Entry was found and updated, otherwise false.
+ * @return the updated Flow Entry if found, otherwise null.
*/
- private boolean updateFlowEntry(FlowEntry flowEntry) {
- if ((! flowEntry.isValidFlowId()) ||
- (! flowEntry.isValidFlowEntryId())) {
- //
- // Ignore events for Flow Entries with invalid Flow ID or
- // Flow Entry ID.
- // This shouldn't happen.
- //
- return true;
- }
-
- FlowPath flowPath = allFlowPaths.get(flowEntry.flowId().value());
- if (flowPath == null)
- return false;
-
+ private FlowEntry updateFlowEntryAdd(FlowPath flowPath,
+ FlowEntry flowEntry) {
//
- // Iterate over all Flow Entries and find a match based on the DPID
+ // Iterate over all Flow Entries and find a match.
//
for (FlowEntry localFlowEntry : flowPath.flowEntries()) {
- if (localFlowEntry.dpid().value() != flowEntry.dpid().value())
+ if (! TopologyManager.isSameFlowEntryDataPath(localFlowEntry,
+ flowEntry)) {
continue;
- //
- // TODO: We might want to check the FlowEntryMatch and
- // FlowEntryActions to double-check it is the same Flow Entry
- //
+ }
//
// Local Flow Entry match found
@@ -467,9 +525,9 @@
flowEntry.flowEntryId().value()) {
//
// Find a local Flow Entry, but the Flow Entry ID doesn't
- // match. Ignore the event.
+ // match. Keep looking.
//
- return true;
+ continue;
}
} else {
// Update the Flow Entry ID
@@ -480,13 +538,44 @@
//
// Update the local Flow Entry.
- // For now we update only the Flow Entry Switch State
//
+ localFlowEntry.setFlowEntryUserState(flowEntry.flowEntryUserState());
localFlowEntry.setFlowEntrySwitchState(flowEntry.flowEntrySwitchState());
- return true;
+ return localFlowEntry;
}
- return false; // Entry not found
+ return null; // Entry not found
+ }
+
+ /**
+ * Update a Flow Entry because of an external ENTRY_REMOVE event.
+ *
+ * @param flowPath the FlowPath for the Flow Entry to update.
+ * @param flowEntry the FlowEntry with the new state.
+ * @return the updated Flow Entry if found, otherwise null.
+ */
+ private FlowEntry updateFlowEntryRemove(FlowPath flowPath,
+ FlowEntry flowEntry) {
+ //
+ // Iterate over all Flow Entries and find a match based on
+ // the Flow Entry ID.
+ //
+ for (FlowEntry localFlowEntry : flowPath.flowEntries()) {
+ if (! localFlowEntry.isValidFlowEntryId())
+ continue;
+ if (localFlowEntry.flowEntryId().value() !=
+ flowEntry.flowEntryId().value()) {
+ continue;
+ }
+ //
+ // Update the local Flow Entry.
+ //
+ localFlowEntry.setFlowEntryUserState(flowEntry.flowEntryUserState());
+ localFlowEntry.setFlowEntrySwitchState(flowEntry.flowEntrySwitchState());
+ return localFlowEntry;
+ }
+
+ return null; // Entry not found
}
/**