Merge branch 'ONOS-ONRC2014-Measurements' of https://github.com/OPENNETWORKINGLAB/ONOS into topodev
diff --git a/src/main/java/net/onrc/onos/datagrid/HazelcastDatagrid.java b/src/main/java/net/onrc/onos/datagrid/HazelcastDatagrid.java
index effbe81..632cc38 100644
--- a/src/main/java/net/onrc/onos/datagrid/HazelcastDatagrid.java
+++ b/src/main/java/net/onrc/onos/datagrid/HazelcastDatagrid.java
@@ -21,10 +21,12 @@
import net.onrc.onos.ofcontroller.proxyarp.ArpMessage;
import net.onrc.onos.ofcontroller.proxyarp.IArpEventHandler;
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;
+import net.onrc.onos.ofcontroller.util.Pair;
import net.onrc.onos.ofcontroller.util.serializers.KryoFactory;
import org.slf4j.Logger;
@@ -79,6 +81,12 @@
private MapFlowIdListener mapFlowIdListener = null;
private String mapFlowIdListenerId = null;
+ // State related to the Flow Entry ID map
+ protected static final String mapFlowEntryIdName = "mapFlowEntryId";
+ private IMap<Long, byte[]> mapFlowEntryId = null;
+ private MapFlowEntryIdListener mapFlowEntryIdListener = null;
+ private String mapFlowEntryIdListenerId = null;
+
// State related to the Network Topology map
protected static final String mapTopologyName = "mapTopology";
private IMap<String, byte[]> mapTopology = null;
@@ -257,7 +265,7 @@
* @param event the notification event for the entry.
*/
public void entryAdded(EntryEvent<Long, byte[]> event) {
- byte[] valueBytes = (byte[])event.getValue();
+ byte[] valueBytes = event.getValue();
//
// Decode the value and deliver the notification
@@ -275,7 +283,7 @@
* @param event the notification event for the entry.
*/
public void entryRemoved(EntryEvent<Long, byte[]> event) {
- byte[] valueBytes = (byte[])event.getValue();
+ byte[] valueBytes = event.getValue();
//
// Decode the value and deliver the notification
@@ -293,7 +301,7 @@
* @param event the notification event for the entry.
*/
public void entryUpdated(EntryEvent<Long, byte[]> event) {
- byte[] valueBytes = (byte[])event.getValue();
+ byte[] valueBytes = event.getValue();
//
// Decode the value and deliver the notification
@@ -316,6 +324,87 @@
}
/**
+ * Class for receiving notifications for FlowEntryId state.
+ *
+ * The datagrid map is:
+ * - Key : FlowEntryId (Long)
+ * - Value : Serialized Switch Dpid (byte[])
+ */
+ class MapFlowEntryIdListener implements EntryListener<Long, byte[]> {
+ /**
+ * Receive a notification that an entry is added.
+ *
+ * @param event the notification event for the entry.
+ */
+ public void entryAdded(EntryEvent<Long, byte[]> event) {
+ Long keyLong = event.getKey();
+ FlowEntryId flowEntryId = new FlowEntryId(keyLong);
+
+ byte[] valueBytes = event.getValue();
+
+ //
+ // Decode the value and deliver the notification
+ //
+ Kryo kryo = kryoFactory.newKryo();
+ Input input = new Input(valueBytes);
+ Dpid dpid = kryo.readObject(input, Dpid.class);
+ kryoFactory.deleteKryo(kryo);
+ flowEventHandlerService.notificationRecvFlowEntryIdAdded(flowEntryId, dpid);
+ }
+
+ /**
+ * Receive a notification that an entry is removed.
+ *
+ * @param event the notification event for the entry.
+ */
+ public void entryRemoved(EntryEvent<Long, byte[]> event) {
+ Long keyLong = event.getKey();
+ FlowEntryId flowEntryId = new FlowEntryId(keyLong);
+
+ byte[] valueBytes = event.getValue();
+
+ //
+ // Decode the value and deliver the notification
+ //
+ Kryo kryo = kryoFactory.newKryo();
+ Input input = new Input(valueBytes);
+ Dpid dpid = kryo.readObject(input, Dpid.class);
+ kryoFactory.deleteKryo(kryo);
+ flowEventHandlerService.notificationRecvFlowEntryIdRemoved(flowEntryId, dpid);
+ }
+
+ /**
+ * Receive a notification that an entry is updated.
+ *
+ * @param event the notification event for the entry.
+ */
+ public void entryUpdated(EntryEvent<Long, byte[]> event) {
+ Long keyLong = event.getKey();
+ FlowEntryId flowEntryId = new FlowEntryId(keyLong);
+
+ byte[] valueBytes = event.getValue();
+
+ //
+ // Decode the value and deliver the notification
+ //
+ Kryo kryo = kryoFactory.newKryo();
+ Input input = new Input(valueBytes);
+ Dpid dpid = kryo.readObject(input, Dpid.class);
+ kryoFactory.deleteKryo(kryo);
+ flowEventHandlerService.notificationRecvFlowEntryIdUpdated(flowEntryId, dpid);
+ }
+
+ /**
+ * Receive a notification that an entry is evicted.
+ *
+ * @param event the notification event for the entry.
+ */
+ public void entryEvicted(EntryEvent<Long, byte[]> event) {
+ // NOTE: We don't use eviction for this map
+ }
+ }
+
+ /**
* Class for receiving notifications for Network Topology state.
*
* The datagrid map is:
@@ -604,6 +693,11 @@
mapFlowId = hazelcastInstance.getMap(mapFlowIdName);
mapFlowIdListenerId = mapFlowId.addEntryListener(mapFlowIdListener, true);
+ // Initialize the FlowEntryId-related map state
+ mapFlowEntryIdListener = new MapFlowEntryIdListener();
+ mapFlowEntryId = hazelcastInstance.getMap(mapFlowEntryIdName);
+ mapFlowEntryIdListenerId = mapFlowEntryId.addEntryListener(mapFlowEntryIdListener, true);
+
// Initialize the Topology-related map state
mapTopologyListener = new MapTopologyListener();
mapTopology = hazelcastInstance.getMap(mapTopologyName);
@@ -636,6 +730,11 @@
mapFlowId = null;
mapFlowIdListener = null;
+ // Clear the FlowEntryId-related map state
+ mapFlowEntryId.removeEntryListener(mapFlowEntryIdListenerId);
+ mapFlowEntryId = null;
+ mapFlowEntryIdListener = null;
+
// Clear the Topology-related map state
mapTopology.removeEntryListener(mapTopologyListenerId);
mapTopology = null;
@@ -920,6 +1019,40 @@
}
/**
+ * Get all Flow Entry IDs that are currently in the datagrid.
+ *
+ * @return all Flow Entry IDs that ae currently in the datagrid.
+ */
+ @Override
+ public Collection<Pair<FlowEntryId, Dpid>> getAllFlowEntryIds() {
+ Collection<Pair<FlowEntryId, Dpid>> allFlowEntryIds =
+ new LinkedList<Pair<FlowEntryId, Dpid>>();
+
+ //
+ // Get all current entries
+ //
+ Kryo kryo = kryoFactory.newKryo();
+ for (Map.Entry<Long, byte[]> entry : mapFlowEntryId.entrySet()) {
+ Long key = entry.getKey();
+ byte[] valueBytes = entry.getValue();
+
+ FlowEntryId flowEntryId = new FlowEntryId(key);
+
+ //
+ // Decode the value
+ //
+ Input input = new Input(valueBytes);
+ Dpid dpid = kryo.readObject(input, Dpid.class);
+
+ Pair<FlowEntryId, Dpid> pair = new Pair(flowEntryId, dpid);
+ allFlowEntryIds.add(pair);
+ }
+ kryoFactory.deleteKryo(kryo);
+
+ return allFlowEntryIds;
+ }
+
+ /**
* Send a notification that a FlowId is added.
*
* @param flowId the FlowId that is added.
@@ -988,6 +1121,78 @@
}
/**
+ * Send a notification that a FlowEntryId is added.
+ *
+ * @param flowEntryId the FlowEntryId that is added.
+ * @param dpid the Switch Dpid.
+ */
+ @Override
+ public void notificationSendFlowEntryIdAdded(FlowEntryId flowEntryId,
+ Dpid dpid) {
+ //
+ // Encode the value
+ //
+ byte[] buffer = new byte[MAX_BUFFER_SIZE];
+ Kryo kryo = kryoFactory.newKryo();
+ Output output = new Output(buffer, -1);
+ kryo.writeObject(output, dpid);
+ byte[] valueBytes = output.toBytes();
+ kryoFactory.deleteKryo(kryo);
+
+ //
+ // Put the entry:
+ // - Key : FlowEntryId (Long)
+ // - Value : Serialized Switch Dpid (byte[])
+ //
+ mapFlowEntryId.putAsync(flowEntryId.value(), valueBytes);
+ }
+
+ /**
+ * Send a notification that a FlowEntryId is removed.
+ *
+ * @param flowEntryId the FlowEntryId that is removed.
+ */
+ @Override
+ public void notificationSendFlowEntryIdRemoved(FlowEntryId flowEntryId) {
+ //
+ // Remove the entry:
+ // - Key : FlowEntryId (Long)
+ // - Value : Serialized Dpid (byte[])
+ //
+ mapFlowEntryId.removeAsync(flowEntryId.value());
+ }
+
+ /**
+ * Send a notification that a FlowEntryId is updated.
+ *
+ * @param flowEntryId the FlowEntryId that is updated.
+ * @param dpid the Switch Dpid.
+ */
+ @Override
+ public void notificationSendFlowEntryIdUpdated(FlowEntryId flowEntryId,
+ Dpid dpid) {
+ // NOTE: Adding an entry with an existing key automatically updates it
+ notificationSendFlowEntryIdAdded(flowEntryId, dpid);
+ }
+
+ /**
+ * Send a notification that all Flow Entry IDs are removed.
+ */
+ @Override
+ public void notificationSendAllFlowEntryIdsRemoved() {
+ //
+ // Remove all entries
+ // NOTE: We remove the entries one-by-one so the per-entry
+ // notifications will be delivered.
+ //
+ // mapFlowEntryId.clear();
+ Set<Long> keySet = mapFlowEntryId.keySet();
+ for (Long key : keySet) {
+ mapFlowEntryId.removeAsync(key);
+ }
+ }
+
+ /**
* Get all Topology Elements that are currently in the datagrid.
*
* @return all Topology Elements that are currently in the datagrid.
diff --git a/src/main/java/net/onrc/onos/datagrid/IDatagridService.java b/src/main/java/net/onrc/onos/datagrid/IDatagridService.java
index d4e7b00..cfc10bc 100644
--- a/src/main/java/net/onrc/onos/datagrid/IDatagridService.java
+++ b/src/main/java/net/onrc/onos/datagrid/IDatagridService.java
@@ -7,10 +7,12 @@
import net.onrc.onos.ofcontroller.proxyarp.ArpMessage;
import net.onrc.onos.ofcontroller.proxyarp.IArpEventHandler;
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;
+import net.onrc.onos.ofcontroller.util.Pair;
/**
* Interface for providing Datagrid Service to other modules.
@@ -136,7 +138,7 @@
/**
* Get all Flow IDs that are currently in the datagrid.
*
- * @return all Flow IDs that are currently in the datagrid.
+ * @return all Flow IDs that ae currently in the datagrid.
*/
Collection<FlowId> getAllFlowIds();
@@ -167,6 +169,42 @@
void notificationSendAllFlowIdsRemoved();
/**
+ * Get all Flow Entry IDs that are currently in the datagrid.
+ *
+ * @return all Flow Entry IDs that ae currently in the datagrid.
+ */
+ Collection<Pair<FlowEntryId, Dpid>> getAllFlowEntryIds();
+
+ /**
+ * Send a notification that a FlowEntryId is added.
+ *
+ * @param flowEntryId the FlowEntryId that is added.
+ * @param dpid the Switch Dpid.
+ */
+ void notificationSendFlowEntryIdAdded(FlowEntryId flowEntryId, Dpid dpid);
+
+ /**
+ * Send a notification that a FlowEntryId is removed.
+ *
+ * @param flowEntryId the FlowEntryId that is removed.
+ */
+ void notificationSendFlowEntryIdRemoved(FlowEntryId flowEntryId);
+
+ /**
+ * Send a notification that a FlowEntryId is updated.
+ *
+ * @param flowEntryId the FlowEntryId that is updated.
+ * @param dpid the Switch Dpid.
+ */
+ void notificationSendFlowEntryIdUpdated(FlowEntryId flowEntryId,
+ Dpid dpid);
+
+ /**
+ * Send a notification that all Flow Entry IDs are removed.
+ */
+ void notificationSendAllFlowEntryIdsRemoved();
+
+ /**
* Get all Topology Elements that are currently in the datagrid.
*
* @return all Topology Elements that are currently in the datagrid.
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 e075bad..3cb026e 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java
@@ -5,7 +5,9 @@
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
+import net.floodlightcontroller.core.IOFSwitch;
import net.floodlightcontroller.util.MACAddress;
import net.onrc.onos.graph.GraphDBOperation;
@@ -23,6 +25,8 @@
* Class for performing Flow-related operations on the Database.
*/
public class FlowDatabaseOperation {
+ static private boolean enableOnrc2014MeasurementsFlows = true;
+
private final static Logger log = LoggerFactory.getLogger(FlowDatabaseOperation.class);
/**
@@ -167,8 +171,10 @@
// flowPath.dataPath().flowEntries()
//
for (FlowEntry flowEntry : flowPath.dataPath().flowEntries()) {
- if (flowEntry.flowEntryUserState() == FlowEntryUserState.FE_USER_DELETE)
- continue; // Skip: all Flow Entries were deleted earlier
+ if (! enableOnrc2014MeasurementsFlows) {
+ if (flowEntry.flowEntryUserState() == FlowEntryUserState.FE_USER_DELETE)
+ continue; // Skip: all Flow Entries were deleted earlier
+ }
if (addFlowEntry(dbHandler, flowObj, flowEntry) == null) {
dbHandler.rollback();
@@ -308,10 +314,14 @@
}
// TODO: Hacks with hard-coded state names!
- if (found)
- flowEntryObj.setUserState("FE_USER_MODIFY");
- else
- flowEntryObj.setUserState("FE_USER_ADD");
+ if (enableOnrc2014MeasurementsFlows) {
+ flowEntryObj.setUserState(flowEntry.flowEntryUserState().toString());
+ } else {
+ if (found)
+ flowEntryObj.setUserState("FE_USER_MODIFY");
+ else
+ flowEntryObj.setUserState("FE_USER_ADD");
+ }
flowEntryObj.setSwitchState(flowEntry.flowEntrySwitchState().toString());
//
// TODO: Take care of the FlowEntryErrorState.
@@ -464,6 +474,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(GraphDBOperation 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(GraphDBOperation 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.
@@ -501,6 +582,60 @@
}
/**
+ * 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(GraphDBOperation 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;
+ }
+
+ /**
* Extract Flow Path State from a Titan Database Object @ref IFlowPath.
*
* @param flowObj the object to extract the Flow Path State from.
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 4c801d6..83f9731 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
@@ -19,6 +19,7 @@
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;
@@ -30,6 +31,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;
@@ -45,12 +48,13 @@
* - Recompute impacted FlowPath using cached Topology.
*/
class FlowEventHandler extends Thread implements IFlowEventHandlerService {
+
+ private boolean enableOnrc2014MeasurementsFlows = true;
+ private boolean enableOnrc2014MeasurementsTopology = true;
+
/** The logger. */
private final static Logger log = LoggerFactory.getLogger(FlowEventHandler.class);
- // Flag to enable feature of acquiring topology information from DB instead of datagrid.
- private final boolean accessDBFlag = false;
-
private GraphDBOperation dbHandler;
private FlowManager flowManager; // The Flow Manager to use
private IDatagridService datagridService; // The Datagrid Service to use
@@ -70,6 +74,8 @@
new LinkedList<EventEntry<FlowEntry>>();
private List<EventEntry<FlowId>> flowIdEvents =
new LinkedList<EventEntry<FlowId>>();
+ private List<EventEntry<Pair<FlowEntryId, Dpid>>> flowEntryIdEvents =
+ new LinkedList<EventEntry<Pair<FlowEntryId, Dpid>>>();
// All internally computed Flow Paths
private Map<Long, FlowPath> allFlowPaths = new HashMap<Long, FlowPath>();
@@ -98,12 +104,10 @@
* @param datagridService the Datagrid Service to use.
*/
FlowEventHandler(FlowManager flowManager,
- IDatagridService datagridService,
- GraphDBOperation dbHandler) {
+ IDatagridService datagridService) {
this.flowManager = flowManager;
this.datagridService = datagridService;
this.topology = new Topology();
- this.dbHandler = dbHandler;
}
/**
@@ -117,6 +121,8 @@
* Startup processing.
*/
private void startup() {
+ this.dbHandler = new GraphDBOperation("");
+
//
// Obtain the initial Topology state
//
@@ -156,6 +162,17 @@
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();
@@ -186,6 +203,7 @@
// - EventEntry<FlowPath>
// - EventEntry<FlowEntry>
// - EventEntry<FlowId>
+ // - EventEntry<Pair<FlowEntryId, Dpid>>
//
for (EventEntry<?> event : collection) {
// Topology event
@@ -220,6 +238,13 @@
flowIdEvents.add(flowIdEventEntry);
continue;
}
+ // FlowEntryId event
+ if (event.eventData() instanceof Pair) {
+ EventEntry<Pair<FlowEntryId, Dpid>> flowEntryIdEventEntry =
+ (EventEntry<Pair<FlowEntryId, Dpid>>)event;
+ flowEntryIdEvents.add(flowEntryIdEventEntry);
+ continue;
+ }
}
collection.clear();
@@ -239,12 +264,66 @@
private void processEvents() {
Collection<FlowEntry> modifiedFlowEntries;
+ if (enableOnrc2014MeasurementsFlows) {
+
+ if (topologyEvents.isEmpty() && flowIdEvents.isEmpty() &&
+ flowEntryIdEvents.isEmpty()) {
+ return; // Nothing to do
+ }
+
+ Map<Long, IOFSwitch> mySwitches = flowManager.getMySwitches();
+
+ // Fetch and prepare my flows
+ prepareMyFlows(mySwitches);
+
+ // 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));
+ }
+ }
+ }
+
+ // Extract my modified Flow Entries
+ modifiedFlowEntries = processFlowEntryIdEvents(mySwitches);
+
+ //
+ // Push the modified state to the Flow Manager
+ //
+ flowManager.pushModifiedFlowState(modifiedFlowPaths.values(),
+ modifiedFlowEntries);
+
+ // Cleanup
+ topologyEvents.clear();
+ flowIdEvents.clear();
+ flowEntryIdEvents.clear();
+ //
+ allFlowPaths.clear();
+ shouldRecomputeFlowPaths.clear();
+ modifiedFlowPaths.clear();
+
+ return;
+ }
+
if (topologyEvents.isEmpty() && flowPathEvents.isEmpty() &&
flowEntryEvents.isEmpty()) {
return; // Nothing to do
}
- processFlowIdEvents();
processFlowPathEvents();
processTopologyEvents();
processUnmatchedFlowEntryAdd();
@@ -284,7 +363,6 @@
topologyEvents.clear();
flowPathEvents.clear();
flowEntryEvents.clear();
- flowIdEvents.clear();
//
shouldRecomputeFlowPaths.clear();
modifiedFlowPaths.clear();
@@ -377,39 +455,123 @@
}
/**
- * Process the Flow ID events.
+ * Prepare my flows.
+ *
+ * @param mySwitches the collection of my switches.
*/
- private void processFlowIdEvents() {
+ private void prepareMyFlows(Map<Long, IOFSwitch> mySwitches) {
+ if (! topologyEvents.isEmpty()) {
+ // Fetch my flows from the database
+ ArrayList<FlowPath> myFlows = FlowDatabaseOperation.getAllMyFlows(dbHandler, mySwitches);
+ for (FlowPath flowPath : myFlows) {
+ log.debug("Found my flow: {}", flowPath);
+
+ allFlowPaths.put(flowPath.flowId().value(), 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 Flow ID events.
+ *
+ * @param mySwitches the collection of my switches.
+ */
+ private void processFlowIdEvents(Map<Long, IOFSwitch> mySwitches) {
//
- // Process all Flow ID events and update the appropriate state
+ // Automatically add all Flow ID events (for the Flows this instance
+ // is responsible for) to the collection of Flows to recompute.
//
for (EventEntry<FlowId> eventEntry : flowIdEvents) {
FlowId flowId = eventEntry.eventData();
log.debug("Flow ID Event: {} {}", eventEntry.eventType(), flowId);
- switch (eventEntry.eventType()) {
- case ENTRY_ADD: {
- //
- // Add a new Flow ID
- //
- // TODO: Implement it!
-
- break;
+ FlowPath flowPath = allFlowPaths.get(flowId.value());
+ if (flowPath == null) {
+ if (! topologyEvents.isEmpty())
+ continue; // Optimization: Not my flow
+ Dpid dpid = FlowDatabaseOperation.getFlowSourceDpid(dbHandler,
+ flowId);
+ if ((dpid != null) && (mySwitches.get(dpid.value()) != null)) {
+ flowPath = FlowDatabaseOperation.getFlow(dbHandler,
+ flowId);
+ }
}
-
- case ENTRY_REMOVE: {
- //
- // Remove an existing Flow ID.
- //
- // TODO: Implement it!
-
- break;
- }
+ if (flowPath != null) {
+ shouldRecomputeFlowPaths.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.
@@ -497,32 +659,70 @@
* 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());
+
+ //
+ // TODO: Fake the unconditional topology read by checking the cache
+ // with the old topology and ignoring topology events that don't
+ // make any impact to the topology.
+ // This is needed aa workaround: if a port is down, we get
+ // up to three additional "Port Down" or "Link Down" events.
+ //
+ 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)
+ break;
+ }
+ if (! isTopologyModified) {
+ log.debug("Ignoring topology events that don't modify the topology");
+ return;
+ }
+
+ topology.readFromDatabase(dbHandler);
+ log.debug("[AFTER] {}", topology.toString());
+ shouldRecomputeFlowPaths.putAll(allFlowPaths);
+ return;
+ }
+
//
// Process all Topology events and update the appropriate state
//
- boolean isTopologyModified = false;
- if (accessDBFlag) {
- log.debug("[BEFORE] {}", topology.toString());
- if (! topology.readFromDatabase(dbHandler)) {
- isTopologyModified = true;
- }
- log.debug("[AFTER] {}", topology.toString());
- } else {
- for (EventEntry<TopologyElement> eventEntry : topologyEvents) {
- TopologyElement topologyElement = eventEntry.eventData();
+ for (EventEntry<TopologyElement> eventEntry : topologyEvents) {
+ TopologyElement topologyElement = eventEntry.eventData();
- log.debug("Topology Event: {} {}", eventEntry.eventType(),
- topologyElement.toString());
+ 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;
- }
- }
+ 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
@@ -789,6 +989,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
//
@@ -1083,6 +1296,55 @@
}
/**
+ * 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.
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 1c9db0c..4230e43 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
@@ -48,6 +48,9 @@
* Flow Manager class for handling the network flows.
*/
public class FlowManager implements IFloodlightModule, IFlowService, INetMapStorage {
+
+ private boolean enableOnrc2014MeasurementsFlows = true;
+
protected GraphDBOperation dbHandlerApi;
protected GraphDBOperation dbHandlerInner;
@@ -224,7 +227,7 @@
// - register with the Datagrid Service
// - startup
//
- flowEventHandler = new FlowEventHandler(this, datagridService, dbHandlerInner);
+ flowEventHandler = new FlowEventHandler(this, datagridService);
datagridService.registerFlowEventHandlerService(flowEventHandler);
flowEventHandler.start();
}
@@ -265,7 +268,12 @@
}
if (FlowDatabaseOperation.addFlow(dbHandlerApi, flowPath)) {
- datagridService.notificationSendFlowAdded(flowPath);
+ if (enableOnrc2014MeasurementsFlows) {
+ datagridService.notificationSendFlowIdAdded(flowPath.flowId());
+ } else {
+ datagridService.notificationSendFlowAdded(flowPath);
+ }
+
return flowPath.flowId();
}
return null;
@@ -279,7 +287,11 @@
@Override
public boolean deleteAllFlows() {
if (FlowDatabaseOperation.deleteAllFlows(dbHandlerApi)) {
- datagridService.notificationSendAllFlowsRemoved();
+ if (enableOnrc2014MeasurementsFlows) {
+ datagridService.notificationSendAllFlowIdsRemoved();
+ } else {
+ datagridService.notificationSendAllFlowsRemoved();
+ }
return true;
}
return false;
@@ -294,7 +306,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;
@@ -312,6 +328,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.
@@ -322,6 +358,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.
@@ -413,6 +461,9 @@
public void flowEntriesPushedToSwitch(
Collection<Pair<IOFSwitch, FlowEntry>> entries) {
+ if (enableOnrc2014MeasurementsFlows)
+ return;
+
//
// Process all entries
//
@@ -482,8 +533,12 @@
// - Flow Paths to the database
//
pushModifiedFlowEntriesToSwitches(modifiedFlowEntries);
- pushModifiedFlowPathsToDatabase(modifiedFlowPaths);
- cleanupDeletedFlowEntriesFromDatagrid(modifiedFlowEntries);
+ if (enableOnrc2014MeasurementsFlows) {
+ writeModifiedFlowPathsToDatabase(modifiedFlowPaths);
+ } else {
+ pushModifiedFlowPathsToDatabase(modifiedFlowPaths);
+ cleanupDeletedFlowEntriesFromDatagrid(modifiedFlowEntries);
+ }
}
/**
@@ -714,6 +769,11 @@
}
} while (retry);
+ if (enableOnrc2014MeasurementsFlows) {
+ // Send the notification
+ datagridService.notificationSendFlowIdRemoved(flowPath.flowId());
+ }
+
continue;
}
@@ -730,10 +790,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)
@@ -759,6 +821,16 @@
log.error("Exception writing Flow Path to Network MAP: ", e);
}
} while (retry);
+
+ if (enableOnrc2014MeasurementsFlows) {
+ // Send the notifications
+ for (FlowEntry flowEntry : flowPath.flowEntries()) {
+ if (flowEntry.flowEntrySwitchState() ==
+ FlowEntrySwitchState.FE_SWITCH_NOT_UPDATED) {
+ datagridService.notificationSendFlowEntryIdAdded(flowEntry.flowEntryId(), flowEntry.dpid());
+ }
+ }
+ }
}
}
}
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 62edf70..04f92bc 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,9 @@
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;
@@ -73,6 +75,32 @@
void notificationRecvFlowIdUpdated(FlowId flowId);
/**
+ * 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/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();
+ }
+ }
+}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/linkdiscovery/internal/LinkDiscoveryManager.java b/src/main/java/net/onrc/onos/ofcontroller/linkdiscovery/internal/LinkDiscoveryManager.java
index 1dbfdcb..3860e05 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/linkdiscovery/internal/LinkDiscoveryManager.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/linkdiscovery/internal/LinkDiscoveryManager.java
@@ -112,7 +112,7 @@
*/
@LogMessageCategory("Network Topology")
public class LinkDiscoveryManager
-implements IOFMessageListener, IOFSwitchListener,
+implements IOFMessageListener, IOFSwitchListener,
ILinkDiscoveryService, IFloodlightModule {
protected IFloodlightProviderService controller;
protected final static Logger log = LoggerFactory.getLogger(LinkDiscoveryManager.class);
@@ -125,7 +125,7 @@
// LLDP and BDDP fields
- private static final byte[] LLDP_STANDARD_DST_MAC_STRING =
+ private static final byte[] LLDP_STANDARD_DST_MAC_STRING =
HexString.fromHexString("01:80:c2:00:00:0e");
private static final long LINK_LOCAL_MASK = 0xfffffffffff0L;
private static final long LINK_LOCAL_VALUE = 0x0180c2000000L;
@@ -135,27 +135,27 @@
private static final String LLDP_BSN_DST_MAC_STRING = "ff:ff:ff:ff:ff:ff";
- // Direction TLVs are used to indicate if the LLDPs were sent
+ // Direction TLVs are used to indicate if the LLDPs were sent
// periodically or in response to a recieved LLDP
private static final byte TLV_DIRECTION_TYPE = 0x73;
private static final short TLV_DIRECTION_LENGTH = 1; // 1 byte
private static final byte TLV_DIRECTION_VALUE_FORWARD[] = {0x01};
private static final byte TLV_DIRECTION_VALUE_REVERSE[] = {0x02};
- private static final LLDPTLV forwardTLV
+ private static final LLDPTLV forwardTLV
= new LLDPTLV().
- setType((byte)TLV_DIRECTION_TYPE).
- setLength((short)TLV_DIRECTION_LENGTH).
+ setType(TLV_DIRECTION_TYPE).
+ setLength(TLV_DIRECTION_LENGTH).
setValue(TLV_DIRECTION_VALUE_FORWARD);
- private static final LLDPTLV reverseTLV
+ private static final LLDPTLV reverseTLV
= new LLDPTLV().
- setType((byte)TLV_DIRECTION_TYPE).
- setLength((short)TLV_DIRECTION_LENGTH).
+ setType(TLV_DIRECTION_TYPE).
+ setLength(TLV_DIRECTION_LENGTH).
setValue(TLV_DIRECTION_VALUE_REVERSE);
// Link discovery task details.
protected SingletonTask discoveryTask;
- protected final int DISCOVERY_TASK_INTERVAL = 1;
+ protected final int DISCOVERY_TASK_INTERVAL = 1;
protected final int LINK_TIMEOUT = 35; // original 35 secs, aggressive 5 secs
protected final int LLDP_TO_ALL_INTERVAL = 15 ; //original 15 seconds, aggressive 2 secs.
protected long lldpClock = 0;
@@ -206,7 +206,7 @@
/* topology aware components are called in the order they were added to the
* the array */
protected ArrayList<ILinkDiscoveryListener> linkDiscoveryAware;
-
+
protected class LinkUpdate extends LDUpdate {
public LinkUpdate(LDUpdate old) {
@@ -263,7 +263,7 @@
*/
protected Map<NodePortTuple, Long> broadcastDomainPortTimeMap;
- /**
+ /**
* Get the LLDP sending period in seconds.
* @return LLDP sending period in seconds.
*/
@@ -283,6 +283,7 @@
return portLinks;
}
+ @Override
public Set<NodePortTuple> getSuppressLLDPsInfo() {
return suppressLinkDiscovery;
}
@@ -291,6 +292,7 @@
* Add a switch port to the suppressed LLDP list.
* Remove any known links on the switch port.
*/
+ @Override
public void AddToSuppressLLDPs(long sw, short port)
{
NodePortTuple npt = new NodePortTuple(sw, port);
@@ -302,7 +304,8 @@
* Remove a switch port from the suppressed LLDP list.
* Discover links on that switchport.
*/
- public void RemoveFromSuppressLLDPs(long sw, short port)
+ @Override
+ public void RemoveFromSuppressLLDPs(long sw, short port)
{
NodePortTuple npt = new NodePortTuple(sw, port);
this.suppressLinkDiscovery.remove(npt);
@@ -317,6 +320,7 @@
return false;
}
+ @Override
public ILinkDiscovery.LinkType getLinkType(Link lt, LinkInfo info) {
if (info.getUnicastValidTime() != null) {
return ILinkDiscovery.LinkType.DIRECT_LINK;
@@ -326,7 +330,7 @@
return ILinkDiscovery.LinkType.INVALID_LINK;
}
-
+
private boolean isLinkDiscoverySuppressed(long sw, short portNumber) {
return this.suppressLinkDiscovery.contains(new NodePortTuple(sw, portNumber));
}
@@ -437,6 +441,7 @@
}
}
+ @Override
public Set<Short> getQuarantinedPorts(long sw) {
Set<Short> qPorts = new HashSet<Short>();
@@ -468,12 +473,12 @@
else operation = UpdateOperation.PORT_DOWN;
LinkUpdate update = new LinkUpdate(new LDUpdate(sw, port, operation));
-
-
+
+
controller.publishUpdate(update);
}
- /**
+ /**
* Send LLDP on known ports
*/
protected void discoverOnKnownLinkPorts() {
@@ -500,7 +505,7 @@
*/
protected IOFSwitch addRemoteSwitch(long sw, short port) {
IOnosRemoteSwitch remotesw = null;
-
+
// add a switch if we have not seen it before
remotesw = remoteSwitches.get(sw);
@@ -510,26 +515,26 @@
remoteSwitches.put(remotesw.getId(), remotesw);
log.debug("addRemoteSwitch(): added fake remote sw {}", remotesw);
}
-
+
// add the port if we have not seen it before
if (remotesw.getPort(port) == null) {
OFPhysicalPort remoteport = new OFPhysicalPort();
remoteport.setPortNumber(port);
remoteport.setName("fake_" + port);
- remoteport.setConfig(0);
+ remoteport.setConfig(0);
remoteport.setState(0);
remotesw.setPort(remoteport);
log.debug("addRemoteSwitch(): added fake remote port {} to sw {}", remoteport, remotesw.getId());
}
-
+
return remotesw;
}
-
+
/**
* Send link discovery message out of a given switch port.
* The discovery message may be a standard LLDP or a modified
- * LLDP, where the dst mac address is set to :ff.
- *
+ * LLDP, where the dst mac address is set to :ff.
+ *
* TODO: The modified LLDP will updated in the future and may
* use a different eth-type.
* @param sw
@@ -565,7 +570,7 @@
if (isLinkDiscoverySuppressed(sw, port)) {
/* Dont send LLDPs out of this port as suppressLLDPs set
- *
+ *
*/
return;
}
@@ -881,9 +886,9 @@
addOrUpdateLink(lt, newLinkInfo);
- // Check if reverse link exists.
- // If it doesn't exist and if the forward link was seen
- // first seen within a small interval, send probe on the
+ // Check if reverse link exists.
+ // If it doesn't exist and if the forward link was seen
+ // first seen within a small interval, send probe on the
// reverse link.
newLinkInfo = links.get(lt);
@@ -927,8 +932,8 @@
protected Command handlePacketIn(long sw, OFPacketIn pi,
FloodlightContext cntx) {
- Ethernet eth =
- IFloodlightProviderService.bcStore.get(cntx,
+ Ethernet eth =
+ IFloodlightProviderService.bcStore.get(cntx,
IFloodlightProviderService.CONTEXT_PI_PAYLOAD);
if(eth.getEtherType() == Ethernet.TYPE_BSN) {
@@ -999,8 +1004,8 @@
newInfo.setFirstSeenTime(oldInfo.getFirstSeenTime());
if (log.isTraceEnabled()) {
- log.trace("addOrUpdateLink: {} {}",
- lt,
+ log.trace("addOrUpdateLink: {} {}",
+ lt,
(newInfo.getMulticastValidTime()!=null) ? "multicast" : "unicast");
}
@@ -1033,7 +1038,7 @@
// Add to portNOFLinks if the unicast valid time is null
if (newInfo.getUnicastValidTime() == null)
addLinkToBroadcastDomain(lt);
-
+
// ONOS: Distinguish added event separately from updated event
updateOperation = UpdateOperation.LINK_ADDED;
linkChanged = true;
@@ -1119,6 +1124,7 @@
return linkChanged;
}
+ @Override
public Map<Long, Set<Link>> getSwitchLinks() {
return this.switchLinks;
}
@@ -1198,7 +1204,7 @@
// ONOS: If we do not control this switch, then we should not process its port status messages
if (!registryService.hasControl(iofSwitch.getId())) return Command.CONTINUE;
-
+
if (log.isTraceEnabled()) {
log.trace("handlePortStatus: Switch {} port #{} reason {}; " +
"config is {} state is {}",
@@ -1225,7 +1231,7 @@
LinkUpdate update = new LinkUpdate(new LDUpdate(sw, port, UpdateOperation.PORT_DOWN));
controller.publishUpdate(update);
linkDeleted = true;
- }
+ }
else if (ps.getReason() ==
(byte)OFPortReason.OFPPR_MODIFY.ordinal()) {
// If ps is a port modification and the port state has changed
@@ -1237,7 +1243,7 @@
assert(linkInfo != null);
Integer updatedSrcPortState = null;
Integer updatedDstPortState = null;
- if (lt.getSrc() == npt.getNodeId() &&
+ if (lt.getSrc() == npt.getNodeId() &&
lt.getSrcPort() == npt.getPortId() &&
(linkInfo.getSrcPortState() !=
ps.getDesc().getState())) {
@@ -1264,7 +1270,7 @@
getLinkType(lt, linkInfo),
operation));
controller.publishUpdate(update);
-
+
linkInfoChanged = true;
}
}
@@ -1378,9 +1384,9 @@
lock.writeLock().unlock();
}
}
-
+
/**
- * We don't react the port changed notifications here. we listen for
+ * We don't react the port changed notifications here. we listen for
* OFPortStatus messages directly. Might consider using this notifier
* instead
*/
@@ -1389,7 +1395,7 @@
// no-op
}
- /**
+ /**
* Delete links incident on a given switch port.
* @param npt
* @param reason
@@ -1409,7 +1415,7 @@
}
}
- /**
+ /**
* Iterates through the list of links and deletes if the
* last discovery message reception time exceeds timeout values.
*/
@@ -1430,7 +1436,7 @@
// Timeout the unicast and multicast LLDP valid times
// independently.
- if ((info.getUnicastValidTime() != null) &&
+ if ((info.getUnicastValidTime() != null) &&
(info.getUnicastValidTime() + (this.LINK_TIMEOUT * 1000) < curTime)){
info.setUnicastValidTime(null);
@@ -1440,7 +1446,7 @@
// the link would be deleted, which would trigger updateClusters().
linkChanged = true;
}
- if ((info.getMulticastValidTime()!= null) &&
+ if ((info.getMulticastValidTime()!= null) &&
(info.getMulticastValidTime()+ (this.LINK_TIMEOUT * 1000) < curTime)) {
info.setMulticastValidTime(null);
// if uTime is not null, then link will remain as openflow
@@ -1451,7 +1457,7 @@
}
// Add to the erase list only if the unicast
// time is null.
- if (info.getUnicastValidTime() == null &&
+ if (info.getUnicastValidTime() == null &&
info.getMulticastValidTime() == null){
eraseList.add(entry.getKey());
} else if (linkChanged) {
@@ -1510,11 +1516,11 @@
srcNpt = new NodePortTuple(lt.getSrc(), lt.getSrcPort());
dstNpt = new NodePortTuple(lt.getDst(), lt.getDstPort());
- if (!portBroadcastDomainLinks.containsKey(lt.getSrc()))
+ if (!portBroadcastDomainLinks.containsKey(srcNpt))
portBroadcastDomainLinks.put(srcNpt, new HashSet<Link>());
portBroadcastDomainLinks.get(srcNpt).add(lt);
- if (!portBroadcastDomainLinks.containsKey(lt.getDst()))
+ if (!portBroadcastDomainLinks.containsKey(dstNpt))
portBroadcastDomainLinks.put(dstNpt, new HashSet<Link>());
portBroadcastDomainLinks.get(dstNpt).add(lt);
}
@@ -1575,7 +1581,7 @@
@Override
public Collection<Class<? extends IFloodlightService>> getModuleServices() {
- Collection<Class<? extends IFloodlightService>> l =
+ Collection<Class<? extends IFloodlightService>> l =
new ArrayList<Class<? extends IFloodlightService>>();
l.add(ILinkDiscoveryService.class);
//l.add(ITopologyService.class);
@@ -1586,7 +1592,7 @@
public Map<Class<? extends IFloodlightService>, IFloodlightService>
getServiceImpls() {
Map<Class<? extends IFloodlightService>,
- IFloodlightService> m =
+ IFloodlightService> m =
new HashMap<Class<? extends IFloodlightService>,
IFloodlightService>();
// We are the class that implements the service
@@ -1596,7 +1602,7 @@
@Override
public Collection<Class<? extends IFloodlightService>> getModuleDependencies() {
- Collection<Class<? extends IFloodlightService>> l =
+ Collection<Class<? extends IFloodlightService>> l =
new ArrayList<Class<? extends IFloodlightService>>();
l.add(IFloodlightProviderService.class);
l.add(IThreadPoolService.class);
@@ -1679,7 +1685,7 @@
log.error("Exception in LLDP send timer.", e);
} finally {
if (!shuttingDown) {
- // Always reschedule link discovery if we're not
+ // Always reschedule link discovery if we're not
// shutting down (no chance of SLAVE role now)
log.trace("Rescheduling discovery task");
discoveryTask.reschedule(DISCOVERY_TASK_INTERVAL,
@@ -1729,7 +1735,7 @@
if ((sw.getChannel() != null) &&
(SocketAddress.class.isInstance(
sw.getChannel().getRemoteAddress()))) {
- evTopoSwitch.ipv4Addr =
+ evTopoSwitch.ipv4Addr =
IPv4.toIPv4Address(((InetSocketAddress)(sw.getChannel().
getRemoteAddress())).getAddress().getAddress());
evTopoSwitch.l4Port =
@@ -1787,10 +1793,12 @@
evTopoCluster = evHistTopologyCluster.put(evTopoCluster, action);
}
+ @Override
public boolean isAutoPortFastFeature() {
return autoPortFastFeature;
}
+ @Override
public void setAutoPortFastFeature(boolean autoPortFastFeature) {
this.autoPortFastFeature = autoPortFastFeature;
}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/topology/Topology.java b/src/main/java/net/onrc/onos/ofcontroller/topology/Topology.java
index dedb589..92da9ba 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/topology/Topology.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/topology/Topology.java
@@ -381,17 +381,13 @@
* Read topology state from the database.
*
* @param dbHandler the Graph Database handler to use.
- * @return true if topology is updated. In other words,
- * topology read from database is different from current topology.
*/
- public boolean readFromDatabase(GraphDBOperation dbHandler) {
+ public void readFromDatabase(GraphDBOperation dbHandler) {
//
// Fetch the relevant info from the Switch and Port vertices
// from the Titan Graph.
//
-
- Map<Long,Node> oldNodesMap = nodesMap;
- nodesMap = new TreeMap<Long,Node>();
+ nodesMap = new TreeMap<Long,Node>();
Iterable<ISwitchObject> activeSwitches = dbHandler.getActiveSwitches();
for (ISwitchObject switchObj : activeSwitches) {
@@ -456,7 +452,6 @@
}
}
dbHandler.commit();
- return ! compareTopology(oldNodesMap, nodesMap);
}
// TODO Merge into loops in readFromDatabase() can reduce execution time.