Renamed the intent package
net.onrc.onos.intent.* => net.onrc.onos.core.intent.*
Change-Id: Id61f79ed52acf3b91af4ebad2515ac5b7d6dc5e1
diff --git a/src/main/java/net/onrc/onos/core/intent/runtime/IPathCalcRuntimeService.java b/src/main/java/net/onrc/onos/core/intent/runtime/IPathCalcRuntimeService.java
new file mode 100644
index 0000000..c6aad89
--- /dev/null
+++ b/src/main/java/net/onrc/onos/core/intent/runtime/IPathCalcRuntimeService.java
@@ -0,0 +1,15 @@
+package net.onrc.onos.core.intent.runtime;
+
+import net.floodlightcontroller.core.module.IFloodlightService;
+import net.onrc.onos.core.intent.IntentMap;
+import net.onrc.onos.core.intent.IntentOperationList;
+
+/**
+ * @author Toshio Koide (t-koide@onlab.us)
+ */
+public interface IPathCalcRuntimeService extends IFloodlightService {
+ public IntentOperationList executeIntentOperations(IntentOperationList list);
+ public IntentMap getHighLevelIntents();
+ public IntentMap getPathIntents();
+ public void purgeIntents();
+}
diff --git a/src/main/java/net/onrc/onos/core/intent/runtime/IntentStateList.java b/src/main/java/net/onrc/onos/core/intent/runtime/IntentStateList.java
new file mode 100644
index 0000000..e6f1180
--- /dev/null
+++ b/src/main/java/net/onrc/onos/core/intent/runtime/IntentStateList.java
@@ -0,0 +1,12 @@
+package net.onrc.onos.core.intent.runtime;
+
+import java.util.HashMap;
+
+import net.onrc.onos.core.intent.Intent.IntentState;
+
+/**
+ * @author Toshio Koide (t-koide@onlab.us)
+ */
+public class IntentStateList extends HashMap<String, IntentState> {
+ private static final long serialVersionUID = -3674903999581438936L;
+}
diff --git a/src/main/java/net/onrc/onos/core/intent/runtime/PathCalcRuntime.java b/src/main/java/net/onrc/onos/core/intent/runtime/PathCalcRuntime.java
new file mode 100644
index 0000000..89a0b1a
--- /dev/null
+++ b/src/main/java/net/onrc/onos/core/intent/runtime/PathCalcRuntime.java
@@ -0,0 +1,141 @@
+package net.onrc.onos.core.intent.runtime;
+
+import java.util.HashMap;
+
+import net.floodlightcontroller.core.module.IFloodlightService;
+import net.onrc.onos.core.intent.ConstrainedBFSTree;
+import net.onrc.onos.core.intent.ConstrainedShortestPathIntent;
+import net.onrc.onos.core.intent.ErrorIntent;
+import net.onrc.onos.core.intent.Intent;
+import net.onrc.onos.core.intent.IntentMap;
+import net.onrc.onos.core.intent.IntentOperation;
+import net.onrc.onos.core.intent.IntentOperationList;
+import net.onrc.onos.core.intent.PathIntent;
+import net.onrc.onos.core.intent.PathIntentMap;
+import net.onrc.onos.core.intent.ShortestPathIntent;
+import net.onrc.onos.core.intent.ErrorIntent.ErrorType;
+import net.onrc.onos.core.intent.Intent.IntentState;
+import net.onrc.onos.core.intent.IntentOperation.Operator;
+import net.onrc.onos.ofcontroller.networkgraph.NetworkGraph;
+import net.onrc.onos.ofcontroller.networkgraph.Path;
+import net.onrc.onos.ofcontroller.networkgraph.Switch;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author Toshio Koide (t-koide@onlab.us)
+ */
+public class PathCalcRuntime implements IFloodlightService {
+ private NetworkGraph graph;
+ private final static Logger log = LoggerFactory.getLogger(PathCalcRuntime.class);
+ public PathCalcRuntime(NetworkGraph g) {
+ this.graph = g;
+ }
+
+ /**
+ * calculate shortest-path and constrained-shortest-path intents into low-level path intents
+ * @param intentOpList IntentOperationList having instances of ShortestPathIntent/ConstrainedShortestPathIntent
+ * @param pathIntents a set of current low-level intents
+ * @return IntentOperationList. PathIntent and/or ErrorIntent instances.
+ */
+ public IntentOperationList calcPathIntents(final IntentOperationList intentOpList, final IntentMap appIntents, final PathIntentMap pathIntents) {
+ IntentOperationList pathIntentOpList = new IntentOperationList();
+ HashMap<Switch, ConstrainedBFSTree> spfTrees = new HashMap<>();
+
+ // TODO optimize locking of NetworkGraph
+ graph.acquireReadLock();
+ log.debug("NetworkGraph: {}", graph.getLinks());
+
+ for (IntentOperation intentOp: intentOpList) {
+ switch (intentOp.operator) {
+ case ADD:
+ if (!(intentOp.intent instanceof ShortestPathIntent)) {
+ log.error("Unsupported intent type: {}", intentOp.intent.getClass().getName());
+ pathIntentOpList.add(Operator.ERROR, new ErrorIntent(
+ ErrorType.UNSUPPORTED_INTENT,
+ "Unsupported intent type.",
+ intentOp.intent));
+ continue;
+ }
+
+ ShortestPathIntent spIntent = (ShortestPathIntent) intentOp.intent;
+ Switch srcSwitch = graph.getSwitch(spIntent.getSrcSwitchDpid());
+ Switch dstSwitch = graph.getSwitch(spIntent.getDstSwitchDpid());
+ if (srcSwitch == null || dstSwitch == null) {
+ log.error("Switch not found. src:{}, dst:{}, NetworkGraph:{}",
+ spIntent.getSrcSwitchDpid(),
+ spIntent.getDstSwitchDpid(),
+ graph.getLinks());
+ pathIntentOpList.add(Operator.ERROR, new ErrorIntent(
+ ErrorType.SWITCH_NOT_FOUND,
+ "Switch not found.",
+ spIntent));
+ continue;
+ }
+
+ double bandwidth = 0.0;
+ ConstrainedBFSTree tree = null;
+ if (spIntent instanceof ConstrainedShortestPathIntent) {
+ bandwidth = ((ConstrainedShortestPathIntent) intentOp.intent).getBandwidth();
+ tree = new ConstrainedBFSTree(srcSwitch, pathIntents, bandwidth);
+ }
+ else {
+ tree = spfTrees.get(srcSwitch);
+ if (tree == null) {
+ tree = new ConstrainedBFSTree(srcSwitch);
+ spfTrees.put(srcSwitch, tree);
+ }
+ }
+ Path path = tree.getPath(dstSwitch);
+ if (path == null) {
+ log.error("Path not found. Intent: {}, NetworkGraph:{}", spIntent.toString(), graph.getLinks());
+ pathIntentOpList.add(Operator.ERROR, new ErrorIntent(
+ ErrorType.PATH_NOT_FOUND,
+ "Path not found.",
+ spIntent));
+ continue;
+ }
+
+ // generate new path-intent ID
+ String oldPathIntentId = spIntent.getPathIntentId();
+ String newPathIntentId;
+ if (oldPathIntentId == null)
+ newPathIntentId = PathIntent.createFirstId(spIntent.getId());
+ else {
+ newPathIntentId = PathIntent.createNextId(oldPathIntentId);
+
+ // Request removal of low-level intent if it exists.
+ pathIntentOpList.add(Operator.REMOVE, new Intent(oldPathIntentId));
+ }
+
+ // create new path-intent
+ PathIntent pathIntent = new PathIntent(newPathIntentId, path, bandwidth, spIntent);
+ pathIntent.setState(IntentState.INST_REQ);
+ spIntent.setPathIntent(pathIntent);
+ pathIntentOpList.add(Operator.ADD, pathIntent);
+
+ break;
+ case REMOVE:
+ ShortestPathIntent targetAppIntent = (ShortestPathIntent) appIntents.getIntent(intentOp.intent.getId());
+ if (targetAppIntent != null) {
+ String pathIntentId = targetAppIntent.getPathIntentId();
+ if (pathIntentId != null) {
+ Intent targetPathIntent = pathIntents.getIntent(pathIntentId);
+ if (targetPathIntent != null) {
+ pathIntentOpList.add(Operator.REMOVE, targetPathIntent);
+ }
+ }
+ }
+ break;
+ case ERROR:
+ // just ignore
+ break;
+ }
+ }
+ // TODO optimize locking of NetworkGraph
+ graph.releaseReadLock();
+
+ return pathIntentOpList;
+ }
+}
\ No newline at end of file
diff --git a/src/main/java/net/onrc/onos/core/intent/runtime/PathCalcRuntimeModule.java b/src/main/java/net/onrc/onos/core/intent/runtime/PathCalcRuntimeModule.java
new file mode 100755
index 0000000..9f6353e
--- /dev/null
+++ b/src/main/java/net/onrc/onos/core/intent/runtime/PathCalcRuntimeModule.java
@@ -0,0 +1,407 @@
+package net.onrc.onos.core.intent.runtime;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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.onrc.onos.core.datagrid.IDatagridService;
+import net.onrc.onos.core.datagrid.IEventChannel;
+import net.onrc.onos.core.datagrid.IEventChannelListener;
+import net.onrc.onos.core.intent.Intent;
+import net.onrc.onos.core.intent.IntentMap;
+import net.onrc.onos.core.intent.IntentOperation;
+import net.onrc.onos.core.intent.IntentOperationList;
+import net.onrc.onos.core.intent.PathIntent;
+import net.onrc.onos.core.intent.PathIntentMap;
+import net.onrc.onos.core.intent.ShortestPathIntent;
+import net.onrc.onos.core.intent.Intent.IntentState;
+import net.onrc.onos.core.intent.IntentOperation.Operator;
+import net.onrc.onos.ofcontroller.networkgraph.DeviceEvent;
+import net.onrc.onos.ofcontroller.networkgraph.INetworkGraphListener;
+import net.onrc.onos.ofcontroller.networkgraph.INetworkGraphService;
+import net.onrc.onos.ofcontroller.networkgraph.LinkEvent;
+import net.onrc.onos.ofcontroller.networkgraph.PortEvent;
+import net.onrc.onos.ofcontroller.networkgraph.SwitchEvent;
+import net.onrc.onos.registry.controller.IControllerRegistryService;
+
+/**
+ * @author Toshio Koide (t-koide@onlab.us)
+ */
+public class PathCalcRuntimeModule implements IFloodlightModule, IPathCalcRuntimeService, INetworkGraphListener, IEventChannelListener<Long, IntentStateList> {
+ class PerfLog {
+ private String step;
+ private long time;
+
+ public PerfLog(String step) {
+ this.step = step;
+ this.time = System.nanoTime();
+ }
+
+ public void logThis() {
+ log.error("Time:{}, Step:{}", time, step);
+ }
+ }
+ class PerfLogger {
+ private LinkedList<PerfLog> logData = new LinkedList<>();
+
+ public PerfLogger(String logPhase) {
+ log("start_" + logPhase);
+ }
+
+ public void log(String step) {
+ logData.add(new PerfLog(step));
+ }
+
+ public void flushLog() {
+ log("finish");
+ for (PerfLog log: logData) {
+ log.logThis();
+ }
+ logData.clear();
+ }
+ }
+ private PathCalcRuntime runtime;
+ private IDatagridService datagridService;
+ private INetworkGraphService networkGraphService;
+ private IntentMap highLevelIntents;
+ private PathIntentMap pathIntents;
+ private IControllerRegistryService controllerRegistry;
+ private PersistIntent persistIntent;
+
+ private IEventChannel<Long, IntentOperationList> opEventChannel;
+ private final ReentrantLock lock = new ReentrantLock();
+ private HashSet<LinkEvent> unmatchedLinkEvents = new HashSet<>();
+ private static final String INTENT_OP_EVENT_CHANNEL_NAME = "onos.pathintent";
+ private static final String INTENT_STATE_EVENT_CHANNEL_NAME = "onos.pathintent_state";
+ private static final Logger log = LoggerFactory.getLogger(PathCalcRuntimeModule.class);
+
+ // ================================================================================
+ // private methods
+ // ================================================================================
+
+ private void reroutePaths(Collection<Intent> oldPaths) {
+ if (oldPaths == null || oldPaths.isEmpty())
+ return;
+
+ IntentOperationList reroutingOperation = new IntentOperationList();
+ for (Intent intent : oldPaths) {
+ PathIntent pathIntent = (PathIntent) intent;
+ if (pathIntent.isPathFrozen())
+ continue;
+ if (pathIntent.getState().equals(IntentState.INST_ACK) && // XXX: path intents in flight
+ !reroutingOperation.contains(pathIntent.getParentIntent())) {
+ reroutingOperation.add(Operator.ADD, pathIntent.getParentIntent());
+ }
+ }
+ executeIntentOperations(reroutingOperation);
+ }
+
+
+ // ================================================================================
+ // IFloodlightModule implementations
+ // ================================================================================
+
+ @Override
+ public Collection<Class<? extends IFloodlightService>> getModuleServices() {
+ Collection<Class<? extends IFloodlightService>> l = new ArrayList<>(1);
+ l.add(IPathCalcRuntimeService.class);
+ return l;
+ }
+
+ @Override
+ public Map<Class<? extends IFloodlightService>, IFloodlightService> getServiceImpls() {
+ Map<Class<? extends IFloodlightService>, IFloodlightService> m = new HashMap<>();
+ m.put(IPathCalcRuntimeService.class, this);
+ return m;
+ }
+
+ @Override
+ public Collection<Class<? extends IFloodlightService>> getModuleDependencies() {
+ Collection<Class<? extends IFloodlightService>> l = new ArrayList<>(2);
+ l.add(IDatagridService.class);
+ l.add(INetworkGraphService.class);
+ return l;
+ }
+
+ @Override
+ public void init(FloodlightModuleContext context) throws FloodlightModuleException {
+ datagridService = context.getServiceImpl(IDatagridService.class);
+ networkGraphService = context.getServiceImpl(INetworkGraphService.class);
+ controllerRegistry = context.getServiceImpl(IControllerRegistryService.class);
+ }
+
+ @Override
+ public void startUp(FloodlightModuleContext context) {
+ highLevelIntents = new IntentMap();
+ runtime = new PathCalcRuntime(networkGraphService.getNetworkGraph());
+ pathIntents = new PathIntentMap();
+ opEventChannel = datagridService.createChannel(INTENT_OP_EVENT_CHANNEL_NAME, Long.class, IntentOperationList.class);
+ datagridService.addListener(INTENT_STATE_EVENT_CHANNEL_NAME, this, Long.class, IntentStateList.class);
+ networkGraphService.registerNetworkGraphListener(this);
+ persistIntent = new PersistIntent(controllerRegistry, networkGraphService);
+ }
+
+ // ================================================================================
+ // IPathCalcRuntimeService implementations
+ // ================================================================================
+
+ @Override
+ public IntentOperationList executeIntentOperations(IntentOperationList list) {
+ if (list == null || list.size() == 0)
+ return null;
+ PerfLogger p = new PerfLogger("executeIntentOperations_" + list.get(0).operator);
+
+ lock.lock(); // TODO optimize locking using smaller steps
+ try {
+ // update the map of high-level intents
+ p.log("begin_updateInMemoryIntents");
+ highLevelIntents.executeOperations(list);
+
+ // change states of high-level intents
+ IntentStateList states = new IntentStateList();
+ for (IntentOperation op : list) {
+ switch (op.operator) {
+ case ADD:
+ switch (op.intent.getState()) {
+ case CREATED:
+ states.put(op.intent.getId(), IntentState.INST_REQ);
+ break;
+ case INST_ACK:
+ states.put(op.intent.getId(), IntentState.REROUTE_REQ);
+ break;
+ default:
+ break;
+ }
+ break;
+ case REMOVE:
+ switch (op.intent.getState()) {
+ case CREATED:
+ states.put(op.intent.getId(), IntentState.DEL_REQ);
+ break;
+ default:
+ break;
+ }
+ break;
+ default:
+ break;
+ }
+ }
+ highLevelIntents.changeStates(states);
+ p.log("end_updateInMemoryIntents");
+
+ // calculate path-intents (low-level operations)
+ p.log("begin_calcPathIntents");
+ IntentOperationList pathIntentOperations = runtime.calcPathIntents(list, highLevelIntents, pathIntents);
+ p.log("end_calcPathIntents");
+
+ // persist calculated low-level operations into data store
+ p.log("begin_persistPathIntents");
+ long key = persistIntent.getKey();
+ persistIntent.persistIfLeader(key, pathIntentOperations);
+ p.log("end_persistPathIntents");
+
+ // remove error-intents and reflect them to high-level intents
+ p.log("begin_removeErrorIntents");
+ states.clear();
+ Iterator<IntentOperation> i = pathIntentOperations.iterator();
+ while (i.hasNext()) {
+ IntentOperation op = i.next();
+ if (op.operator.equals(Operator.ERROR)) {
+ states.put(op.intent.getId(), IntentState.INST_NACK);
+ i.remove();
+ }
+ }
+ highLevelIntents.changeStates(states);
+ p.log("end_removeErrorIntents");
+
+ // update the map of path intents and publish the path operations
+ p.log("begin_updateInMemoryPathIntents");
+ pathIntents.executeOperations(pathIntentOperations);
+ p.log("end_updateInMemoryPathIntents");
+
+ // XXX Demo special: add a complete path to remove operation
+ p.log("begin_addPathToRemoveOperation");
+ for (IntentOperation op: pathIntentOperations) {
+ if(op.operator.equals(Operator.REMOVE)) {
+ op.intent = pathIntents.getIntent(op.intent.getId());
+ }
+ if (op.intent instanceof PathIntent) {
+ log.debug("operation: {}, intent:{}", op.operator, op.intent);
+ }
+ }
+ p.log("end_addPathToRemoveOperation");
+
+ // send notification
+ p.log("begin_sendNotification");
+ // XXX: Send notifications using the same key every time
+ // and receive them by entryAdded() and entryUpdated()
+ opEventChannel.addEntry(0L, pathIntentOperations);
+ p.log("end_sendNotification");
+ //opEventChannel.removeEntry(key);
+ return pathIntentOperations;
+ }
+ finally {
+ p.flushLog();
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public IntentMap getHighLevelIntents() {
+ return highLevelIntents;
+ }
+
+ @Override
+ public IntentMap getPathIntents() {
+ return pathIntents;
+ }
+
+ @Override
+ public void purgeIntents() {
+ highLevelIntents.purge();
+ pathIntents.purge();
+ }
+
+ // ================================================================================
+ // INetworkGraphListener implementations
+ // ================================================================================
+
+ @Override
+ public void networkGraphEvents(Collection<SwitchEvent> addedSwitchEvents,
+ Collection<SwitchEvent> removedSwitchEvents,
+ Collection<PortEvent> addedPortEvents,
+ Collection<PortEvent> removedPortEvents,
+ Collection<LinkEvent> addedLinkEvents,
+ Collection<LinkEvent> removedLinkEvents,
+ Collection<DeviceEvent> addedDeviceEvents,
+ Collection<DeviceEvent> removedDeviceEvents) {
+
+ PerfLogger p = new PerfLogger("networkGraphEvents");
+ HashSet<Intent> affectedPaths = new HashSet<>();
+
+ boolean rerouteAll = false;
+ for(LinkEvent le : addedLinkEvents) {
+ LinkEvent rev = new LinkEvent(le.getDst().getDpid(), le.getDst().getNumber(), le.getSrc().getDpid(), le.getSrc().getNumber());
+ if(unmatchedLinkEvents.contains(rev)) {
+ rerouteAll = true;
+ unmatchedLinkEvents.remove(rev);
+ log.debug("Found matched LinkEvent: {} {}", rev, le);
+ }
+ else {
+ unmatchedLinkEvents.add(le);
+ log.debug("Adding unmatched LinkEvent: {}", le);
+ }
+ }
+ for(LinkEvent le : removedLinkEvents) {
+ if (unmatchedLinkEvents.contains(le)) {
+ unmatchedLinkEvents.remove(le);
+ log.debug("Removing LinkEvent: {}", le);
+ }
+ }
+ if(unmatchedLinkEvents.size() > 0) {
+ log.debug("Unmatched link events: {} events", unmatchedLinkEvents.size());
+ }
+
+ if ( rerouteAll ) {//addedLinkEvents.size() > 0) { // ||
+// addedPortEvents.size() > 0 ||
+// addedSwitchEvents.size() > 0) {
+ p.log("begin_getAllIntents");
+ affectedPaths.addAll(getPathIntents().getAllIntents());
+ p.log("end_getAllIntents");
+ }
+ else if (removedSwitchEvents.size() > 0 ||
+ removedLinkEvents.size() > 0 ||
+ removedPortEvents.size() > 0) {
+ p.log("begin_getIntentsByLink");
+ for (LinkEvent linkEvent: removedLinkEvents)
+ affectedPaths.addAll(pathIntents.getIntentsByLink(linkEvent));
+ p.log("end_getIntentsByLink");
+
+ p.log("begin_getIntentsByPort");
+ for (PortEvent portEvent: removedPortEvents)
+ affectedPaths.addAll(pathIntents.getIntentsByPort(portEvent.getDpid(), portEvent.getNumber()));
+ p.log("end_getIntentsByPort");
+
+ p.log("begin_getIntentsByDpid");
+ for (SwitchEvent switchEvent: removedSwitchEvents)
+ affectedPaths.addAll(pathIntents.getIntentsByDpid(switchEvent.getDpid()));
+ p.log("end_getIntentsByDpid");
+ }
+ p.log("begin_reroutePaths");
+ reroutePaths(affectedPaths);
+ p.log("end_reroutePaths");
+ p.flushLog();
+ }
+
+ // ================================================================================
+ // IEventChannelListener implementations
+ // ================================================================================
+
+ @Override
+ public void entryAdded(IntentStateList value) {
+ entryUpdated(value);
+ }
+
+ @Override
+ public void entryRemoved(IntentStateList value) {
+ // do nothing
+ }
+
+ @Override
+ public void entryUpdated(IntentStateList value) {
+ // TODO draw state transition diagram in multiple ONOS instances and update this method
+ PerfLogger p = new PerfLogger("entryUpdated");
+ lock.lock(); // TODO optimize locking using smaller steps
+ try {
+ // reflect state changes of path-level intent into application-level intents
+ p.log("begin_changeStateByNotification");
+ IntentStateList highLevelIntentStates = new IntentStateList();
+ IntentStateList pathIntentStates = new IntentStateList();
+ for (Entry<String, IntentState> entry: value.entrySet()) {
+ PathIntent pathIntent = (PathIntent) pathIntents.getIntent(entry.getKey());
+ if (pathIntent == null) continue;
+
+ Intent parentIntent = pathIntent.getParentIntent();
+ if (parentIntent == null ||
+ !(parentIntent instanceof ShortestPathIntent) ||
+ !((ShortestPathIntent) parentIntent).getPathIntentId().equals(pathIntent.getId()))
+ continue;
+
+ IntentState state = entry.getValue();
+ switch (state) {
+ //case INST_REQ:
+ case INST_ACK:
+ case INST_NACK:
+ //case DEL_REQ:
+ case DEL_ACK:
+ case DEL_PENDING:
+ highLevelIntentStates.put(parentIntent.getId(), state);
+ pathIntentStates.put(entry.getKey(), entry.getValue());
+ break;
+ default:
+ break;
+ }
+ }
+ highLevelIntents.changeStates(highLevelIntentStates);
+ pathIntents.changeStates(pathIntentStates);
+ p.log("end_changeStateByNotification");
+ }
+ finally {
+ p.flushLog();
+ lock.unlock();
+ }
+ }
+}
diff --git a/src/main/java/net/onrc/onos/core/intent/runtime/PersistIntent.java b/src/main/java/net/onrc/onos/core/intent/runtime/PersistIntent.java
new file mode 100755
index 0000000..134f437
--- /dev/null
+++ b/src/main/java/net/onrc/onos/core/intent/runtime/PersistIntent.java
@@ -0,0 +1,129 @@
+/*
+ * To change this template, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package net.onrc.onos.core.intent.runtime;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Output;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicLong;
+
+import net.onrc.onos.core.datagrid.web.IntentResource;
+import net.onrc.onos.core.datastore.DataStoreClient;
+import net.onrc.onos.core.datastore.IKVTable;
+import net.onrc.onos.core.datastore.ObjectExistsException;
+import net.onrc.onos.core.intent.IntentOperationList;
+import net.onrc.onos.ofcontroller.networkgraph.INetworkGraphService;
+import net.onrc.onos.ofcontroller.networkgraph.NetworkGraph;
+import net.onrc.onos.ofcontroller.util.serializers.KryoFactory;
+import net.onrc.onos.registry.controller.IControllerRegistryService;
+import net.onrc.onos.registry.controller.IdBlock;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ * @author nickkaranatsios
+ */
+public class PersistIntent {
+ private final static Logger log = LoggerFactory.getLogger(IntentResource.class);
+ private long range = 10000L;
+ private final IControllerRegistryService controllerRegistry;
+ NetworkGraph graph = null;
+ private final static String intentJournal = "G:IntentJournal";
+ private final static int valueStoreLimit = 1024 * 1024;
+ private IKVTable table;
+ private Kryo kryo;
+ private ByteArrayOutputStream stream;
+ private Output output = null;
+ private AtomicLong nextId = null;
+ private long rangeEnd;
+ private IdBlock idBlock = null;
+
+
+ public PersistIntent(final IControllerRegistryService controllerRegistry, INetworkGraphService ng) {
+ this.controllerRegistry = controllerRegistry;
+ this.graph = ng.getNetworkGraph();
+ table = DataStoreClient.getClient().getTable(intentJournal);
+ stream = new ByteArrayOutputStream(1024);
+ output = new Output(stream);
+ kryo = (new KryoFactory()).newKryo();
+ }
+
+ public long getKey() {
+ long key;
+ if (idBlock == null) {
+ key = getNextBlock();
+ } else {
+ key = nextId.incrementAndGet();
+ if (key >= rangeEnd) {
+ key = getNextBlock();
+ }
+ }
+ return key;
+ }
+
+ private long getNextBlock() {
+ // XXX This method is not thread safe, may lose allocated IdBlock
+ idBlock = controllerRegistry.allocateUniqueIdBlock(range);
+ nextId = new AtomicLong(idBlock.getStart());
+ rangeEnd = idBlock.getEnd();
+ return nextId.get();
+ }
+
+ public boolean persistIfLeader(long key, IntentOperationList operations) {
+ boolean leader = true;
+ boolean ret = false;
+ // TODO call controllerRegistry.isClusterLeader()
+ if (leader) {
+ try {
+ // reserve key 10 entries for multi-write if size over 1MB
+ key *= 10;
+ kryo.writeObject(output, operations);
+ output.close();
+ ByteBuffer keyBytes = ByteBuffer.allocate(8).putLong(key);
+ byte[] buffer = stream.toByteArray();
+ int total = buffer.length;
+ if ((total >= valueStoreLimit )) {
+ int writeCount = total / valueStoreLimit;
+ int remainder = total % valueStoreLimit;
+ int upperIndex = 0;
+ for (int i = 0; i < writeCount; i++, key++) {
+ keyBytes.clear();
+ keyBytes.putLong(key);
+ keyBytes.flip();
+ upperIndex = (i * valueStoreLimit + valueStoreLimit) - 1;
+ log.debug("writing using indexes {}:{}", (i*valueStoreLimit) ,upperIndex);
+ table.create(keyBytes.array(), Arrays.copyOfRange(buffer, i * valueStoreLimit, upperIndex));
+ }
+ if (remainder > 0) {
+ keyBytes.clear();
+ keyBytes.putLong(key);
+ keyBytes.flip();
+ log.debug("writing using indexes {}:{}" ,upperIndex ,total);
+ table.create(keyBytes.array(), Arrays.copyOfRange(buffer, upperIndex + 1, total - 1));
+ }
+ } else {
+ keyBytes.flip();
+ table.create(keyBytes.array(), buffer);
+ }
+ log.debug("key is {} value length is {}", key, buffer.length);
+ stream.reset();
+ stream.close();
+ log.debug("persist operations to ramcloud size of operations: {}", operations.size());
+ ret = true;
+ } catch (ObjectExistsException ex) {
+ log.warn("Failed to store intent journal with key " + key);
+ } catch (IOException ex) {
+ log.error("Failed to close the stream");
+ }
+ }
+ return ret;
+ }
+}
diff --git a/src/main/java/net/onrc/onos/core/intent/runtime/PlanCalcRuntime.java b/src/main/java/net/onrc/onos/core/intent/runtime/PlanCalcRuntime.java
new file mode 100644
index 0000000..7c92115
--- /dev/null
+++ b/src/main/java/net/onrc/onos/core/intent/runtime/PlanCalcRuntime.java
@@ -0,0 +1,155 @@
+package net.onrc.onos.core.intent.runtime;
+
+import java.util.ArrayList;
+import java.util.Collection;
+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.Set;
+
+import net.floodlightcontroller.util.MACAddress;
+import net.onrc.onos.core.intent.FlowEntry;
+import net.onrc.onos.core.intent.Intent;
+import net.onrc.onos.core.intent.IntentOperation;
+import net.onrc.onos.core.intent.IntentOperationList;
+import net.onrc.onos.core.intent.PathIntent;
+import net.onrc.onos.core.intent.ShortestPathIntent;
+import net.onrc.onos.core.intent.IntentOperation.Operator;
+import net.onrc.onos.ofcontroller.networkgraph.LinkEvent;
+//import net.onrc.onos.ofcontroller.networkgraph.NetworkGraph;
+
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ * @author Brian O'Connor <bocon@onlab.us>
+ *
+ */
+
+public class PlanCalcRuntime {
+
+// NetworkGraph graph;
+ private final static Logger log = LoggerFactory.getLogger(PlanCalcRuntime.class);
+
+ public PlanCalcRuntime(/*NetworkGraph graph*/) {
+// this.graph = graph;
+ }
+
+ public List<Set<FlowEntry>> computePlan(IntentOperationList intentOps) {
+ long start = System.nanoTime();
+ List<Collection<FlowEntry>> flowEntries = computeFlowEntries(intentOps);
+ long step1 = System.nanoTime();
+ List<Set<FlowEntry>> plan = buildPhases(flowEntries);
+ long step2 = System.nanoTime();
+ log.error("MEASUREMENT: Compute flow entries: {} ns, Build phases: {} ns",
+ (step1 - start), (step2 - step1));
+ return plan;
+ }
+
+ private List<Collection<FlowEntry>> computeFlowEntries(IntentOperationList intentOps) {
+ List<Collection<FlowEntry>> flowEntries = new LinkedList<>();
+ for(IntentOperation i : intentOps) {
+ if(!(i.intent instanceof PathIntent)) {
+ log.warn("Not a path intent: {}", i);
+ continue;
+ }
+ PathIntent intent = (PathIntent) i.intent;
+ Intent parent = intent.getParentIntent();
+ long srcPort, dstPort;
+ long lastDstSw = -1, lastDstPort = -1;
+ MACAddress srcMac, dstMac;
+ if(parent instanceof ShortestPathIntent) {
+ ShortestPathIntent pathIntent = (ShortestPathIntent) parent;
+// Switch srcSwitch = graph.getSwitch(pathIntent.getSrcSwitchDpid());
+// srcPort = srcSwitch.getPort(pathIntent.getSrcPortNumber());
+ srcPort = pathIntent.getSrcPortNumber();
+ srcMac = MACAddress.valueOf(pathIntent.getSrcMac());
+ dstMac = MACAddress.valueOf(pathIntent.getDstMac());
+// Switch dstSwitch = graph.getSwitch(pathIntent.getDstSwitchDpid());
+ lastDstSw = pathIntent.getDstSwitchDpid();
+// lastDstPort = dstSwitch.getPort(pathIntent.getDstPortNumber());
+ lastDstPort = pathIntent.getDstPortNumber();
+ }
+ else {
+ log.warn("Unsupported Intent: {}", parent);
+ continue;
+ }
+ List<FlowEntry> entries = new ArrayList<>();
+ for(LinkEvent linkEvent : intent.getPath()) {
+// Link link = graph.getLink(linkEvent.getSrc().getDpid(),
+// linkEvent.getSrc().getNumber(),
+// linkEvent.getDst().getDpid(),
+// linkEvent.getDst().getNumber());
+// Switch sw = link.getSrcSwitch();
+ long sw = linkEvent.getSrc().getDpid();
+// dstPort = link.getSrcPort();
+ dstPort = linkEvent.getSrc().getNumber();
+ FlowEntry fe = new FlowEntry(sw, srcPort, dstPort, srcMac, dstMac, i.operator);
+ entries.add(fe);
+// srcPort = link.getDstPort();
+ srcPort = linkEvent.getDst().getNumber();
+ }
+ if(lastDstSw >= 0 && lastDstPort >= 0) {
+ //Switch sw = lastDstPort.getSwitch();
+ long sw = lastDstSw;
+ dstPort = lastDstPort;
+ FlowEntry fe = new FlowEntry(sw, srcPort, dstPort, srcMac, dstMac, i.operator);
+ entries.add(fe);
+ }
+ // install flow entries in reverse order
+ Collections.reverse(entries);
+ flowEntries.add(entries);
+ }
+ return flowEntries;
+ }
+
+ private List<Set<FlowEntry>> buildPhases(List<Collection<FlowEntry>> flowEntries) {
+ Map<FlowEntry, Integer> map = new HashMap<>();
+ List<Set<FlowEntry>> plan = new ArrayList<>();
+ for(Collection<FlowEntry> c : flowEntries) {
+ for(FlowEntry e : c) {
+ Integer i = map.get(e);
+ if(i == null) {
+ i = Integer.valueOf(0);
+ }
+ switch(e.getOperator()) {
+ case ADD:
+ i += 1;
+ break;
+ case REMOVE:
+ i -= 1;
+ break;
+ default:
+ break;
+ }
+ map.put(e, i);
+ // System.out.println(e + " " + e.getOperator());
+ }
+ }
+
+ // really simple first iteration of plan
+ //TODO: optimize the map in phases
+ Set<FlowEntry> phase = new HashSet<>();
+ for(FlowEntry e : map.keySet()) {
+ Integer i = map.get(e);
+ if(i == 0) {
+ continue;
+ }
+ else if(i > 0) {
+ e.setOperator(Operator.ADD);
+ }
+ else if(i < 0) {
+ e.setOperator(Operator.REMOVE);
+ }
+ phase.add(e);
+ }
+ plan.add(phase);
+
+ return plan;
+ }
+}
diff --git a/src/main/java/net/onrc/onos/core/intent/runtime/PlanInstallModule.java b/src/main/java/net/onrc/onos/core/intent/runtime/PlanInstallModule.java
new file mode 100644
index 0000000..bbc303d
--- /dev/null
+++ b/src/main/java/net/onrc/onos/core/intent/runtime/PlanInstallModule.java
@@ -0,0 +1,201 @@
+package net.onrc.onos.core.intent.runtime;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import net.floodlightcontroller.core.IFloodlightProviderService;
+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.onrc.onos.core.datagrid.IDatagridService;
+import net.onrc.onos.core.datagrid.IEventChannel;
+import net.onrc.onos.core.datagrid.IEventChannelListener;
+import net.onrc.onos.core.intent.FlowEntry;
+import net.onrc.onos.core.intent.IntentOperation;
+import net.onrc.onos.core.intent.IntentOperationList;
+import net.onrc.onos.core.intent.Intent.IntentState;
+import net.onrc.onos.ofcontroller.flowprogrammer.IFlowPusherService;
+import net.onrc.onos.ofcontroller.networkgraph.INetworkGraphService;
+//import net.onrc.onos.ofcontroller.networkgraph.NetworkGraph;
+
+
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PlanInstallModule implements IFloodlightModule {
+ protected volatile IFloodlightProviderService floodlightProvider;
+ protected volatile INetworkGraphService networkGraph;
+ protected volatile IDatagridService datagridService;
+ protected volatile IFlowPusherService flowPusher;
+ private PlanCalcRuntime planCalc;
+ private PlanInstallRuntime planInstall;
+ private EventListener eventListener;
+ private IEventChannel<Long, IntentStateList> intentStateChannel;
+ private final static Logger log = LoggerFactory.getLogger(PlanInstallModule.class);
+
+
+ private static final String PATH_INTENT_CHANNEL_NAME = "onos.pathintent";
+ private static final String INTENT_STATE_EVENT_CHANNEL_NAME = "onos.pathintent_state";
+
+
+ @Override
+ public void init(FloodlightModuleContext context)
+ throws FloodlightModuleException {
+ floodlightProvider = context.getServiceImpl(IFloodlightProviderService.class);
+ networkGraph = context.getServiceImpl(INetworkGraphService.class);
+ datagridService = context.getServiceImpl(IDatagridService.class);
+ flowPusher = context.getServiceImpl(IFlowPusherService.class);
+// NetworkGraph graph = networkGraph.getNetworkGraph();
+ planCalc = new PlanCalcRuntime();
+ planInstall = new PlanInstallRuntime(floodlightProvider, flowPusher);
+ eventListener = new EventListener();
+ }
+
+ class EventListener extends Thread
+ implements IEventChannelListener<Long, IntentOperationList> {
+
+ private BlockingQueue<IntentOperationList> intentQueue = new LinkedBlockingQueue<>();
+ private Long key = Long.valueOf(0);
+
+ @Override
+ public void run() {
+ while(true) {
+ try {
+ IntentOperationList intents = intentQueue.take();
+ //TODO: consider draining the remaining intent lists
+ // and processing in one big batch
+// List<IntentOperationList> remaining = new LinkedList<>();
+// intentQueue.drainTo(remaining);
+
+ processIntents(intents);
+ } catch (InterruptedException e) {
+ log.warn("Error taking from intent queue: {}", e.getMessage());
+ }
+ }
+ }
+
+ private void processIntents(IntentOperationList intents) {
+ log("start_processIntents");
+ log.debug("Processing OperationList {}", intents);
+ log("begin_computePlan");
+ List<Set<FlowEntry>> plan = planCalc.computePlan(intents);
+ log("end_computePlan");
+ log.debug("Plan: {}", plan);
+ log("begin_installPlan");
+ boolean success = planInstall.installPlan(plan);
+ log("end_installPlan");
+
+ log("begin_sendInstallNotif");
+ sendNotifications(intents, true, success);
+ log("end_sendInstallNotif");
+ log("finish");
+ }
+
+ private void sendNotifications(IntentOperationList intents, boolean installed, boolean success) {
+ IntentStateList states = new IntentStateList();
+ for(IntentOperation i : intents) {
+ IntentState newState;
+ switch(i.operator) {
+ case REMOVE:
+ if(installed) {
+ newState = success ? IntentState.DEL_ACK : IntentState.DEL_PENDING;
+ }
+ else {
+ newState = IntentState.DEL_REQ;
+ }
+ break;
+ case ADD:
+ default:
+ if(installed) {
+ newState = success ? IntentState.INST_ACK : IntentState.INST_NACK;
+ }
+ else {
+ newState = IntentState.INST_REQ;
+ }
+ break;
+ }
+ states.put(i.intent.getId(), newState);
+ }
+ intentStateChannel.addEntry(key, states);
+ // XXX: Send notifications using the same key every time
+ // and receive them by entryAdded() and entryUpdated()
+ // key += 1;
+ }
+
+ @Override
+ public void entryAdded(IntentOperationList value) {
+ entryUpdated(value);
+ }
+
+ @Override
+ public void entryRemoved(IntentOperationList value) {
+ // This channel is a queue, so this method is not needed
+ }
+
+ @Override
+ public void entryUpdated(IntentOperationList value) {
+ log("start_intentNotifRecv");
+ log("begin_sendReceivedNotif");
+ sendNotifications(value, false, false);
+ log("end_sendReceivedNotif");
+ log("finish");
+
+ log.debug("Added OperationList {}", value);
+ try {
+ intentQueue.put(value);
+ } catch (InterruptedException e) {
+ log.warn("Error putting to intent queue: {}", e.getMessage());
+ }
+ }
+ }
+
+ public static void log(String step) {
+ log.error("Time:{}, Step:{}", System.nanoTime(), step);
+ }
+
+ @Override
+ public void startUp(FloodlightModuleContext context) {
+ // start subscriber
+ datagridService.addListener(PATH_INTENT_CHANNEL_NAME,
+ eventListener,
+ Long.class,
+ IntentOperationList.class);
+ eventListener.start();
+ // start publisher
+ intentStateChannel = datagridService.createChannel(INTENT_STATE_EVENT_CHANNEL_NAME,
+ Long.class,
+ IntentStateList.class);
+ }
+
+ @Override
+ public Collection<Class<? extends IFloodlightService>> getModuleDependencies() {
+ Collection<Class<? extends IFloodlightService>> l =
+ new ArrayList<Class<? extends IFloodlightService>>();
+ l.add(IFloodlightProviderService.class);
+ l.add(INetworkGraphService.class);
+ l.add(IDatagridService.class);
+ l.add(IFlowPusherService.class);
+ return l;
+ }
+
+ @Override
+ public Collection<Class<? extends IFloodlightService>> getModuleServices() {
+ // no services, for now
+ return null;
+ }
+
+ @Override
+ public Map<Class<? extends IFloodlightService>, IFloodlightService> getServiceImpls() {
+ // no services, for now
+ return null;
+ }
+
+}
diff --git a/src/main/java/net/onrc/onos/core/intent/runtime/PlanInstallRuntime.java b/src/main/java/net/onrc/onos/core/intent/runtime/PlanInstallRuntime.java
new file mode 100644
index 0000000..e3bdc41
--- /dev/null
+++ b/src/main/java/net/onrc/onos/core/intent/runtime/PlanInstallRuntime.java
@@ -0,0 +1,156 @@
+package net.onrc.onos.core.intent.runtime;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+
+import net.floodlightcontroller.core.IFloodlightProviderService;
+import net.floodlightcontroller.core.IOFSwitch;
+import net.floodlightcontroller.core.internal.OFMessageFuture;
+import net.onrc.onos.core.intent.FlowEntry;
+import net.onrc.onos.ofcontroller.flowprogrammer.IFlowPusherService;
+//import net.onrc.onos.ofcontroller.networkgraph.NetworkGraph;
+import net.onrc.onos.ofcontroller.util.Pair;
+
+import org.openflow.protocol.OFBarrierReply;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ * @author Brian O'Connor <bocon@onlab.us>
+ *
+ */
+
+public class PlanInstallRuntime {
+// NetworkGraph graph;
+ IFlowPusherService pusher;
+ IFloodlightProviderService provider;
+ private final static Logger log = LoggerFactory.getLogger(PlanInstallRuntime.class);
+
+ public PlanInstallRuntime(//NetworkGraph graph,
+ IFloodlightProviderService provider,
+ IFlowPusherService pusher) {
+// this.graph = graph;
+ this.provider = provider;
+ this.pusher = pusher;
+ }
+
+ private static class FlowModCount {
+ IOFSwitch sw;
+ long modFlows = 0;
+ long delFlows = 0;
+ long errors = 0;
+
+ FlowModCount(IOFSwitch sw) {
+ this.sw = sw;
+ }
+
+ void addFlowEntry(FlowEntry entry) {
+ switch(entry.getOperator()){
+ case ADD:
+ modFlows++;
+ break;
+ case ERROR:
+ errors++;
+ break;
+ case REMOVE:
+ delFlows++;
+ break;
+ default:
+ break;
+ }
+ }
+
+ public String toString() {
+ return "sw:" + sw.getStringId() + ": modify " + modFlows + " delete " + delFlows + " error " + errors;
+ }
+
+ static Map<IOFSwitch, FlowModCount> map = new HashMap<>();
+ static void countFlowEntry(IOFSwitch sw, FlowEntry entry) {
+ FlowModCount count = map.get(sw);
+ if(count == null) {
+ count = new FlowModCount(sw);
+ map.put(sw, count);
+ }
+ count.addFlowEntry(entry);
+ }
+ static void startCount() {
+ map.clear();
+ }
+ static void printCount() {
+ String result = "FLOWMOD COUNT:\n";
+ for(FlowModCount count : map.values()) {
+ result += count.toString() + '\n';
+ }
+ if(map.values().isEmpty()) {
+ result += "No flow mods installed\n";
+ }
+ log.error(result);
+ }
+ }
+
+ public boolean installPlan(List<Set<FlowEntry>> plan) {
+ long start = System.nanoTime();
+ Map<Long,IOFSwitch> switches = provider.getSwitches();
+
+ log.debug("IOFSwitches: {}", switches);
+
+ FlowModCount.startCount();
+ for(Set<FlowEntry> phase : plan) {
+ Set<Pair<IOFSwitch, net.onrc.onos.ofcontroller.util.FlowEntry>> entries = new HashSet<>();
+ Set<IOFSwitch> modifiedSwitches = new HashSet<>();
+
+ long step1 = System.nanoTime();
+ // convert flow entries and create pairs
+ for(FlowEntry entry : phase) {
+ IOFSwitch sw = switches.get(entry.getSwitch());
+ if(sw == null) {
+ // no active switch, skip this flow entry
+ log.debug("Skipping flow entry: {}", entry);
+ continue;
+ }
+ entries.add(new Pair<>(sw, entry.getFlowEntry()));
+ modifiedSwitches.add(sw);
+ FlowModCount.countFlowEntry(sw, entry);
+ }
+ long step2 = System.nanoTime();
+
+ // push flow entries to switches
+ log.debug("Pushing flow entries: {}", entries);
+ pusher.pushFlowEntries(entries);
+ long step3 = System.nanoTime();
+
+ // TODO: insert a barrier after each phase on each modifiedSwitch
+ // TODO: wait for confirmation messages before proceeding
+ List<Pair<IOFSwitch,OFMessageFuture<OFBarrierReply>>> barriers = new ArrayList<>();
+ for(IOFSwitch sw : modifiedSwitches) {
+ barriers.add(new Pair<>(sw, pusher.barrierAsync(sw)));
+ }
+ for(Pair<IOFSwitch,OFMessageFuture<OFBarrierReply>> pair : barriers) {
+ IOFSwitch sw = pair.first;
+ OFMessageFuture<OFBarrierReply> future = pair.second;
+ try {
+ future.get();
+ } catch (InterruptedException | ExecutionException e) {
+ log.error("Barrier message not received for sw: {}", sw);
+ }
+ }
+ long step4 = System.nanoTime();
+ log.error("MEASUREMENT: convert: {} ns, push: {} ns, barrierWait: {} ns",
+ step2 - step1, step3 - step2, step4 - step3);
+
+ }
+ long end = System.nanoTime();
+ log.error("MEASUREMENT: Install plan: {} ns", (end-start));
+ FlowModCount.printCount();
+
+ // TODO: we assume that the plan installation succeeds for now
+ return true;
+ }
+
+}