Merge master branch into syncdev
Conflicts:
src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
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 d06c62c..926788f 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java
@@ -326,6 +326,44 @@
}
/**
+ * Delete a flow entry from the Network MAP.
+ *
+ * @param dbHandler the Graph Database handler to use.
+ * @param flowObj the corresponding Flow Path object for the Flow Entry.
+ * @param flowEntry the Flow Entry to delete.
+ * @return true on success, otherwise false.
+ */
+ static boolean deleteFlowEntry(GraphDBOperation dbHandler,
+ IFlowPath flowObj,
+ FlowEntry flowEntry) {
+ IFlowEntry flowEntryObj = null;
+ try {
+ flowEntryObj = dbHandler.searchFlowEntry(flowEntry.flowEntryId());
+ } catch (Exception e) {
+ log.error(":deleteFlowEntry FlowEntryId:{} failed",
+ flowEntry.flowEntryId().toString());
+ return false;
+ }
+ //
+ // TODO: Don't print an error for now, because multiple controller
+ // instances might be deleting the same flow entry.
+ //
+ /*
+ if (flowEntryObj == null) {
+ log.error(":deleteFlowEntry FlowEntryId:{} failed: FlowEntry object not found",
+ flowEntry.flowEntryId().toString());
+ return false;
+ }
+ */
+ if (flowEntryObj == null)
+ return true;
+
+ flowObj.removeFlowEntry(flowEntryObj);
+ dbHandler.removeFlowEntry(flowEntryObj);
+ return true;
+ }
+
+ /**
* Delete all previously added flows.
*
* @param dbHandler the Graph Database handler to use.
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
new file mode 100644
index 0000000..0e9887a
--- /dev/null
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
@@ -0,0 +1,867 @@
+package net.onrc.onos.ofcontroller.flowmanager;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import net.floodlightcontroller.core.IOFSwitch;
+import net.onrc.onos.datagrid.IDatagridService;
+import net.onrc.onos.ofcontroller.topology.Topology;
+import net.onrc.onos.ofcontroller.topology.TopologyElement;
+import net.onrc.onos.ofcontroller.topology.TopologyManager;
+import net.onrc.onos.ofcontroller.util.DataPath;
+import net.onrc.onos.ofcontroller.util.EventEntry;
+import net.onrc.onos.ofcontroller.util.FlowEntry;
+import net.onrc.onos.ofcontroller.util.FlowEntryAction;
+import net.onrc.onos.ofcontroller.util.FlowEntryActions;
+import net.onrc.onos.ofcontroller.util.FlowEntryId;
+import net.onrc.onos.ofcontroller.util.FlowEntryMatch;
+import net.onrc.onos.ofcontroller.util.FlowEntrySwitchState;
+import net.onrc.onos.ofcontroller.util.FlowEntryUserState;
+import net.onrc.onos.ofcontroller.util.FlowId;
+import net.onrc.onos.ofcontroller.util.FlowPath;
+import net.onrc.onos.ofcontroller.util.FlowPathUserState;
+
+import org.slf4j.Logger;
+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.
+ * - Detect FlowPaths impacted by Topology change.
+ * - Recompute impacted FlowPath using cached Topology.
+ */
+class FlowEventHandler extends Thread implements IFlowEventHandlerService {
+ /** The logger. */
+ private final static Logger log = LoggerFactory.getLogger(FlowEventHandler.class);
+
+ 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 =
+ new LinkedBlockingQueue<EventEntry<?>>();
+
+ // The pending Topology, FlowPath, and FlowEntry events
+ private List<EventEntry<TopologyElement>> topologyEvents =
+ new LinkedList<EventEntry<TopologyElement>>();
+ private List<EventEntry<FlowPath>> flowPathEvents =
+ new LinkedList<EventEntry<FlowPath>>();
+ private List<EventEntry<FlowEntry>> flowEntryEvents =
+ new LinkedList<EventEntry<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>();
+
+
+ /**
+ * Constructor for a given Flow Manager and Datagrid Service.
+ *
+ * @param flowManager the Flow Manager to use.
+ * @param datagridService the Datagrid Service to use.
+ */
+ FlowEventHandler(FlowManager flowManager,
+ IDatagridService datagridService) {
+ this.flowManager = flowManager;
+ this.datagridService = datagridService;
+ this.topology = new Topology();
+ }
+
+ /**
+ * Get the network topology.
+ *
+ * @return the network topology.
+ */
+ protected Topology getTopology() { return this.topology; }
+
+ /**
+ * Startup processing.
+ */
+ private void startup() {
+ //
+ // Obtain the initial Topology state
+ //
+ Collection<TopologyElement> topologyElements =
+ datagridService.getAllTopologyElements();
+ for (TopologyElement topologyElement : topologyElements) {
+ EventEntry<TopologyElement> eventEntry =
+ new EventEntry<TopologyElement>(EventEntry.Type.ENTRY_ADD, topologyElement);
+ topologyEvents.add(eventEntry);
+ }
+ //
+ // Obtain the initial Flow Path state
+ //
+ Collection<FlowPath> flowPaths = datagridService.getAllFlows();
+ for (FlowPath flowPath : flowPaths) {
+ EventEntry<FlowPath> eventEntry =
+ new EventEntry<FlowPath>(EventEntry.Type.ENTRY_ADD, flowPath);
+ flowPathEvents.add(eventEntry);
+ }
+ //
+ // Obtain the initial FlowEntry state
+ //
+ Collection<FlowEntry> flowEntries = datagridService.getAllFlowEntries();
+ for (FlowEntry flowEntry : flowEntries) {
+ EventEntry<FlowEntry> eventEntry =
+ new EventEntry<FlowEntry>(EventEntry.Type.ENTRY_ADD, flowEntry);
+ flowEntryEvents.add(eventEntry);
+ }
+
+ // Process the initial events (if any)
+ processEvents();
+ }
+
+ /**
+ * Run the thread.
+ */
+ @Override
+ public void run() {
+ startup();
+
+ //
+ // The main loop
+ //
+ Collection<EventEntry<?>> collection = new LinkedList<EventEntry<?>>();
+ try {
+ while (true) {
+ EventEntry<?> eventEntry = networkEvents.take();
+ collection.add(eventEntry);
+ networkEvents.drainTo(collection);
+
+ //
+ // Demultiplex all events:
+ // - EventEntry<TopologyElement>
+ // - EventEntry<FlowPath>
+ // - EventEntry<FlowEntry>
+ //
+ for (EventEntry<?> event : collection) {
+ if (event.eventData() instanceof TopologyElement) {
+ EventEntry<TopologyElement> topologyEventEntry =
+ (EventEntry<TopologyElement>)event;
+ topologyEvents.add(topologyEventEntry);
+ } else if (event.eventData() instanceof FlowPath) {
+ EventEntry<FlowPath> flowPathEventEntry =
+ (EventEntry<FlowPath>)event;
+ flowPathEvents.add(flowPathEventEntry);
+ } else if (event.eventData() instanceof FlowEntry) {
+ EventEntry<FlowEntry> flowEntryEventEntry =
+ (EventEntry<FlowEntry>)event;
+ flowEntryEvents.add(flowEntryEventEntry);
+ }
+ }
+ collection.clear();
+
+ // Process the events (if any)
+ processEvents();
+ }
+ } catch (Exception exception) {
+ log.debug("Exception processing Network Events: ", exception);
+ }
+ }
+
+ /**
+ * Process the events (if any)
+ */
+ private void processEvents() {
+ List<FlowPathEntryPair> modifiedFlowEntries;
+
+ if (topologyEvents.isEmpty() && flowPathEvents.isEmpty() &&
+ flowEntryEvents.isEmpty()) {
+ return; // Nothing to do
+ }
+
+ 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) {
+ if (recomputeFlowPath(flowPath))
+ modifiedFlowPaths.add(flowPath);
+ }
+
+ modifiedFlowEntries = extractModifiedFlowEntries(modifiedFlowPaths);
+
+ // Assign missing Flow Entry IDs
+ assignFlowEntryId(modifiedFlowEntries);
+
+ //
+ // Push the modified Flow Entries to switches, datagrid and database
+ //
+ flowManager.pushModifiedFlowEntriesToSwitches(modifiedFlowEntries);
+ flowManager.pushModifiedFlowEntriesToDatagrid(modifiedFlowEntries);
+ flowManager.pushModifiedFlowEntriesToDatabase(modifiedFlowEntries);
+ flowManager.pushModifiedFlowEntriesToDatabase(updatedFlowEntries);
+ flowManager.pushModifiedFlowEntriesToDatabase(unmatchedDeleteFlowEntries);
+
+ //
+ // Remove Flow Entries that were deleted
+ //
+ for (FlowPath flowPath : modifiedFlowPaths)
+ flowPath.dataPath().removeDeletedFlowEntries();
+
+ // Cleanup
+ topologyEvents.clear();
+ flowPathEvents.clear();
+ flowEntryEvents.clear();
+ //
+ newFlowPaths.clear();
+ recomputeFlowPaths.clear();
+ modifiedFlowPaths.clear();
+ updatedFlowEntries.clear();
+ unmatchedDeleteFlowEntries.clear();
+ }
+
+ /**
+ * Extract the modified Flow Entries.
+ */
+ 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) {
+ FlowPathEntryPair flowPair =
+ new FlowPathEntryPair(flowPath, flowEntry);
+ modifiedFlowEntries.add(flowPair);
+ }
+ }
+ }
+ return modifiedFlowEntries;
+ }
+
+ /**
+ * Assign the Flow Entry ID as needed.
+ */
+ private void assignFlowEntryId(List<FlowPathEntryPair> modifiedFlowEntries) {
+ if (modifiedFlowEntries.isEmpty())
+ return;
+
+ Map<Long, IOFSwitch> mySwitches = flowManager.getMySwitches();
+
+ //
+ // Assign the Flow Entry ID only for Flow Entries for my switches
+ //
+ 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)
+ continue;
+ if (! flowEntry.isValidFlowEntryId()) {
+ long id = flowManager.getNextFlowEntryId();
+ flowEntry.setFlowEntryId(new FlowEntryId(id));
+ }
+ }
+ }
+
+ /**
+ * Process the Flow Path events.
+ */
+ private void processFlowPathEvents() {
+ //
+ // Process all Flow Path events and update the appropriate state
+ //
+ for (EventEntry<FlowPath> eventEntry : flowPathEvents) {
+ FlowPath flowPath = eventEntry.eventData();
+
+ log.debug("Flow Event: {} {}", eventEntry.eventType(),
+ flowPath.toString());
+
+ switch (eventEntry.eventType()) {
+ case ENTRY_ADD: {
+ //
+ // Add a new Flow Path
+ //
+ 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.
+ //
+ break;
+ }
+
+ switch (flowPath.flowPathType()) {
+ case FP_TYPE_SHORTEST_PATH:
+ //
+ // Reset the Data Path, in case it was set already, because
+ // we are going to recompute it anyway.
+ //
+ flowPath.flowEntries().clear();
+ recomputeFlowPaths.add(flowPath);
+ break;
+ case FP_TYPE_EXPLICIT_PATH:
+ //
+ // Mark all Flow Entries for installation in the switches.
+ //
+ for (FlowEntry flowEntry : flowPath.flowEntries()) {
+ flowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
+ }
+ modifiedFlowPaths.add(flowPath);
+ break;
+ }
+ newFlowPaths.add(flowPath);
+
+ break;
+ }
+
+ case ENTRY_REMOVE: {
+ //
+ // Remove an existing Flow Path.
+ //
+ // Find the Flow Path, and mark the Flow and its Flow Entries
+ // for deletion.
+ //
+ FlowPath existingFlowPath =
+ allFlowPaths.get(flowPath.flowId().value());
+ if (existingFlowPath == null)
+ continue; // Nothing to do
+
+ existingFlowPath.setFlowPathUserState(FlowPathUserState.FP_USER_DELETE);
+ for (FlowEntry flowEntry : existingFlowPath.flowEntries()) {
+ flowEntry.setFlowEntryUserState(FlowEntryUserState.FE_USER_DELETE);
+ flowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
+ }
+
+ allFlowPaths.remove(existingFlowPath.flowId().value());
+ modifiedFlowPaths.add(existingFlowPath);
+
+ break;
+ }
+ }
+ }
+ }
+
+ /**
+ * Process the Topology events.
+ */
+ private void processTopologyEvents() {
+ //
+ // Process all Topology events and update the appropriate state
+ //
+ boolean isTopologyModified = false;
+ for (EventEntry<TopologyElement> eventEntry : topologyEvents) {
+ TopologyElement topologyElement = eventEntry.eventData();
+
+ log.debug("Topology Event: {} {}", eventEntry.eventType(),
+ topologyElement.toString());
+
+ switch (eventEntry.eventType()) {
+ case ENTRY_ADD:
+ isTopologyModified |= topology.addTopologyElement(topologyElement);
+ break;
+ case ENTRY_REMOVE:
+ isTopologyModified |= topology.removeTopologyElement(topologyElement);
+ break;
+ }
+ }
+ if (isTopologyModified) {
+ // TODO: For now, if the topology changes, we recompute all Flows
+ recomputeFlowPaths.addAll(allFlowPaths.values());
+ }
+ }
+
+ /**
+ * Process the Flow Entry events.
+ */
+ private void processFlowEntryEvents() {
+ FlowPathEntryPair flowPair;
+ FlowPath flowPath;
+ FlowEntry updatedFlowEntry;
+
+ //
+ // Update Flow Entries with previously unmatched Flow Entry updates
+ //
+ 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);
+ }
+ unmatchedFlowEntryAdd = remainingUpdates;
+ }
+
+ //
+ // Process all Flow Entry events and update the appropriate state
+ //
+ 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:
+ 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:
+ 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 ENTRY_ADD 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 updateFlowEntryAdd(FlowPath flowPath,
+ FlowEntry flowEntry) {
+ //
+ // Iterate over all Flow Entries and find a match.
+ //
+ for (FlowEntry localFlowEntry : flowPath.flowEntries()) {
+ if (! TopologyManager.isSameFlowEntryDataPath(localFlowEntry,
+ flowEntry)) {
+ continue;
+ }
+
+ //
+ // Local Flow Entry match found
+ //
+ if (localFlowEntry.isValidFlowEntryId()) {
+ if (localFlowEntry.flowEntryId().value() !=
+ flowEntry.flowEntryId().value()) {
+ //
+ // Find a local Flow Entry, but the Flow Entry ID doesn't
+ // match. Keep looking.
+ //
+ continue;
+ }
+ } else {
+ // Update the Flow Entry ID
+ FlowEntryId flowEntryId =
+ new FlowEntryId(flowEntry.flowEntryId().value());
+ localFlowEntry.setFlowEntryId(flowEntryId);
+ }
+
+ //
+ // Update the local Flow Entry.
+ //
+ localFlowEntry.setFlowEntryUserState(flowEntry.flowEntryUserState());
+ localFlowEntry.setFlowEntrySwitchState(flowEntry.flowEntrySwitchState());
+ return localFlowEntry;
+ }
+
+ 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
+ }
+
+ /**
+ * Recompute a Flow Path.
+ *
+ * @param flowPath the Flow Path to recompute.
+ * @return true if the recomputed Flow Path has changed, otherwise false.
+ */
+ private boolean recomputeFlowPath(FlowPath flowPath) {
+ boolean hasChanged = false;
+
+ //
+ // Test whether the Flow Path needs to be recomputed
+ //
+ switch (flowPath.flowPathType()) {
+ case FP_TYPE_UNKNOWN:
+ return false; // Can't recompute on Unknown FlowType
+ case FP_TYPE_SHORTEST_PATH:
+ break;
+ case FP_TYPE_EXPLICIT_PATH:
+ return false; // An explicit path never changes
+ }
+
+ DataPath oldDataPath = flowPath.dataPath();
+
+ // Compute the new path
+ DataPath newDataPath = TopologyManager.computeNetworkPath(topology,
+ flowPath);
+ if (newDataPath == null) {
+ // We need the DataPath to compare the paths
+ newDataPath = new DataPath();
+ }
+ newDataPath.applyFlowPathFlags(flowPath.flowPathFlags());
+
+ //
+ // Test whether the new path is same
+ //
+ if (oldDataPath.flowEntries().size() !=
+ newDataPath.flowEntries().size()) {
+ hasChanged = true;
+ } else {
+ Iterator<FlowEntry> oldIter = oldDataPath.flowEntries().iterator();
+ Iterator<FlowEntry> newIter = newDataPath.flowEntries().iterator();
+ while (oldIter.hasNext() && newIter.hasNext()) {
+ FlowEntry oldFlowEntry = oldIter.next();
+ FlowEntry newFlowEntry = newIter.next();
+ if (! TopologyManager.isSameFlowEntryDataPath(oldFlowEntry,
+ newFlowEntry)) {
+ hasChanged = true;
+ break;
+ }
+ }
+ }
+ if (! hasChanged)
+ return hasChanged;
+
+ //
+ // Merge the changes in the path:
+ // - If a Flow Entry for a switch is in the old data path, but not
+ // in the new data path, then mark it for deletion.
+ // - If a Flow Entry for a switch is in the new data path, but not
+ // in the old data path, then mark it for addition.
+ // - If a Flow Entry for a switch is in both the old and the new
+ // data path, but it has changed, e.g., the incoming and/or outgoing
+ // port(s), then mark the old Flow Entry for deletion, and mark
+ // the new Flow Entry for addition.
+ // - If a Flow Entry for a switch is in both the old and the new
+ // data path, and it hasn't changed, then just keep it.
+ //
+ // NOTE: We use the Switch DPID of each entry to match the entries
+ //
+ Map<Long, FlowEntry> oldFlowEntriesMap = new HashMap<Long, FlowEntry>();
+ Map<Long, FlowEntry> newFlowEntriesMap = new HashMap<Long, FlowEntry>();
+ ArrayList<FlowEntry> finalFlowEntries = new ArrayList<FlowEntry>();
+ List<FlowEntry> deletedFlowEntries = new LinkedList<FlowEntry>();
+
+ // Prepare maps with the Flow Entries, so they are fast to lookup
+ for (FlowEntry flowEntry : oldDataPath.flowEntries())
+ oldFlowEntriesMap.put(flowEntry.dpid().value(), flowEntry);
+ for (FlowEntry flowEntry : newDataPath.flowEntries())
+ newFlowEntriesMap.put(flowEntry.dpid().value(), flowEntry);
+
+ //
+ // Find the old Flow Entries that should be deleted
+ //
+ for (FlowEntry oldFlowEntry : oldDataPath.flowEntries()) {
+ FlowEntry newFlowEntry =
+ newFlowEntriesMap.get(oldFlowEntry.dpid().value());
+ if (newFlowEntry == null) {
+ // The old Flow Entry should be deleted: not on the path
+ oldFlowEntry.setFlowEntryUserState(FlowEntryUserState.FE_USER_DELETE);
+ oldFlowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
+ deletedFlowEntries.add(oldFlowEntry);
+ }
+ }
+
+ //
+ // Find the new Flow Entries that should be added or updated
+ //
+ int idx = 0;
+ for (FlowEntry newFlowEntry : newDataPath.flowEntries()) {
+ FlowEntry oldFlowEntry =
+ oldFlowEntriesMap.get(newFlowEntry.dpid().value());
+
+ if ((oldFlowEntry != null) &&
+ TopologyManager.isSameFlowEntryDataPath(oldFlowEntry,
+ newFlowEntry)) {
+ //
+ // Both Flow Entries are same
+ //
+ finalFlowEntries.add(oldFlowEntry);
+ idx++;
+ continue;
+ }
+
+ if (oldFlowEntry != null) {
+ //
+ // The old Flow Entry should be deleted: path diverges
+ //
+ oldFlowEntry.setFlowEntryUserState(FlowEntryUserState.FE_USER_DELETE);
+ oldFlowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
+ deletedFlowEntries.add(oldFlowEntry);
+ }
+
+ //
+ // Add the new Flow Entry
+ //
+ //
+ // NOTE: Assign only the Flow ID.
+ // The Flow Entry ID is assigned later only for the Flow Entries
+ // this instance is responsible for.
+ //
+ newFlowEntry.setFlowId(new FlowId(flowPath.flowId().value()));
+
+ // Set the incoming port matching
+ FlowEntryMatch flowEntryMatch = new FlowEntryMatch();
+ newFlowEntry.setFlowEntryMatch(flowEntryMatch);
+ flowEntryMatch.enableInPort(newFlowEntry.inPort());
+
+ //
+ // Set the actions:
+ // If the first Flow Entry, copy the Flow Path actions to it.
+ //
+ FlowEntryActions flowEntryActions = newFlowEntry.flowEntryActions();
+ if ((idx == 0) && (flowPath.flowEntryActions() != null)) {
+ FlowEntryActions flowActions =
+ new FlowEntryActions(flowPath.flowEntryActions());
+ for (FlowEntryAction action : flowActions.actions())
+ flowEntryActions.addAction(action);
+ }
+ idx++;
+
+ //
+ // Add the outgoing port output action
+ //
+ FlowEntryAction flowEntryAction = new FlowEntryAction();
+ flowEntryAction.setActionOutput(newFlowEntry.outPort());
+ flowEntryActions.addAction(flowEntryAction);
+
+ //
+ // Set the state of the new Flow Entry
+ //
+ newFlowEntry.setFlowEntryUserState(FlowEntryUserState.FE_USER_ADD);
+ newFlowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
+ finalFlowEntries.add(newFlowEntry);
+ }
+
+ //
+ // Replace the old Flow Entries with the new Flow Entries.
+ // Note that the Flow Entries that will be deleted are added at
+ // the end.
+ //
+ finalFlowEntries.addAll(deletedFlowEntries);
+ flowPath.dataPath().setFlowEntries(finalFlowEntries);
+
+ return hasChanged;
+ }
+
+ /**
+ * Receive a notification that a Flow is added.
+ *
+ * @param flowPath the Flow that is added.
+ */
+ @Override
+ public void notificationRecvFlowAdded(FlowPath flowPath) {
+ EventEntry<FlowPath> eventEntry =
+ new EventEntry<FlowPath>(EventEntry.Type.ENTRY_ADD, flowPath);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a Flow is removed.
+ *
+ * @param flowPath the Flow that is removed.
+ */
+ @Override
+ public void notificationRecvFlowRemoved(FlowPath flowPath) {
+ EventEntry<FlowPath> eventEntry =
+ new EventEntry<FlowPath>(EventEntry.Type.ENTRY_REMOVE, flowPath);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a Flow is updated.
+ *
+ * @param flowPath the Flow that is updated.
+ */
+ @Override
+ public void notificationRecvFlowUpdated(FlowPath flowPath) {
+ // NOTE: The ADD and UPDATE events are processed in same way
+ EventEntry<FlowPath> eventEntry =
+ new EventEntry<FlowPath>(EventEntry.Type.ENTRY_ADD, flowPath);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a FlowEntry is added.
+ *
+ * @param flowEntry the FlowEntry that is added.
+ */
+ @Override
+ public void notificationRecvFlowEntryAdded(FlowEntry flowEntry) {
+ EventEntry<FlowEntry> eventEntry =
+ new EventEntry<FlowEntry>(EventEntry.Type.ENTRY_ADD, flowEntry);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a FlowEntry is removed.
+ *
+ * @param flowEntry the FlowEntry that is removed.
+ */
+ @Override
+ public void notificationRecvFlowEntryRemoved(FlowEntry flowEntry) {
+ EventEntry<FlowEntry> eventEntry =
+ new EventEntry<FlowEntry>(EventEntry.Type.ENTRY_REMOVE, flowEntry);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a FlowEntry is updated.
+ *
+ * @param flowEntry the FlowEntry that is updated.
+ */
+ @Override
+ public void notificationRecvFlowEntryUpdated(FlowEntry flowEntry) {
+ // NOTE: The ADD and UPDATE events are processed in same way
+ EventEntry<FlowEntry> eventEntry =
+ new EventEntry<FlowEntry>(EventEntry.Type.ENTRY_ADD, flowEntry);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a Topology Element is added.
+ *
+ * @param topologyElement the Topology Element that is added.
+ */
+ @Override
+ public void notificationRecvTopologyElementAdded(TopologyElement topologyElement) {
+ EventEntry<TopologyElement> eventEntry =
+ new EventEntry<TopologyElement>(EventEntry.Type.ENTRY_ADD, topologyElement);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a Topology Element is removed.
+ *
+ * @param topologyElement the Topology Element that is removed.
+ */
+ @Override
+ public void notificationRecvTopologyElementRemoved(TopologyElement topologyElement) {
+ EventEntry<TopologyElement> eventEntry =
+ new EventEntry<TopologyElement>(EventEntry.Type.ENTRY_REMOVE, topologyElement);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a Topology Element is updated.
+ *
+ * @param topologyElement the Topology Element that is updated.
+ */
+ @Override
+ public void notificationRecvTopologyElementUpdated(TopologyElement topologyElement) {
+ // NOTE: The ADD and UPDATE events are processed in same way
+ EventEntry<TopologyElement> eventEntry =
+ new EventEntry<TopologyElement>(EventEntry.Type.ENTRY_ADD, topologyElement);
+ networkEvents.add(eventEntry);
+ }
+}
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 1ef4431..ab4be1a 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
@@ -18,6 +18,7 @@
import net.floodlightcontroller.core.module.IFloodlightModule;
import net.floodlightcontroller.core.module.IFloodlightService;
import net.floodlightcontroller.restserver.IRestApiService;
+import net.floodlightcontroller.util.OFMessageDamper;
import net.onrc.onos.datagrid.IDatagridService;
import net.onrc.onos.graph.GraphDBOperation;
import net.onrc.onos.ofcontroller.core.INetMapStorage;
@@ -26,33 +27,49 @@
import net.onrc.onos.ofcontroller.floodlightlistener.INetworkGraphService;
import net.onrc.onos.ofcontroller.flowmanager.web.FlowWebRoutable;
import net.onrc.onos.ofcontroller.flowprogrammer.FlowPusher;
-import net.onrc.onos.ofcontroller.flowprogrammer.IFlowPusherService;
import net.onrc.onos.ofcontroller.topology.ITopologyNetService;
import net.onrc.onos.ofcontroller.topology.Topology;
import net.onrc.onos.ofcontroller.topology.TopologyElement;
import net.onrc.onos.ofcontroller.util.*;
-import org.openflow.protocol.OFMessage;
+import org.openflow.protocol.OFType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Flow Manager class for handling the network flows.
*/
-public class FlowManager implements IFloodlightModule, IFlowService, INetMapStorage,
- IFlowPusherService {
+public class FlowManager implements IFloodlightModule, IFlowService, INetMapStorage {
- protected GraphDBOperation dbHandler;
+ //
+ // TODO: A temporary variable to switch between the poll-based and
+ // notification mechanism for the Flow Manager.
+ //
+ private final static boolean enableNotifications = false;
+
+ protected GraphDBOperation dbHandlerApi;
+ protected GraphDBOperation dbHandlerInner;
protected volatile IFloodlightProviderService floodlightProvider;
protected volatile ITopologyNetService topologyNetService;
protected volatile IDatagridService datagridService;
protected IRestApiService restApi;
protected FloodlightModuleContext context;
- protected PathComputation pathComputation;
+ protected FlowEventHandler flowEventHandler;
protected FlowPusher pusher;
+ private static final int NUM_PUSHER_THREAD = 1;
+// LEGACY
+// protected OFMessageDamper messageDamper;
+//
+// //
+// // TODO: Values copied from elsewhere (class LearningSwitch).
+// // The local copy should go away!
+// //
+// protected static final int OFMESSAGE_DAMPER_CAPACITY = 50000; // TODO: find sweet spot
+// protected static final int OFMESSAGE_DAMPER_TIMEOUT = 250; // ms
+
// Flow Entry ID generation state
private static Random randomGenerator = new Random();
private static int nextFlowEntryIdPrefix = 0;
@@ -76,7 +93,7 @@
runImpl();
} catch (Exception e) {
log.debug("Exception processing All Flow Entries from the Network MAP: ", e);
- dbHandler.rollback();
+ dbHandlerInner.rollback();
return;
}
}
@@ -107,10 +124,8 @@
// switches.
//
Iterable<IFlowEntry> allFlowEntries =
- dbHandler.getAllSwitchNotUpdatedFlowEntries();
+ dbHandlerInner.getAllSwitchNotUpdatedFlowEntries();
for (IFlowEntry flowEntryObj : allFlowEntries) {
- log.debug("flowEntryobj : {}", flowEntryObj);
-
counterAllFlowEntries++;
String dpidStr = flowEntryObj.getSwitchDpid();
@@ -122,7 +137,7 @@
continue; // Ignore the entry: not my switch
IFlowPath flowObj =
- dbHandler.getFlowPathByFlowEntry(flowEntryObj);
+ dbHandlerInner.getFlowPathByFlowEntry(flowEntryObj);
if (flowObj == null)
continue; // Should NOT happen
if (flowObj.getFlowId() == null)
@@ -145,15 +160,13 @@
}
counterMyNotUpdatedFlowEntries++;
}
-
- log.debug("addFlowEntries : {}", addFlowEntries);
//
// Process the Flow Entries that need to be added
//
for (IFlowEntry flowEntryObj : addFlowEntries) {
IFlowPath flowObj =
- dbHandler.getFlowPathByFlowEntry(flowEntryObj);
+ dbHandlerInner.getFlowPathByFlowEntry(flowEntryObj);
if (flowObj == null)
continue; // Should NOT happen
if (flowObj.getFlowId() == null)
@@ -177,16 +190,16 @@
while (! deleteFlowEntries.isEmpty()) {
IFlowEntry flowEntryObj = deleteFlowEntries.poll();
IFlowPath flowObj =
- dbHandler.getFlowPathByFlowEntry(flowEntryObj);
+ dbHandlerInner.getFlowPathByFlowEntry(flowEntryObj);
if (flowObj == null) {
log.debug("Did not find FlowPath to be deleted");
continue;
}
flowObj.removeFlowEntry(flowEntryObj);
- dbHandler.removeFlowEntry(flowEntryObj);
+ dbHandlerInner.removeFlowEntry(flowEntryObj);
}
- dbHandler.commit();
+ dbHandlerInner.commit();
long estimatedTime = System.nanoTime() - startTime;
double rate = 0.0;
@@ -211,7 +224,7 @@
runImpl();
} catch (Exception e) {
log.debug("Exception processing All Flows from the Network MAP: ", e);
- dbHandler.rollback();
+ dbHandlerInner.rollback();
return;
}
}
@@ -238,7 +251,7 @@
// Flow Paths this controller is responsible for.
//
Topology topology = topologyNetService.newDatabaseTopology();
- Iterable<IFlowPath> allFlowPaths = dbHandler.getAllFlowPaths();
+ Iterable<IFlowPath> allFlowPaths = dbHandlerInner.getAllFlowPaths();
for (IFlowPath flowPathObj : allFlowPaths) {
counterAllFlowPaths++;
if (flowPathObj == null)
@@ -340,12 +353,12 @@
//
while (! deleteFlows.isEmpty()) {
IFlowPath flowPathObj = deleteFlows.poll();
- dbHandler.removeFlowPath(flowPathObj);
+ dbHandlerInner.removeFlowPath(flowPathObj);
}
topologyNetService.dropTopology(topology);
- dbHandler.commit();
+ dbHandlerInner.commit();
long estimatedTime = System.nanoTime() - startTime;
double rate = 0.0;
@@ -368,7 +381,8 @@
*/
@Override
public void init(String conf) {
- dbHandler = new GraphDBOperation(conf);
+ dbHandlerApi = new GraphDBOperation(conf);
+ dbHandlerInner = new GraphDBOperation(conf);
}
/**
@@ -383,9 +397,9 @@
*/
@Override
public void close() {
- datagridService.deregisterPathComputationService(pathComputation);
- dbHandler.close();
- pusher.stop();
+ datagridService.deregisterFlowEventHandlerService(flowEventHandler);
+ dbHandlerApi.close();
+ dbHandlerInner.close();
}
/**
@@ -447,10 +461,14 @@
topologyNetService = context.getServiceImpl(ITopologyNetService.class);
datagridService = context.getServiceImpl(IDatagridService.class);
restApi = context.getServiceImpl(IRestApiService.class);
-
- pusher = new FlowPusher();
+
+// LEGACY
+// messageDamper = new OFMessageDamper(OFMESSAGE_DAMPER_CAPACITY,
+// EnumSet.of(OFType.FLOW_MOD),
+// OFMESSAGE_DAMPER_TIMEOUT);
+
+ pusher = new FlowPusher(NUM_PUSHER_THREAD);
pusher.init(null, floodlightProvider.getOFMessageFactory(), null);
-
this.init("");
mapReaderScheduler = Executors.newScheduledThreadPool(1);
@@ -493,20 +511,22 @@
nextFlowEntryIdPrefix = randomGenerator.nextInt();
pusher.start();
-
+
//
- // Create the Path Computation thread and register it with the
+ // Create the Flow Event Handler thread and register it with the
// Datagrid Service
//
- pathComputation = new PathComputation(this, datagridService);
- datagridService.registerPathComputationService(pathComputation);
+ flowEventHandler = new FlowEventHandler(this, datagridService);
+ datagridService.registerFlowEventHandlerService(flowEventHandler);
// Schedule the threads and periodic tasks
- pathComputation.start();
- mapReaderScheduler.scheduleAtFixedRate(
+ flowEventHandler.start();
+ if (! enableNotifications) {
+ mapReaderScheduler.scheduleAtFixedRate(
mapReader, 3, 3, TimeUnit.SECONDS);
- shortestPathReconcileScheduler.scheduleAtFixedRate(
+ shortestPathReconcileScheduler.scheduleAtFixedRate(
shortestPathReconcile, 3, 3, TimeUnit.SECONDS);
+ }
}
/**
@@ -519,7 +539,7 @@
@Override
public boolean addFlow(FlowPath flowPath, FlowId flowId) {
//
- // NOTE: We need to explicitly initialize the Flow Entry Switch State,
+ // NOTE: We need to explicitly initialize some of the state,
// in case the application didn't do it.
//
for (FlowEntry flowEntry : flowPath.flowEntries()) {
@@ -527,9 +547,11 @@
FlowEntrySwitchState.FE_SWITCH_UNKNOWN) {
flowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
}
+ if (! flowEntry.isValidFlowId())
+ flowEntry.setFlowId(new FlowId(flowPath.flowId().value()));
}
- if (FlowDatabaseOperation.addFlow(this, dbHandler, flowPath, flowId)) {
+ if (FlowDatabaseOperation.addFlow(this, dbHandlerApi, flowPath, flowId)) {
datagridService.notificationSendFlowAdded(flowPath);
return true;
}
@@ -544,8 +566,20 @@
* @return the added Flow Entry object on success, otherwise null.
*/
private IFlowEntry addFlowEntry(IFlowPath flowObj, FlowEntry flowEntry) {
- return FlowDatabaseOperation.addFlowEntry(this, dbHandler, flowObj,
- flowEntry);
+ return FlowDatabaseOperation.addFlowEntry(this, dbHandlerInner,
+ flowObj, flowEntry);
+ }
+
+ /**
+ * Delete a flow entry from the Network MAP.
+ *
+ * @param flowObj the corresponding Flow Path object for the Flow Entry.
+ * @param flowEntry the Flow Entry to delete.
+ * @return true on success, otherwise false.
+ */
+ private boolean deleteFlowEntry(IFlowPath flowObj, FlowEntry flowEntry) {
+ return FlowDatabaseOperation.deleteFlowEntry(dbHandlerInner,
+ flowObj, flowEntry);
}
/**
@@ -555,7 +589,7 @@
*/
@Override
public boolean deleteAllFlows() {
- if (FlowDatabaseOperation.deleteAllFlows(dbHandler)) {
+ if (FlowDatabaseOperation.deleteAllFlows(dbHandlerApi)) {
datagridService.notificationSendAllFlowsRemoved();
return true;
}
@@ -570,7 +604,7 @@
*/
@Override
public boolean deleteFlow(FlowId flowId) {
- if (FlowDatabaseOperation.deleteFlow(dbHandler, flowId)) {
+ if (FlowDatabaseOperation.deleteFlow(dbHandlerApi, flowId)) {
datagridService.notificationSendFlowRemoved(flowId);
return true;
}
@@ -584,7 +618,7 @@
*/
@Override
public boolean clearAllFlows() {
- if (FlowDatabaseOperation.clearAllFlows(dbHandler)) {
+ if (FlowDatabaseOperation.clearAllFlows(dbHandlerApi)) {
datagridService.notificationSendAllFlowsRemoved();
return true;
}
@@ -599,7 +633,7 @@
*/
@Override
public boolean clearFlow(FlowId flowId) {
- if (FlowDatabaseOperation.clearFlow(dbHandler, flowId)) {
+ if (FlowDatabaseOperation.clearFlow(dbHandlerApi, flowId)) {
datagridService.notificationSendFlowRemoved(flowId);
return true;
}
@@ -614,7 +648,7 @@
*/
@Override
public FlowPath getFlow(FlowId flowId) {
- return FlowDatabaseOperation.getFlow(dbHandler, flowId);
+ return FlowDatabaseOperation.getFlow(dbHandlerApi, flowId);
}
/**
@@ -624,7 +658,7 @@
*/
@Override
public ArrayList<FlowPath> getAllFlows() {
- return FlowDatabaseOperation.getAllFlows(dbHandler);
+ return FlowDatabaseOperation.getAllFlows(dbHandlerApi);
}
/**
@@ -638,7 +672,7 @@
@Override
public ArrayList<FlowPath> getAllFlows(CallerId installerId,
DataPathEndpoints dataPathEndpoints) {
- return FlowDatabaseOperation.getAllFlows(dbHandler, installerId,
+ return FlowDatabaseOperation.getAllFlows(dbHandlerApi, installerId,
dataPathEndpoints);
}
@@ -650,7 +684,8 @@
*/
@Override
public ArrayList<FlowPath> getAllFlows(DataPathEndpoints dataPathEndpoints) {
- return FlowDatabaseOperation.getAllFlows(dbHandler, dataPathEndpoints);
+ return FlowDatabaseOperation.getAllFlows(dbHandlerApi,
+ dataPathEndpoints);
}
/**
@@ -663,7 +698,7 @@
@Override
public ArrayList<IFlowPath> getAllFlowsSummary(FlowId flowId,
int maxFlows) {
- return FlowDatabaseOperation.getAllFlowsSummary(dbHandler, flowId,
+ return FlowDatabaseOperation.getAllFlowsSummary(dbHandlerApi, flowId,
maxFlows);
}
@@ -673,7 +708,7 @@
* @return all Flows information, without the associated Flow Entries.
*/
public ArrayList<IFlowPath> getAllFlowsWithoutFlowEntries() {
- return FlowDatabaseOperation.getAllFlowsWithoutFlowEntries(dbHandler);
+ return FlowDatabaseOperation.getAllFlowsWithoutFlowEntries(dbHandlerApi);
}
/**
@@ -700,6 +735,24 @@
}
/**
+ * Get the collection of my switches.
+ *
+ * @return the collection of my switches.
+ */
+ public Map<Long, IOFSwitch> getMySwitches() {
+ return floodlightProvider.getSwitches();
+ }
+
+ /**
+ * Get the network topology.
+ *
+ * @return the network topology.
+ */
+ public Topology getTopology() {
+ return flowEventHandler.getTopology();
+ }
+
+ /**
* Reconcile a flow.
*
* @param flowObj the flow that needs to be reconciliated.
@@ -707,6 +760,7 @@
* @return true on success, otherwise false.
*/
private boolean reconcileFlow(IFlowPath flowObj, DataPath newDataPath) {
+ String flowIdStr = flowObj.getFlowId();
//
// Set the incoming port matching and the outgoing port output
@@ -714,6 +768,8 @@
//
int idx = 0;
for (FlowEntry flowEntry : newDataPath.flowEntries()) {
+ flowEntry.setFlowId(new FlowId(flowIdStr));
+
// Mark the Flow Entry as not updated in the switch
flowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
// Set the incoming port matching
@@ -788,6 +844,11 @@
private boolean installFlowEntry(IOFSwitch mySwitch, IFlowPath flowObj,
IFlowEntry flowEntryObj) {
return pusher.add(mySwitch, flowObj, flowEntryObj);
+
+// LEGACY
+// return FlowSwitchOperation.installFlowEntry(
+// floodlightProvider.getOFMessageFactory(),
+// messageDamper, mySwitch, flowObj, flowEntryObj);
}
/**
@@ -801,6 +862,11 @@
private boolean installFlowEntry(IOFSwitch mySwitch, FlowPath flowPath,
FlowEntry flowEntry) {
return pusher.add(mySwitch, flowPath, flowEntry);
+
+// LEGACY
+// return FlowSwitchOperation.installFlowEntry(
+// floodlightProvider.getOFMessageFactory(),
+// messageDamper, mySwitch, flowPath, flowEntry);
}
/**
@@ -821,130 +887,217 @@
}
/**
- * Push the modified Flow Entries of a collection of Flow Paths.
- * Only the Flow Entries to switches controlled by this instance
+ * Push modified Flow Entries to switches.
+ *
+ * NOTE: Only the Flow Entries to switches controlled by this instance
* are pushed.
*
- * NOTE: Currently, we write to both the Network MAP and the switches.
- *
- * @param modifiedFlowPaths the collection of Flow Paths with the modified
- * Flow Entries.
+ * @param modifiedFlowEntries the collection of modified Flow Entries.
*/
- public void pushModifiedFlowEntries(Collection<FlowPath> modifiedFlowPaths) {
-
+ public void pushModifiedFlowEntriesToSwitches(
+ Collection<FlowPathEntryPair> modifiedFlowEntries) {
// TODO: For now, the pushing of Flow Entries is disabled
- if (true)
+ if (! enableNotifications)
return;
- Map<Long, IOFSwitch> mySwitches = floodlightProvider.getSwitches();
+ if (modifiedFlowEntries.isEmpty())
+ return;
- for (FlowPath flowPath : modifiedFlowPaths) {
- IFlowPath flowObj = dbHandler.searchFlowPath(flowPath.flowId());
- if (flowObj == null) {
- String logMsg = "Cannot find Network MAP entry for Flow Path " +
- flowPath.flowId();
+ Map<Long, IOFSwitch> mySwitches = getMySwitches();
+
+ for (FlowPathEntryPair flowPair : modifiedFlowEntries) {
+ FlowPath flowPath = flowPair.flowPath;
+ FlowEntry flowEntry = flowPair.flowEntry;
+
+ IOFSwitch mySwitch = mySwitches.get(flowEntry.dpid().value());
+ if (mySwitch == null)
+ continue;
+
+ log.debug("Pushing Flow Entry To Switch: {}", flowEntry.toString());
+
+ //
+ // Install the Flow Entry into the switch
+ //
+ if (! installFlowEntry(mySwitch, flowPath, flowEntry)) {
+ String logMsg = "Cannot install Flow Entry " +
+ flowEntry.flowEntryId() +
+ " from Flow Path " + flowPath.flowId() +
+ " on switch " + flowEntry.dpid();
log.error(logMsg);
continue;
}
- for (FlowEntry flowEntry : flowPath.flowEntries()) {
- if (flowEntry.flowEntrySwitchState() !=
- FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED) {
- continue; // No need to update the entry
- }
+ //
+ // NOTE: Here we assume that the switch has been
+ // successfully updated.
+ //
+ flowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_UPDATED);
+ }
+ }
- IOFSwitch mySwitch = mySwitches.get(flowEntry.dpid().value());
+ /**
+ * Push modified Flow Entries to the datagrid.
+ *
+ * @param modifiedFlowEntries the collection of modified Flow Entries.
+ */
+ public void pushModifiedFlowEntriesToDatagrid(
+ Collection<FlowPathEntryPair> modifiedFlowEntries) {
+ // TODO: For now, the pushing of Flow Entries is disabled
+ if (! enableNotifications)
+ return;
+
+ if (modifiedFlowEntries.isEmpty())
+ return;
+
+ Map<Long, IOFSwitch> mySwitches = getMySwitches();
+
+ for (FlowPathEntryPair flowPair : modifiedFlowEntries) {
+ FlowEntry flowEntry = flowPair.flowEntry;
+
+ IOFSwitch mySwitch = mySwitches.get(flowEntry.dpid().value());
+
+ //
+ // TODO: For now Flow Entries are removed by all instances,
+ // even if this Flow Entry is not for our switches.
+ //
+ // This is needed to handle the case a switch going down:
+ // it has no Master controller instance, hence no
+ // controller instance will cleanup its flow entries.
+ // This is sub-optimal: we need to elect a controller
+ // instance to handle the cleanup of such orphaned flow
+ // entries.
+ //
+ if (mySwitch == null) {
+ if (flowEntry.flowEntryUserState() !=
+ FlowEntryUserState.FE_USER_DELETE) {
+ continue;
+ }
+ if (! flowEntry.isValidFlowEntryId())
+ continue;
+ }
+
+ log.debug("Pushing Flow Entry To Datagrid: {}", flowEntry.toString());
+ //
+ // Write the Flow Entry to the Datagrid
+ //
+ switch (flowEntry.flowEntryUserState()) {
+ case FE_USER_ADD:
if (mySwitch == null)
- continue; // Ignore the entry: not my switch
+ break; // Install only flow entries for my switches
+ datagridService.notificationSendFlowEntryAdded(flowEntry);
+ break;
+ case FE_USER_MODIFY:
+ if (mySwitch == null)
+ break; // Install only flow entries for my switches
+ datagridService.notificationSendFlowEntryUpdated(flowEntry);
+ break;
+ case FE_USER_DELETE:
+ datagridService.notificationSendFlowEntryRemoved(flowEntry.flowEntryId());
+ break;
+ }
+ }
+ }
- //
- // Assign the FlowEntry ID if needed
- //
- if ((flowEntry.flowEntryId() == null) ||
- (flowEntry.flowEntryId().value() == 0)) {
- long id = getNextFlowEntryId();
- flowEntry.setFlowEntryId(new FlowEntryId(id));
- }
+ /**
+ * Push Flow Entries 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.
+ *
+ * @param modifiedFlowEntries the collection of Flow Entries to push.
+ */
+ public void pushModifiedFlowEntriesToDatabase(
+ Collection<FlowPathEntryPair> modifiedFlowEntries) {
+ // TODO: For now, the pushing of Flow Entries is disabled
+ if (! enableNotifications)
+ return;
- //
- // Install the Flow Entry into the switch
- //
- if (! installFlowEntry(mySwitch, flowPath, flowEntry)) {
- String logMsg = "Cannot install Flow Entry " +
- flowEntry.flowEntryId() +
- " from Flow Path " + flowPath.flowId() +
- " on switch " + flowEntry.dpid();
- log.error(logMsg);
- continue;
- }
+ if (modifiedFlowEntries.isEmpty())
+ return;
- //
- // NOTE: Here we assume that the switch has been successfully
- // updated.
- //
- flowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_UPDATED);
+ Map<Long, IOFSwitch> mySwitches = getMySwitches();
- //
- // Write the Flow Entry to the Network Map
- //
+ for (FlowPathEntryPair flowPair : modifiedFlowEntries) {
+ FlowPath flowPath = flowPair.flowPath;
+ FlowEntry flowEntry = flowPair.flowEntry;
+
+ if (! flowEntry.isValidFlowEntryId())
+ continue;
+
+ //
+ // Push the changes only on the instance responsible for the
+ // first switch.
+ //
+ Dpid srcDpid = flowPath.dataPath().srcPort().dpid();
+ IOFSwitch mySrcSwitch = mySwitches.get(srcDpid.value());
+ if (mySrcSwitch == null)
+ continue;
+
+ log.debug("Pushing Flow Entry To Database: {}", flowEntry.toString());
+ //
+ // Write the Flow Entry to the Network Map
+ //
+ // 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 {
- if (addFlowEntry(flowObj, flowEntry) == null) {
- String logMsg = "Cannot write to Network MAP Flow Entry " +
- flowEntry.flowEntryId() +
- " from Flow Path " + flowPath.flowId() +
- " on switch " + flowEntry.dpid();
+ //
+ // 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);
- continue;
+ 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) {
- String logMsg = "Exception writing Flow Entry to Network MAP";
- log.debug(logMsg);
- dbHandler.rollback();
- continue;
+ 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) {
+ }
}
}
}
-
- dbHandler.commit();
}
-
- @Override
- public void addMessage(long dpid, OFMessage msg) {
- IOFSwitch sw = floodlightProvider.getSwitches().get(dpid);
- if (sw == null) {
- return;
- }
-
- pusher.add(sw, msg);
- }
-
- @Override
- public boolean suspend(long dpid) {
- IOFSwitch sw = floodlightProvider.getSwitches().get(dpid);
- if (sw == null) {
- return false;
- }
-
- return pusher.suspend(sw);
- }
-
- @Override
- public boolean resume(long dpid) {
- IOFSwitch sw = floodlightProvider.getSwitches().get(dpid);
- if (sw == null) {
- return false;
- }
-
- return pusher.resume(sw);
- }
-
- @Override
- public boolean isSuspended(long dpid) {
- IOFSwitch sw = floodlightProvider.getSwitches().get(dpid);
- if (sw == null) {
- return false;
- }
-
- return pusher.isSuspended(sw);
- }
}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowEventHandlerService.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowEventHandlerService.java
new file mode 100644
index 0000000..78562e1
--- /dev/null
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowEventHandlerService.java
@@ -0,0 +1,73 @@
+package net.onrc.onos.ofcontroller.flowmanager;
+
+import net.onrc.onos.ofcontroller.topology.TopologyElement;
+import net.onrc.onos.ofcontroller.util.FlowEntry;
+import net.onrc.onos.ofcontroller.util.FlowPath;
+
+/**
+ * Interface for providing Flow Event Handler Service to other modules.
+ */
+public interface IFlowEventHandlerService {
+ /**
+ * Receive a notification that a Flow is added.
+ *
+ * @param flowPath the Flow that is added.
+ */
+ void notificationRecvFlowAdded(FlowPath flowPath);
+
+ /**
+ * Receive a notification that a Flow is removed.
+ *
+ * @param flowPath the Flow that is removed.
+ */
+ void notificationRecvFlowRemoved(FlowPath flowPath);
+
+ /**
+ * Receive a notification that a Flow is updated.
+ *
+ * @param flowPath the Flow that is updated.
+ */
+ void notificationRecvFlowUpdated(FlowPath flowPath);
+
+ /**
+ * Receive a notification that a FlowEntry is added.
+ *
+ * @param flowEntry the FlowEntry that is added.
+ */
+ void notificationRecvFlowEntryAdded(FlowEntry flowEntry);
+
+ /**
+ * Receive a notification that a FlowEntry is removed.
+ *
+ * @param flowEntry the FlowEntry that is removed.
+ */
+ void notificationRecvFlowEntryRemoved(FlowEntry flowEntry);
+
+ /**
+ * Receive a notification that a FlowEntry is updated.
+ *
+ * @param flowEntry the FlowEntry that is updated.
+ */
+ void notificationRecvFlowEntryUpdated(FlowEntry flowEntry);
+
+ /**
+ * Receive a notification that a Topology Element is added.
+ *
+ * @param topologyElement the Topology Element that is added.
+ */
+ void notificationRecvTopologyElementAdded(TopologyElement topologyElement);
+
+ /**
+ * Receive a notification that a Topology Element is removed.
+ *
+ * @param topologyElement the Topology Element that is removed.
+ */
+ void notificationRecvTopologyElementRemoved(TopologyElement topologyElement);
+
+ /**
+ * Receive a notification that a Topology Element is updated.
+ *
+ * @param topologyElement the Topology Element that is updated.
+ */
+ void notificationRecvTopologyElementUpdated(TopologyElement topologyElement);
+}
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 1f8cd5b..8d362d1 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowService.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowService.java
@@ -4,6 +4,7 @@
import net.floodlightcontroller.core.module.IFloodlightService;
import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IFlowPath;
+import net.onrc.onos.ofcontroller.topology.Topology;
import net.onrc.onos.ofcontroller.util.CallerId;
import net.onrc.onos.ofcontroller.util.DataPathEndpoints;
import net.onrc.onos.ofcontroller.util.FlowId;
@@ -112,4 +113,11 @@
* @return the added shortest-path flow on success, otherwise null.
*/
FlowPath addAndMaintainShortestPathFlow(FlowPath flowPath);
+
+ /**
+ * Get the network topology.
+ *
+ * @return the network topology.
+ */
+ Topology getTopology();
}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IPathComputationService.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IPathComputationService.java
deleted file mode 100644
index 1bc0be1..0000000
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IPathComputationService.java
+++ /dev/null
@@ -1,51 +0,0 @@
-package net.onrc.onos.ofcontroller.flowmanager;
-
-import net.onrc.onos.ofcontroller.topology.TopologyElement;
-import net.onrc.onos.ofcontroller.util.FlowPath;
-
-/**
- * Interface for providing Path Computation Service to other modules.
- */
-public interface IPathComputationService {
- /**
- * Receive a notification that a Flow is added.
- *
- * @param flowPath the flow that is added.
- */
- void notificationRecvFlowAdded(FlowPath flowPath);
-
- /**
- * Receive a notification that a Flow is removed.
- *
- * @param flowPath the flow that is removed.
- */
- void notificationRecvFlowRemoved(FlowPath flowPath);
-
- /**
- * Receive a notification that a Flow is updated.
- *
- * @param flowPath the flow that is updated.
- */
- void notificationRecvFlowUpdated(FlowPath flowPath);
-
- /**
- * Receive a notification that a Topology Element is added.
- *
- * @param topologyElement the Topology Element that is added.
- */
- void notificationRecvTopologyElementAdded(TopologyElement topologyElement);
-
- /**
- * Receive a notification that a Topology Element is removed.
- *
- * @param topologyElement the Topology Element that is removed.
- */
- void notificationRecvTopologyElementRemoved(TopologyElement topologyElement);
-
- /**
- * Receive a notification that a Topology Element is updated.
- *
- * @param topologyElement the Topology Element that is updated.
- */
- void notificationRecvTopologyElementUpdated(TopologyElement topologyElement);
-}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/PathComputation.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/PathComputation.java
deleted file mode 100644
index ae14e09..0000000
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/PathComputation.java
+++ /dev/null
@@ -1,495 +0,0 @@
-package net.onrc.onos.ofcontroller.flowmanager;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import net.onrc.onos.datagrid.IDatagridService;
-import net.onrc.onos.ofcontroller.topology.ShortestPath;
-import net.onrc.onos.ofcontroller.topology.Topology;
-import net.onrc.onos.ofcontroller.topology.TopologyElement;
-import net.onrc.onos.ofcontroller.topology.TopologyManager;
-import net.onrc.onos.ofcontroller.util.DataPath;
-import net.onrc.onos.ofcontroller.util.EventEntry;
-import net.onrc.onos.ofcontroller.util.FlowEntry;
-import net.onrc.onos.ofcontroller.util.FlowEntryAction;
-import net.onrc.onos.ofcontroller.util.FlowEntryActions;
-import net.onrc.onos.ofcontroller.util.FlowEntryMatch;
-import net.onrc.onos.ofcontroller.util.FlowEntrySwitchState;
-import net.onrc.onos.ofcontroller.util.FlowEntryUserState;
-import net.onrc.onos.ofcontroller.util.FlowPath;
-import net.onrc.onos.ofcontroller.util.FlowPathUserState;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Class for implementing the Path Computation and Path Maintenance.
- */
-class PathComputation extends Thread implements IPathComputationService {
- /** The logger. */
- private final static Logger log = LoggerFactory.getLogger(PathComputation.class);
-
- 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>();
-
- // The queue with Flow Path and Topology Element updates
- private BlockingQueue<EventEntry<?>> networkEvents =
- new LinkedBlockingQueue<EventEntry<?>>();
-
- // The pending Topology and Flow Path events
- private List<EventEntry<TopologyElement>> topologyEvents =
- new LinkedList<EventEntry<TopologyElement>>();
- private List<EventEntry<FlowPath>> flowPathEvents =
- new LinkedList<EventEntry<FlowPath>>();
-
- /**
- * Constructor for a given Flow Manager and Datagrid Service.
- *
- * @param flowManager the Flow Manager to use.
- * @param datagridService the Datagrid Service to use.
- */
- PathComputation(FlowManager flowManager,
- IDatagridService datagridService) {
- this.flowManager = flowManager;
- this.datagridService = datagridService;
- this.topology = new Topology();
- }
-
- /**
- * Run the thread.
- */
- @Override
- public void run() {
- //
- // Obtain the initial Topology state
- //
- Collection<TopologyElement> topologyElements =
- datagridService.getAllTopologyElements();
- for (TopologyElement topologyElement : topologyElements) {
- EventEntry<TopologyElement> eventEntry =
- new EventEntry<TopologyElement>(EventEntry.Type.ENTRY_ADD, topologyElement);
- topologyEvents.add(eventEntry);
- }
- //
- // Obtain the initial Flow Path state
- //
- Collection<FlowPath> flowPaths = datagridService.getAllFlows();
- for (FlowPath flowPath : flowPaths) {
- EventEntry<FlowPath> eventEntry =
- new EventEntry<FlowPath>(EventEntry.Type.ENTRY_ADD, flowPath);
- flowPathEvents.add(eventEntry);
- }
- // Process the events (if any)
- processEvents();
-
- //
- // The main loop
- //
- Collection<EventEntry<?>> collection = new LinkedList<EventEntry<?>>();
- try {
- while (true) {
- EventEntry<?> eventEntry = networkEvents.take();
- collection.add(eventEntry);
- networkEvents.drainTo(collection);
-
- //
- // Demultiplex all events:
- // - EventEntry<TopologyElement>
- // - EventEntry<FlowPath>
- //
- for (EventEntry<?> event : collection) {
- if (event.eventData() instanceof TopologyElement) {
- EventEntry<TopologyElement> topologyEventEntry =
- (EventEntry<TopologyElement>)event;
- topologyEvents.add(topologyEventEntry);
- } else if (event.eventData() instanceof FlowPath) {
- EventEntry<FlowPath> flowPathEventEntry =
- (EventEntry<FlowPath>)event;
- flowPathEvents.add(flowPathEventEntry);
- }
- }
- collection.clear();
-
- // Process the events (if any)
- processEvents();
- }
- } catch (Exception exception) {
- log.debug("Exception processing Network Events: ", exception);
- }
- }
-
- /**
- * Process the events (if any)
- */
- private void processEvents() {
- List<FlowPath> newFlowPaths = new LinkedList<FlowPath>();
- List<FlowPath> recomputeFlowPaths = new LinkedList<FlowPath>();
- List<FlowPath> modifiedFlowPaths = new LinkedList<FlowPath>();
-
- if (topologyEvents.isEmpty() && flowPathEvents.isEmpty())
- return; // Nothing to do
-
- //
- // Process the Flow Path events
- //
- for (EventEntry<FlowPath> eventEntry : flowPathEvents) {
- FlowPath flowPath = eventEntry.eventData();
-
- switch (eventEntry.eventType()) {
- case ENTRY_ADD: {
- //
- // Add a new Flow Path
- //
- 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.
- //
- break;
- }
-
- switch (flowPath.flowPathType()) {
- case FP_TYPE_SHORTEST_PATH:
- //
- // Reset the Data Path, in case it was set already, because
- // we are going to recompute it anyway.
- //
- flowPath.flowEntries().clear();
- recomputeFlowPaths.add(flowPath);
- break;
- case FP_TYPE_EXPLICIT_PATH:
- //
- // Mark all Flow Entries for installation in the switches.
- //
- for (FlowEntry flowEntry : flowPath.flowEntries()) {
- flowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
- }
- modifiedFlowPaths.add(flowPath);
- break;
- }
- newFlowPaths.add(flowPath);
-
- break;
- }
-
- case ENTRY_REMOVE: {
- //
- // Remove an existing Flow Path.
- //
- // Find the Flow Path, and mark the Flow and its Flow Entries
- // for deletion.
- //
- FlowPath existingFlowPath =
- allFlowPaths.get(flowPath.flowId().value());
- if (existingFlowPath == null)
- continue; // Nothing to do
-
- existingFlowPath.setFlowPathUserState(FlowPathUserState.FP_USER_DELETE);
- for (FlowEntry flowEntry : existingFlowPath.flowEntries()) {
- flowEntry.setFlowEntryUserState(FlowEntryUserState.FE_USER_DELETE);
- flowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
- }
-
- allFlowPaths.remove(existingFlowPath.flowId());
- modifiedFlowPaths.add(existingFlowPath);
-
- break;
- }
- }
- }
-
- //
- // Process the topology events
- //
- boolean isTopologyModified = false;
- for (EventEntry<TopologyElement> eventEntry : topologyEvents) {
- TopologyElement topologyElement = eventEntry.eventData();
- switch (eventEntry.eventType()) {
- case ENTRY_ADD:
- isTopologyModified = topology.addTopologyElement(topologyElement);
- break;
- case ENTRY_REMOVE:
- isTopologyModified = topology.removeTopologyElement(topologyElement);
- break;
- }
- }
- if (isTopologyModified) {
- // TODO: For now, if the topology changes, we recompute all Flows
- recomputeFlowPaths.addAll(allFlowPaths.values());
- }
-
- // Add all new Flows
- for (FlowPath flowPath : newFlowPaths) {
- allFlowPaths.put(flowPath.flowId().value(), flowPath);
- }
-
- // Recompute all affected Flow Paths and keep only the modified
- for (FlowPath flowPath : recomputeFlowPaths) {
- if (recomputeFlowPath(flowPath))
- modifiedFlowPaths.add(flowPath);
- }
-
- //
- // Push the Flow Entries that have been modified
- //
- flowManager.pushModifiedFlowEntries(modifiedFlowPaths);
-
- // Cleanup
- topologyEvents.clear();
- flowPathEvents.clear();
- }
-
- /**
- * Recompute a Flow Path.
- *
- * @param flowPath the Flow Path to recompute.
- * @return true if the recomputed Flow Path has changed, otherwise false.
- */
- private boolean recomputeFlowPath(FlowPath flowPath) {
- boolean hasChanged = false;
-
- //
- // Test whether the Flow Path needs to be recomputed
- //
- switch (flowPath.flowPathType()) {
- case FP_TYPE_SHORTEST_PATH:
- break;
- case FP_TYPE_EXPLICIT_PATH:
- return false; // An explicit path never changes
- }
-
- DataPath oldDataPath = flowPath.dataPath();
-
- // Compute the new path
- DataPath newDataPath = TopologyManager.computeNetworkPath(topology,
- flowPath);
- if (newDataPath == null) {
- // We need the DataPath to compare the paths
- newDataPath = new DataPath();
- }
- newDataPath.applyFlowPathFlags(flowPath.flowPathFlags());
-
- //
- // Test whether the new path is same
- //
- if (oldDataPath.flowEntries().size() !=
- newDataPath.flowEntries().size()) {
- hasChanged = true;
- } else {
- Iterator<FlowEntry> oldIter = oldDataPath.flowEntries().iterator();
- Iterator<FlowEntry> newIter = newDataPath.flowEntries().iterator();
- while (oldIter.hasNext() && newIter.hasNext()) {
- FlowEntry oldFlowEntry = oldIter.next();
- FlowEntry newFlowEntry = newIter.next();
- if (! TopologyManager.isSameFlowEntryDataPath(oldFlowEntry,
- newFlowEntry)) {
- hasChanged = true;
- break;
- }
- }
- }
- if (! hasChanged)
- return hasChanged;
-
- //
- // Merge the changes in the path:
- // - If a Flow Entry for a switch is in the old data path, but not
- // in the new data path, then mark it for deletion.
- // - If a Flow Entry for a switch is in the new data path, but not
- // in the old data path, then mark it for addition.
- // - If a Flow Entry for a switch is in both the old and the new
- // data path, but it has changed, e.g., the incoming and/or outgoing
- // port(s), then mark the old Flow Entry for deletion, and mark
- // the new Flow Entry for addition.
- // - If a Flow Entry for a switch is in both the old and the new
- // data path, and it hasn't changed, then just keep it.
- //
- // NOTE: We use the Switch DPID of each entry to match the entries
- //
- Map<Long, FlowEntry> oldFlowEntriesMap = new HashMap<Long, FlowEntry>();
- Map<Long, FlowEntry> newFlowEntriesMap = new HashMap<Long, FlowEntry>();
- ArrayList<FlowEntry> finalFlowEntries = new ArrayList<FlowEntry>();
- List<FlowEntry> deletedFlowEntries = new LinkedList<FlowEntry>();
-
- // Prepare maps with the Flow Entries, so they are fast to lookup
- for (FlowEntry flowEntry : oldDataPath.flowEntries())
- oldFlowEntriesMap.put(flowEntry.dpid().value(), flowEntry);
- for (FlowEntry flowEntry : newDataPath.flowEntries())
- newFlowEntriesMap.put(flowEntry.dpid().value(), flowEntry);
-
- //
- // Find the old Flow Entries that should be deleted
- //
- for (FlowEntry oldFlowEntry : oldDataPath.flowEntries()) {
- FlowEntry newFlowEntry =
- newFlowEntriesMap.get(oldFlowEntry.dpid().value());
- if (newFlowEntry == null) {
- // The old Flow Entry should be deleted
- oldFlowEntry.setFlowEntryUserState(FlowEntryUserState.FE_USER_DELETE);
- oldFlowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
- deletedFlowEntries.add(oldFlowEntry);
- }
- }
-
- //
- // Find the new Flow Entries that should be added or updated
- //
- int idx = 0;
- for (FlowEntry newFlowEntry : newDataPath.flowEntries()) {
- FlowEntry oldFlowEntry =
- oldFlowEntriesMap.get(newFlowEntry.dpid().value());
-
- if ((oldFlowEntry != null) &&
- TopologyManager.isSameFlowEntryDataPath(oldFlowEntry,
- newFlowEntry)) {
- //
- // Both Flow Entries are same
- //
- finalFlowEntries.add(oldFlowEntry);
- idx++;
- continue;
- }
-
- if (oldFlowEntry != null) {
- //
- // The old Flow Entry should be deleted
- //
- oldFlowEntry.setFlowEntryUserState(FlowEntryUserState.FE_USER_DELETE);
- oldFlowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
- deletedFlowEntries.add(oldFlowEntry);
- }
-
- //
- // Add the new Flow Entry
- //
-
- // Set the incoming port matching
- FlowEntryMatch flowEntryMatch = new FlowEntryMatch();
- newFlowEntry.setFlowEntryMatch(flowEntryMatch);
- flowEntryMatch.enableInPort(newFlowEntry.inPort());
-
- //
- // Set the actions:
- // If the first Flow Entry, copy the Flow Path actions to it.
- //
- FlowEntryActions flowEntryActions = newFlowEntry.flowEntryActions();
- if ((idx == 0) && (flowPath.flowEntryActions() != null)) {
- FlowEntryActions flowActions =
- new FlowEntryActions(flowPath.flowEntryActions());
- for (FlowEntryAction action : flowActions.actions())
- flowEntryActions.addAction(action);
- }
- idx++;
-
- //
- // Add the outgoing port output action
- //
- FlowEntryAction flowEntryAction = new FlowEntryAction();
- flowEntryAction.setActionOutput(newFlowEntry.outPort());
- flowEntryActions.addAction(flowEntryAction);
-
- //
- // Set the state of the new Flow Entry
- //
- newFlowEntry.setFlowEntryUserState(FlowEntryUserState.FE_USER_ADD);
- newFlowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED);
- finalFlowEntries.add(newFlowEntry);
- }
-
- //
- // Replace the old Flow Entries with the new Flow Entries.
- // Note that the Flow Entries that will be deleted are added at
- // the end.
- //
- for (FlowEntry flowEntry : deletedFlowEntries)
- finalFlowEntries.add(flowEntry);
- flowPath.dataPath().setFlowEntries(finalFlowEntries);
-
- return hasChanged;
- }
-
- /**
- * Receive a notification that a Flow is added.
- *
- * @param flowPath the flow that is added.
- */
- @Override
- public void notificationRecvFlowAdded(FlowPath flowPath) {
- EventEntry<FlowPath> eventEntry =
- new EventEntry<FlowPath>(EventEntry.Type.ENTRY_ADD, flowPath);
- networkEvents.add(eventEntry);
- }
-
- /**
- * Receive a notification that a Flow is removed.
- *
- * @param flowPath the flow that is removed.
- */
- @Override
- public void notificationRecvFlowRemoved(FlowPath flowPath) {
- EventEntry<FlowPath> eventEntry =
- new EventEntry<FlowPath>(EventEntry.Type.ENTRY_REMOVE, flowPath);
- networkEvents.add(eventEntry);
- }
-
- /**
- * Receive a notification that a Flow is updated.
- *
- * @param flowPath the flow that is updated.
- */
- @Override
- public void notificationRecvFlowUpdated(FlowPath flowPath) {
- // NOTE: The ADD and UPDATE events are processed in same way
- EventEntry<FlowPath> eventEntry =
- new EventEntry<FlowPath>(EventEntry.Type.ENTRY_ADD, flowPath);
- networkEvents.add(eventEntry);
- }
-
- /**
- * Receive a notification that a Topology Element is added.
- *
- * @param topologyElement the Topology Element that is added.
- */
- @Override
- public void notificationRecvTopologyElementAdded(TopologyElement topologyElement) {
- EventEntry<TopologyElement> eventEntry =
- new EventEntry<TopologyElement>(EventEntry.Type.ENTRY_ADD, topologyElement);
- networkEvents.add(eventEntry);
- }
-
- /**
- * Receive a notification that a Topology Element is removed.
- *
- * @param topologyElement the Topology Element that is removed.
- */
- @Override
- public void notificationRecvTopologyElementRemoved(TopologyElement topologyElement) {
- EventEntry<TopologyElement> eventEntry =
- new EventEntry<TopologyElement>(EventEntry.Type.ENTRY_REMOVE, topologyElement);
- networkEvents.add(eventEntry);
- }
-
- /**
- * Receive a notification that a Topology Element is updated.
- *
- * @param topologyElement the Topology Element that is updated.
- */
- @Override
- public void notificationRecvTopologyElementUpdated(TopologyElement topologyElement) {
- // NOTE: The ADD and UPDATE events are processed in same way
- EventEntry<TopologyElement> eventEntry =
- new EventEntry<TopologyElement>(EventEntry.Type.ENTRY_ADD, topologyElement);
- networkEvents.add(eventEntry);
- }
-}