Merge branch 'ONOS-ONRC2014-Measurements' of github.com:OPENNETWORKINGLAB/ONOS into RAMCloud-merge
Conflicts:
src/main/java/net/onrc/onos/graph/GraphDBOperation.java
src/main/java/net/onrc/onos/graph/IDBOperation.java
src/main/java/net/onrc/onos/graph/LocalTopologyEventListener.java
src/main/java/net/onrc/onos/ofcontroller/core/internal/DeviceStorageImpl.java
src/main/java/net/onrc/onos/ofcontroller/core/internal/LinkStorageImpl.java
src/main/java/net/onrc/onos/ofcontroller/core/internal/TopoLinkServiceImpl.java
src/main/java/net/onrc/onos/ofcontroller/core/internal/TopoSwitchServiceImpl.java
src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java
src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
src/main/java/net/onrc/onos/ofcontroller/topology/Topology.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 af7bca8..1f0c163 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java
@@ -3,9 +3,11 @@
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.LinkedList;
import java.util.Map;
+import net.floodlightcontroller.core.IOFSwitch;
import net.floodlightcontroller.util.MACAddress;
import net.onrc.onos.graph.DBOperation;
import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IBaseObject;
@@ -139,7 +141,7 @@
//
flowProp.setFlowId(flowPath.flowId().toString());
if ( measureONOSFlowTimeProp ) {
- numPropsSet += 2;
+ numPropsSet += 1;
}
//
@@ -613,11 +615,7 @@
}
}
- // TODO: Hacks with hard-coded state names!
- if (found)
- flowProp.setUserState("FE_USER_MODIFY");
- else
- flowProp.setUserState("FE_USER_ADD");
+ flowProp.setUserState(flowEntry.flowEntryUserState().toString());
flowProp.setSwitchState(flowEntry.flowEntrySwitchState().toString());
if (measureONOSFlowEntryTimeProp) {
numProperties += 2;
@@ -804,6 +802,77 @@
}
/**
+ * Get a previously added flow entry.
+ *
+ * @param dbHandler the Graph Database handler to use.
+ * @param flowEntryId the Flow Entry ID of the flow entry to get.
+ * @return the Flow Entry if found, otherwise null.
+ */
+ static FlowEntry getFlowEntry(DBOperation dbHandler,
+ FlowEntryId flowEntryId) {
+ IFlowEntry flowEntryObj = null;
+ try {
+ flowEntryObj = dbHandler.searchFlowEntry(flowEntryId);
+ } catch (Exception e) {
+ // TODO: handle exceptions
+ dbHandler.rollback();
+ log.error(":getFlowEntry FlowEntryId:{} failed", flowEntryId);
+ return null;
+ }
+ if (flowEntryObj == null) {
+ dbHandler.commit();
+ return null; // Flow not found
+ }
+
+ //
+ // Extract the Flow Entry state
+ //
+ FlowEntry flowEntry = extractFlowEntry(flowEntryObj);
+ dbHandler.commit();
+
+ return flowEntry;
+ }
+
+ /**
+ * Get the source switch DPID of a previously added flow.
+ *
+ * @param dbHandler the Graph Database handler to use.
+ * @param flowId the Flow ID of the flow to get.
+ * @return the source switch DPID if found, otherwise null.
+ */
+ static Dpid getFlowSourceDpid(DBOperation dbHandler, FlowId flowId) {
+ IFlowPath flowObj = null;
+ try {
+ flowObj = dbHandler.searchFlowPath(flowId);
+ } catch (Exception e) {
+ // TODO: handle exceptions
+ dbHandler.rollback();
+ log.error(":getFlowSourceDpid FlowId:{} failed", flowId);
+ return null;
+ }
+ if (flowObj == null) {
+ dbHandler.commit();
+ return null; // Flow not found
+ }
+
+ //
+ // Extract the Flow Source DPID
+ //
+ String srcSwitchStr = flowObj.getSrcSwitch();
+ if (srcSwitchStr == null) {
+ // TODO: A work-around, becauuse of some bogus database objects
+ dbHandler.commit();
+ return null;
+ }
+
+ Dpid dpid = new Dpid(srcSwitchStr);
+
+ dbHandler.commit();
+
+ return dpid;
+ }
+
+ /**
* Get all installed flows by all installers.
*
* @param dbHandler the Graph Database handler to use.
@@ -841,12 +910,88 @@
}
/**
+ * Get all installed flows whose Source Switch is controlled by this
+ * instance.
+ *
+ * @param dbHandler the Graph Database handler to use.
+ * @param mySwitches the collection of the switches controlled by this
+ * instance.
+ * @return the Flow Paths if found, otherwise null.
+ */
+ static ArrayList<FlowPath> getAllMyFlows(DBOperation dbHandler,
+ Map<Long, IOFSwitch> mySwitches) {
+ Iterable<IFlowPath> flowPathsObj = null;
+ ArrayList<FlowPath> flowPaths = new ArrayList<FlowPath>();
+
+ try {
+ flowPathsObj = dbHandler.getAllFlowPaths();
+ } catch (Exception e) {
+ // TODO: handle exceptions
+ dbHandler.rollback();
+ log.error(":getAllMyFlowPaths failed");
+ return flowPaths;
+ }
+ if ((flowPathsObj == null) || (flowPathsObj.iterator().hasNext() == false)) {
+ dbHandler.commit();
+ return flowPaths; // No Flows found
+ }
+
+ for (IFlowPath flowObj : flowPathsObj) {
+ //
+ // Extract the Source Switch DPID and ignore if the switch
+ // is not controlled by this instance.
+ //
+ String srcSwitchStr = flowObj.getSrcSwitch();
+ if (srcSwitchStr == null) {
+ // TODO: A work-around, becauuse of some bogus database objects
+ continue;
+ }
+ Dpid dpid = new Dpid(srcSwitchStr);
+ if (mySwitches.get(dpid.value()) == null)
+ continue;
+
+ //
+ // Extract the Flow state
+ //
+ FlowPath flowPath = extractFlowPath(flowObj);
+ if (flowPath != null)
+ flowPaths.add(flowPath);
+ }
+
+ dbHandler.commit();
+
+ return flowPaths;
+ }
+
+ /**
+ * Get a subset of installed flows.
+ *
+ * @param dbHandler the Graph Database handler to use.
+ * @param flowIds the collection of Flow IDs to get.
+ * @return the Flow Paths if found, otherwise null.
+ */
+ static ArrayList<FlowPath> getFlows(DBOperation dbHandler,
+ Collection<FlowId> flowIds) {
+ ArrayList<FlowPath> flowPaths = new ArrayList<FlowPath>();
+
+ // TODO: This implementation should use threads
+ for (FlowId flowId : flowIds) {
+ FlowPath flowPath = getFlow(dbHandler, flowId);
+ if (flowPath != null)
+ flowPaths.add(flowPath);
+ }
+ // dbHandler.commit();
+
+ return flowPaths;
+ }
+
+ /**
* Extract Flow Path State from a Titan Database Object @ref IFlowPath.
*
* @param flowObj the object to extract the Flow Path State from.
* @return the extracted Flow Path State.
*/
- private static FlowPath extractFlowPath(IFlowPath flowObj) {
+ static FlowPath extractFlowPath(IFlowPath flowObj) {
//
// Extract the Flow state
//
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 001fb3c..cf9b67d 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
@@ -15,11 +15,15 @@
import java.util.concurrent.LinkedBlockingQueue;
import net.floodlightcontroller.core.IOFSwitch;
+import net.floodlightcontroller.core.IOFSwitchListener;
import net.onrc.onos.datagrid.IDatagridService;
+import net.onrc.onos.graph.DBOperation;
+import net.onrc.onos.graph.GraphDBManager;
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.Dpid;
import net.onrc.onos.ofcontroller.util.EventEntry;
import net.onrc.onos.ofcontroller.util.FlowEntry;
import net.onrc.onos.ofcontroller.util.FlowEntryAction;
@@ -31,6 +35,8 @@
import net.onrc.onos.ofcontroller.util.FlowId;
import net.onrc.onos.ofcontroller.util.FlowPath;
import net.onrc.onos.ofcontroller.util.FlowPathUserState;
+import net.onrc.onos.ofcontroller.util.Pair;
+import net.onrc.onos.ofcontroller.util.Port;
import net.onrc.onos.ofcontroller.util.serializers.KryoFactory;
import com.esotericsoftware.kryo2.Kryo;
@@ -46,17 +52,16 @@
* - Detect FlowPaths impacted by Topology change.
* - Recompute impacted FlowPath using cached Topology.
*/
-class FlowEventHandler extends Thread implements IFlowEventHandlerService {
+class FlowEventHandler extends Thread implements IFlowEventHandlerService,
+ IOFSwitchListener {
+
+ private boolean enableOnrc2014MeasurementsFlows = true;
+ private boolean enableOnrc2014MeasurementsTopology = true;
+
/** The logger. */
private final static Logger log = LoggerFactory.getLogger(FlowEventHandler.class);
- // Flag to refresh Topology object periodically
- private final static boolean refreshTopology = false;
- // Refresh delay(ms)
- private final static long refreshTopologyDelay = 5000;
- // Refresh interval(ms)
- private final static long refreshTopologyInterval = 2000;
- private Timer refreshTopologyTimer;
+ private DBOperation dbHandler;
private FlowManager flowManager; // The Flow Manager to use
private IDatagridService datagridService; // The Datagrid Service to use
@@ -74,6 +79,12 @@
new LinkedList<EventEntry<FlowPath>>();
private List<EventEntry<FlowEntry>> flowEntryEvents =
new LinkedList<EventEntry<FlowEntry>>();
+ private List<EventEntry<Pair<FlowId, Dpid>>> flowIdEvents =
+ new LinkedList<EventEntry<Pair<FlowId, Dpid>>>();
+ private List<EventEntry<Pair<FlowEntryId, Dpid>>> flowEntryIdEvents =
+ new LinkedList<EventEntry<Pair<FlowEntryId, Dpid>>>();
+ private List<EventEntry<Dpid>> switchDpidEvents =
+ new LinkedList<EventEntry<Dpid>>();
// All internally computed Flow Paths
private Map<Long, FlowPath> allFlowPaths = new HashMap<Long, FlowPath>();
@@ -119,6 +130,8 @@
* Startup processing.
*/
private void startup() {
+ this.dbHandler = GraphDBManager.getDBOperation("ramcloud", "/tmp/ramcloudconf");
+
//
// Obtain the initial Topology state
//
@@ -148,32 +161,33 @@
flowEntryEvents.add(eventEntry);
}
+ //
+ // Obtain the initial FlowId state
+ //
+ Collection<Pair<FlowId, Dpid>> flowIds =
+ datagridService.getAllFlowIds();
+ for (Pair<FlowId, Dpid> pair : flowIds) {
+ EventEntry<Pair<FlowId, Dpid>> eventEntry =
+ new EventEntry<Pair<FlowId, Dpid>>(EventEntry.Type.ENTRY_ADD, pair);
+ flowIdEvents.add(eventEntry);
+ }
+
+ //
+ // Obtain the initial FlowEntryId state
+ //
+ Collection<Pair<FlowEntryId, Dpid>> flowEntryIds =
+ datagridService.getAllFlowEntryIds();
+ for (Pair<FlowEntryId, Dpid> pair : flowEntryIds) {
+ EventEntry<Pair<FlowEntryId, Dpid>> eventEntry =
+ new EventEntry<Pair<FlowEntryId, Dpid>>(EventEntry.Type.ENTRY_ADD, pair);
+ flowEntryIdEvents.add(eventEntry);
+ }
+
// Process the initial events (if any)
synchronized (allFlowPaths) {
processEvents();
}
- if (refreshTopology) {
- refreshTopologyTimer = new Timer();
- refreshTopologyTimer.schedule(new TimerTask() {
- @Override
- public void run() {
- PerfMon pm = PerfMon.getInstance();
- log.debug("[BEFORE] {}", topology);
- long begin, end;
- synchronized(topology) {
- begin = System.nanoTime();
- pm.read_whole_topology_start();
- topology.readFromDatabase(flowManager.dbHandlerInner);
- pm.read_whole_topology_end();
- end = System.nanoTime();
- }
- // FIXME level raised for measurement. Was debug
- log.error("[AFTER] {}", topology);
- log.error("refresh takes : {}[us]", (end - begin) / 1000.0);
- }
- }, refreshTopologyDelay, refreshTopologyInterval);
- }
}
/**
@@ -199,12 +213,15 @@
// - EventEntry<TopologyElement>
// - EventEntry<FlowPath>
// - EventEntry<FlowEntry>
+ // - EventEntry<Pair<FlowId, Dpid>>
+ // - EventEntry<Pair<FlowEntryId, Dpid>>
//
for (EventEntry<?> event : collection) {
// Topology event
if (event.eventData() instanceof TopologyElement) {
EventEntry<TopologyElement> topologyEventEntry =
(EventEntry<TopologyElement>)event;
+
topologyEvents.add(topologyEventEntry);
continue;
}
@@ -224,6 +241,34 @@
flowEntryEvents.add(flowEntryEventEntry);
continue;
}
+
+ // FlowId event
+ if (event.eventData() instanceof Pair) {
+ EventEntry<Pair<FlowId, Dpid>> flowIdEventEntry =
+ (EventEntry<Pair<FlowId, Dpid>>)event;
+ flowIdEvents.add(flowIdEventEntry);
+ continue;
+ }
+
+ // Switch Dpid event
+ if (event.eventData() instanceof Dpid) {
+ EventEntry<Dpid> switchDpidEventEntry =
+ (EventEntry<Dpid>)event;
+ switchDpidEvents.add(switchDpidEventEntry);
+ continue;
+ }
+
+ // FlowEntryId event
+ // TODO: Fix the code below if we need again to handle
+ // the FlowEntryId events
+ /*
+ if (event.eventData() instanceof Pair) {
+ EventEntry<Pair<FlowEntryId, Dpid>> flowEntryIdEventEntry =
+ (EventEntry<Pair<FlowEntryId, Dpid>>)event;
+ flowEntryIdEvents.add(flowEntryIdEventEntry);
+ continue;
+ }
+ */
}
collection.clear();
@@ -236,13 +281,89 @@
log.debug("Exception processing Network Events: ", exception);
}
}
-
+
/**
* Process the events (if any)
*/
private void processEvents() {
Collection<FlowEntry> modifiedFlowEntries;
+ if (enableOnrc2014MeasurementsFlows) {
+
+ if (topologyEvents.isEmpty() && flowIdEvents.isEmpty() &&
+ switchDpidEvents.isEmpty()) {
+ return; // Nothing to do
+ }
+
+ Map<Long, IOFSwitch> mySwitches = flowManager.getMySwitches();
+
+ // Process the Switch Dpid events
+ processSwitchDpidEvents();
+
+ // Process the Flow ID events
+ processFlowIdEvents(mySwitches);
+
+ // Fetch the topology
+ processTopologyEvents();
+
+ // Recompute all affected Flow Paths and keep only the modified
+ for (FlowPath flowPath : shouldRecomputeFlowPaths.values()) {
+ if (recomputeFlowPath(flowPath))
+ modifiedFlowPaths.put(flowPath.flowId().value(), flowPath);
+ }
+
+ // Assign the Flow Entry ID as needed
+ for (FlowPath flowPath : modifiedFlowPaths.values()) {
+ for (FlowEntry flowEntry : flowPath.flowEntries()) {
+ if (! flowEntry.isValidFlowEntryId()) {
+ long id = flowManager.getNextFlowEntryId();
+ flowEntry.setFlowEntryId(new FlowEntryId(id));
+ }
+ }
+ }
+
+ //
+ // Push the modified state to the database
+ //
+ for (FlowPath flowPath : modifiedFlowPaths.values()) {
+ //
+ // Delete the Flow Path from the Network Map
+ //
+ if (flowPath.flowPathUserState() ==
+ FlowPathUserState.FP_USER_DELETE) {
+ log.debug("Deleting Flow Path From Database: {}", flowPath);
+ // TODO: For now the deleting of a Flow Path is blocking
+ ParallelFlowDatabaseOperation.deleteFlow(dbHandler,
+ flowPath.flowId());
+ // Send the notifications for the deleted Flow Entries
+ for (FlowEntry flowEntry : flowPath.flowEntries()) {
+ datagridService.notificationSendFlowEntryRemoved(flowEntry.flowEntryId());
+ }
+
+ continue;
+ }
+
+ log.debug("Pushing Flow Path To Database: {}", flowPath);
+ //
+ // Write the Flow Path to the Network Map
+ //
+ ParallelFlowDatabaseOperation.addFlow(dbHandler, flowPath,
+ datagridService);
+ }
+
+ // Cleanup
+ topologyEvents.clear();
+ flowIdEvents.clear();
+ switchDpidEvents.clear();
+ //
+ // NOTE: Keep a cache with my Flow Paths
+ // allFlowPaths.clear();
+ shouldRecomputeFlowPaths.clear();
+ modifiedFlowPaths.clear();
+
+ return;
+ }
+
if (topologyEvents.isEmpty() && flowPathEvents.isEmpty() &&
flowEntryEvents.isEmpty()) {
return; // Nothing to do
@@ -379,6 +500,246 @@
}
/**
+ * Fix a flow fetched from the database.
+ *
+ * @param flowPath the Flow to fix.
+ */
+ private void fixFlowFromDatabase(FlowPath flowPath) {
+ //
+ // TODO: Bug workaround / fix :
+ // method FlowDatabaseOperation.extractFlowEntry() doesn't
+ // fetch the inPort and outPort, hence we assign them here.
+ //
+ // Assign the inPort and outPort for the Flow Entries
+ for (FlowEntry flowEntry : flowPath.flowEntries()) {
+ // Set the inPort
+ do {
+ if (flowEntry.inPort() != null)
+ break;
+ if (flowEntry.flowEntryMatch() == null)
+ break;
+ Port inPort = new Port(flowEntry.flowEntryMatch().inPort().value());
+ flowEntry.setInPort(inPort);
+ } while (false);
+
+ // Set the outPort
+ do {
+ if (flowEntry.outPort() != null)
+ break;
+ for (FlowEntryAction fa : flowEntry.flowEntryActions().actions()) {
+ if (fa.actionOutput() != null) {
+ Port outPort = new Port(fa.actionOutput().port().value());
+ flowEntry.setOutPort(outPort);
+ break;
+ }
+ }
+ } while (false);
+ }
+ }
+
+ /**
+ * Process the Switch Dpid events.
+ */
+ private void processSwitchDpidEvents() {
+ Map<Long, Dpid> addedSwitches = new HashMap<Long, Dpid>();
+ Map<Long, Dpid> removedSwitches = new HashMap<Long, Dpid>();
+
+ //
+ // Process all Switch Dpid events and update the appropriate state
+ //
+ for (EventEntry<Dpid> eventEntry : switchDpidEvents) {
+ Dpid dpid = eventEntry.eventData();
+
+ log.debug("SwitchDpid Event: {} {}", eventEntry.eventType(), dpid);
+
+ // Compute the final set of added and removed switches
+ switch (eventEntry.eventType()) {
+ case ENTRY_ADD:
+ addedSwitches.put(dpid.value(), dpid);
+ removedSwitches.remove(dpid.value());
+ break;
+ case ENTRY_REMOVE:
+ addedSwitches.remove(dpid.value());
+ removedSwitches.put(dpid.value(), dpid);
+ break;
+ }
+ }
+
+ //
+ // Remove the Flows from the local cache if the removed
+ // switch is the Source Switch.
+ //
+ // TODO: This search can be expensive for a large number of flows
+ // and should be optmized.
+ //
+ List<FlowId> deleteFlowIds = new LinkedList<FlowId>();
+ for (Dpid switchDpid : removedSwitches.values()) {
+ for (FlowPath flowPath : allFlowPaths.values()) {
+ Dpid srcDpid = flowPath.dataPath().srcPort().dpid();
+ if (srcDpid.value() == switchDpid.value())
+ deleteFlowIds.add(flowPath.flowId());
+ }
+ }
+ //
+ // Remove the Flows from the local cache
+ //
+ for (FlowId flowId : deleteFlowIds)
+ allFlowPaths.remove(flowId.value());
+
+ // Get the Flows for the added switches
+ Collection<FlowPath> flowPaths =
+ ParallelFlowDatabaseOperation.getFlowsForSwitches(dbHandler,
+ addedSwitches.values());
+ for (FlowPath flowPath : flowPaths) {
+ allFlowPaths.put(flowPath.flowId().value(), flowPath);
+ }
+ }
+
+ /**
+ * Process the Flow ID events.
+ *
+ * @param mySwitches the collection of my switches.
+ */
+ private void processFlowIdEvents(Map<Long, IOFSwitch> mySwitches) {
+ List<FlowId> shouldFetchMyFlowIds = new LinkedList<FlowId>();
+
+ //
+ // Process all Flow Id events and update the appropriate state
+ //
+ for (EventEntry<Pair<FlowId, Dpid>> eventEntry : flowIdEvents) {
+ Pair<FlowId, Dpid> pair = eventEntry.eventData();
+ FlowId flowId = pair.first;
+ Dpid dpid = pair.second;
+
+ log.debug("Flow ID Event: {} {} {}", eventEntry.eventType(),
+ flowId, dpid);
+
+ //
+ // Ignore Flows if the Source Switch is not controlled by this
+ // instance.
+ //
+ if (mySwitches.get(dpid.value()) == null)
+ continue;
+
+ switch (eventEntry.eventType()) {
+ case ENTRY_ADD: {
+ //
+ // Add a new Flow Path
+ //
+ if (allFlowPaths.get(flowId.value()) != null) {
+ //
+ // TODO: What to do if the Flow Path already exists?
+ // Fow now, we just re-add it.
+ //
+ }
+ shouldFetchMyFlowIds.add(flowId);
+
+ 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(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);
+ }
+
+ // Remove the Flow Path from the internal state
+ Long key = existingFlowPath.flowId().value();
+ allFlowPaths.remove(key);
+ shouldRecomputeFlowPaths.remove(key);
+ modifiedFlowPaths.put(key, existingFlowPath);
+
+ break;
+ }
+ }
+ }
+
+ // Get my Flows
+ Collection<FlowPath> myFlows =
+ ParallelFlowDatabaseOperation.getFlows(dbHandler,
+ shouldFetchMyFlowIds);
+
+ for (FlowPath flowPath : myFlows) {
+ fixFlowFromDatabase(flowPath);
+
+ 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();
+ shouldRecomputeFlowPaths.put(flowPath.flowId().value(),
+ 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.put(flowPath.flowId().value(), flowPath);
+ break;
+ case FP_TYPE_UNKNOWN:
+ log.error("FlowPath event with unknown type");
+ break;
+ }
+ allFlowPaths.put(flowPath.flowId().value(), flowPath);
+ }
+ }
+
+ /**
+ * Process the Flow Entry ID events.
+ *
+ * @param mySwitches the collection of my switches.
+ * @return a collection of modified Flow Entries this instance needs
+ * to push to its own switches.
+ */
+ private Collection<FlowEntry> processFlowEntryIdEvents(Map<Long, IOFSwitch> mySwitches) {
+ List<FlowEntry> modifiedFlowEntries = new LinkedList<FlowEntry>();
+
+ //
+ // Process all Flow ID events and update the appropriate state
+ //
+ for (EventEntry<Pair<FlowEntryId, Dpid>> eventEntry : flowEntryIdEvents) {
+ Pair<FlowEntryId, Dpid> pair = eventEntry.eventData();
+ FlowEntryId flowEntryId = pair.first;
+ Dpid dpid = pair.second;
+
+ log.debug("Flow Entry ID Event: {} {} {}", eventEntry.eventType(),
+ flowEntryId, dpid);
+
+ if (mySwitches.get(dpid.value()) == null)
+ continue;
+
+ // Fetch the Flow Entry
+ FlowEntry flowEntry = FlowDatabaseOperation.getFlowEntry(dbHandler,
+ flowEntryId);
+ if (flowEntry == null) {
+ log.debug("Flow Entry ID {} : Flow Entry not found!",
+ flowEntryId);
+ continue;
+ }
+ modifiedFlowEntries.add(flowEntry);
+ }
+
+ return modifiedFlowEntries;
+ }
+
+ /**
* Process the Flow Path events.
*/
private void processFlowPathEvents() {
@@ -464,15 +825,34 @@
* Process the Topology events.
*/
private void processTopologyEvents() {
+ boolean isTopologyModified = false;
+
+ if (enableOnrc2014MeasurementsTopology) {
+ if (topologyEvents.isEmpty())
+ return;
+
+ // TODO: Code for debugging purpose only
+ for (EventEntry<TopologyElement> eventEntry : topologyEvents) {
+ TopologyElement topologyElement = eventEntry.eventData();
+ log.debug("Topology Event: {} {}", eventEntry.eventType(),
+ topologyElement.toString());
+ }
+
+ log.debug("[BEFORE] {}", topology.toString());
+ topology.readFromDatabase(dbHandler);
+ log.debug("[AFTER] {}", topology.toString());
+ shouldRecomputeFlowPaths.putAll(allFlowPaths);
+ return;
+ }
+
//
// 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);
+ topologyElement.toString());
switch (eventEntry.eventType()) {
case ENTRY_ADD:
@@ -752,6 +1132,19 @@
private boolean recomputeFlowPath(FlowPath flowPath) {
boolean hasChanged = false;
+ if (enableOnrc2014MeasurementsFlows) {
+ // Cleanup the deleted Flow Entries from the earlier iteration
+ flowPath.dataPath().removeDeletedFlowEntries();
+
+ //
+ // TODO: Fake it that the Flow Entries have been already pushed
+ // into the switches, so we don't push them again.
+ //
+ for (FlowEntry flowEntry : flowPath.flowEntries()) {
+ flowEntry.setFlowEntrySwitchState(FlowEntrySwitchState.FE_SWITCH_UPDATED);
+ }
+ }
+
//
// Test whether the Flow Path needs to be recomputed
//
@@ -772,7 +1165,6 @@
newDataPath = TopologyManager.computeNetworkPath(topology,
flowPath);
}
-
if (newDataPath == null) {
// We need the DataPath to compare the paths
newDataPath = new DataPath();
@@ -982,6 +1374,13 @@
*/
@Override
public void notificationRecvFlowEntryAdded(FlowEntry flowEntry) {
+ if (enableOnrc2014MeasurementsFlows) {
+ Collection entries = new ArrayList();
+ entries.add(flowEntry);
+ flowManager.pushModifiedFlowEntriesToSwitches(entries);
+ return;
+ }
+
EventEntry<FlowEntry> eventEntry =
new EventEntry<FlowEntry>(EventEntry.Type.ENTRY_ADD, flowEntry);
networkEvents.add(eventEntry);
@@ -994,6 +1393,19 @@
*/
@Override
public void notificationRecvFlowEntryRemoved(FlowEntry flowEntry) {
+ if (enableOnrc2014MeasurementsFlows) {
+ //
+ // NOTE: Must update the state to DELETE, because
+ // the notification contains the original state.
+ //
+ flowEntry.setFlowEntryUserState(FlowEntryUserState.FE_USER_DELETE);
+
+ Collection entries = new ArrayList();
+ entries.add(flowEntry);
+ flowManager.pushModifiedFlowEntriesToSwitches(entries);
+ return;
+ }
+
EventEntry<FlowEntry> eventEntry =
new EventEntry<FlowEntry>(EventEntry.Type.ENTRY_REMOVE, flowEntry);
networkEvents.add(eventEntry);
@@ -1006,6 +1418,13 @@
*/
@Override
public void notificationRecvFlowEntryUpdated(FlowEntry flowEntry) {
+ if (enableOnrc2014MeasurementsFlows) {
+ Collection entries = new ArrayList();
+ entries.add(flowEntry);
+ flowManager.pushModifiedFlowEntriesToSwitches(entries);
+ return;
+ }
+
// NOTE: The ADD and UPDATE events are processed in same way
EventEntry<FlowEntry> eventEntry =
new EventEntry<FlowEntry>(EventEntry.Type.ENTRY_ADD, flowEntry);
@@ -1013,6 +1432,101 @@
}
/**
+ * Receive a notification that a FlowId is added.
+ *
+ * @param flowId the FlowId that is added.
+ * @param dpid the Source Switch Dpid for the corresponding Flow.
+ */
+ @Override
+ public void notificationRecvFlowIdAdded(FlowId flowId, Dpid dpid) {
+ Pair flowIdPair = new Pair(flowId, dpid);
+
+ EventEntry<Pair<FlowId, Dpid>> eventEntry =
+ new EventEntry<Pair<FlowId, Dpid>>(EventEntry.Type.ENTRY_ADD, flowIdPair);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a FlowId is removed.
+ *
+ * @param flowId the FlowId that is removed.
+ * @param dpid the Source Switch Dpid for the corresponding Flow.
+ */
+ @Override
+ public void notificationRecvFlowIdRemoved(FlowId flowId, Dpid dpid) {
+ Pair flowIdPair = new Pair(flowId, dpid);
+
+ EventEntry<Pair<FlowId, Dpid>> eventEntry =
+ new EventEntry<Pair<FlowId, Dpid>>(EventEntry.Type.ENTRY_REMOVE, flowIdPair);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a FlowId is updated.
+ *
+ * @param flowId the FlowId that is updated.
+ * @param dpid the Source Switch Dpid for the corresponding Flow.
+ */
+ @Override
+ public void notificationRecvFlowIdUpdated(FlowId flowId, Dpid dpid) {
+ Pair flowIdPair = new Pair(flowId, dpid);
+
+ // NOTE: The ADD and UPDATE events are processed in same way
+ EventEntry<Pair<FlowId, Dpid>> eventEntry =
+ new EventEntry<Pair<FlowId, Dpid>>(EventEntry.Type.ENTRY_ADD, flowIdPair);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a FlowEntryId is added.
+ *
+ * @param flowEntryId the FlowEntryId that is added.
+ * @param dpid the Switch Dpid for the corresponding Flow Entry.
+ */
+ @Override
+ public void notificationRecvFlowEntryIdAdded(FlowEntryId flowEntryId,
+ Dpid dpid) {
+ Pair flowEntryIdPair = new Pair(flowEntryId, dpid);
+
+ EventEntry<Pair<FlowEntryId, Dpid>> eventEntry =
+ new EventEntry<Pair<FlowEntryId, Dpid>>(EventEntry.Type.ENTRY_ADD, flowEntryIdPair);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a FlowEntryId is removed.
+ *
+ * @param flowEntryId the FlowEntryId that is removed.
+ * @param dpid the Switch Dpid for the corresponding Flow Entry.
+ */
+ @Override
+ public void notificationRecvFlowEntryIdRemoved(FlowEntryId flowEntryId,
+ Dpid dpid) {
+ Pair flowEntryIdPair = new Pair(flowEntryId, dpid);
+
+ EventEntry<Pair<FlowEntryId, Dpid>> eventEntry =
+ new EventEntry<Pair<FlowEntryId, Dpid>>(EventEntry.Type.ENTRY_REMOVE, flowEntryIdPair);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a FlowEntryId is updated.
+ *
+ * @param flowEntryId the FlowEntryId that is updated.
+ * @param dpid the Switch Dpid for the corresponding Flow Entry.
+ */
+ @Override
+ public void notificationRecvFlowEntryIdUpdated(FlowEntryId flowEntryId,
+ Dpid dpid) {
+ Pair flowEntryIdPair = new Pair(flowEntryId, dpid);
+
+ // NOTE: The ADD and UPDATE events are processed in same way
+ EventEntry<Pair<FlowEntryId, Dpid>> eventEntry =
+ new EventEntry<Pair<FlowEntryId, Dpid>>(EventEntry.Type.ENTRY_ADD, flowEntryIdPair);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
* Receive a notification that a Topology Element is added.
*
* @param topologyElement the Topology Element that is added.
@@ -1050,6 +1564,40 @@
}
/**
+ * Receive a notification that a switch is added to this instance.
+ *
+ * @param sw the switch that is added.
+ */
+ @Override
+ public void addedSwitch(IOFSwitch sw) {
+ Dpid dpid = new Dpid(sw.getId());
+ EventEntry<Dpid> eventEntry =
+ new EventEntry<Dpid>(EventEntry.Type.ENTRY_ADD, dpid);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that a switch is removed from this instance.
+ *
+ * @param sw the switch that is removed.
+ */
+ @Override
+ public void removedSwitch(IOFSwitch sw) {
+ Dpid dpid = new Dpid(sw.getId());
+ EventEntry<Dpid> eventEntry =
+ new EventEntry<Dpid>(EventEntry.Type.ENTRY_REMOVE, dpid);
+ networkEvents.add(eventEntry);
+ }
+
+ /**
+ * Receive a notification that the ports on a switch have changed.
+ */
+ @Override
+ public void switchPortChanged(Long switchId) {
+ // Nothing to do
+ }
+
+ /**
* Get a sorted copy of all Flow Paths.
*
* @return a sorted copy of all Flow Paths.
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 611b5c9..fc5ae34 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
@@ -8,6 +8,7 @@
import java.util.Map;
import java.util.Random;
import java.util.SortedMap;
+import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
@@ -50,8 +51,9 @@
* Flow Manager class for handling the network flows.
*/
public class FlowManager implements IFloodlightModule, IFlowService, INetMapStorage {
- // flag to use FlowPusher instead of FlowSwitchOperation/MessageDamper
- private final static boolean enableFlowPusher = false;
+
+ private boolean enableOnrc2014MeasurementsFlows = true;
+
protected DBOperation dbHandlerApi;
protected DBOperation dbHandlerInner;
@@ -98,7 +100,7 @@
* Shutdown the Flow Manager operation.
*/
@Override
- public void finalize() {
+ protected void finalize() {
close();
}
@@ -107,6 +109,7 @@
*/
@Override
public void close() {
+ floodlightProvider.removeOFSwitchListener(flowEventHandler);
datagridService.deregisterFlowEventHandlerService(flowEventHandler);
dbHandlerApi.close();
dbHandlerInner.close();
@@ -233,6 +236,7 @@
// - startup
//
flowEventHandler = new FlowEventHandler(this, datagridService);
+ floodlightProvider.addOFSwitchListener(flowEventHandler);
datagridService.registerFlowEventHandlerService(flowEventHandler);
flowEventHandler.start();
}
@@ -273,7 +277,13 @@
}
if (FlowDatabaseOperation.addFlow(dbHandlerApi, flowPath)) {
- datagridService.notificationSendFlowAdded(flowPath);
+ if (enableOnrc2014MeasurementsFlows) {
+ datagridService.notificationSendFlowIdAdded(flowPath.flowId(),
+ flowPath.dataPath().srcPort().dpid());
+ } else {
+ datagridService.notificationSendFlowAdded(flowPath);
+ }
+
return flowPath.flowId();
}
return null;
@@ -287,7 +297,11 @@
@Override
public boolean deleteAllFlows() {
if (FlowDatabaseOperation.deleteAllFlows(dbHandlerApi)) {
- datagridService.notificationSendAllFlowsRemoved();
+ if (enableOnrc2014MeasurementsFlows) {
+ datagridService.notificationSendAllFlowIdsRemoved();
+ } else {
+ datagridService.notificationSendAllFlowsRemoved();
+ }
return true;
}
return false;
@@ -302,7 +316,11 @@
@Override
public boolean deleteFlow(FlowId flowId) {
if (FlowDatabaseOperation.deleteFlow(dbHandlerApi, flowId)) {
- datagridService.notificationSendFlowRemoved(flowId);
+ if (enableOnrc2014MeasurementsFlows) {
+ datagridService.notificationSendFlowIdRemoved(flowId);
+ } else {
+ datagridService.notificationSendFlowRemoved(flowId);
+ }
return true;
}
return false;
@@ -320,6 +338,26 @@
}
/**
+ * Get a previously added flow entry.
+ *
+ * @param flowEntryId the Flow Entry ID of the flow entry to get.
+ * @return the Flow Entry if found, otherwise null.
+ */
+ public FlowEntry getFlowEntry(FlowEntryId flowEntryId) {
+ return FlowDatabaseOperation.getFlowEntry(dbHandlerApi, flowEntryId);
+ }
+
+ /**
+ * Get the source switch DPID of a previously added flow.
+ *
+ * @param flowId the Flow ID of the flow to get.
+ * @return the source switch DPID if found, otherwise null.
+ */
+ public Dpid getFlowSourceDpid(FlowId flowId) {
+ return FlowDatabaseOperation.getFlowSourceDpid(dbHandlerApi, flowId);
+ }
+
+ /**
* Get all installed flows by all installers.
*
* @return the Flow Paths if found, otherwise null.
@@ -330,6 +368,18 @@
}
/**
+ * Get all installed flows whose Source Switch is controlled by this
+ * instance.
+ *
+ * @param mySwitches the collection of the switches controlled by this
+ * instance.
+ * @return the Flow Paths if found, otherwise null.
+ */
+ public ArrayList<FlowPath> getAllMyFlows(Map<Long, IOFSwitch> mySwitches) {
+ return FlowDatabaseOperation.getAllMyFlows(dbHandlerApi, mySwitches);
+ }
+
+ /**
* Get summary of all installed flows by all installers in a given range.
*
* @param flowId the Flow ID of the first flow in the flow range to get.
@@ -341,7 +391,17 @@
int maxFlows) {
ArrayList<FlowPath> flowPaths = new ArrayList<FlowPath>();
SortedMap<Long, FlowPath> sortedFlowPaths =
- flowEventHandler.getAllFlowPathsCopy();
+ new TreeMap<Long, FlowPath>();
+
+ if (enableOnrc2014MeasurementsFlows) {
+ Collection<FlowPath> databaseFlowPaths =
+ ParallelFlowDatabaseOperation.getAllFlows(dbHandlerApi);
+ for (FlowPath flowPath : databaseFlowPaths) {
+ sortedFlowPaths.put(flowPath.flowId().value(), flowPath);
+ }
+ } else {
+ sortedFlowPaths = flowEventHandler.getAllFlowPathsCopy();
+ }
//
// Truncate each Flow Path and Flow Entry
@@ -421,6 +481,9 @@
public void flowEntriesPushedToSwitch(
Collection<Pair<IOFSwitch, FlowEntry>> entries) {
+ if (enableOnrc2014MeasurementsFlows)
+ return;
+
//
// Process all entries
//
@@ -490,8 +553,12 @@
// - Flow Paths to the database
//
pushModifiedFlowEntriesToSwitches(modifiedFlowEntries);
- pushModifiedFlowPathsToDatabase(modifiedFlowPaths);
- cleanupDeletedFlowEntriesFromDatagrid(modifiedFlowEntries);
+ if (enableOnrc2014MeasurementsFlows) {
+ writeModifiedFlowPathsToDatabase(modifiedFlowPaths);
+ } else {
+ pushModifiedFlowPathsToDatabase(modifiedFlowPaths);
+ cleanupDeletedFlowEntriesFromDatagrid(modifiedFlowEntries);
+ }
}
/**
@@ -502,7 +569,7 @@
*
* @param modifiedFlowEntries the collection of modified Flow Entries.
*/
- private void pushModifiedFlowEntriesToSwitches(
+ void pushModifiedFlowEntriesToSwitches(
Collection<FlowEntry> modifiedFlowEntries) {
if (modifiedFlowEntries.isEmpty())
return;
@@ -679,7 +746,7 @@
*
* @param modifiedFlowPaths the collection of Flow Paths to write.
*/
- private void writeModifiedFlowPathsToDatabase(
+ void writeModifiedFlowPathsToDatabase(
Collection<FlowPath> modifiedFlowPaths) {
if (modifiedFlowPaths.isEmpty())
return;
@@ -719,7 +786,7 @@
retry = true;
} catch (Exception e) {
log.error("Exception deleting Flow Path from Network MAP: {}", e);
- }
+ }
} while (retry);
continue;
@@ -738,10 +805,12 @@
allValid = false;
break;
}
- if (flowEntry.flowEntrySwitchState() !=
- FlowEntrySwitchState.FE_SWITCH_UPDATED) {
- allValid = false;
- break;
+ if (! enableOnrc2014MeasurementsFlows) {
+ if (flowEntry.flowEntrySwitchState() !=
+ FlowEntrySwitchState.FE_SWITCH_UPDATED) {
+ allValid = false;
+ break;
+ }
}
}
if (! allValid)
@@ -777,6 +846,36 @@
//log.error("Performance %% Flow path total time {} : {}", endTime - startTime, flowPath.toString());
}
} while (retry);
+
+ if (enableOnrc2014MeasurementsFlows) {
+ // Send the notifications
+
+ for (FlowEntry flowEntry : flowPath.flowEntries()) {
+ if (flowEntry.flowEntrySwitchState() !=
+ FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED) {
+ continue;
+ }
+ // datagridService.notificationSendFlowEntryIdAdded(flowEntry.flowEntryId(), flowEntry.dpid());
+
+ //
+ // Write the Flow Entry to the Datagrid
+ //
+ switch (flowEntry.flowEntryUserState()) {
+ case FE_USER_ADD:
+ datagridService.notificationSendFlowEntryAdded(flowEntry);
+ break;
+ case FE_USER_MODIFY:
+ datagridService.notificationSendFlowEntryUpdated(flowEntry);
+ break;
+ case FE_USER_DELETE:
+ datagridService.notificationSendFlowEntryRemoved(flowEntry.flowEntryId());
+ break;
+ case FE_USER_UNKNOWN:
+ assert(false);
+ break;
+ }
+ }
+ }
}
}
}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowEventHandlerService.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowEventHandlerService.java
index 78562e1..a44a898 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowEventHandlerService.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowEventHandlerService.java
@@ -1,7 +1,10 @@
package net.onrc.onos.ofcontroller.flowmanager;
import net.onrc.onos.ofcontroller.topology.TopologyElement;
+import net.onrc.onos.ofcontroller.util.Dpid;
import net.onrc.onos.ofcontroller.util.FlowEntry;
+import net.onrc.onos.ofcontroller.util.FlowEntryId;
+import net.onrc.onos.ofcontroller.util.FlowId;
import net.onrc.onos.ofcontroller.util.FlowPath;
/**
@@ -51,6 +54,56 @@
void notificationRecvFlowEntryUpdated(FlowEntry flowEntry);
/**
+ * Receive a notification that a FlowId is added.
+ *
+ * @param flowId the FlowId that is added.
+ * @param dpid the Source Switch Dpid for the corresponding Flow.
+ */
+ void notificationRecvFlowIdAdded(FlowId flowId, Dpid dpid);
+
+ /**
+ * Receive a notification that a FlowId is removed.
+ *
+ * @param flowId the FlowId that is removed.
+ * @param dpid the Source Switch Dpid for the corresponding Flow.
+ */
+ void notificationRecvFlowIdRemoved(FlowId flowId, Dpid dpid);
+
+ /**
+ * Receive a notification that a FlowId is updated.
+ *
+ * @param flowId the FlowId that is updated.
+ * @param dpid the Source Switch Dpid for the corresponding Flow.
+ */
+ void notificationRecvFlowIdUpdated(FlowId flowId, Dpid dpid);
+
+ /**
+ * Receive a notification that a FlowEntryId is added.
+ *
+ * @param flowEntryId the FlowEntryId that is added.
+ * @param dpid the Switch Dpid for the corresponding Flow Entry.
+ */
+ void notificationRecvFlowEntryIdAdded(FlowEntryId flowEntryId, Dpid dpid);
+
+ /**
+ * Receive a notification that a FlowEntryId is removed.
+ *
+ * @param flowEntryId the FlowEntryId that is removed.
+ * @param dpid the Switch Dpid for the corresponding Flow Entry.
+ */
+ void notificationRecvFlowEntryIdRemoved(FlowEntryId flowEntryId,
+ Dpid dpid);
+
+ /**
+ * Receive a notification that a FlowEntryId is updated.
+ *
+ * @param flowEntryId the FlowEntryId that is updated.
+ * @param dpid the Switch Dpid for the corresponding Flow Entry.
+ */
+ void notificationRecvFlowEntryIdUpdated(FlowEntryId flowEntryId,
+ Dpid dpid);
+
+ /**
* Receive a notification that a Topology Element is added.
*
* @param topologyElement the Topology Element that is added.
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/ParallelFlowDatabaseOperation.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/ParallelFlowDatabaseOperation.java
new file mode 100644
index 0000000..212c59f
--- /dev/null
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/ParallelFlowDatabaseOperation.java
@@ -0,0 +1,367 @@
+package net.onrc.onos.ofcontroller.flowmanager;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import net.onrc.onos.datagrid.IDatagridService;
+import net.onrc.onos.graph.DBOperation;
+import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IFlowPath;
+import net.onrc.onos.ofcontroller.util.Dpid;
+import net.onrc.onos.ofcontroller.util.FlowEntry;
+import net.onrc.onos.ofcontroller.util.FlowEntrySwitchState;
+import net.onrc.onos.ofcontroller.util.FlowId;
+import net.onrc.onos.ofcontroller.util.FlowPath;
+
+/**
+ * Class for performing parallel Flow-related operations on the Database.
+ *
+ * This class is mostly a wrapper of FlowDatabaseOperation with a thread pool
+ * for parallelization.
+ *
+ * @author Brian O'Connor <brian@onlab.us>
+ */
+public class ParallelFlowDatabaseOperation extends FlowDatabaseOperation {
+ private final static Logger log = LoggerFactory.getLogger(FlowDatabaseOperation.class);
+
+ private final static int numThreads = Integer.valueOf(System.getProperty("parallelFlowDatabase.numThreads", "32"));
+ private final static ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+
+ /**
+ * Get all installed flows by first querying the database for all FlowPaths
+ * and then populating them from the database in parallel.
+ *
+ * @param dbHandler the Graph Database handler to use.
+ * @return the Flow Paths if found, otherwise an empty list.
+ */
+ static ArrayList<FlowPath> getAllFlows(DBOperation dbHandler) {
+ Iterable<IFlowPath> flowPathsObj = null;
+ ArrayList<FlowPath> flowPaths = new ArrayList<FlowPath>();
+
+ try {
+ flowPathsObj = dbHandler.getAllFlowPaths();
+ } catch (Exception e) {
+ // TODO: handle exceptions
+ dbHandler.rollback();
+ log.error(":getAllFlowPaths failed");
+ return flowPaths;
+ }
+ if ((flowPathsObj == null) || (flowPathsObj.iterator().hasNext() == false)) {
+ dbHandler.commit();
+ return flowPaths; // No Flows found
+ }
+
+ CompletionService<FlowPath> tasks = new ExecutorCompletionService<>(executor);
+ int numTasks = 0;
+ for(IFlowPath flowObj : flowPathsObj) {
+ tasks.submit(new ExtractFlowTask(flowObj));
+ numTasks++;
+ }
+ for(int i = 0; i < numTasks; i++) {
+ try {
+ FlowPath flowPath = tasks.take().get();
+ if(flowPath != null) {
+ flowPaths.add(flowPath);
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ log.error("Error reading FlowPath from IFlowPath object");
+ }
+ }
+ dbHandler.commit();
+ return flowPaths;
+ }
+
+ /**
+ * Query the database for all flow paths that have their source switch
+ * in the provided collection
+ *
+ * Note: this function is implemented naively and inefficiently
+ *
+ * @param dbHandler the Graph Database handler to use.
+ * @param switches a collection of switches whose flow paths you want
+ * @return the Flow Paths if found, otherwise an empty list.
+ */
+ static ArrayList<FlowPath> getFlowsForSwitches(DBOperation dbHandler, Collection<Dpid> switches) {
+ Iterable<IFlowPath> flowPathsObj = null;
+ ArrayList<FlowPath> flowPaths = new ArrayList<FlowPath>();
+
+ try {
+ flowPathsObj = dbHandler.getAllFlowPaths();
+ } catch (Exception e) {
+ // TODO: handle exceptions
+ dbHandler.rollback();
+ log.error(":getAllFlowPaths failed");
+ return flowPaths;
+ }
+ if ((flowPathsObj == null) || (flowPathsObj.iterator().hasNext() == false)) {
+ dbHandler.commit();
+ return flowPaths; // No Flows found
+ }
+
+ // convert the collection of switch dpids into a set of strings
+ Set<String> switchSet = new HashSet<>();
+ for(Dpid dpid : switches) {
+ switchSet.add(dpid.toString());
+ }
+
+ CompletionService<FlowPath> tasks = new ExecutorCompletionService<>(executor);
+ int numTasks = 0;
+ for(IFlowPath flowObj : flowPathsObj) {
+ if(switchSet.contains(flowObj.getSrcSwitch())) {
+ tasks.submit(new ExtractFlowTask(flowObj));
+ numTasks++;
+ }
+ }
+ for(int i = 0; i < numTasks; i++) {
+ try {
+ FlowPath flowPath = tasks.take().get();
+ if(flowPath != null) {
+ flowPaths.add(flowPath);
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ log.error("Error reading FlowPath from IFlowPath object");
+ }
+ }
+ dbHandler.commit();
+ return flowPaths;
+ }
+
+ /**
+ * The basic parallelization unit for extracting FlowEntries from the database.
+ *
+ * This is simply a wrapper for FlowDatabaseOperation.extractFlowPath()
+ */
+ private final static class ExtractFlowTask implements Callable<FlowPath> {
+ private final IFlowPath flowObj;
+
+ ExtractFlowTask(IFlowPath flowObj){
+ this.flowObj = flowObj;
+ }
+ @Override
+ public FlowPath call() throws Exception {
+ return extractFlowPath(flowObj);
+ }
+ }
+
+ /**
+ * Get a subset of installed flows in parallel.
+ *
+ * @param dbHandler the Graph Database handler to use.
+ * @param flowIds the collection of Flow IDs to get.
+ * @return the Flow Paths if found, otherwise an empty list.
+ */
+ static ArrayList<FlowPath> getFlows(DBOperation dbHandler,
+ Collection<FlowId> flowIds) {
+ ArrayList<FlowPath> flowPaths = new ArrayList<FlowPath>();
+
+ CompletionService<FlowPath> tasks = new ExecutorCompletionService<>(executor);
+ int numTasks = 0;
+ for (FlowId flowId : flowIds) {
+ tasks.submit(new GetFlowTask(dbHandler, flowId));
+ numTasks++;
+ }
+ for(int i = 0; i < numTasks; i++) {
+ try {
+ FlowPath flowPath = tasks.take().get();
+ if(flowPath != null) {
+ flowPaths.add(flowPath);
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ log.error("Error reading FlowPath from database");
+ }
+ }
+ // TODO: should we commit?
+ //dbHandler.commit();
+ return flowPaths;
+ }
+
+ /**
+ * The basic parallelization unit for getting FlowEntries.
+ *
+ * This is simply a wrapper for FlowDatabaseOperation.getFlow()
+ */
+ private final static class GetFlowTask implements Callable<FlowPath> {
+ private final DBOperation dbHandler;
+ private final FlowId flowId;
+
+ GetFlowTask(DBOperation dbHandler, FlowId flowId) {
+ this.dbHandler = dbHandler;
+ this.flowId = flowId;
+ }
+ @Override
+ public FlowPath call() throws Exception{
+ return getFlow(dbHandler, flowId);
+ }
+ }
+
+ /**
+ * Add a flow by creating a database task, then waiting for the result.
+ * Mostly, a wrapper for FlowDatabaseOperation.addFlow() which overs little
+ * performance benefit.
+ *
+ * @param dbHandler the Graph Database handler to use.
+ * @param flowPath the Flow Path to install.
+ * @return true on success, otherwise false.
+ */
+ static boolean addFlow(DBOperation dbHandler, FlowPath flowPath) {
+ Future<Boolean> result = executor.submit(new AddFlowTask(dbHandler, flowPath, null));
+ // NOTE: This function is blocking
+ try {
+ return result.get();
+ } catch (InterruptedException | ExecutionException e) {
+ return false;
+ }
+ }
+
+ /**
+ * Add a flow asynchronously by creating a database task.
+ *
+ * @param dbHandler the Graph Database handler to use.
+ * @param flowPath the Flow Path to install.
+ * @param datagridService the notification service for when the task is completed
+ * @return true always
+ */
+ static boolean addFlow(DBOperation dbHandler, FlowPath flowPath, IDatagridService datagridService) {
+ executor.submit(new AddFlowTask(dbHandler, flowPath, datagridService));
+ // TODO: If we need the results, submit returns a Future that contains
+ // the result.
+ return true;
+
+ }
+
+ /**
+ * The basic parallelization unit for adding FlowPaths.
+ *
+ * This is simply a wrapper for FlowDatabaseOperation.addFlow(),
+ * which also sends a notification if a datagrid services is provided
+ */
+ private final static class AddFlowTask implements Callable<Boolean> {
+ private final DBOperation dbHandler;
+ private final FlowPath flowPath;
+ private final IDatagridService datagridService;
+
+ AddFlowTask(DBOperation dbHandler,
+ FlowPath flowPath,
+ IDatagridService datagridService) {
+ this.dbHandler = dbHandler;
+ this.flowPath = flowPath;
+ this.datagridService = datagridService;
+ }
+
+ @Override
+ public Boolean call() throws Exception {
+ boolean success = FlowDatabaseOperation.addFlow(dbHandler, flowPath);
+ if(success) {
+ if(datagridService != null) {
+ // Send notifications for each Flow Entry
+ for (FlowEntry flowEntry : flowPath.flowEntries()) {
+ if (flowEntry.flowEntrySwitchState() !=
+ FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED) {
+ continue;
+ }
+ //
+ // Write the Flow Entry to the Datagrid
+ //
+ switch (flowEntry.flowEntryUserState()) {
+ case FE_USER_ADD:
+ datagridService.notificationSendFlowEntryAdded(flowEntry);
+ break;
+ case FE_USER_MODIFY:
+ datagridService.notificationSendFlowEntryUpdated(flowEntry);
+ break;
+ case FE_USER_DELETE:
+ datagridService.notificationSendFlowEntryRemoved(flowEntry.flowEntryId());
+ break;
+ case FE_USER_UNKNOWN:
+ assert(false);
+ break;
+ }
+ }
+ }
+ }
+ else {
+ log.error("Error adding flow path {} to database", flowPath);
+ }
+ return success;
+
+ }
+ }
+
+ /**
+ * Delete a previously added flow by creating a database task, then waiting
+ * for the result.
+ *
+ * Mostly, a wrapper for FlowDatabaseOperation.addFlow() which overs little
+ * performance benefit.
+ *
+ * @param dbHandler the Graph Database handler to use.
+ * @param flowId the Flow ID of the flow to delete.
+ * @return true on success, otherwise false.
+ */
+ static boolean deleteFlow(DBOperation dbHandler, FlowId flowId) {
+ Future<Boolean> result = executor.submit(new DeleteFlowTask(dbHandler, flowId, null));
+ // NOTE: This function is blocking
+ try {
+ return result.get();
+ } catch (InterruptedException | ExecutionException e) {
+ return false;
+ }
+ }
+
+ /**
+ * Delete a previously added flow asynchronously by creating a database task.
+ *
+ * @param dbHandler the Graph Database handler to use.
+ * @param flowId the Flow ID of the flow to delete.
+ * @param datagridService the notification service for when the task is completed
+ * @return true always
+ */
+ static boolean deleteFlow(DBOperation dbHandler, FlowId flowId, IDatagridService datagridService) {
+ executor.submit(new DeleteFlowTask(dbHandler, flowId, datagridService));
+ // TODO: If we need the results, submit returns a Future that contains
+ // the result.
+ return true;
+ }
+
+ /**
+ * The basic parallelization unit for deleting FlowPaths.
+ *
+ * This is simply a wrapper for FlowDatabaseOperation.deleteFlow(),
+ * which also sends a notification if a datagrid services is provided
+ */
+ private final static class DeleteFlowTask implements Callable<Boolean> {
+ private final DBOperation dbHandler;
+ private final FlowId flowId;
+ private final IDatagridService datagridService;
+
+ DeleteFlowTask(DBOperation dbHandler, FlowId flowId, IDatagridService datagridService) {
+ this.dbHandler = dbHandler;
+ this.flowId = flowId;
+ this.datagridService = datagridService;
+ }
+ @Override
+ public Boolean call() throws Exception{
+ boolean success = FlowDatabaseOperation.deleteFlow(dbHandler, flowId);
+ if(success) {
+ if(datagridService != null) {
+ datagridService.notificationSendFlowIdRemoved(flowId);
+ }
+ }
+ else {
+ log.error("Error removing flow path {} from database", flowId);
+ }
+ return success;
+ }
+ }
+}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/PerformanceMonitor.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/PerformanceMonitor.java
new file mode 100644
index 0000000..13319e7
--- /dev/null
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/PerformanceMonitor.java
@@ -0,0 +1,162 @@
+package net.onrc.onos.ofcontroller.flowmanager;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Class for collecting performance measurements
+ */
+public class PerformanceMonitor {
+ private final static Map<String, Measurement> map = new ConcurrentHashMap<String, Measurement>();;
+ private final static Logger log = LoggerFactory.getLogger(PerformanceMonitor.class);
+ private static long overhead;
+
+ /**
+ * Start a performance measurement, identified by a tag
+ *
+ * Note: Only a single measurement can use the same tag at a time.
+ *
+ * @param tag for performance measurement
+ */
+ public static void start(String tag) {
+ long start = System.nanoTime();
+ Measurement m = new Measurement();
+ if(map.put(tag, m) != null) {
+ // if there was a previous entry, we have just overwritten it
+ log.error("Tag {} already exists", tag);
+ }
+ m.start();
+ overhead += System.nanoTime() - start;
+ }
+
+ /**
+ * Stop a performance measurement.
+ *
+ * You must have already started a measurement with tag.
+ *
+ * @param tag for performance measurement
+ */
+ public static void stop(String tag) {
+ long time = System.nanoTime();
+ Measurement m = map.get(tag);
+ if(m == null) {
+ log.error("Tag {} does not exist", tag);
+ }
+ else {
+ map.get(tag).stop(time);
+ }
+ overhead += System.nanoTime() - time;
+ }
+
+ /**
+ * Find a measurement, identified by tag, and return the result
+ *
+ * @param tag for performance measurement
+ * @return the time in nanoseconds
+ */
+ public static long result(String tag) {
+ Measurement m = map.get(tag);
+ if(m != null) {
+ return m.elapsed();
+ }
+ else {
+ return -1;
+ }
+ }
+
+ /**
+ * Clear all performance measurements.
+ */
+ public static void clear() {
+ map.clear();
+ overhead = 0;
+ }
+
+ /**
+ * Write all performance measurements to the log
+ */
+ public static void report() {
+ double overheadMilli = overhead / Math.pow(10, 6);
+ log.error("Performance Results: {} with measurement overhead: {} ms", map, overheadMilli);
+ }
+
+ /**
+ * A single performance measurement
+ */
+ static class Measurement {
+ long start;
+ long stop;
+
+ /**
+ * Start the measurement
+ */
+ public void start() {
+ start = System.nanoTime();
+ }
+
+ /**
+ * Stop the measurement
+ */
+ public void stop() {
+ stop = System.nanoTime();
+ }
+
+ /**
+ * Stop the measurement at a specific time
+ * @param time to stop
+ */
+ public void stop(long time){
+ stop = time;
+ }
+
+ /**
+ * Compute the elapsed time of the measurement in nanoseconds
+ *
+ * @return the measurement time in nanoseconds, or -1 if the measurement is stil running.
+ */
+ public long elapsed() {
+ if(stop == 0) {
+ return -1;
+ }
+ else {
+ return stop - start;
+ }
+ }
+
+ /**
+ * Returns the number of milliseconds for the measurement as a String.
+ */
+ public String toString() {
+ double milli = elapsed() / Math.pow(10, 6);
+ return Double.toString(milli) + "ms";
+ }
+ }
+
+ public static void main(String args[]){
+ // test the measurement overhead
+ String tag;
+ for(int i = 0; i < 100; i++){
+ tag = "foo foo foo";
+ start(tag); stop(tag);
+ tag = "bar";
+ start(tag); stop(tag);
+ tag = "baz";
+ start(tag); stop(tag);
+ report();
+ clear();
+ }
+ for(int i = 0; i < 100; i++){
+ tag = "a";
+ start(tag); stop(tag);
+ tag = "b";
+ start(tag); stop(tag);
+ tag = "c";
+ start(tag); stop(tag);
+ report();
+ clear();
+ }
+ }
+}