diff --git a/.gitignore b/.gitignore
index 291274fb43c0e74e3561ad875248da3ab44e9ede..b79ebe2942680058581bffac633a1294b5e282f4 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,6 +1,8 @@
*~
+*.class
.classpath
.project
+.pydevproject
.settings
.DS_Store
target
diff --git a/src/main/java/net/floodlightcontroller/core/IOFSwitch.java b/src/main/java/net/floodlightcontroller/core/IOFSwitch.java
index e1ba7ef84ef32c53068a2e66d646ea6c1b846f2c..023769ee40f054477b81c0a6bdc56d3f053701fe 100644
--- a/src/main/java/net/floodlightcontroller/core/IOFSwitch.java
+++ b/src/main/java/net/floodlightcontroller/core/IOFSwitch.java
@@ -22,6 +22,7 @@ import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;
+import java.util.concurrent.locks.Lock;
import net.floodlightcontroller.core.types.MacVlanPair;
import net.floodlightcontroller.util.TimedCache;
@@ -243,12 +244,14 @@ public interface IOFSwitch {
/**
* Check if the switch is still connected;
+ * Only call while holding processMessageLock
* @return whether the switch is still disconnected
*/
public boolean isConnected();
/**
* Set whether the switch is connected
+ * Only call while holding modifySwitchLock
* @param connected whether the switch is connected
*/
public void setConnected(boolean connected);
@@ -304,4 +307,19 @@ public interface IOFSwitch {
* @return
*/
public TimedCache<Long> getTimedCache();
+
+ /**
+ * Return a lock that need to be held while processing a message. Multiple threads
+ * can hold this lock.
+ * @return
+ */
+ public Lock processMessageLock();
+
+ /**
+ * Return a lock that needs to be held while the switch is removed asynchronously, i.e.,
+ * the removing is not triggered by events on this switch's channel.
+ * Mutex with processMessageLock
+ * @return
+ */
+ public Lock asyncRemoveSwitchLock();
}
diff --git a/src/main/java/net/floodlightcontroller/core/internal/Controller.java b/src/main/java/net/floodlightcontroller/core/internal/Controller.java
index ac7e59f83eaeb0a06f1a33d434b00cc3105acfd1..7ce55563744d0907e5a2f91b5f5ca4a277ec25ad 100644
--- a/src/main/java/net/floodlightcontroller/core/internal/Controller.java
+++ b/src/main/java/net/floodlightcontroller/core/internal/Controller.java
@@ -329,6 +329,7 @@ public class Controller
@Override
public void channelIdle(ChannelHandlerContext ctx, IdleStateEvent e)
throws Exception {
+ // Can write to switch without holding processLock
OFMessage m = factory.getMessage(OFType.ECHO_REQUEST);
e.getChannel().write(m);
}
@@ -471,101 +472,109 @@ public class Controller
*/
protected void processOFMessage(OFMessage m)
throws IOException, SwitchStateException {
- switch (m.getType()) {
- case HELLO:
- log.debug("HELLO from {}", sw);
- if (state.hsState.equals(HandshakeState.START)) {
- state.hsState = HandshakeState.HELLO;
- sendHelloConfiguration();
- } else {
- throw new SwitchStateException("Unexpected HELLO from " + sw);
- }
- break;
- case ECHO_REQUEST:
- OFEchoReply reply =
- (OFEchoReply) factory.getMessage(OFType.ECHO_REPLY);
- reply.setXid(m.getXid());
- sw.write(reply, null);
- break;
- case ECHO_REPLY:
- break;
- case FEATURES_REPLY:
- log.debug("Features Reply from {}", sw);
- if (state.hsState.equals(HandshakeState.HELLO)) {
- sw.setFeaturesReply((OFFeaturesReply) m);
- sendFeatureReplyConfiguration();
- state.hsState = HandshakeState.FEATURES_REPLY;
- // uncomment to enable "dumb" switches like cbench
- // state.hsState = HandshakeState.READY;
- // addSwitch(sw);
- } else {
- String em = "Unexpected FEATURES_REPLY from " + sw;
- throw new SwitchStateException(em);
- }
- break;
- case GET_CONFIG_REPLY:
- if (!state.hsState.equals(HandshakeState.FEATURES_REPLY)) {
- String em = "Unexpected GET_CONFIG_REPLY from " + sw;
- throw new SwitchStateException(em);
- }
- OFGetConfigReply cr = (OFGetConfigReply) m;
- if (cr.getMissSendLength() == (short)0xffff) {
- log.debug("Config Reply from {} confirms " +
- "miss length set to 0xffff", sw);
- } else {
- log.warn("Config Reply from {} has " +
- "miss length set to {}",
- sw, cr.getMissSendLength());
- }
- state.hasGetConfigReply = true;
- if (state.hasDescription && state.hasGetConfigReply) {
- addSwitch(sw);
- state.hsState = HandshakeState.READY;
- }
- break;
- case ERROR:
- OFError error = (OFError) m;
- logError(sw, error);
- break;
- case STATS_REPLY:
- if (state.hsState.ordinal() <
- HandshakeState.FEATURES_REPLY.ordinal()) {
- String em = "Unexpected STATS_REPLY from " + sw;
- throw new SwitchStateException(em);
- }
- sw.deliverStatisticsReply(m);
- if (sw.hasAttribute(IOFSwitch.SWITCH_DESCRIPTION_FUTURE)) {
- processSwitchDescReply();
- }
- break;
- /*
- * "Trivial" server to test raw low-level throughput
- case PACKET_IN:
- OFPacketIn pi = (OFPacketIn)m;
-
- OFFlowMod fm =
- (OFFlowMod)factory.getMessage(OFType.FLOW_MOD);
- OFMatch match = new OFMatch();
- match.loadFromPacket(pi.getPacketData(), pi.getInPort());
- fm.setBufferId(pi.getBufferId());
- fm.setMatch(match);
- sw.write(fm, null);
- break;
- */
- case PORT_STATUS:
- boolean swadded =
- state.hsState.equals(HandshakeState.READY);
- handlePortStatusMessage(sw, (OFPortStatus)m, swadded);
- // fall through
- default:
- if (!state.hsState.equals(HandshakeState.READY)) {
- log.debug("Ignoring message type {} received " +
- "from switch {} before switch is " +
- "fully configured.", m.getType(), sw);
+ sw.processMessageLock().lock();
+ try {
+ if (!sw.isConnected())
+ return;
+ switch (m.getType()) {
+ case HELLO:
+ log.debug("HELLO from {}", sw);
+ if (state.hsState.equals(HandshakeState.START)) {
+ state.hsState = HandshakeState.HELLO;
+ sendHelloConfiguration();
+ } else {
+ throw new SwitchStateException("Unexpected HELLO from " + sw);
+ }
break;
- }
- handleMessage(sw, m, null);
- break;
+ case ECHO_REQUEST:
+ OFEchoReply reply =
+ (OFEchoReply) factory.getMessage(OFType.ECHO_REPLY);
+ reply.setXid(m.getXid());
+ sw.write(reply, null);
+ break;
+ case ECHO_REPLY:
+ break;
+ case FEATURES_REPLY:
+ log.debug("Features Reply from {}", sw);
+ if (state.hsState.equals(HandshakeState.HELLO)) {
+ sw.setFeaturesReply((OFFeaturesReply) m);
+ sendFeatureReplyConfiguration();
+ state.hsState = HandshakeState.FEATURES_REPLY;
+ // uncomment to enable "dumb" switches like cbench
+ // state.hsState = HandshakeState.READY;
+ // addSwitch(sw);
+ } else {
+ String em = "Unexpected FEATURES_REPLY from " + sw;
+ throw new SwitchStateException(em);
+ }
+ break;
+ case GET_CONFIG_REPLY:
+ if (!state.hsState.equals(HandshakeState.FEATURES_REPLY)) {
+ String em = "Unexpected GET_CONFIG_REPLY from " + sw;
+ throw new SwitchStateException(em);
+ }
+ OFGetConfigReply cr = (OFGetConfigReply) m;
+ if (cr.getMissSendLength() == (short)0xffff) {
+ log.debug("Config Reply from {} confirms " +
+ "miss length set to 0xffff", sw);
+ } else {
+ log.warn("Config Reply from {} has " +
+ "miss length set to {}",
+ sw, cr.getMissSendLength());
+ }
+ state.hasGetConfigReply = true;
+ if (state.hasDescription && state.hasGetConfigReply) {
+ addSwitch(sw);
+ state.hsState = HandshakeState.READY;
+ }
+ break;
+ case ERROR:
+ OFError error = (OFError) m;
+ logError(sw, error);
+ break;
+ case STATS_REPLY:
+ if (state.hsState.ordinal() <
+ HandshakeState.FEATURES_REPLY.ordinal()) {
+ String em = "Unexpected STATS_REPLY from " + sw;
+ throw new SwitchStateException(em);
+ }
+ sw.deliverStatisticsReply(m);
+ if (sw.hasAttribute(IOFSwitch.SWITCH_DESCRIPTION_FUTURE)) {
+ processSwitchDescReply();
+ }
+ break;
+ /*
+ * "Trivial" server to test raw low-level throughput
+ case PACKET_IN:
+ OFPacketIn pi = (OFPacketIn)m;
+
+ OFFlowMod fm =
+ (OFFlowMod)factory.getMessage(OFType.FLOW_MOD);
+ OFMatch match = new OFMatch();
+ match.loadFromPacket(pi.getPacketData(), pi.getInPort());
+ fm.setBufferId(pi.getBufferId());
+ fm.setMatch(match);
+ sw.write(fm, null);
+ break;
+ */
+ case PORT_STATUS:
+ boolean swadded =
+ state.hsState.equals(HandshakeState.READY);
+ handlePortStatusMessage(sw, (OFPortStatus)m, swadded);
+ // fall through
+ default:
+ if (!state.hsState.equals(HandshakeState.READY)) {
+ log.debug("Ignoring message type {} received " +
+ "from switch {} before switch is " +
+ "fully configured.", m.getType(), sw);
+ break;
+ }
+ handleMessage(sw, m, null);
+ break;
+ }
+ }
+ finally {
+ sw.processMessageLock().unlock();
}
}
}
@@ -878,6 +887,8 @@ public class Controller
* @param sw the new switch
*/
protected void addSwitch(IOFSwitch sw) {
+ // TODO: is it save to modify the HashMap without holding
+ // the old switch's lock
IOFSwitch oldSw = this.switches.put(sw.getId(), sw);
if (sw == oldSw) {
// Note == for object equality, not .equals for value
@@ -888,21 +899,27 @@ public class Controller
log.info("Switch handshake successful: {}", sw);
if (oldSw != null) {
- log.error("New switch connection {} for already-connected switch {}",
- sw, oldSw);
- oldSw.setConnected(false);
- updateInactiveSwitchInfo(oldSw);
-
- // we need to clean out old switch state definitively
- // before adding the new switch
- if (switchListeners != null) {
- for (IOFSwitchListener listener : switchListeners) {
- listener.removedSwitch(oldSw);
+ oldSw.asyncRemoveSwitchLock().lock();
+ try {
+ log.error("New switch connection {} for already-connected switch {}",
+ sw, oldSw);
+ oldSw.setConnected(false);
+ updateInactiveSwitchInfo(oldSw);
+
+ // we need to clean out old switch state definitively
+ // before adding the new switch
+ if (switchListeners != null) {
+ for (IOFSwitchListener listener : switchListeners) {
+ listener.removedSwitch(oldSw);
+ }
}
+ // will eventually trigger a removeSwitch(), which will cause
+ // a "Not removing Switch ... already removed debug message.
+ oldSw.getChannel().close();
+ }
+ finally {
+ oldSw.asyncRemoveSwitchLock().unlock();
}
- // will eventually trigger a removeSwitch(), which will cause
- // a "Not removing Switch ... already removed debug message.
- oldSw.getChannel().close();
}
updateActiveSwitchInfo(sw);
@@ -919,12 +936,15 @@ public class Controller
* @param sw the switch that has disconnected
*/
protected void removeSwitch(IOFSwitch sw) {
+ // No need to acquire the asyncRemoveSwitch lock, since
+ // this method is only called after netty has processed all
+ // pending messages
if (!this.switches.remove(sw.getId(), sw) ||
(sw.isConnected() == false)) {
log.debug("Not removing switch {}; already removed", sw);
return;
}
-
+
sw.setConnected(false);
updateInactiveSwitchInfo(sw);
Update update = new Update(sw, false);
diff --git a/src/main/java/net/floodlightcontroller/core/internal/OFSwitchImpl.java b/src/main/java/net/floodlightcontroller/core/internal/OFSwitchImpl.java
index 46a3c9bf91182ec70cabc61929c40bcbaa52951d..afdd7beb97645bde699b1566609b6df2ff17ae93 100644
--- a/src/main/java/net/floodlightcontroller/core/internal/OFSwitchImpl.java
+++ b/src/main/java/net/floodlightcontroller/core/internal/OFSwitchImpl.java
@@ -26,6 +26,8 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import net.floodlightcontroller.core.FloodlightContext;
import net.floodlightcontroller.core.IFloodlightProvider;
@@ -72,6 +74,7 @@ public class OFSwitchImpl implements IOFSwitch {
protected Map<Integer,OFStatisticsFuture> statsFutureMap;
protected boolean connected;
protected TimedCache<Long> timedCache;
+ protected ReentrantReadWriteLock lock;
public static IOFSwitchFeatures switchFeatures;
@@ -87,6 +90,7 @@ public class OFSwitchImpl implements IOFSwitch {
this.connected = true;
this.statsFutureMap = new ConcurrentHashMap<Integer,OFStatisticsFuture>();
this.timedCache = new TimedCache<Long>(100, 5*1000 ); // 5 seconds interval
+ this.lock = new ReentrantReadWriteLock();
// Defaults properties for an ideal switch
this.setAttribute(PROP_FASTWILDCARDS, (Integer) OFMatch.OFPFW_ALL);
@@ -360,4 +364,14 @@ public class OFSwitchImpl implements IOFSwitch {
public TimedCache<Long> getTimedCache() {
return timedCache;
}
+
+ @Override
+ public Lock processMessageLock() {
+ return lock.readLock();
+ }
+
+ @Override
+ public Lock asyncRemoveSwitchLock() {
+ return lock.writeLock();
+ }
}
diff --git a/src/main/java/net/floodlightcontroller/jython/Server.java b/src/main/java/net/floodlightcontroller/jython/Server.java
index cd330fa27c21d53a37ae427d0c9dd93ffc09d320..47ec1a52a930a4cf6c1734627c2154c50d91fbaa 100644
--- a/src/main/java/net/floodlightcontroller/jython/Server.java
+++ b/src/main/java/net/floodlightcontroller/jython/Server.java
@@ -1,9 +1,12 @@
package net.floodlightcontroller.jython;
import java.net.URL;
+import java.util.HashMap;
import java.util.Map;
import org.python.util.PythonInterpreter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* This class starts a thread that runs a jython interpreter that
@@ -13,6 +16,8 @@ import org.python.util.PythonInterpreter;
*
*/
public class Server extends Thread {
+ protected static Logger log = LoggerFactory.getLogger(Server.class);
+
int port;
Map<String, Object> locals;
@@ -23,7 +28,11 @@ public class Server extends Thread {
public Server(int port_, Map<String, Object> locals_) {
this.port = port_ ;
this.locals = locals_;
-
+ if (this.locals == null) {
+ this.locals = new HashMap<String, Object>();
+ }
+ this.locals.put("log", Server.log);
+ this.setName("debugserver");
}
/**
diff --git a/src/main/java/net/floodlightcontroller/topology/internal/TopologyImpl.java b/src/main/java/net/floodlightcontroller/topology/internal/TopologyImpl.java
index ce1c5926c451e0295822ed9a2a9eed1da4ceeaff..19c1015a26fa1a441814a6beaed42a70bec60345 100644
--- a/src/main/java/net/floodlightcontroller/topology/internal/TopologyImpl.java
+++ b/src/main/java/net/floodlightcontroller/topology/internal/TopologyImpl.java
@@ -858,7 +858,10 @@ public class TopologyImpl implements IOFMessageListener, IOFSwitchListener,
*/
@Override
public void addedSwitch(IOFSwitch sw) {
- sendLLDPs(sw);
+ // It's probably overkill to send LLDP from all switches, but we don't
+ // know which switches might be connected to the new switch.
+ // Need to optimize when supporting a large number of switches.
+ sendLLDPs();
}
/**
diff --git a/src/main/python/debugserver.py b/src/main/python/debugserver.py
index 2f6346f028a2bfacbfd6640a7acb7505ec74bb20..d8c81f9d49c3ff1c1e77f3b04acefa9dbfbf93e5 100644
--- a/src/main/python/debugserver.py
+++ b/src/main/python/debugserver.py
@@ -1,11 +1,23 @@
#!/usr/bin/env python
import sys
+from threading import currentThread
from SocketServer import BaseRequestHandler, TCPServer
from code import InteractiveConsole
_locals = None
+class DebugLogger(object):
+ def do_print(self, *args):
+ for i in args:
+ print i,
+ print
+ info = do_print
+ warn = do_print
+ debug = do_print
+_log = DebugLogger()
+
+
class DebugConsole(InteractiveConsole):
def __init__(self, request):
self.request = request
@@ -26,7 +38,8 @@ class DebugConsole(InteractiveConsole):
class DebugServerHandler(BaseRequestHandler):
def __init__(self, request, client_address, server):
- print 'Open connection to DebugServer from: %s' % str(client_address)
+ currentThread()._thread.setName("debugserver-%s:%d" % client_address)
+ _log.debug('Open connection to DebugServer from %s:%d' % client_address)
BaseRequestHandler.__init__(self, request, client_address, server)
def handle(self):
@@ -36,15 +49,23 @@ class DebugServerHandler(BaseRequestHandler):
self.request.close()
class DebugServer(TCPServer):
+ daemon_threads = True
+ allow_reuse_address = True
+
def handle_error(self, request, client_address):
- print 'Closing connection to DebugServer from: %s' % str(client_address)
+ _log.debug('Closing connection to DebugServer from %s:%d' % client_address)
request.close()
def run_server(port=6655, host='0.0.0.0', locals=locals()):
+ currentThread()._thread.setName("debugserver-main")
+
global _locals
_locals = locals
+ if "log" in locals.keys():
+ global _log
+ _log = locals["log"]
- print "Starting DebugServer on port %d" % port
+ _log.info("Starting DebugServer on port %d" % port)
server = DebugServer(('', port), DebugServerHandler)
try:
server.serve_forever()
diff --git a/src/test/java/net/floodlightcontroller/core/internal/ControllerTest.java b/src/test/java/net/floodlightcontroller/core/internal/ControllerTest.java
index fe686ab7e8109f703cca7a23cffd14a3beea57f5..17993d3adfdb7c94a3d7df0539dde352925fe5c5 100644
--- a/src/test/java/net/floodlightcontroller/core/internal/ControllerTest.java
+++ b/src/test/java/net/floodlightcontroller/core/internal/ControllerTest.java
@@ -27,6 +27,7 @@ import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import net.floodlightcontroller.core.FloodlightContext;
import net.floodlightcontroller.core.IFloodlightProvider;
@@ -390,9 +391,11 @@ public class ControllerTest extends FloodlightTestCase {
@Test
public void testAddSwitch() throws Exception {
controller.switches = new ConcurrentHashMap<Long, IOFSwitch>();
+ ReentrantReadWriteLock rwlock = new ReentrantReadWriteLock();
IOFSwitch oldsw = createMock(IOFSwitch.class);
expect(oldsw.getId()).andReturn(0L).anyTimes();
+ expect(oldsw.asyncRemoveSwitchLock()).andReturn(rwlock.writeLock()).anyTimes();
oldsw.setConnected(false);
expect(oldsw.getFeaturesReply()).andReturn(new OFFeaturesReply()).anyTimes();
expect(oldsw.getStringId()).andReturn("00:00:00:00:00:00:00").anyTimes();