Merge ONOS master
diff --git a/cluster-mgmt/template/onsdemo_edge_template.py b/cluster-mgmt/template/onsdemo_edge_template.py
index e340f38..c3d0287 100755
--- a/cluster-mgmt/template/onsdemo_edge_template.py
+++ b/cluster-mgmt/template/onsdemo_edge_template.py
@@ -65,7 +65,7 @@
switch.append(sw)
for i in range (NR_NODES):
- host.append(self.addHost( 'host%d' % (int(i)+1) ))
+ host.append(self.addHost( 'host%d.%d' % (NWID, int(i)+1) ))
for i in range (NR_NODES):
self.addLink(host[i], switch[i])
@@ -117,7 +117,7 @@
host = []
for i in range (NR_NODES):
- host.append(net.get( 'host%d' % (int(i)+1) ))
+ host.append(net.get( 'host%d.%d' % (NWID, (int(i)+1)) ))
net.start()
diff --git a/kryo2/pom.xml b/kryo2/pom.xml
index 1beb87d..788f952 100644
--- a/kryo2/pom.xml
+++ b/kryo2/pom.xml
@@ -7,7 +7,7 @@
<version>2.22</version>
<packaging>jar</packaging>
- <name>kyro2</name>
+ <name>kryo2</name>
<url>http://maven.apache.org</url>
<properties>
diff --git a/src/main/java/net/onrc/onos/datagrid/HazelcastDatagrid.java b/src/main/java/net/onrc/onos/datagrid/HazelcastDatagrid.java
index 180cbe9..775f952 100644
--- a/src/main/java/net/onrc/onos/datagrid/HazelcastDatagrid.java
+++ b/src/main/java/net/onrc/onos/datagrid/HazelcastDatagrid.java
@@ -607,6 +607,29 @@
}
/**
+ * Get a Flow for a given Flow ID.
+ *
+ * @param flowId the Flow ID of the Flow to get.
+ * @return the Flow if found, otherwise null.
+ */
+ @Override
+ public FlowPath getFlow(FlowId flowId) {
+ byte[] valueBytes = mapFlow.get(flowId.value());
+ if (valueBytes == null)
+ return null;
+
+ Kryo kryo = kryoFactory.newKryo();
+ //
+ // Decode the value
+ //
+ Input input = new Input(valueBytes);
+ FlowPath flowPath = kryo.readObject(input, FlowPath.class);
+ kryoFactory.deleteKryo(kryo);
+
+ return flowPath;
+ }
+
+ /**
* Send a notification that a Flow is added.
*
* @param flowPath the Flow that is added.
@@ -702,6 +725,29 @@
}
/**
+ * Get a Flow Entry for a given Flow Entry ID.
+ *
+ * @param flowEntryId the Flow Entry ID of the Flow Entry to get.
+ * @return the Flow Entry if found, otherwise null.
+ */
+ @Override
+ public FlowEntry getFlowEntry(FlowEntryId flowEntryId) {
+ byte[] valueBytes = mapFlowEntry.get(flowEntryId.value());
+ if (valueBytes == null)
+ return null;
+
+ Kryo kryo = kryoFactory.newKryo();
+ //
+ // Decode the value
+ //
+ Input input = new Input(valueBytes);
+ FlowEntry flowEntry = kryo.readObject(input, FlowEntry.class);
+ kryoFactory.deleteKryo(kryo);
+
+ return flowEntry;
+ }
+
+ /**
* Send a notification that a FlowEntry is added.
*
* @param flowEntry the FlowEntry that is added.
diff --git a/src/main/java/net/onrc/onos/datagrid/IDatagridService.java b/src/main/java/net/onrc/onos/datagrid/IDatagridService.java
index 9361341..034fe25 100644
--- a/src/main/java/net/onrc/onos/datagrid/IDatagridService.java
+++ b/src/main/java/net/onrc/onos/datagrid/IDatagridService.java
@@ -50,7 +50,7 @@
* @param arpEventHandler The ARP event handler to de-register.
*/
public void deregisterArpEventHandler(IArpEventHandler arpEventHandler);
-
+
/**
* Get all Flows that are currently in the datagrid.
*
@@ -59,6 +59,14 @@
Collection<FlowPath> getAllFlows();
/**
+ * Get a Flow for a given Flow ID.
+ *
+ * @param flowId the Flow ID of the Flow to get.
+ * @return the Flow if found, otherwise null.
+ */
+ FlowPath getFlow(FlowId flowId);
+
+ /**
* Send a notification that a Flow is added.
*
* @param flowPath the Flow that is added.
@@ -92,6 +100,14 @@
Collection<FlowEntry> getAllFlowEntries();
/**
+ * Get a Flow Entry for a given Flow Entry ID.
+ *
+ * @param flowEntryId the Flow Entry ID of the Flow Entry to get.
+ * @return the Flow Entry if found, otherwise null.
+ */
+ FlowEntry getFlowEntry(FlowEntryId flowEntryId);
+
+ /**
* Send a notification that a FlowEntry is added.
*
* @param flowEntry the FlowEntry that is added.
diff --git a/src/main/java/net/onrc/onos/flow/FlowManagerImpl.java b/src/main/java/net/onrc/onos/flow/FlowManagerImpl.java
deleted file mode 100644
index 92948b7..0000000
--- a/src/main/java/net/onrc/onos/flow/FlowManagerImpl.java
+++ /dev/null
@@ -1,329 +0,0 @@
-package net.onrc.onos.flow;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Queue;
-import java.util.Set;
-
-import org.openflow.util.HexString;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.tinkerpop.blueprints.Direction;
-import com.tinkerpop.blueprints.Vertex;
-
-import net.floodlightcontroller.core.IOFSwitch;
-import net.onrc.onos.graph.DBOperation;
-import net.onrc.onos.graph.LocalTopologyEventListener;
-import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IFlowEntry;
-import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IFlowPath;
-import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IPortObject;
-import net.onrc.onos.ofcontroller.core.ISwitchStorage.SwitchState;
-import net.onrc.onos.ofcontroller.util.DataPath;
-import net.onrc.onos.ofcontroller.util.Dpid;
-import net.onrc.onos.ofcontroller.util.FlowEntry;
-import net.onrc.onos.ofcontroller.util.FlowEntryAction;
-import net.onrc.onos.ofcontroller.util.FlowEntryActions;
-import net.onrc.onos.ofcontroller.util.FlowEntryMatch;
-import net.onrc.onos.ofcontroller.util.FlowPath;
-import net.onrc.onos.ofcontroller.util.Port;
-import net.onrc.onos.ofcontroller.util.SwitchPort;
-
-public class FlowManagerImpl implements IFlowManager {
-
- protected final static Logger log = LoggerFactory.getLogger(LocalTopologyEventListener.class);
- protected DBOperation op;
-
- @Override
- public void createFlow(IPortObject src_port, IPortObject dest_port) {
- // TODO Auto-generated method stub
-
- }
-
- @Override
- public Iterable<FlowPath> getFlows(IPortObject src_port,
- IPortObject dest_port) {
- // TODO Auto-generated method stub
- return null;
- }
-
- @Override
- public Iterable<FlowPath> getOutFlows(IPortObject port) {
- // TODO Auto-generated method stub
- List<FlowPath> flowPaths = new ArrayList<FlowPath> ();
- Iterable<IFlowEntry> flowEntries = port.getOutFlowEntries();
-
- for(IFlowEntry fe: flowEntries) {
- IFlowPath flow = fe.getFlow();
- FlowPath flowPath = new FlowPath(flow);
- flowPaths.add(flowPath);
- }
- return flowPaths;
- }
-
- @Override
- public void reconcileFlows(IPortObject src_port) {
- // TODO Auto-generated method stub
-
- log.debug("Reconcile Flows for Port removed: {}:{}",src_port.getSwitch().getDPID(),src_port.getNumber());
- Iterable<IFlowEntry> flowEntries = src_port.getOutFlowEntries();
-
- for(IFlowEntry fe: flowEntries) {
- IFlowPath flow = fe.getFlow();
- reconcileFlow(flow);
- }
- }
-
- private void reconcileFlow(IFlowPath flow) {
- // TODO Auto-generated method stub
- String src_dpid = flow.getSrcSwitch();
- String dst_dpid = flow.getDstSwitch();
- Short src_port = flow.getSrcPort();
- Short dst_port = flow.getDstPort();
- IPortObject src = op.searchPort(src_dpid, src_port);
- IPortObject dst = op.searchPort(dst_dpid, dst_port);
- if (src != null && dst != null) {
- FlowPath newFlow = this.computeFlowPath(src,dst);
- installFlow(newFlow);
- removeFlow(flow);
- }
-
- }
-
- private void removeFlow(IFlowPath flow) {
- // TODO Auto-generated method stub
-
- }
-
- private void installFlow(FlowPath newFlow) {
- // TODO Auto-generated method stub
-
- }
-
- @Override
- public void reconcileFlow(IPortObject src_port, IPortObject dest_port) {
- // TODO Auto-generated method stub
-
- }
-
- @Override
- public FlowPath computeFlowPath(IPortObject src_port, IPortObject dest_port) {
- // TODO Auto-generated method stub
- DataPath dataPath = new DataPath();
-
- // FIXME: Bad idea to use FloodLight data structures (SwitchPort)
-
- dataPath.setSrcPort(new SwitchPort(new Dpid(src_port.getSwitch().getDPID()),
- new Port(src_port.getNumber())));
- dataPath.setDstPort(new SwitchPort(new Dpid(src_port.getSwitch().getDPID()),
- new Port(src_port.getNumber())));
-
- if (src_port.getSwitch().equals(dest_port.getSwitch())) {
- // on same switch create quick path
- FlowEntry flowEntry = new FlowEntry();
- flowEntry.setDpid(new Dpid(src_port.getSwitch().getDPID()));
- flowEntry.setInPort(new Port(src_port.getNumber()));
- flowEntry.setOutPort(new Port(src_port.getNumber()));
- flowEntry.setFlowEntryMatch(new FlowEntryMatch());
- flowEntry.flowEntryMatch().enableInPort(flowEntry.inPort());
-
- // Set the outgoing port output action
- FlowEntryActions flowEntryActions = flowEntry.flowEntryActions();
- FlowEntryAction flowEntryAction = new FlowEntryAction();
- flowEntryAction.setActionOutput(flowEntry.outPort());
- flowEntryActions.addAction(flowEntryAction);
- dataPath.flowEntries().add(flowEntry);
-
- FlowPath flowPath = new FlowPath();
- flowPath.setDataPath(dataPath);
-
- return flowPath;
- }
- Vertex v_src = src_port.getSwitch().asVertex();
- Vertex v_dest = dest_port.getSwitch().asVertex();
-
- //
- // Implement the Shortest Path computation by using Breath First Search
- //
- Set<Vertex> visitedSet = new HashSet<Vertex>();
- Queue<Vertex> processingList = new LinkedList<Vertex>();
- Map<Vertex, Vertex> previousVertexMap = new HashMap<Vertex, Vertex>();
-
- processingList.add(v_src);
- visitedSet.add(v_src);
- Boolean path_found = false;
- while (! processingList.isEmpty()) {
- Vertex nextVertex = processingList.poll();
- if (v_dest.equals(nextVertex)) {
- path_found = true;
- break;
- }
- for (Vertex parentPort : nextVertex.getVertices(Direction.OUT, "on")) {
- for (Vertex childPort : parentPort.getVertices(Direction.OUT, "link")) {
- for (Vertex child : childPort.getVertices(Direction.IN, "on")) {
- // Ignore inactive switches
- String state = child.getProperty("state").toString();
- if (! state.equals(SwitchState.ACTIVE.toString()))
- continue;
-
- if (! visitedSet.contains(child)) {
- previousVertexMap.put(parentPort, nextVertex);
- previousVertexMap.put(childPort, parentPort);
- previousVertexMap.put(child, childPort);
- visitedSet.add(child);
- processingList.add(child);
- }
- }
- }
- }
- }
- if (! path_found) {
- return null; // No path found
- }
-
- List<Vertex> resultPath = new LinkedList<Vertex>();
- Vertex previousVertex = v_dest;
- resultPath.add(v_dest);
- while (! v_src.equals(previousVertex)) {
- Vertex currentVertex = previousVertexMap.get(previousVertex);
- resultPath.add(currentVertex);
- previousVertex = currentVertex;
- }
- Collections.reverse(resultPath);
-
- // Loop through the result and prepare the return result
- // as a list of Flow Entries.
- //
- long nodeId = 0;
- short portId = 0;
- Port inPort = new Port(src_port.getNumber());
- Port outPort = new Port();
- int idx = 0;
- for (Vertex v: resultPath) {
- String type = v.getProperty("type").toString();
- // System.out.println("type: " + type);
- if (type.equals("port")) {
- //String number = v.getProperty("number").toString();
- // System.out.println("number: " + number);
-
- Object obj = v.getProperty("number");
- // String class_str = obj.getClass().toString();
- if (obj instanceof Short) {
- portId = (Short)obj;
- } else if (obj instanceof Integer) {
- Integer int_nodeId = (Integer)obj;
- portId = int_nodeId.shortValue();
- // int int_nodeId = (Integer)obj;
- // portId = (short)int_nodeId.;
- }
- } else if (type.equals("switch")) {
- String dpid = v.getProperty("dpid").toString();
- nodeId = HexString.toLong(dpid);
-
- // System.out.println("dpid: " + dpid);
- }
- idx++;
- if (idx == 1) {
- continue;
- }
- int mod = idx % 3;
- if (mod == 0) {
- // Setup the incoming port
- inPort = new Port(portId);
- continue;
- }
- if (mod == 2) {
- // Setup the outgoing port, and add the Flow Entry
- outPort = new Port(portId);
-
- FlowEntry flowEntry = new FlowEntry();
- flowEntry.setDpid(new Dpid(nodeId));
- flowEntry.setInPort(inPort);
- flowEntry.setOutPort(outPort);
- flowEntry.setFlowEntryMatch(new FlowEntryMatch());
- flowEntry.flowEntryMatch().enableInPort(flowEntry.inPort());
-
- // Set the outgoing port output action
- FlowEntryActions flowEntryActions = flowEntry.flowEntryActions();
- FlowEntryAction flowEntryAction = new FlowEntryAction();
- flowEntryAction.setActionOutput(flowEntry.outPort());
- flowEntryActions.addAction(flowEntryAction);
- dataPath.flowEntries().add(flowEntry);
- continue;
- }
- }
- if (idx > 0) {
- // Add the last Flow Entry
- FlowEntry flowEntry = new FlowEntry();
- flowEntry.setDpid(new Dpid(nodeId));
- flowEntry.setInPort(inPort);
- flowEntry.setOutPort(new Port(dest_port.getNumber()));
- flowEntry.setFlowEntryMatch(new FlowEntryMatch());
- flowEntry.flowEntryMatch().enableInPort(flowEntry.inPort());
-
- // Set the outgoing port output action
- FlowEntryActions flowEntryActions = flowEntry.flowEntryActions();
- FlowEntryAction flowEntryAction = new FlowEntryAction();
- flowEntryAction.setActionOutput(flowEntry.outPort());
- flowEntryActions.addAction(flowEntryAction);
- dataPath.flowEntries().add(flowEntry);
- // TODO (BOC): why is this twice?
- dataPath.flowEntries().add(flowEntry);
- }
-
-
- if (dataPath.flowEntries().size() > 0) {
- FlowPath flowPath = new FlowPath();
- flowPath.setDataPath(dataPath);
-
- return flowPath;
- }
- return null;
-
- }
-
- @Override
- public Iterable<FlowEntry> getFlowEntries(FlowPath flow) {
- // TODO Auto-generated method stub
- return null;
- }
-
-
- @Override
- public boolean installRemoteFlowEntry(FlowPath flowPath,
- FlowEntry entry) {
- // TODO Auto-generated method stub
- return false;
- }
-
- @Override
- public boolean removeRemoteFlowEntry(FlowPath flowPath,
- FlowEntry entry) {
- return false;
- // TODO Auto-generated method stub
-
- }
-
- @Override
- public boolean installFlowEntry(IOFSwitch mySwitch,
- FlowPath flowPath,
- FlowEntry flowEntry) {
- // TODO Auto-generated method stub
- return false;
- }
-
- @Override
- public boolean removeFlowEntry(IOFSwitch mySwitch,
- FlowPath flowPath,
- FlowEntry flowEntry) {
- // TODO Auto-generated method stub
- return false;
- }
-
-
-}
diff --git a/src/main/java/net/onrc/onos/flow/IFlowManager.java b/src/main/java/net/onrc/onos/flow/IFlowManager.java
deleted file mode 100644
index 598da85..0000000
--- a/src/main/java/net/onrc/onos/flow/IFlowManager.java
+++ /dev/null
@@ -1,126 +0,0 @@
-package net.onrc.onos.flow;
-
-import net.floodlightcontroller.core.IOFSwitch;
-import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IPortObject;
-import net.onrc.onos.ofcontroller.util.FlowEntry;
-import net.onrc.onos.ofcontroller.util.FlowPath;
-
-public interface IFlowManager {
- /**
- * Create a Flow from port to port.
- *
- * TODO: We don't need it for now.
- *
- * @param src_port the source port.
- * @param dest_port the destination port.
- */
- public void createFlow(IPortObject src_port, IPortObject dest_port);
-
- /**
- * Get all Flows matching a source and a destination port.
- *
- * TODO: Pankaj might be implementing it later.
- *
- * @param src_port the source port to match.
- * @param dest_port the destination port to match.
- * @return all flows matching the source and the destination port.
- */
- public Iterable<FlowPath> getFlows(IPortObject src_port,
- IPortObject dest_port);
-
- /**
- * Get all Flows going out from a port.
- *
- * TODO: We need it now: Pankaj
- *
- * @param port the port to match.
- * @return the list of flows that are going out from the port.
- */
- public Iterable<FlowPath> getOutFlows(IPortObject port);
-
- /**
- * Reconcile all flows on inactive switch port.
- *
- * @param portObject the port that has become inactive.
- */
- public void reconcileFlows(IPortObject portObject);
-
- /**
- * Reconcile all flows between a source and a destination port.
- *
- * TODO: We don't need it for now.
- *
- * @param src_port the source port.
- * @param dest_port the destination port.
- */
- public void reconcileFlow(IPortObject src_port, IPortObject dest_port);
-
- /**
- * Compute the shortest path between a source and a destination ports.
- *
- * @param src_port the source port.
- * @param dest_port the destination port.
- * @return the computed shortest path between the source and the
- * destination ports. The flow entries in the path itself would
- * contain the incoming port matching and the outgoing port output
- * actions set. However, the path itself will NOT have the Flow ID,
- * Installer ID, and any additional matching conditions for the
- * flow entries (e.g., source or destination MAC address, etc).
- */
- public FlowPath computeFlowPath(IPortObject src_port,
- IPortObject dest_port);
-
- /**
- * Get all Flow Entries of a Flow.
- *
- * @param flow the flow whose flow entries should be returned.
- * @return the flow entries of the flow.
- */
- public Iterable<FlowEntry> getFlowEntries(FlowPath flow);
-
- /**
- * Install a Flow Entry on a switch.
- *
- * @param mySwitch the switch to install the Flow Entry into.
- * @param flowPath the flow path for the flow entry to install.
- * @param flowEntry the flow entry to install.
- * @return true on success, otherwise false.
- */
- public boolean installFlowEntry(IOFSwitch mySwitch, FlowPath flowPath,
- FlowEntry flowEntry);
-
- /**
- * Remove a Flow Entry from a switch.
- *
- * @param mySwitch the switch to remove the Flow Entry from.
- * @param flowPath the flow path for the flow entry to remove.
- * @param flowEntry the flow entry to remove.
- * @return true on success, otherwise false.
- */
- public boolean removeFlowEntry(IOFSwitch mySwitch, FlowPath flowPath,
- FlowEntry flowEntry);
-
- /**
- * Install a Flow Entry on a remote controller.
- *
- * TODO: We need it now: Jono
- * - For now it will make a REST call to the remote controller.
- * - Internally, it needs to know the name of the remote controller.
- *
- * @param flowPath the flow path for the flow entry to install.
- * @param flowEntry the flow entry to install.
- * @return true on success, otherwise false.
- */
- public boolean installRemoteFlowEntry(FlowPath flowPath,
- FlowEntry flowEntry);
-
- /**
- * Remove a flow entry on a remote controller.
- *
- * @param flowPath the flow path for the flow entry to remove.
- * @param flowEntry the flow entry to remove.
- * @return true on success, otherwise false.
- */
- public boolean removeRemoteFlowEntry(FlowPath flowPath,
- FlowEntry flowEntry);
-}
diff --git a/src/main/java/net/onrc/onos/graph/LocalTopologyEventListener.java b/src/main/java/net/onrc/onos/graph/LocalTopologyEventListener.java
index 66e02bc..2778b55 100644
--- a/src/main/java/net/onrc/onos/graph/LocalTopologyEventListener.java
+++ b/src/main/java/net/onrc/onos/graph/LocalTopologyEventListener.java
@@ -1,7 +1,5 @@
package net.onrc.onos.graph;
-import net.onrc.onos.flow.FlowManagerImpl;
-import net.onrc.onos.flow.IFlowManager;
import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IPortObject;
import org.slf4j.Logger;
@@ -53,9 +51,12 @@
src_port.getNumber(),
dest_port.getSwitch().getDPID(),
dest_port.getNumber()});
- IFlowManager manager = new FlowManagerImpl();
// TODO: Find the flows and add to reconcile queue
- manager.reconcileFlows(src_port);
+ //
+ // NOTE: Old code/logic.
+ //
+ // IFlowService flowManager = ...
+ // flowManager.reconcileFlows(src_port);
}
}
@@ -81,8 +82,11 @@
IPortObject src_port = (IPortObject) conn.getFramedGraph().frame(vertex, IPortObject.class);
log.debug("TopologyEvents: Port removed: {}:{}",src_port.getSwitch().getDPID(),src_port.getNumber());
- IFlowManager manager = new FlowManagerImpl();
- manager.reconcileFlows(src_port);
+
+ // NOTE: Old code/logic.
+ //
+ // IFlowService flowManager = ...
+ // flowManager.reconcileFlows(src_port);
}
}
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 fd88538..1bea72d 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowDatabaseOperation.java
@@ -12,8 +12,6 @@
import net.floodlightcontroller.util.MACAddress;
import net.onrc.onos.graph.DBOperation;
-import net.onrc.onos.graph.GraphDBOperation;
-
import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IFlowEntry;
import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IFlowPath;
import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IPortObject;
@@ -191,8 +189,7 @@
//
// Assign the FlowEntry ID.
//
- if ((flowEntry.flowEntryId() == null) ||
- (flowEntry.flowEntryId().value() == 0)) {
+ if (! flowEntry.isValidFlowEntryId()) {
long id = flowManager.getNextFlowEntryId();
flowEntry.setFlowEntryId(new FlowEntryId(id));
}
@@ -592,7 +589,7 @@
* @param maxFlows the maximum number of flows to be returned.
* @return the Flow Paths if found, otherwise null.
*/
- static ArrayList<IFlowPath> getAllFlowsSummary(DBOperation dbHandler,
+ static ArrayList<FlowPath> getAllFlowsSummary(DBOperation dbHandler,
FlowId flowId,
int maxFlows) {
//
@@ -602,61 +599,32 @@
// We should use the appropriate Titan/Gremlin query to filter-out
// the flows as appropriate.
//
- ArrayList<IFlowPath> flowPathsWithoutFlowEntries =
- getAllFlowsWithoutFlowEntries(dbHandler);
-
- Collections.sort(flowPathsWithoutFlowEntries,
- new Comparator<IFlowPath>() {
- @Override
- public int compare(IFlowPath first, IFlowPath second) {
- long result =
- new FlowId(first.getFlowId()).value()
- - new FlowId(second.getFlowId()).value();
- if (result > 0) {
- return 1;
- } else if (result < 0) {
- return -1;
- } else {
- return 0;
- }
- }
- }
- );
-
- return flowPathsWithoutFlowEntries;
+ ArrayList<FlowPath> flowPaths = getAllFlowsWithDataPathSummary(dbHandler);
+ Collections.sort(flowPaths);
+ return flowPaths;
}
/**
- * Get all Flows information, without the associated Flow Entries.
+ * Get all Flows information, with Data Path summary for the Flow Entries.
*
* @param dbHandler the Graph Database handler to use.
- * @return all Flows information, without the associated Flow Entries.
+ * @return all Flows information, with Data Path summary for the Flow
+ * Entries.
*/
- static ArrayList<IFlowPath> getAllFlowsWithoutFlowEntries(DBOperation dbHandler) {
- Iterable<IFlowPath> flowPathsObj = null;
- ArrayList<IFlowPath> flowPathsObjArray = new ArrayList<IFlowPath>();
+ static ArrayList<FlowPath> getAllFlowsWithDataPathSummary(DBOperation dbHandler) {
+ ArrayList<FlowPath> flowPaths = getAllFlows(dbHandler);
- // TODO: Remove this op.commit() flow, because it is not needed?
- dbHandler.commit();
+ // Truncate each Flow Path and Flow Entry
+ for (FlowPath flowPath : flowPaths) {
+ flowPath.setFlowEntryMatch(null);
+ flowPath.setFlowEntryActions(null);
+ for (FlowEntry flowEntry : flowPath.flowEntries()) {
+ flowEntry.setFlowEntryMatch(null);
+ flowEntry.setFlowEntryActions(null);
+ }
+ }
- try {
- flowPathsObj = dbHandler.getAllFlowPaths();
- } catch (Exception e) {
- // TODO: handle exceptions
- dbHandler.rollback();
- log.error(":getAllFlowPaths failed");
- return flowPathsObjArray; // No Flows found
- }
- if ((flowPathsObj == null) || (flowPathsObj.iterator().hasNext() == false)) {
- return flowPathsObjArray; // No Flows found
- }
-
- for (IFlowPath flowObj : flowPathsObj)
- flowPathsObjArray.add(flowObj);
-
- // conn.endTx(Transaction.COMMIT);
-
- return flowPathsObjArray;
+ return flowPaths;
}
/**
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 0e9887a..eccf40b 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowEventHandler.java
@@ -171,18 +171,28 @@
// - EventEntry<FlowEntry>
//
for (EventEntry<?> event : collection) {
+ // Topology event
if (event.eventData() instanceof TopologyElement) {
EventEntry<TopologyElement> topologyEventEntry =
(EventEntry<TopologyElement>)event;
topologyEvents.add(topologyEventEntry);
- } else if (event.eventData() instanceof FlowPath) {
+ continue;
+ }
+
+ // FlowPath event
+ if (event.eventData() instanceof FlowPath) {
EventEntry<FlowPath> flowPathEventEntry =
(EventEntry<FlowPath>)event;
flowPathEvents.add(flowPathEventEntry);
- } else if (event.eventData() instanceof FlowEntry) {
+ continue;
+ }
+
+ // FlowEntry event
+ if (event.eventData() instanceof FlowEntry) {
EventEntry<FlowEntry> flowEntryEventEntry =
(EventEntry<FlowEntry>)event;
flowEntryEvents.add(flowEntryEventEntry);
+ continue;
}
}
collection.clear();
@@ -710,9 +720,18 @@
//
newFlowEntry.setFlowId(new FlowId(flowPath.flowId().value()));
- // Set the incoming port matching
- FlowEntryMatch flowEntryMatch = new FlowEntryMatch();
+ //
+ // Allocate the FlowEntryMatch by copying the default one
+ // from the FlowPath (if set).
+ //
+ FlowEntryMatch flowEntryMatch = null;
+ if (flowPath.flowEntryMatch() != null)
+ flowEntryMatch = new FlowEntryMatch(flowPath.flowEntryMatch());
+ else
+ flowEntryMatch = new FlowEntryMatch();
newFlowEntry.setFlowEntryMatch(flowEntryMatch);
+
+ // Set the incoming port matching
flowEntryMatch.enableInPort(newFlowEntry.inPort());
//
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 20fe5d2..a6b7f32 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/FlowManager.java
@@ -2,7 +2,6 @@
import java.util.ArrayList;
import java.util.Collection;
-import java.util.EnumSet;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
@@ -53,15 +52,6 @@
protected IFlowPusherService pusher;
- protected OFMessageDamper messageDamper;
-
- //
- // TODO: Values copied from elsewhere (class LearningSwitch).
- // The local copy should go away!
- //
- protected static final int OFMESSAGE_DAMPER_CAPACITY = 50000; // TODO: find sweet spot
- protected static final int OFMESSAGE_DAMPER_TIMEOUT = 250; // ms
-
// Flow Entry ID generation state
private static Random randomGenerator = new Random();
private static int nextFlowEntryIdPrefix = 0;
@@ -165,14 +155,7 @@
floodlightProvider = context.getServiceImpl(IFloodlightProviderService.class);
datagridService = context.getServiceImpl(IDatagridService.class);
restApi = context.getServiceImpl(IRestApiService.class);
-
- if (enableFlowPusher) {
- pusher = context.getServiceImpl(IFlowPusherService.class);
- } else {
- messageDamper = new OFMessageDamper(OFMESSAGE_DAMPER_CAPACITY,
- EnumSet.of(OFType.FLOW_MOD),
- OFMESSAGE_DAMPER_TIMEOUT);
- }
+ pusher = context.getServiceImpl(IFlowPusherService.class);
this.init("","");
}
@@ -369,22 +352,13 @@
* @return the Flow Paths if found, otherwise null.
*/
@Override
- public ArrayList<IFlowPath> getAllFlowsSummary(FlowId flowId,
- int maxFlows) {
+ public ArrayList<FlowPath> getAllFlowsSummary(FlowId flowId,
+ int maxFlows) {
return FlowDatabaseOperation.getAllFlowsSummary(dbHandlerApi, flowId,
maxFlows);
}
/**
- * Get all Flows information, without the associated Flow Entries.
- *
- * @return all Flows information, without the associated Flow Entries.
- */
- public ArrayList<IFlowPath> getAllFlowsWithoutFlowEntries() {
- return FlowDatabaseOperation.getAllFlowsWithoutFlowEntries(dbHandlerApi);
- }
-
- /**
* Add and maintain a shortest-path flow.
*
* NOTE: The Flow Path argument does NOT contain flow entries.
@@ -426,58 +400,13 @@
}
/**
- * Install a Flow Entry on a switch.
+ * Inform the Flow Manager that a Flow Entry on switch expired.
*
- * @param mySwitch the switch to install the Flow Entry into.
- * @param flowObj the flow path object for the flow entry to install.
- * @param flowEntryObj the flow entry object to install.
- * @return true on success, otherwise false.
+ * @param sw the switch the Flow Entry expired on.
+ * @param flowEntryId the Flow Entry ID of the expired Flow Entry.
*/
- private boolean installFlowEntry(IOFSwitch mySwitch, IFlowPath flowObj,
- IFlowEntry flowEntryObj) {
- if (enableFlowPusher) {
- return pusher.add(mySwitch, flowObj, flowEntryObj);
- } else {
- return FlowSwitchOperation.installFlowEntry(
- floodlightProvider.getOFMessageFactory(),
- messageDamper, mySwitch, flowObj, flowEntryObj);
- }
- }
-
- /**
- * Install a Flow Entry on a switch.
- *
- * @param mySwitch the switch to install the Flow Entry into.
- * @param flowPath the flow path for the flow entry to install.
- * @param flowEntry the flow entry to install.
- * @return true on success, otherwise false.
- */
- private boolean installFlowEntry(IOFSwitch mySwitch, FlowPath flowPath,
- FlowEntry flowEntry) {
- if (enableFlowPusher) {
- return pusher.add(mySwitch, flowPath, flowEntry);
- } else {
- return FlowSwitchOperation.installFlowEntry(
- floodlightProvider.getOFMessageFactory(),
- messageDamper, mySwitch, flowPath, flowEntry);
- }
- }
-
- /**
- * Remove a Flow Entry from a switch.
- *
- * @param mySwitch the switch to remove the Flow Entry from.
- * @param flowPath the flow path for the flow entry to remove.
- * @param flowEntry the flow entry to remove.
- * @return true on success, otherwise false.
- */
- private boolean removeFlowEntry(IOFSwitch mySwitch, FlowPath flowPath,
- FlowEntry flowEntry) {
- //
- // The installFlowEntry() method implements both installation
- // and removal of flow entries.
- //
- return (installFlowEntry(mySwitch, flowPath, flowEntry));
+ public void flowEntryOnSwitchExpired(IOFSwitch sw, FlowEntryId flowEntryId) {
+ // TODO: Not implemented yet
}
/**
@@ -506,9 +435,9 @@
log.debug("Pushing Flow Entry To Switch: {}", flowEntry.toString());
//
- // Install the Flow Entry into the switch
+ // Push the Flow Entry into the switch
//
- if (! installFlowEntry(mySwitch, flowPath, flowEntry)) {
+ if (! pusher.add(mySwitch, flowPath, flowEntry)) {
String logMsg = "Cannot install Flow Entry " +
flowEntry.flowEntryId() +
" from Flow Path " + flowPath.flowId() +
@@ -540,6 +469,9 @@
for (FlowPathEntryPair flowPair : modifiedFlowEntries) {
FlowEntry flowEntry = flowPair.flowEntry;
+ if (! flowEntry.isValidFlowEntryId())
+ continue;
+
IOFSwitch mySwitch = mySwitches.get(flowEntry.dpid().value());
//
@@ -558,8 +490,6 @@
FlowEntryUserState.FE_USER_DELETE) {
continue;
}
- if (! flowEntry.isValidFlowEntryId())
- continue;
}
log.debug("Pushing Flow Entry To Datagrid: {}", flowEntry.toString());
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowService.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowService.java
index b06d844..8d2b797 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowService.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/IFlowService.java
@@ -2,11 +2,12 @@
import java.util.ArrayList;
+import net.floodlightcontroller.core.IOFSwitch;
import net.floodlightcontroller.core.module.IFloodlightService;
-import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IFlowPath;
import net.onrc.onos.ofcontroller.topology.Topology;
import net.onrc.onos.ofcontroller.util.CallerId;
import net.onrc.onos.ofcontroller.util.DataPathEndpoints;
+import net.onrc.onos.ofcontroller.util.FlowEntryId;
import net.onrc.onos.ofcontroller.util.FlowId;
import net.onrc.onos.ofcontroller.util.FlowPath;
@@ -82,7 +83,7 @@
* @param maxFlows number of flows to return
* @return the Flow Paths if found, otherwise null.
*/
- ArrayList<IFlowPath> getAllFlowsSummary(FlowId flowId, int maxFlows);
+ ArrayList<FlowPath> getAllFlowsSummary(FlowId flowId, int maxFlows);
/**
* Add and maintain a shortest-path flow.
@@ -113,4 +114,12 @@
* @return unique flow ID
*/
public long getNextFlowEntryId();
+
+ /**
+ * Inform the Flow Manager that a Flow Entry on switch expired.
+ *
+ * @param sw the switch the Flow Entry expired on.
+ * @param flowEntryId the Flow Entry ID of the expired Flow Entry.
+ */
+ public void flowEntryOnSwitchExpired(IOFSwitch sw, FlowEntryId flowEntryId);
}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/web/GetSummaryFlowsResource.java b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/web/GetSummaryFlowsResource.java
index 89e5b01..58f82a9 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowmanager/web/GetSummaryFlowsResource.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowmanager/web/GetSummaryFlowsResource.java
@@ -2,8 +2,8 @@
import java.util.ArrayList;
-import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IFlowPath;
import net.onrc.onos.ofcontroller.flowmanager.IFlowService;
+import net.onrc.onos.ofcontroller.util.FlowPath;
import net.onrc.onos.ofcontroller.util.FlowId;
import org.restlet.resource.Get;
@@ -31,8 +31,8 @@
* @return the collection of Flow states if any found, otherwise null.
*/
@Get("json")
- public ArrayList<IFlowPath> retrieve() {
- ArrayList<IFlowPath> result = null;
+ public ArrayList<FlowPath> retrieve() {
+ ArrayList<FlowPath> result = null;
FlowId flowId;
int maxFlows = 0;
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowProgrammer.java b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowProgrammer.java
index b567a87..461d231 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowProgrammer.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowProgrammer.java
@@ -5,32 +5,61 @@
import java.util.HashMap;
import java.util.Map;
+import org.openflow.protocol.OFFlowRemoved;
+import org.openflow.protocol.OFMessage;
+import org.openflow.protocol.OFType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import net.floodlightcontroller.core.FloodlightContext;
import net.floodlightcontroller.core.IFloodlightProviderService;
+import net.floodlightcontroller.core.IOFMessageListener;
+import net.floodlightcontroller.core.IOFSwitch;
+import net.floodlightcontroller.core.IOFSwitchListener;
import net.floodlightcontroller.core.module.FloodlightModuleContext;
import net.floodlightcontroller.core.module.FloodlightModuleException;
import net.floodlightcontroller.core.module.IFloodlightModule;
import net.floodlightcontroller.core.module.IFloodlightService;
+import net.floodlightcontroller.restserver.IRestApiService;
+import net.onrc.onos.ofcontroller.flowprogrammer.web.FlowProgrammerWebRoutable;
+import net.onrc.onos.ofcontroller.flowmanager.IFlowService;
+import net.onrc.onos.ofcontroller.util.FlowEntryId;
+import net.onrc.onos.registry.controller.IControllerRegistryService;
-public class FlowProgrammer implements IFloodlightModule {
+/**
+ * FlowProgrammer is a module responsible to maintain flows installed to switches.
+ * FlowProgrammer consists of FlowPusher and FlowSynchronizer.
+ * FlowPusher manages the rate of installation, and FlowSynchronizer synchronizes
+ * flows between GraphDB and switches.
+ * FlowProgrammer also watch the event of addition/deletion of switches to
+ * start/stop synchronization. When a switch is added to network, FlowProgrammer
+ * immediately kicks synchronization to keep switch's flow table latest state.
+ * Adversely, when a switch is removed from network, FlowProgrammer immediately
+ * stops synchronization.
+ * @author Brian
+ *
+ */
+public class FlowProgrammer implements IFloodlightModule,
+ IOFMessageListener,
+ IOFSwitchListener {
@SuppressWarnings("unused")
- private final static Logger log = LoggerFactory.getLogger(FlowProgrammer.class);
-
- private static final boolean enableFlowSync = false;
-
+ // flag to enable FlowSynchronizer
+ private static final boolean enableFlowSync = false;
+ protected static Logger log = LoggerFactory.getLogger(FlowProgrammer.class);
protected volatile IFloodlightProviderService floodlightProvider;
+ protected volatile IControllerRegistryService registryService;
+ protected volatile IRestApiService restApi;
+ protected volatile IFlowService flowManager;
protected FlowPusher pusher;
private static final int NUM_PUSHER_THREAD = 1;
protected FlowSynchronizer synchronizer;
-
+
public FlowProgrammer() {
pusher = new FlowPusher(NUM_PUSHER_THREAD);
if (enableFlowSync) {
- synchronizer = new FlowSynchronizer();
+ synchronizer = new FlowSynchronizer();
}
}
@@ -38,18 +67,21 @@
public void init(FloodlightModuleContext context)
throws FloodlightModuleException {
floodlightProvider = context.getServiceImpl(IFloodlightProviderService.class);
+ registryService = context.getServiceImpl(IControllerRegistryService.class);
+ restApi = context.getServiceImpl(IRestApiService.class);
+ flowManager = context.getServiceImpl(IFlowService.class);
pusher.init(null, context, floodlightProvider.getOFMessageFactory(), null);
if (enableFlowSync) {
- synchronizer.init(context);
+ synchronizer.init(pusher);
}
}
@Override
public void startUp(FloodlightModuleContext context) {
+ restApi.addRestletRoutable(new FlowProgrammerWebRoutable());
pusher.start();
- if (enableFlowSync) {
- synchronizer.startUp(context);
- }
+ floodlightProvider.addOFMessageListener(OFType.FLOW_REMOVED, this);
+ floodlightProvider.addOFSwitchListener(this);
}
@Override
@@ -58,7 +90,7 @@
new ArrayList<Class<? extends IFloodlightService>>();
l.add(IFlowPusherService.class);
if (enableFlowSync) {
- l.add(IFlowSyncService.class);
+ l.add(IFlowSyncService.class);
}
return l;
}
@@ -71,7 +103,7 @@
IFloodlightService>();
m.put(IFlowPusherService.class, pusher);
if (enableFlowSync) {
- m.put(IFlowSyncService.class, synchronizer);
+ m.put(IFlowSyncService.class, synchronizer);
}
return m;
}
@@ -81,7 +113,65 @@
Collection<Class<? extends IFloodlightService>> l =
new ArrayList<Class<? extends IFloodlightService>>();
l.add(IFloodlightProviderService.class);
+ l.add(IRestApiService.class);
return l;
}
+
+ @Override
+ public String getName() {
+ // TODO Auto-generated method stub
+ return "FlowProgrammer";
+ }
+
+ @Override
+ public boolean isCallbackOrderingPrereq(OFType type, String name) {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public boolean isCallbackOrderingPostreq(OFType type, String name) {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public Command receive(IOFSwitch sw, OFMessage msg, FloodlightContext cntx) {
+ switch (msg.getType()) {
+ case FLOW_REMOVED:
+ OFFlowRemoved flowMsg = (OFFlowRemoved) msg;
+ log.debug("Got flow removed from "+ sw.getId() +": "+ flowMsg.getCookie());
+ FlowEntryId id = new FlowEntryId(flowMsg.getCookie());
+ flowManager.flowEntryOnSwitchExpired(sw, id);
+ break;
+ default:
+ break;
+ }
+
+ return Command.CONTINUE;
+ }
+
+ @Override
+ public void addedSwitch(IOFSwitch sw) {
+ log.debug("Switch added: {}", sw.getId());
+
+ if (enableFlowSync && registryService.hasControl(sw.getId())) {
+ synchronizer.synchronize(sw);
+ }
+ }
+
+ @Override
+ public void removedSwitch(IOFSwitch sw) {
+ log.debug("Switch removed: {}", sw.getId());
+
+ if (enableFlowSync) {
+ synchronizer.interrupt(sw);
+ }
+ }
+
+ @Override
+ public void switchPortChanged(Long switchId) {
+ // TODO Auto-generated method stub
+ }
}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowPusher.java b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowPusher.java
index f43a83e..8f2469b 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowPusher.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowPusher.java
@@ -5,6 +5,7 @@
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -40,16 +41,20 @@
import net.onrc.onos.ofcontroller.util.Port;
/**
- * FlowPusher intermediates FlowManager/FlowSynchronizer and switches to push OpenFlow
- * messages to switches in proper rate.
+ * FlowPusher is a implementation of FlowPusherService.
+ * FlowPusher assigns one message queue instance for each one switch.
+ * Number of message processing threads is configurable by constructor, and
+ * one thread can handle multiple message queues. Each queue will be assigned to
+ * a thread according to hash function defined by getHash().
+ * Each processing thread reads messages from queues and sends it to switches
+ * in round-robin. Processing thread also calculates rate of sending to suppress
+ * excessive message sending.
* @author Naoki Shiota
*
*/
public class FlowPusher implements IFlowPusherService, IOFMessageListener {
private final static Logger log = LoggerFactory.getLogger(FlowPusher.class);
- private static boolean barrierIfEmpty = false;
-
// NOTE: Below are moved from FlowManager.
// TODO: Values copied from elsewhere (class LearningSwitch).
// The local copy should go away!
@@ -57,10 +62,6 @@
protected static final int OFMESSAGE_DAMPER_CAPACITY = 50000; // TODO: find sweet spot
protected static final int OFMESSAGE_DAMPER_TIMEOUT = 250; // ms
- // Interval of sleep when queue is empty
- protected static final long SLEEP_MILLI_SEC = 10;
- protected static final int SLEEP_NANO_SEC = 0;
-
// Number of messages sent to switch at once
protected static final int MAX_MESSAGE_SEND = 100;
@@ -74,7 +75,7 @@
}
/**
- * Message queue attached to a switch.
+ * SwitchQueue represents message queue attached to a switch.
* This consists of queue itself and variables used for limiting sending rate.
* @author Naoki Shiota
*
@@ -83,11 +84,14 @@
private class SwitchQueue extends ArrayDeque<OFMessage> {
QueueState state;
- // Max rate of sending message (bytes/sec). 0 implies no limitation.
+ // Max rate of sending message (bytes/ms). 0 implies no limitation.
long max_rate = 0; // 0 indicates no limitation
long last_sent_time = 0;
long last_sent_size = 0;
+ // "To be deleted" flag
+ boolean toBeDeleted = false;
+
/**
* Check if sending rate is within the rate
* @param current Current time
@@ -125,7 +129,10 @@
private FloodlightContext context = null;
private BasicFactory factory = null;
+
+ // Map of threads versus dpid
private Map<Long, FlowPusherThread> threadMap = null;
+ // Map of Future objects versus dpid and transaction ID.
private Map<Long, Map<Integer, OFBarrierReplyFuture>>
barrierFutures = new HashMap<Long, Map<Integer, OFBarrierReplyFuture>>();
@@ -138,14 +145,13 @@
*/
private class FlowPusherThread extends Thread {
private Map<IOFSwitch,SwitchQueue> queues
- = new HashMap<IOFSwitch,SwitchQueue>();
+ = new HashMap<IOFSwitch,SwitchQueue>();
+ // reusable latch used for waiting for arrival of message
private Semaphore mutex = new Semaphore(0);
@Override
public void run() {
- log.debug("Begin Flow Pusher Process");
-
while (true) {
try {
// wait for message pushed to queue
@@ -156,14 +162,16 @@
return;
}
- Set< Map.Entry<IOFSwitch,SwitchQueue> > entries;
+ // for safety of concurrent access, copy all key objects
+ Set<IOFSwitch> keys = new HashSet<IOFSwitch>(queues.size());
synchronized (queues) {
- entries = queues.entrySet();
+ for (IOFSwitch sw : queues.keySet()) {
+ keys.add(sw);
+ }
}
- for (Map.Entry<IOFSwitch,SwitchQueue> entry : entries) {
- IOFSwitch sw = entry.getKey();
- SwitchQueue queue = entry.getValue();
+ for (IOFSwitch sw : keys) {
+ SwitchQueue queue = queues.get(sw);
// Skip if queue is suspended
if (sw == null || queue == null ||
@@ -171,43 +179,62 @@
continue;
}
- // check sending rate and determine it to be sent or not
- long current_time = System.nanoTime();
- long size = 0;
-
synchronized (queue) {
- if (queue.isSendable(current_time)) {
- int i = 0;
- while (! queue.isEmpty()) {
- // Number of messages excess the limit
- if (i >= MAX_MESSAGE_SEND) {
- // Messages remains in queue
- mutex.release();
- break;
- }
- ++i;
-
- OFMessage msg = queue.poll();
- try {
- messageDamper.write(sw, msg, context);
- log.debug("Pusher sends message : {}", msg);
- size += msg.getLength();
- } catch (IOException e) {
- e.printStackTrace();
- log.error("Exception in sending message ({}) : {}", msg, e);
+ processQueue(sw, queue, MAX_MESSAGE_SEND);
+ if (queue.isEmpty()) {
+ // remove queue if flagged to be.
+ if (queue.toBeDeleted) {
+ synchronized (queues) {
+ queues.remove(sw);
}
}
- sw.flush();
- queue.logSentData(current_time, size);
-
- if (queue.isEmpty() && barrierIfEmpty) {
- barrier(sw);
+ } else {
+ // if some messages remains in queue, latch down
+ if (mutex.availablePermits() == 0) {
+ mutex.release();
}
}
}
}
}
}
+
+ /**
+ * Read messages from queue and send them to the switch.
+ * If number of messages excess the limit, stop sending messages.
+ * @param sw Switch to which messages will be sent.
+ * @param queue Queue of messages.
+ * @param max_msg Limitation of number of messages to be sent. If set to 0,
+ * all messages in queue will be sent.
+ */
+ private void processQueue(IOFSwitch sw, SwitchQueue queue, long max_msg) {
+ // check sending rate and determine it to be sent or not
+ long current_time = System.currentTimeMillis();
+ long size = 0;
+
+ if (queue.isSendable(current_time)) {
+ int i = 0;
+ while (! queue.isEmpty()) {
+ // Number of messages excess the limit
+ if (0 < max_msg && max_msg <= i) {
+ break;
+ }
+ ++i;
+
+ OFMessage msg = queue.poll();
+ try {
+ messageDamper.write(sw, msg, context);
+ log.debug("Pusher sends message : {}", msg);
+ size += msg.getLength();
+ } catch (IOException e) {
+ e.printStackTrace();
+ log.error("Exception in sending message ({}) : {}", msg, e);
+ }
+ }
+ sw.flush();
+ queue.logSentData(current_time, size);
+ }
+ }
}
/**
@@ -247,7 +274,7 @@
if (damper != null) {
messageDamper = damper;
} else {
- // use default value
+ // use default values
messageDamper = new OFMessageDamper(OFMESSAGE_DAMPER_CAPACITY,
EnumSet.of(OFType.FLOW_MOD),
OFMESSAGE_DAMPER_TIMEOUT);
@@ -272,10 +299,6 @@
}
}
- /**
- * Suspend sending messages to switch.
- * @param sw
- */
@Override
public boolean suspend(IOFSwitch sw) {
SwitchQueue queue = getQueue(sw);
@@ -293,9 +316,6 @@
}
}
- /**
- * Resume sending messages to switch.
- */
@Override
public boolean resume(IOFSwitch sw) {
SwitchQueue queue = getQueue(sw);
@@ -313,9 +333,6 @@
}
}
- /**
- * Check if given switch is suspended.
- */
@Override
public boolean isSuspended(IOFSwitch sw) {
SwitchQueue queue = getQueue(sw);
@@ -341,11 +358,7 @@
}
}
- /**
- * Set sending rate to a switch.
- * @param sw Switch.
- * @param rate Rate in bytes/sec.
- */
+ @Override
public void setRate(IOFSwitch sw, long rate) {
SwitchQueue queue = getQueue(sw);
if (queue == null) {
@@ -353,27 +366,66 @@
}
if (rate > 0) {
+ log.debug("rate for {} is set to {}", sw.getId(), rate);
queue.max_rate = rate;
}
}
+
+ @Override
+ public boolean createQueue(IOFSwitch sw) {
+ SwitchQueue queue = getQueue(sw);
+ if (queue != null) {
+ return false;
+ }
+
+ FlowPusherThread proc = getProcess(sw);
+ queue = new SwitchQueue();
+ queue.state = QueueState.READY;
+ synchronized (proc.queues) {
+ proc.queues.put(sw, queue);
+ }
+
+ return true;
+ }
+
+ @Override
+ public boolean deleteQueue(IOFSwitch sw) {
+ return deleteQueue(sw, false);
+ }
- /**
- * Add OFMessage to queue of the switch.
- * @param sw Switch to which message is sent.
- * @param msg Message to be sent.
- * @return true if succeed.
- */
+ @Override
+ public boolean deleteQueue(IOFSwitch sw, boolean forceStop) {
+ FlowPusherThread proc = getProcess(sw);
+
+ if (forceStop) {
+ synchronized (proc.queues) {
+ SwitchQueue queue = proc.queues.remove(sw);
+ if (queue == null) {
+ return false;
+ }
+ }
+ return true;
+ } else {
+ SwitchQueue queue = getQueue(sw);
+ if (queue == null) {
+ return false;
+ }
+ synchronized (queue) {
+ queue.toBeDeleted = true;
+ }
+ return true;
+ }
+ }
+
@Override
public boolean add(IOFSwitch sw, OFMessage msg) {
FlowPusherThread proc = getProcess(sw);
SwitchQueue queue = proc.queues.get(sw);
+ // create queue at first addition of message
if (queue == null) {
- queue = new SwitchQueue();
- queue.state = QueueState.READY;
- synchronized (proc) {
- proc.queues.put(sw, queue);
- }
+ createQueue(sw);
+ queue = getQueue(sw);
}
synchronized (queue) {
@@ -388,13 +440,6 @@
return true;
}
- /**
- * Create OFMessage from given flow information and add it to the queue.
- * @param sw Switch to which message is sent.
- * @param flowObj FlowPath.
- * @param flowEntryObj FlowEntry.
- * @return true if succeed.
- */
@Override
public boolean add(IOFSwitch sw, IFlowPath flowObj, IFlowEntry flowEntryObj) {
log.debug("sending : {}, {}", sw, flowObj);
@@ -700,13 +745,6 @@
return true;
}
- /**
- * Create OFMessage from given flow information and add it to the queue.
- * @param sw Switch to which message is sent.
- * @param flowPath FlowPath.
- * @param flowEntry FlowEntry.
- * @return true if secceed.
- */
@Override
public boolean add(IOFSwitch sw, FlowPath flowPath, FlowEntry flowEntry) {
//
@@ -740,7 +778,6 @@
//
OFMatch match = new OFMatch();
match.setWildcards(OFMatch.OFPFW_ALL);
- FlowEntryMatch flowPathMatch = flowPath.flowEntryMatch();
FlowEntryMatch flowEntryMatch = flowEntry.flowEntryMatch();
// Match the Incoming Port
@@ -752,9 +789,6 @@
// Match the Source MAC address
MACAddress matchSrcMac = flowEntryMatch.srcMac();
- if ((matchSrcMac == null) && (flowPathMatch != null)) {
- matchSrcMac = flowPathMatch.srcMac();
- }
if (matchSrcMac != null) {
match.setDataLayerSource(matchSrcMac.toString());
match.setWildcards(match.getWildcards() & ~OFMatch.OFPFW_DL_SRC);
@@ -762,9 +796,6 @@
// Match the Destination MAC address
MACAddress matchDstMac = flowEntryMatch.dstMac();
- if ((matchDstMac == null) && (flowPathMatch != null)) {
- matchDstMac = flowPathMatch.dstMac();
- }
if (matchDstMac != null) {
match.setDataLayerDestination(matchDstMac.toString());
match.setWildcards(match.getWildcards() & ~OFMatch.OFPFW_DL_DST);
@@ -772,9 +803,6 @@
// Match the Ethernet Frame Type
Short matchEthernetFrameType = flowEntryMatch.ethernetFrameType();
- if ((matchEthernetFrameType == null) && (flowPathMatch != null)) {
- matchEthernetFrameType = flowPathMatch.ethernetFrameType();
- }
if (matchEthernetFrameType != null) {
match.setDataLayerType(matchEthernetFrameType);
match.setWildcards(match.getWildcards() & ~OFMatch.OFPFW_DL_TYPE);
@@ -782,9 +810,6 @@
// Match the VLAN ID
Short matchVlanId = flowEntryMatch.vlanId();
- if ((matchVlanId == null) && (flowPathMatch != null)) {
- matchVlanId = flowPathMatch.vlanId();
- }
if (matchVlanId != null) {
match.setDataLayerVirtualLan(matchVlanId);
match.setWildcards(match.getWildcards() & ~OFMatch.OFPFW_DL_VLAN);
@@ -792,9 +817,6 @@
// Match the VLAN priority
Byte matchVlanPriority = flowEntryMatch.vlanPriority();
- if ((matchVlanPriority == null) && (flowPathMatch != null)) {
- matchVlanPriority = flowPathMatch.vlanPriority();
- }
if (matchVlanPriority != null) {
match.setDataLayerVirtualLanPriorityCodePoint(matchVlanPriority);
match.setWildcards(match.getWildcards()
@@ -803,27 +825,18 @@
// Match the Source IPv4 Network prefix
IPv4Net matchSrcIPv4Net = flowEntryMatch.srcIPv4Net();
- if ((matchSrcIPv4Net == null) && (flowPathMatch != null)) {
- matchSrcIPv4Net = flowPathMatch.srcIPv4Net();
- }
if (matchSrcIPv4Net != null) {
match.setFromCIDR(matchSrcIPv4Net.toString(), OFMatch.STR_NW_SRC);
}
// Natch the Destination IPv4 Network prefix
IPv4Net matchDstIPv4Net = flowEntryMatch.dstIPv4Net();
- if ((matchDstIPv4Net == null) && (flowPathMatch != null)) {
- matchDstIPv4Net = flowPathMatch.dstIPv4Net();
- }
if (matchDstIPv4Net != null) {
match.setFromCIDR(matchDstIPv4Net.toString(), OFMatch.STR_NW_DST);
}
// Match the IP protocol
Byte matchIpProto = flowEntryMatch.ipProto();
- if ((matchIpProto == null) && (flowPathMatch != null)) {
- matchIpProto = flowPathMatch.ipProto();
- }
if (matchIpProto != null) {
match.setNetworkProtocol(matchIpProto);
match.setWildcards(match.getWildcards() & ~OFMatch.OFPFW_NW_PROTO);
@@ -831,9 +844,6 @@
// Match the IP ToS (DSCP field, 6 bits)
Byte matchIpToS = flowEntryMatch.ipToS();
- if ((matchIpToS == null) && (flowPathMatch != null)) {
- matchIpToS = flowPathMatch.ipToS();
- }
if (matchIpToS != null) {
match.setNetworkTypeOfService(matchIpToS);
match.setWildcards(match.getWildcards() & ~OFMatch.OFPFW_NW_TOS);
@@ -841,9 +851,6 @@
// Match the Source TCP/UDP port
Short matchSrcTcpUdpPort = flowEntryMatch.srcTcpUdpPort();
- if ((matchSrcTcpUdpPort == null) && (flowPathMatch != null)) {
- matchSrcTcpUdpPort = flowPathMatch.srcTcpUdpPort();
- }
if (matchSrcTcpUdpPort != null) {
match.setTransportSource(matchSrcTcpUdpPort);
match.setWildcards(match.getWildcards() & ~OFMatch.OFPFW_TP_SRC);
@@ -851,9 +858,6 @@
// Match the Destination TCP/UDP port
Short matchDstTcpUdpPort = flowEntryMatch.dstTcpUdpPort();
- if ((matchDstTcpUdpPort == null) && (flowPathMatch != null)) {
- matchDstTcpUdpPort = flowPathMatch.dstTcpUdpPort();
- }
if (matchDstTcpUdpPort != null) {
match.setTransportDestination(matchDstTcpUdpPort);
match.setWildcards(match.getWildcards() & ~OFMatch.OFPFW_TP_DST);
@@ -1049,11 +1053,8 @@
OFBarrierRequest msg = (OFBarrierRequest) factory.getMessage(OFType.BARRIER_REQUEST);
msg.setXid(sw.getNextTransactionId());
- add(sw, msg);
- // TODO create Future object of message
OFBarrierReplyFuture future = new OFBarrierReplyFuture(threadPool, sw, msg.getXid());
-
synchronized (barrierFutures) {
Map<Integer,OFBarrierReplyFuture> map = barrierFutures.get(sw.getId());
if (map == null) {
@@ -1061,12 +1062,18 @@
barrierFutures.put(sw.getId(), map);
}
map.put(msg.getXid(), future);
- log.debug("Inserted future for {}", msg.getXid());
}
+ add(sw, msg);
+
return future;
}
+ /**
+ * Get a queue attached to a switch.
+ * @param sw Switch object
+ * @return Queue object
+ */
protected SwitchQueue getQueue(IOFSwitch sw) {
if (sw == null) {
return null;
@@ -1075,12 +1082,22 @@
return getProcess(sw).queues.get(sw);
}
+ /**
+ * Get a hash value correspondent to a switch.
+ * @param sw Switch object
+ * @return Hash value
+ */
protected long getHash(IOFSwitch sw) {
// This code assumes DPID is sequentially assigned.
// TODO consider equalization algorithm
return sw.getId() % number_thread;
}
-
+
+ /**
+ * Get a Thread object which processes the queue attached to a switch.
+ * @param sw Switch object
+ * @return Thread object
+ */
protected FlowPusherThread getProcess(IOFSwitch sw) {
long hash = getHash(sw);
@@ -1106,11 +1123,13 @@
public Command receive(IOFSwitch sw, OFMessage msg, FloodlightContext cntx) {
Map<Integer,OFBarrierReplyFuture> map = barrierFutures.get(sw.getId());
if (map == null) {
+ log.debug("null map for {} : {}", sw.getId(), barrierFutures);
return Command.CONTINUE;
}
OFBarrierReplyFuture future = map.get(msg.getXid());
if (future == null) {
+ log.debug("null future for {} : {}", msg.getXid(), map);
return Command.CONTINUE;
}
@@ -1119,4 +1138,5 @@
return Command.CONTINUE;
}
+
}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowSynchronizer.java b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowSynchronizer.java
index f3f5c50..b3c8e84 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowSynchronizer.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/FlowSynchronizer.java
@@ -21,87 +21,68 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import net.floodlightcontroller.core.IFloodlightProviderService;
import net.floodlightcontroller.core.IOFSwitch;
-import net.floodlightcontroller.core.IOFSwitchListener;
-import net.floodlightcontroller.core.module.FloodlightModuleContext;
-import net.floodlightcontroller.core.module.FloodlightModuleException;
import net.onrc.onos.graph.GraphDBOperation;
import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.IFlowEntry;
import net.onrc.onos.ofcontroller.core.INetMapTopologyObjects.ISwitchObject;
import net.onrc.onos.ofcontroller.util.Dpid;
import net.onrc.onos.ofcontroller.util.FlowEntryId;
-import net.onrc.onos.registry.controller.IControllerRegistryService;
-public class FlowSynchronizer implements IFlowSyncService, IOFSwitchListener {
+/**
+ * FlowSynchronizer is an implementation of FlowSyncService.
+ * In addition to IFlowSyncService, FlowSynchronizer periodically reads flow
+ * tables from switches and compare them with GraphDB to drop unnecessary
+ * flows and/or to install missing flows.
+ * @author Brian
+ *
+ */
+public class FlowSynchronizer implements IFlowSyncService {
- protected static Logger log = LoggerFactory.getLogger(FlowSynchronizer.class);
- protected IFloodlightProviderService floodlightProvider;
- protected IControllerRegistryService registryService;
- protected IFlowPusherService pusher;
+ private static Logger log = LoggerFactory.getLogger(FlowSynchronizer.class);
private GraphDBOperation dbHandler;
- private Map<IOFSwitch, Thread> switchThread = new HashMap<IOFSwitch, Thread>();
+ protected IFlowPusherService pusher;
+ private Map<IOFSwitch, Thread> switchThreads;
public FlowSynchronizer() {
dbHandler = new GraphDBOperation("");
+ switchThreads = new HashMap<IOFSwitch, Thread>();
}
+ @Override
public void synchronize(IOFSwitch sw) {
- Synchroizer sync = new Synchroizer(sw);
+ Synchronizer sync = new Synchronizer(sw);
Thread t = new Thread(sync);
- switchThread.put(sw, t);
+ switchThreads.put(sw, t);
t.start();
}
-
+
@Override
- public void addedSwitch(IOFSwitch sw) {
- log.debug("Switch added: {}", sw.getId());
-
- if (registryService.hasControl(sw.getId())) {
- synchronize(sw);
- }
- }
-
- @Override
- public void removedSwitch(IOFSwitch sw) {
- log.debug("Switch removed: {}", sw.getId());
-
- Thread t = switchThread.remove(sw);
+ public void interrupt(IOFSwitch sw) {
+ Thread t = switchThreads.remove(sw);
if(t != null) {
t.interrupt();
- }
-
+ }
}
- @Override
- public void switchPortChanged(Long switchId) {
- // TODO Auto-generated method stub
+ /**
+ * Initialize Synchronizer.
+ * @param pusherService FlowPusherService used for sending messages.
+ */
+ public void init(IFlowPusherService pusherService) {
+ pusher = pusherService;
}
- @Override
- public String getName() {
- return "FlowSynchronizer";
- }
-
- //@Override
- public void init(FloodlightModuleContext context)
- throws FloodlightModuleException {
- floodlightProvider = context.getServiceImpl(IFloodlightProviderService.class);
- registryService = context.getServiceImpl(IControllerRegistryService.class);
- pusher = context.getServiceImpl(IFlowPusherService.class);
- }
-
- //@Override
- public void startUp(FloodlightModuleContext context) {
- floodlightProvider.addOFSwitchListener(this);
- }
-
- protected class Synchroizer implements Runnable {
+ /**
+ * Synchronizer represents main thread of synchronization.
+ * @author Brian
+ *
+ */
+ protected class Synchronizer implements Runnable {
IOFSwitch sw;
ISwitchObject swObj;
- public Synchroizer(IOFSwitch sw) {
+ public Synchronizer(IOFSwitch sw) {
this.sw = sw;
Dpid dpid = new Dpid(sw.getId());
this.swObj = dbHandler.searchSwitch(dpid.toString());
@@ -109,11 +90,21 @@
@Override
public void run() {
+ // TODO: stop adding other flow entries while synchronizing
+ //pusher.suspend(sw);
Set<FlowEntryWrapper> graphEntries = getFlowEntriesFromGraph();
Set<FlowEntryWrapper> switchEntries = getFlowEntriesFromSwitch();
compare(graphEntries, switchEntries);
+ //pusher.resume(sw);
}
+ /**
+ * Compare flows entries in GraphDB and switch to pick up necessary
+ * messages.
+ * After picking up, picked messages are added to FlowPusher.
+ * @param graphEntries Flow entries in GraphDB.
+ * @param switchEntries Flow entries in switch.
+ */
private void compare(Set<FlowEntryWrapper> graphEntries, Set<FlowEntryWrapper> switchEntries) {
int added = 0, removed = 0, skipped = 0;
for(FlowEntryWrapper entry : switchEntries) {
@@ -137,6 +128,10 @@
"Flow entries skipped " + skipped);
}
+ /**
+ * Read GraphDB to get FlowEntries associated with a switch.
+ * @return set of FlowEntries
+ */
private Set<FlowEntryWrapper> getFlowEntriesFromGraph() {
Set<FlowEntryWrapper> entries = new HashSet<FlowEntryWrapper>();
for(IFlowEntry entry : swObj.getFlowEntries()) {
@@ -146,6 +141,10 @@
return entries;
}
+ /**
+ * Read flow table from switch and derive FlowEntries from table.
+ * @return set of FlowEntries
+ */
private Set<FlowEntryWrapper> getFlowEntriesFromSwitch() {
int lengthU = 0;
@@ -192,48 +191,75 @@
}
+ /**
+ * FlowEntryWrapper represents abstract FlowEntry which is embodied
+ * by FlowEntryId (from GraphDB) or OFFlowStatisticsReply (from switch).
+ * @author Brian
+ *
+ */
class FlowEntryWrapper {
- FlowEntryId id;
- IFlowEntry iflowEntry;
+ FlowEntryId flowEntryId;
OFFlowStatisticsReply statisticsReply;
public FlowEntryWrapper(IFlowEntry entry) {
- iflowEntry = entry;
- id = new FlowEntryId(entry.getFlowEntryId());
+ flowEntryId = new FlowEntryId(entry.getFlowEntryId());
}
public FlowEntryWrapper(OFFlowStatisticsReply entry) {
+ flowEntryId = new FlowEntryId(entry.getCookie());
statisticsReply = entry;
- id = new FlowEntryId(entry.getCookie());
}
+ /**
+ * Install this FlowEntry to a switch via FlowPusher.
+ * @param sw Switch to which flow will be installed.
+ */
public void addToSwitch(IOFSwitch sw) {
- if(iflowEntry != null) {
- pusher.add(sw, iflowEntry.getFlow(), iflowEntry);
- }
- else if(statisticsReply != null) {
- log.error("Adding existing flow entry {} to sw {}",
+ if (statisticsReply != null) {
+ log.error("Error adding existing flow entry {} to sw {}",
statisticsReply.getCookie(), sw.getId());
+ return;
}
+
+ // Get the Flow Entry state from the Network Graph
+ IFlowEntry iFlowEntry = null;
+ try {
+ iFlowEntry = dbHandler.searchFlowEntry(flowEntryId);
+ } catch (Exception e) {
+ log.error("Error finding flow entry {} in Network Graph",
+ flowEntryId);
+ return;
+ }
+ if (iFlowEntry == null) {
+ log.error("Cannot add flow entry {} to sw {} : flow entry not found",
+ flowEntryId, sw.getId());
+ return;
+ }
+
+ pusher.add(sw, iFlowEntry.getFlow(), iFlowEntry);
}
- public void removeFromSwitch(IOFSwitch sw){
- if(iflowEntry != null) {
- log.error("Removing non-existent flow entry {} from sw {}",
- iflowEntry.getFlowEntryId(), sw.getId());
+ /**
+ * Remove this FlowEntry from a switch via FlowPusher.
+ * @param sw Switch from which flow will be removed.
+ */
+ public void removeFromSwitch(IOFSwitch sw) {
+ if (statisticsReply == null) {
+ log.error("Error removing non-existent flow entry {} from sw {}",
+ flowEntryId, sw.getId());
+ return;
+ }
- }
- else if(statisticsReply != null) {
- // Convert Statistics Reply to Flow Mod, then write it
- OFFlowMod fm = new OFFlowMod();
- fm.setCookie(statisticsReply.getCookie());
- fm.setCommand(OFFlowMod.OFPFC_DELETE_STRICT);
- fm.setLengthU(OFFlowMod.MINIMUM_LENGTH);
- fm.setMatch(statisticsReply.getMatch());
- fm.setPriority(statisticsReply.getPriority());
- fm.setOutPort(OFPort.OFPP_NONE);
- pusher.add(sw, fm);
- }
+ // Convert Statistics Reply to Flow Mod, then write it
+ OFFlowMod fm = new OFFlowMod();
+ fm.setCookie(statisticsReply.getCookie());
+ fm.setCommand(OFFlowMod.OFPFC_DELETE_STRICT);
+ fm.setLengthU(OFFlowMod.MINIMUM_LENGTH);
+ fm.setMatch(statisticsReply.getMatch());
+ fm.setPriority(statisticsReply.getPriority());
+ fm.setOutPort(OFPort.OFPP_NONE);
+
+ pusher.add(sw, fm);
}
/**
@@ -241,7 +267,7 @@
*/
@Override
public int hashCode() {
- return id.hashCode();
+ return flowEntryId.hashCode();
}
/**
@@ -249,22 +275,21 @@
* the same value; otherwise, returns false.
*
* @param Object to compare
+ * @return true if the object has the same Flow Entry ID.
*/
@Override
public boolean equals(Object obj){
if(obj.getClass() == this.getClass()) {
FlowEntryWrapper entry = (FlowEntryWrapper) obj;
// TODO: we need to actually compare the match + actions
- return this.id.equals(entry.id);
+ return this.flowEntryId.equals(entry.flowEntryId);
}
return false;
}
@Override
public String toString() {
- return id.toString();
+ return flowEntryId.toString();
}
}
}
-
-
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/IFlowPusherService.java b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/IFlowPusherService.java
index 94d6e35..20a6249 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/IFlowPusherService.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/IFlowPusherService.java
@@ -11,8 +11,44 @@
import net.onrc.onos.ofcontroller.util.FlowEntry;
import net.onrc.onos.ofcontroller.util.FlowPath;
+/**
+ * FlowPusherService is a service to send message to switches in proper rate.
+ * Conceptually a queue is attached to each switch, and FlowPusherService
+ * read a message from queue and send it to switch in order.
+ * To guarantee message has been installed, FlowPusherService can add barrier
+ * message to queue and can notify when barrier message is sent to switch.
+ * @author Naoki Shiota
+ *
+ */
public interface IFlowPusherService extends IFloodlightService {
/**
+ * Create a queue correspondent to the switch.
+ * @param sw Switch to which new queue is attached.
+ * @return true if new queue is successfully created.
+ */
+ boolean createQueue(IOFSwitch sw);
+
+ /**
+ * Delete a queue correspondent to the switch.
+ * Messages remains in queue will be all sent before queue is deleted.
+ * @param sw Switch of which queue is deleted.
+ * @return true if queue is successfully deleted.
+ */
+ boolean deleteQueue(IOFSwitch sw);
+
+ /**
+ * Delete a queue correspondent to the switch.
+ * By setting force flag on, queue will be deleted immediately.
+ * @param sw Switch of which queue is deleted.
+ * @param forceStop If this flag is set to true, queue will be deleted
+ * immediately regardless of any messages in the queue.
+ * If false, all messages will be sent to switch and queue will
+ * be deleted after that.
+ * @return true if queue is successfully deleted or flagged to be deleted.
+ */
+ boolean deleteQueue(IOFSwitch sw, boolean forceStop);
+
+ /**
* Add a message to the queue of the switch.
* @param sw Switch to which message is pushed.
* @param msg Message object to be added.
@@ -39,6 +75,13 @@
boolean add(IOFSwitch sw, IFlowPath flowObj, IFlowEntry flowEntryObj);
/**
+ * Set sending rate to a switch.
+ * @param sw Switch.
+ * @param rate Rate in bytes/ms.
+ */
+ public void setRate(IOFSwitch sw, long rate);
+
+ /**
* Add BARRIER message to queue and wait for reply.
* @param sw Switch to which barrier message is pushed.
* @return BARRIER_REPLY message sent from switch.
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/IFlowSyncService.java b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/IFlowSyncService.java
index 1e71f6a..4e6efaf 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/IFlowSyncService.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/IFlowSyncService.java
@@ -5,9 +5,13 @@
import net.floodlightcontroller.core.module.IFloodlightService;
/**
- * @author bocon
+ * FlowSyncService is a service to synchronize GraphDB and switch's flow table.
+ * FlowSyncService offers APIs to trigger and interrupt synchronization explicitly.
+ * @author Brian
*
*/
public interface IFlowSyncService extends IFloodlightService {
public void synchronize(IOFSwitch sw);
+
+ public void interrupt(IOFSwitch sw);
}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/FlowProgrammerWebRoutable.java b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/FlowProgrammerWebRoutable.java
new file mode 100644
index 0000000..21e5bfb
--- /dev/null
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/FlowProgrammerWebRoutable.java
@@ -0,0 +1,26 @@
+package net.onrc.onos.ofcontroller.flowprogrammer.web;
+
+import org.restlet.Context;
+import org.restlet.Restlet;
+import org.restlet.routing.Router;
+
+import net.floodlightcontroller.restserver.RestletRoutable;
+
+public class FlowProgrammerWebRoutable implements RestletRoutable {
+
+ @Override
+ public Restlet getRestlet(Context context) {
+ Router router = new Router(context);
+ router.attach("/setrate/{dpid}/{rate}/json", SetPushRateResource.class);
+ router.attach("/suspend/{dpid}/json", SuspendPusherResource.class);
+ router.attach("/resume/{dpid}/json", ResumePusherResource.class);
+ router.attach("/barrier/{dpid}/json", SendBarrierResource.class);
+ return router;
+ }
+
+ @Override
+ public String basePath() {
+ return "/wm/fprog";
+ }
+
+}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/PusherResource.java b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/PusherResource.java
new file mode 100644
index 0000000..4e1c0fc
--- /dev/null
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/PusherResource.java
@@ -0,0 +1,33 @@
+package net.onrc.onos.ofcontroller.flowprogrammer.web;
+
+import org.restlet.resource.ServerResource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import net.floodlightcontroller.core.IFloodlightProviderService;
+import net.onrc.onos.ofcontroller.flowprogrammer.IFlowPusherService;
+
+public class PusherResource extends ServerResource {
+ protected final static Logger log = LoggerFactory.getLogger(PusherResource.class);
+
+ protected IFloodlightProviderService provider;
+ protected IFlowPusherService pusher;
+
+ protected boolean init() {
+ provider = (IFloodlightProviderService)
+ getContext().getAttributes().
+ get(IFloodlightProviderService.class.getCanonicalName());
+ if (provider == null) {
+ log.debug("ONOS FloodlightProvider not found");
+ return false;
+ }
+
+ pusher = (IFlowPusherService)getContext().getAttributes().
+ get(IFlowPusherService.class.getCanonicalName());
+ if (pusher == null) {
+ log.debug("ONOS FlowPusherService not found");
+ return false;
+ }
+ return true;
+ }
+}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/ResumePusherResource.java b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/ResumePusherResource.java
new file mode 100644
index 0000000..ca1ec00
--- /dev/null
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/ResumePusherResource.java
@@ -0,0 +1,41 @@
+package net.onrc.onos.ofcontroller.flowprogrammer.web;
+
+import net.floodlightcontroller.core.IOFSwitch;
+
+import org.openflow.util.HexString;
+import org.restlet.resource.Get;
+
+/**
+ * FlowProgrammer REST API implementation: Resume sending message to switch.
+ *
+ * GET /wm/fprog/resume/{dpid}/json"
+ */
+public class ResumePusherResource extends PusherResource {
+ /**
+ * Implement the API.
+ *
+ * @return true if succeeded, false if failed.
+ */
+ @Get("json")
+ public boolean retrieve() {
+ if (! init()) {
+ return false;
+ }
+
+ long dpid;
+ try {
+ dpid = HexString.toLong((String)getRequestAttributes().get("dpid"));
+ } catch (NumberFormatException e) {
+ log.error("Invalid number format");
+ return false;
+ }
+
+ IOFSwitch sw = provider.getSwitches().get(dpid);
+ if (sw == null) {
+ log.error("Invalid dpid");
+ return false;
+ }
+
+ return pusher.resume(sw);
+ }
+}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/SendBarrierResource.java b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/SendBarrierResource.java
new file mode 100644
index 0000000..9c348ff
--- /dev/null
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/SendBarrierResource.java
@@ -0,0 +1,41 @@
+package net.onrc.onos.ofcontroller.flowprogrammer.web;
+
+import net.floodlightcontroller.core.IOFSwitch;
+
+import org.openflow.protocol.OFBarrierReply;
+import org.openflow.util.HexString;
+import org.restlet.resource.Get;
+
+/**
+ * FlowProgrammer REST API implementation: Send barrier message to switch.
+ *
+ * GET /wm/fprog/barrier/{dpid}/json"
+ */
+public class SendBarrierResource extends PusherResource {
+ /**
+ * Implement the API.
+ *
+ * @return true if succeeded, false if failed.
+ */
+ @Get("json")
+ public OFBarrierReply retrieve() {
+ if (! init()) {
+ return null;
+ }
+ long dpid;
+ try {
+ dpid = HexString.toLong((String)getRequestAttributes().get("dpid"));
+ } catch (NumberFormatException e) {
+ log.error("Invalid number format");
+ return null;
+ }
+
+ IOFSwitch sw = provider.getSwitches().get(dpid);
+ if (sw == null) {
+ log.error("Invalid dpid");
+ return null;
+ }
+
+ return pusher.barrier(sw);
+ }
+}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/SetPushRateResource.java b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/SetPushRateResource.java
new file mode 100644
index 0000000..08a728e
--- /dev/null
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/SetPushRateResource.java
@@ -0,0 +1,47 @@
+package net.onrc.onos.ofcontroller.flowprogrammer.web;
+
+import net.floodlightcontroller.core.IOFSwitch;
+
+import org.openflow.util.HexString;
+import org.restlet.resource.Get;
+
+/**
+ * FlowProgrammer REST API implementation: Set sending rate to the switch.
+ *
+ * GET /wm/fprog/setrate/{dpid}/{rate}/json"
+ */
+public class SetPushRateResource extends PusherResource {
+
+ /**
+ * Implement the API.
+ *
+ * @return true if succeeded, false if failed.
+ */
+ @Get("json")
+ public boolean retrieve() {
+ if (! init()) {
+ return false;
+ }
+
+ long dpid;
+ long rate;
+
+ try {
+ dpid = HexString.toLong((String)getRequestAttributes().get("dpid"));
+ rate = Long.valueOf((String)getRequestAttributes().get("rate"));
+ } catch (NumberFormatException e) {
+ log.error("Invalid number format");
+ return false;
+ }
+
+ IOFSwitch sw = provider.getSwitches().get(dpid);
+ if (sw == null) {
+ log.error("Invalid dpid");
+ return false;
+ }
+
+ pusher.setRate(sw, rate);
+
+ return true;
+ }
+}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/SuspendPusherResource.java b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/SuspendPusherResource.java
new file mode 100644
index 0000000..39d245b
--- /dev/null
+++ b/src/main/java/net/onrc/onos/ofcontroller/flowprogrammer/web/SuspendPusherResource.java
@@ -0,0 +1,46 @@
+package net.onrc.onos.ofcontroller.flowprogrammer.web;
+
+import net.floodlightcontroller.core.IOFSwitch;
+
+import org.openflow.util.HexString;
+import org.restlet.resource.Get;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * FlowProgrammer REST API implementation: Suspend sending message to switch.
+ *
+ * GET /wm/fprog/suspend/{dpid}/json"
+ */
+public class SuspendPusherResource extends PusherResource {
+
+ protected final static Logger log = LoggerFactory.getLogger(SetPushRateResource.class);
+
+ /**
+ * Implement the API.
+ *
+ * @return true if succeeded, false if failed.
+ */
+ @Get("json")
+ public boolean retrieve() {
+ if (! init()) {
+ return false;
+ }
+
+ long dpid;
+ try {
+ dpid = HexString.toLong((String)getRequestAttributes().get("dpid"));
+ } catch (NumberFormatException e) {
+ log.error("Invalid number format");
+ return false;
+ }
+
+ IOFSwitch sw = provider.getSwitches().get(dpid);
+ if (sw == null) {
+ log.error("Invalid dpid");
+ return false;
+ }
+
+ return pusher.suspend(sw);
+ }
+}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/util/FlowEntry.java b/src/main/java/net/onrc/onos/ofcontroller/util/FlowEntry.java
index 15a6233..98dbd88 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/util/FlowEntry.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/util/FlowEntry.java
@@ -136,10 +136,11 @@
*
* @return true if the Flow ID is valid, otherwise false.
*/
+ @JsonIgnore
public boolean isValidFlowId() {
if (this.flowId == null)
return false;
- return (this.flowId.value() != 0);
+ return (this.flowId.isValid());
}
/**
@@ -165,10 +166,11 @@
*
* @return true if the Flow Entry ID is valid, otherwise false.
*/
+ @JsonIgnore
public boolean isValidFlowEntryId() {
if (this.flowEntryId == null)
return false;
- return (this.flowEntryId.value() != 0);
+ return (this.flowEntryId.isValid());
}
/**
diff --git a/src/main/java/net/onrc/onos/ofcontroller/util/FlowEntryActions.java b/src/main/java/net/onrc/onos/ofcontroller/util/FlowEntryActions.java
index 53aab66..7d9688b 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/util/FlowEntryActions.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/util/FlowEntryActions.java
@@ -2,6 +2,7 @@
import java.util.ArrayList;
+import org.codehaus.jackson.annotate.JsonIgnore;
import org.codehaus.jackson.annotate.JsonProperty;
/**
@@ -79,6 +80,7 @@
*
* @return true if the set of actions is empty, otherwise false.
*/
+ @JsonIgnore
public Boolean isEmpty() {
return actions.isEmpty();
}
diff --git a/src/main/java/net/onrc/onos/ofcontroller/util/FlowEntryId.java b/src/main/java/net/onrc/onos/ofcontroller/util/FlowEntryId.java
index 29efe4c..f5728b0 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/util/FlowEntryId.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/util/FlowEntryId.java
@@ -5,6 +5,7 @@
import net.onrc.onos.ofcontroller.util.serializers.FlowEntryIdDeserializer;
import net.onrc.onos.ofcontroller.util.serializers.FlowEntryIdSerializer;
+import org.codehaus.jackson.annotate.JsonIgnore;
import org.codehaus.jackson.map.annotate.JsonDeserialize;
import org.codehaus.jackson.map.annotate.JsonSerialize;
@@ -20,7 +21,7 @@
* Default constructor.
*/
public FlowEntryId() {
- this.value = 0;
+ this.value = -1;
}
/**
@@ -66,7 +67,17 @@
public void setValue(long value) {
this.value = value;
}
-
+
+ /**
+ * Test whether the Flow Entry ID is valid.
+ *
+ * @return true if the Flow Entry ID is valid, otherwise false.
+ */
+ @JsonIgnore
+ public boolean isValid() {
+ return (this.value() != -1);
+ }
+
/**
* Returns true of the object is another Flow Entry ID with
* the same value; otherwise, returns false.
diff --git a/src/main/java/net/onrc/onos/ofcontroller/util/FlowId.java b/src/main/java/net/onrc/onos/ofcontroller/util/FlowId.java
index de955ba..d90e96f 100644
--- a/src/main/java/net/onrc/onos/ofcontroller/util/FlowId.java
+++ b/src/main/java/net/onrc/onos/ofcontroller/util/FlowId.java
@@ -5,6 +5,7 @@
import net.onrc.onos.ofcontroller.util.serializers.FlowIdDeserializer;
import net.onrc.onos.ofcontroller.util.serializers.FlowIdSerializer;
+import org.codehaus.jackson.annotate.JsonIgnore;
import org.codehaus.jackson.map.annotate.JsonDeserialize;
import org.codehaus.jackson.map.annotate.JsonSerialize;
@@ -20,7 +21,7 @@
* Default constructor.
*/
public FlowId() {
- this.value = 0;
+ this.value = -1;
}
/**
@@ -68,6 +69,16 @@
}
/**
+ * Test whether the Flow ID is valid.
+ *
+ * @return true if the Flow ID is valid, otherwise false.
+ */
+ @JsonIgnore
+ public boolean isValid() {
+ return (this.value() != -1);
+ }
+
+ /**
* Convert the Flow ID value to a hexadecimal string.
*
* @return the Flow ID value to a hexadecimal string.
diff --git a/src/test/java/net/onrc/onos/ofcontroller/flowmanager/FlowManagerTest.java b/src/test/java/net/onrc/onos/ofcontroller/flowmanager/FlowManagerTest.java
index c2db225..bbe7e31 100644
--- a/src/test/java/net/onrc/onos/ofcontroller/flowmanager/FlowManagerTest.java
+++ b/src/test/java/net/onrc/onos/ofcontroller/flowmanager/FlowManagerTest.java
@@ -463,35 +463,35 @@
*/
@Test
public final void testGetAllFlowsSummarySuccessNormally() throws Exception {
- final String getAllFlowsWithoutFlowEntries = "getAllFlowsWithoutFlowEntries";
+ final String getAllFlowsWithDataPathSummary = "getAllFlowsWithDataPathSummary";
// create mock objects
- FlowManager fm = createPartialMockAndInvokeDefaultConstructor(FlowManager.class, getAllFlowsWithoutFlowEntries);
- IFlowPath flowPath1 = createIFlowPathMock(1, "", "FP_TYPE_SHORTEST_PATH", "FP_USER_ADD", 0, 1, 2, 3, 4);
- IFlowPath flowPath2 = createIFlowPathMock(5, "", "FP_TYPE_SHORTEST_PATH", "FP_USER_ADD", 0, 2, 3, 4, 5);
- IFlowPath flowPath3 = createIFlowPathMock(10, "", "FP_TYPE_SHORTEST_PATH", "FP_USER_ADD", 0, 3, 4, 5, 6);
+ FlowManager fm = createPartialMockAndInvokeDefaultConstructor(FlowManager.class, getAllFlowsWithDataPathSummary);
+ FlowPath flowPath1 = createTestFlowPath(1, "", "FP_TYPE_SHORTEST_PATH", "FP_USER_ADD", 0, 1, 2, 3, 4);
+ FlowPath flowPath2 = createTestFlowPath(5, "", "FP_TYPE_SHORTEST_PATH", "FP_USER_ADD", 0, 2, 3, 4, 5);
+ FlowPath flowPath3 = createTestFlowPath(10, "", "FP_TYPE_SHORTEST_PATH", "FP_USER_ADD", 0, 3, 4, 5, 6);
// instantiate required objects
- ArrayList<IFlowPath> flows = new ArrayList<IFlowPath>();
+ ArrayList<FlowPath> flows = new ArrayList<FlowPath>();
flows.add(flowPath3);
flows.add(flowPath1);
flows.add(flowPath2);
// setup expectations
expectInitWithContext();
- expectPrivate(fm, getAllFlowsWithoutFlowEntries).andReturn(flows);
+ expectPrivate(fm, getAllFlowsWithDataPathSummary).andReturn(flows);
// start the test
replayAll();
fm.init(context);
- ArrayList<IFlowPath> returnedFlows = fm.getAllFlowsSummary(null, 0);
+ ArrayList<FlowPath> returnedFlows = fm.getAllFlowsSummary(null, 0);
// verify the test
verifyAll();
assertEquals(3, returnedFlows.size());
- assertEquals(1, new FlowId(returnedFlows.get(0).getFlowId()).value());
- assertEquals(5, new FlowId(returnedFlows.get(1).getFlowId()).value());
- assertEquals(10, new FlowId(returnedFlows.get(2).getFlowId()).value());
+ assertEquals(1, new FlowId(returnedFlows.get(0).flowId().value()).value());
+ assertEquals(5, new FlowId(returnedFlows.get(1).flowId().value()).value());
+ assertEquals(10, new FlowId(returnedFlows.get(2).flowId().value()).value());
}
/**
@@ -786,41 +786,6 @@
// other methods
/**
- * Test method for {@link FlowManager#getAllFlowsWithoutFlowEntries()}.
- * @throws Exception
- */
- @Test
- public final void testGetAllFlowsWithoutFlowEntriesSuccessNormally() throws Exception {
- // create mock objects
- IFlowPath iFlowPath1 = createIFlowPathMock(1, "caller id", "FP_TYPE_SHORTEST_PATH", "FP_USER_ADD", 0, 1, 1, 2, 2);
- IFlowPath iFlowPath2 = createIFlowPathMock(2, "caller id", "FP_TYPE_SHORTEST_PATH", "FP_USER_ADD", 0, 2, 5, 3, 5);
-
- // instantiate required objects
- ArrayList<IFlowPath> flowPaths = new ArrayList<IFlowPath>();
- flowPaths.add(iFlowPath1);
- flowPaths.add(iFlowPath2);
- FlowManager fm = new FlowManager();
-
- // setup expectations
- expectInitWithContext();
- op.commit();
- expect(op.getAllFlowPaths()).andReturn(flowPaths);
-
- // start the test
- replayAll();
-
- fm.init(context);
- ArrayList<IFlowPath> result = fm.getAllFlowsWithoutFlowEntries();
-
- // verify the test
- verifyAll();
- assertEquals(iFlowPath1, result.get(0));
- assertEquals(iFlowPath2, result.get(1));
-
- // TODO: does this method just return the replica of the flow paths?
- }
-
- /**
* Test method for {@link FlowManager#reconcileFlow(IFlowPath, DataPath)}.
* @throws Exception
*/
diff --git a/web/ons-demo/js/app.js b/web/ons-demo/js/app.js
index 94c41e2..d869de7 100644
--- a/web/ons-demo/js/app.js
+++ b/web/ons-demo/js/app.js
@@ -1,6 +1,17 @@
/*global d3, document∆*/
+function updateFlow(model) {
+ model.flows.forEach(function (flow) {
+ flow.flowId = flow.flowId.value;
+ flow.installerId = flow.installerId.value;
+ flow.dstDpid = flow.dataPath.dstPort.dpid.value;
+ flow.srcDpid = flow.dataPath.srcPort.dpid.value;
+ flow.dstPort = flow.dataPath.dstPort.port.value;
+ flow.srcPort = flow.dataPath.srcPort.port.value;
+ });
+}
+
function sync() {
var d = Date.now();
@@ -8,6 +19,7 @@
// console.log('Update time: ' + (Date.now() - d)/1000 + 's');
if (newModel) {
+ updateFlow(newModel);
var modelChanged = false;
var newModelString = JSON.stringify(newModel);
if (!modelString || newModelString != modelString) {
diff --git a/web/pusher.py b/web/pusher.py
new file mode 100755
index 0000000..f53e3ea
--- /dev/null
+++ b/web/pusher.py
@@ -0,0 +1,145 @@
+#! /usr/bin/env python
+
+
+import pprint
+import os
+import sys
+import subprocess
+import json
+import argparse
+import io
+import time
+
+from flask import Flask, json, Response, render_template, make_response, request
+
+## Global Var ##
+ControllerIP="127.0.0.1"
+ControllerPort=8080
+
+DEBUG=0
+pp = pprint.PrettyPrinter(indent=4)
+
+app = Flask(__name__)
+
+## Worker Functions ##
+def log_error(txt):
+ print '%s' % (txt)
+
+def debug(txt):
+ if DEBUG:
+ print '%s' % (txt)
+
+# @app.route("/wm/fprog/setrate/<dpid>/<rate>/json")
+# Sample output:
+# "true"
+
+
+def set_rate(dpid,rate):
+ try:
+ command = "curl -s \"http://%s:%s/wm/fprog/setrate/%s/%s/json\"" % (ControllerIP, ControllerPort, dpid, rate)
+ debug("set_rate %s" % command)
+
+ result = os.popen(command).read()
+ debug("result %s" % result)
+ if result == "false":
+ print "Failed to set rate"
+ return;
+ except:
+ log_error("Controller IF has issue")
+ exit(1)
+
+ print "Sending rate to %s is successfully set to %s" % (dpid, rate)
+
+def suspend(dpid):
+ try:
+ command = "curl -s \"http://%s:%s/wm/fprog/suspend/%s/json\"" % (ControllerIP, ControllerPort, dpid)
+ debug("suspend %s" % command)
+
+ result = os.popen(command).read()
+ debug("result %s" % result)
+ if result == "false":
+ print "Failed to suspend"
+ return;
+ except:
+ log_error("Controller IF has issue")
+ exit(1)
+
+ print "DPID %s is successfully suspended" % dpid
+
+def resume(dpid):
+ try:
+ command = "curl -s \"http://%s:%s/wm/fprog/resume/%s/json\"" % (ControllerIP, ControllerPort, dpid)
+ debug("resume %s" % command)
+
+ result = os.popen(command).read()
+ debug("result %s" % result)
+ if result == "false":
+ print "Failed to resume"
+ return;
+ except:
+ log_error("Controller IF has issue")
+ exit(1)
+
+ print "DPID %s is successfully resumed" % dpid
+
+def barrier(dpid):
+ try:
+ command = "curl -s \"http://%s:%s/wm/fprog/barrier/%s/json\"" % (ControllerIP, ControllerPort, dpid)
+ debug("barrier %s" % command)
+
+ result = os.popen(command).read()
+ debug("result %s" % result)
+ if result == "false":
+ print "Failed to send barrier"
+ return;
+ except:
+ log_error("Controller IF has issue")
+ exit(1)
+
+ print "Barrier reply from %s : %s" % (dpid, result)
+
+
+if __name__ == "__main__":
+ usage_msg1 = "Usage:\n"
+ usage_msg2 = "%s rate <dpid> <rate> : Set sending rate[bytes/ms] to the switch\n" % (sys.argv[0])
+ usage_msg3 = " suspend <dpid> : Suspend sending message to the switch\n"
+ usage_msg4 = " resume <dpid> : Resume sending message to the switch\n"
+ usage_msg5 = " barrier <dpid> : Send barrier message to the switch\n"
+ usage_msg = usage_msg1 + usage_msg2 + usage_msg3 + usage_msg4 + usage_msg5;
+
+ app.debug = True;
+
+ # Usage info
+ if len(sys.argv) > 1 and (sys.argv[1] == "-h" or sys.argv[1] == "--help"):
+ print(usage_msg)
+ exit(0)
+
+ # Check arguments
+ if len(sys.argv) < 2:
+ log_error(usage_msg)
+ exit(1)
+
+ # Do the work
+ if sys.argv[1] == "rate":
+ if len(sys.argv) < 4:
+ log_error(usage_msg)
+ exit(1)
+ set_rate(sys.argv[2], sys.argv[3])
+ elif sys.argv[1] == "suspend":
+ if len(sys.argv) < 3:
+ log_error(usage_msg)
+ exit(1)
+ suspend(sys.argv[2])
+ elif sys.argv[1] == "resume":
+ if len(sys.argv) < 3:
+ log_error(usage_msg)
+ exit(1)
+ resume(sys.argv[2])
+ elif sys.argv[1] == "barrier":
+ if len(sys.argv) < 3:
+ log_error(usage_msg)
+ exit(1)
+ barrier(sys.argv[2])
+ else:
+ log_error(usage_msg)
+ exit(1)
diff --git a/web/topology_rest.py b/web/topology_rest.py
index bac3113..b3a415e 100755
--- a/web/topology_rest.py
+++ b/web/topology_rest.py
@@ -720,8 +720,8 @@
# print "Debug: Controller command %s called %s" % (cmd, controller_name)
else:
# No longer use -i to specify keys (use .ssh/config to specify it)
- start_onos="ssh %s ONOS/start-onos.sh start" % (controller_name)
- stop_onos="ssh %s ONOS/start-onos.sh stop" % (controller_name)
+ start_onos="ssh %s \"cd ONOS; ./start-onos.sh start\"" % (controller_name)
+ stop_onos="ssh %s \"cd ONOS; ./start-onos.sh stop\"" % (controller_name)
# start_onos="ssh -i ~/.ssh/onlabkey.pem %s ONOS/start-onos.sh start" % (controller_name)
# stop_onos="ssh -i ~/.ssh/onlabkey.pem %s ONOS/start-onos.sh stop" % (controller_name)
@@ -960,7 +960,7 @@
parsedResult = json.loads(result)
if len(parsedResult) > 0:
if parsedResult[-1].has_key('flowId'):
- flow_nr = int(parsedResult[-1]['flowId'], 16)
+ flow_nr = int(parsedResult[-1]['flowId']['value'], 16)
else:
flow_nr = -1 # first flow
print "first flow"