diff --git a/logback.xml b/logback.xml
index f5163007d4c2af4ea21c0c62b023614e2689ee79..0485717cddefe39bd795685712cce3e4c0af18ad 100644
--- a/logback.xml
+++ b/logback.xml
@@ -7,9 +7,8 @@
<root level="DEBUG">
<appender-ref ref="STDOUT" />
</root>
- <logger name="org" level="DEBUG"/>
+ <logger name="io" level="INFO"></logger> <!-- Netty logging -->
<logger name="LogService" level="DEBUG"/> <!-- Restlet access logging -->
- <logger name="net.floodlightcontroller" level="DEBUG"/>
- <logger name="net.floodlightcontroller.logging" level="DEBUG"/>
+ <logger name="net.floodlightcontroller" level="INFO"/>
<logger name="org.sdnplatform" level="INFO"/>
</configuration>
diff --git a/src/main/java/net/floodlightcontroller/core/IOFConnection.java b/src/main/java/net/floodlightcontroller/core/IOFConnection.java
index 21855fd33aeb96714451ae38d5dee1b88dc16383..2537d40a7984c6843ed8b4b206e4cdce53034711 100644
--- a/src/main/java/net/floodlightcontroller/core/IOFConnection.java
+++ b/src/main/java/net/floodlightcontroller/core/IOFConnection.java
@@ -22,12 +22,6 @@ public interface IOFConnection extends IOFMessageWriter {
*/
Date getConnectedSince();
- /**
- * Flush all flows queued for this switch in the current thread.
- * NOTE: The contract is limited to the current thread
- */
- void flush();
-
/** @return the DatapathId of the switch associated with the connection */
DatapathId getDatapathId();
diff --git a/src/main/java/net/floodlightcontroller/core/IOFSwitch.java b/src/main/java/net/floodlightcontroller/core/IOFSwitch.java
index 9e51dd1f46124c8e8633c31df8d737c654f51054..cc138ecc6207be57b387c89c630faeb92b224aad 100644
--- a/src/main/java/net/floodlightcontroller/core/IOFSwitch.java
+++ b/src/main/java/net/floodlightcontroller/core/IOFSwitch.java
@@ -290,13 +290,6 @@ public interface IOFSwitch extends IOFMessageWriter {
*/
OFFactory getOFFactory();
- /**
- * Flush all flows queued for this switch on all connections that were written by the current thread.
- *
- *
- */
- void flush();
-
/**
* Gets the OF connections for this switch instance
* @return Collection of IOFConnection
diff --git a/src/main/java/net/floodlightcontroller/core/internal/Controller.java b/src/main/java/net/floodlightcontroller/core/internal/Controller.java
index fa8194649cc4fdc862fe584361592d3a94ff9aca..3a15697b68a5cfb0524716d5b8211ea2a7681157 100644
--- a/src/main/java/net/floodlightcontroller/core/internal/Controller.java
+++ b/src/main/java/net/floodlightcontroller/core/internal/Controller.java
@@ -133,7 +133,8 @@ public class Controller implements IFloodlightProviderService, IStorageSourceLis
// Configuration options
private static TransportPort openFlowPort = TransportPort.of(6653); // new registered OF port number
private static Set<IPv4Address> openFlowAddresses = new HashSet<IPv4Address>();
- protected int workerThreads = 0;
+ public static final int SEND_BUFFER_SIZE = 4 * 1024 * 1024;
+ protected int workerThreads = 16;
// The id for this controller node. Should be unique for each controller
// node in a controller cluster.
@@ -143,8 +144,7 @@ public class Controller implements IFloodlightProviderService, IStorageSourceLis
// if they should operate in ACTIVE / STANDBY
protected volatile HARole notifiedRole;
- private static final String
- INITIAL_ROLE_CHANGE_DESCRIPTION = "Controller startup.";
+ private static final String INITIAL_ROLE_CHANGE_DESCRIPTION = "Controller startup.";
/**
* NOTE: roleManager is not 'final' because it's initialized at run time
* based on parameters that are only available in init()
@@ -176,10 +176,6 @@ public class Controller implements IFloodlightProviderService, IStorageSourceLis
FLOW_COLUMN_ACCESS_PRIORITY,
FLOW_COLUMN_CORE_PRIORITY
};
-
- // Perf. related configuration
- protected static final int SEND_BUFFER_SIZE = 128 * 1024;
- public static final int BATCH_MAX_SIZE = 1; //TODO @Ryan this was 100. Causes packet_out messages to stall until 100 accumulated...
protected static final boolean ALWAYS_DECODE_ETH = true;
// Set of port name prefixes that will be classified as uplink ports,
@@ -590,7 +586,6 @@ public class Controller implements IFloodlightProviderService, IStorageSourceLis
if (m == null)
throw new NullPointerException("OFMessage must not be null");
- // FIXME floodlight context not supported any more
FloodlightContext bc = new FloodlightContext();
List<IOFMessageListener> listeners = null;
diff --git a/src/main/java/net/floodlightcontroller/core/internal/NullConnection.java b/src/main/java/net/floodlightcontroller/core/internal/NullConnection.java
index 2db236e45f64a6e475e4729d3e0653604f6d918a..75ce28049cc4b086d5bf1c5ac3a6d3872b48d0c9 100644
--- a/src/main/java/net/floodlightcontroller/core/internal/NullConnection.java
+++ b/src/main/java/net/floodlightcontroller/core/internal/NullConnection.java
@@ -77,11 +77,6 @@ public class NullConnection implements IOFConnectionBackend, IOFMessageWriter {
// noop
}
- @Override
- public void flush() {
- // noop
- }
-
@Override
public <R extends OFMessage> ListenableFuture<R> writeRequest(OFRequest<R> request) {
return Futures.immediateFailedFuture(new SwitchDisconnectedException(getDatapathId()));
diff --git a/src/main/java/net/floodlightcontroller/core/internal/OFConnection.java b/src/main/java/net/floodlightcontroller/core/internal/OFConnection.java
index 791ced36bf710eb2a59f2d2def57a1f68f5539a7..1039a0e933926577efeb5b26f229c66431146cf4 100644
--- a/src/main/java/net/floodlightcontroller/core/internal/OFConnection.java
+++ b/src/main/java/net/floodlightcontroller/core/internal/OFConnection.java
@@ -50,7 +50,6 @@ import org.projectfloodlight.openflow.protocol.OFRequest;
import org.projectfloodlight.openflow.protocol.OFStatsReply;
import org.projectfloodlight.openflow.protocol.OFStatsReplyFlags;
import org.projectfloodlight.openflow.protocol.OFStatsRequest;
-import org.projectfloodlight.openflow.protocol.OFType;
import org.projectfloodlight.openflow.types.DatapathId;
import org.projectfloodlight.openflow.types.OFAuxId;
import org.projectfloodlight.openflow.types.U64;
@@ -71,323 +70,329 @@ import com.google.common.util.concurrent.ListenableFuture;
* @author Andreas Wundsam <andreas.wundsam@bigswitch.com>
*/
public class OFConnection implements IOFConnection, IOFConnectionBackend{
- private static final Logger logger = LoggerFactory.getLogger(OFConnection.class);
- private final DatapathId dpid;
- private final OFFactory factory;
- private final Channel channel;
- private final OFAuxId auxId;
- private final Timer timer;
-
- private final Date connectedSince;
-
- private final Map<Long, Deliverable<?>> xidDeliverableMap;
-
- protected final static ThreadLocal<List<OFMessage>> localMsgBuffer =
- new ThreadLocal<List<OFMessage>>();
-
- private static final long DELIVERABLE_TIME_OUT = 60;
- private static final TimeUnit DELIVERABLE_TIME_OUT_UNIT = TimeUnit.SECONDS;
-
- private final OFConnectionCounters counters;
- private IOFConnectionListener listener;
-
- private volatile U64 latency;
-
- public OFConnection(@Nonnull DatapathId dpid,
- @Nonnull OFFactory factory,
- @Nonnull Channel channel,
- @Nonnull OFAuxId auxId,
- @Nonnull IDebugCounterService debugCounters,
- @Nonnull Timer timer) {
- Preconditions.checkNotNull(dpid, "dpid");
- Preconditions.checkNotNull(factory, "factory");
- Preconditions.checkNotNull(channel, "channel");
- Preconditions.checkNotNull(timer, "timer");
- Preconditions.checkNotNull(debugCounters);
-
- this.listener = NullConnectionListener.INSTANCE;
- this.dpid = dpid;
- this.factory = factory;
- this.channel = channel;
- this.auxId = auxId;
- this.connectedSince = new Date();
- this.xidDeliverableMap = new ConcurrentHashMap<>();
- this.counters = new OFConnectionCounters(debugCounters, dpid, this.auxId);
- this.timer = timer;
- this.latency = U64.ZERO;
- }
-
- @Override
- public void write(OFMessage m) {
- if (!isConnected()) {
- if (logger.isDebugEnabled())
- logger.debug("{}: not connected - dropping message {}", this, m);
- return;
- }
- if (logger.isDebugEnabled())
- logger.debug("{}: send {}", this, m);
- List<OFMessage> msgBuffer = localMsgBuffer.get();
- if (msgBuffer == null) {
- msgBuffer = new ArrayList<OFMessage>();
- localMsgBuffer.set(msgBuffer);
- }
-
- counters.updateWriteStats(m);
- msgBuffer.add(m);
-
- if ((msgBuffer.size() >= Controller.BATCH_MAX_SIZE) || ((m.getType() != OFType.PACKET_OUT) && (m.getType() != OFType.FLOW_MOD))) {
- this.write(msgBuffer);
- localMsgBuffer.set(null);
- }
- }
-
- @Override
- public <R extends OFMessage> ListenableFuture<R> writeRequest(OFRequest<R> request) {
- if (!isConnected())
- return Futures.immediateFailedFuture(new SwitchDisconnectedException(getDatapathId()));
-
- DeliverableListenableFuture<R> future = new DeliverableListenableFuture<R>();
- xidDeliverableMap.put(request.getXid(), future);
- listener.messageWritten(this, request);
- write(request);
- return future;
- }
-
- @Override
- public void write(Iterable<OFMessage> msglist) {
- if (!isConnected()) {
- if (logger.isDebugEnabled())
- logger.debug(this.toString() + " : not connected - dropping {} element msglist {} ",
- Iterables.size(msglist),
- String.valueOf(msglist).substring(0, 80));
- return;
- }
- for (OFMessage m : msglist) {
- if (logger.isTraceEnabled())
- logger.trace("{}: send {}", this, m);
- counters.updateWriteStats(m);
- }
- this.channel.writeAndFlush(msglist);
- }
-
- // Notifies the connection object that the channel has been disconnected
- public void disconnected() {
- SwitchDisconnectedException exception = new SwitchDisconnectedException(getDatapathId());
- for (Long xid : xidDeliverableMap.keySet()) {
- // protect against other mechanisms running at the same time
- // (timeout)
- Deliverable<?> removed = xidDeliverableMap.remove(xid);
- if (removed != null) {
- removed.deliverError(exception);
- }
- }
- }
-
- @Override
- public void disconnect() {
- this.channel.disconnect();
- this.counters.uninstallCounters();
- }
-
- @Override
- public String toString() {
- String channelString = (channel != null) ? String.valueOf(channel.remoteAddress()): "?";
- return "OFConnection [" + getDatapathId() + "(" + getAuxId() + ")" + "@" + channelString + "]";
- }
-
- @Override
- public Date getConnectedSince() {
- return connectedSince;
- }
-
- @Override
- public <REPLY extends OFStatsReply> ListenableFuture<List<REPLY>> writeStatsRequest(
- OFStatsRequest<REPLY> request) {
- if (!isConnected())
- return Futures.immediateFailedFuture(new SwitchDisconnectedException(getDatapathId()));
-
- final DeliverableListenableFuture<List<REPLY>> future =
- new DeliverableListenableFuture<List<REPLY>>();
-
- Deliverable<REPLY> deliverable = new Deliverable<REPLY>() {
- private final List<REPLY> results = Collections
- .synchronizedList(new ArrayList<REPLY>());
-
- @Override
- public void deliver(REPLY reply) {
- results.add(reply);
- if (!reply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
- // done
- future.deliver(results);
- }
- }
-
- @Override
- public void deliverError(Throwable cause) {
- future.deliverError(cause);
- }
-
- @Override
- public boolean isDone() {
- return future.isDone();
- }
-
- @Override
- public boolean cancel(boolean mayInterruptIfRunning) {
- return future.cancel(mayInterruptIfRunning);
- }
- };
-
- registerDeliverable(request.getXid(), deliverable);
- this.write(request);
- return future;
- }
-
- private void registerDeliverable(long xid, Deliverable<?> deliverable) {
- this.xidDeliverableMap.put(xid, deliverable);
- timer.newTimeout(new TimeOutDeliverable(xid), DELIVERABLE_TIME_OUT, DELIVERABLE_TIME_OUT_UNIT);
- }
-
- public boolean handleGenericDeliverable(OFMessage reply) {
- counters.updateReadStats(reply);
- @SuppressWarnings("unchecked")
- Deliverable<OFMessage> deliverable =
- (Deliverable<OFMessage>) this.xidDeliverableMap.get(reply.getXid());
- if (deliverable != null) {
- if(reply instanceof OFErrorMsg) {
- deliverable.deliverError(new OFErrorMsgException((OFErrorMsg) reply));
- } else {
- deliverable.deliver(reply);
- }
- if (deliverable.isDone())
- this.xidDeliverableMap.remove(reply.getXid());
- return true;
- } else {
- return false;
- }
- }
-
- @Override
- public void cancelAllPendingRequests() {
- /*
- * we don't need to be synchronized here. Even if another thread
- * modifies the map while we're cleaning up the future will eventually
- * timeout
- */
- for (Deliverable<?> d : xidDeliverableMap.values()) {
- d.cancel(true);
- }
- xidDeliverableMap.clear();
- }
-
- @Override
- public boolean isConnected() {
- return channel.isActive();
- }
-
- @Override
- public void flush() {
- List<OFMessage> msglist = localMsgBuffer.get();
- if ((msglist != null) && (msglist.size() > 0)) {
- this.write(msglist);
- localMsgBuffer.set(null);
- }
- }
-
- @Override
- public SocketAddress getRemoteInetAddress() {
- return channel.remoteAddress();
- }
-
- @Override
- public SocketAddress getLocalInetAddress() {
- return channel.localAddress();
- }
-
- public boolean deliverResponse(OFMessage m) {
- if (handleGenericDeliverable(m))
- return true;
- else
- return false;
- }
-
- @Override
- public boolean isWritable() {
- return channel.isWritable();
- }
-
- @Override
- public DatapathId getDatapathId() {
- return dpid;
- }
-
- @Override
- public OFAuxId getAuxId() {
- return auxId;
- }
-
- Set<Long> getPendingRequestIds() {
- return ImmutableSet.copyOf(xidDeliverableMap.keySet());
- }
-
- @Override
- public OFFactory getOFFactory() {
- return this.factory;
- }
-
- /**
- * Timeout class instantiated for deliverables. Will throw a timeout exception
- * if proper responses are not received in time.
- *
- */
- private class TimeOutDeliverable implements TimerTask {
- private final long xid;
-
- public TimeOutDeliverable(long xid) {
- this.xid = xid;
- }
-
- @Override
- public void run(Timeout timeout) throws Exception {
- Deliverable<?> removed = xidDeliverableMap.remove(xid);
- if (removed != null && !removed.isDone()) {
- removed.deliverError(new TimeoutException(
- "timeout - did not receive answer for xid " + xid));
- }
-
- }
- }
-
- public IOFConnectionListener getListener() {
- return listener;
- }
-
- /** set the connection listener
- * <p>
- * Note: this is assumed to be called from the Connection's IO Thread.
- *
- * @param listener
- */
- @Override
- public void setListener(IOFConnectionListener listener) {
- this.listener = listener;
- }
-
- public void messageReceived(OFMessage m) {
- // Check if message was a response for a xid waiting at the switch
- if(!deliverResponse(m)){
- listener.messageReceived(this, m);
- }
- }
-
- @Override
- public U64 getLatency() {
- return this.latency;
- }
-
- @Override
- public void updateLatency(U64 latency) {
- if (latency == null) {
+ private static final Logger logger = LoggerFactory.getLogger(OFConnection.class);
+ private final DatapathId dpid;
+ private final OFFactory factory;
+
+ /** CAREFUL CAREFUL CAREFUL:
+ *
+ * Netty4 does not guarantee order of messages that are written into the channel any more.
+ * To ensure messages do not get reordered, never directly call {@link Channel#write(Object)}.
+ *
+ * Instead, use {@link #write(Iterable)}, which queue up write request on the EventLoop,
+ * to make sure they are handled in order.
+ */
+ private final Channel channel;
+
+ private final OFAuxId auxId;
+ private final Timer timer;
+
+ private final Date connectedSince;
+
+ private final Map<Long, Deliverable<?>> xidDeliverableMap;
+
+ private static final long DELIVERABLE_TIME_OUT = 60;
+ private static final TimeUnit DELIVERABLE_TIME_OUT_UNIT = TimeUnit.SECONDS;
+
+ private final OFConnectionCounters counters;
+ private IOFConnectionListener listener;
+
+ private volatile U64 latency;
+
+ /**
+ * Used to write messages to ensure order w/Netty4.
+ * It also ensures we do not reuse the array, since
+ * Netty4 will write the object, not the items.
+ */
+ private class WriteMessageTask implements Runnable {
+ private final Iterable<OFMessage> msglist;
+
+ public WriteMessageTask(Iterable<OFMessage> msglist) {
+ this.msglist = msglist;
+ }
+
+ @Override
+ public void run() {
+ for (OFMessage m : msglist) {
+ if (logger.isTraceEnabled())
+ logger.trace("{}: send {}", this, m);
+ counters.updateWriteStats(m);
+ }
+ channel.writeAndFlush(msglist);
+ }
+ }
+
+ public OFConnection(@Nonnull DatapathId dpid,
+ @Nonnull OFFactory factory,
+ @Nonnull Channel channel,
+ @Nonnull OFAuxId auxId,
+ @Nonnull IDebugCounterService debugCounters,
+ @Nonnull Timer timer) {
+ Preconditions.checkNotNull(dpid, "dpid");
+ Preconditions.checkNotNull(factory, "factory");
+ Preconditions.checkNotNull(channel, "channel");
+ Preconditions.checkNotNull(timer, "timer");
+ Preconditions.checkNotNull(debugCounters);
+
+ this.listener = NullConnectionListener.INSTANCE;
+ this.dpid = dpid;
+ this.factory = factory;
+ this.channel = channel;
+ this.auxId = auxId;
+ this.connectedSince = new Date();
+ this.xidDeliverableMap = new ConcurrentHashMap<>();
+ this.counters = new OFConnectionCounters(debugCounters, dpid, this.auxId);
+ this.timer = timer;
+ this.latency = U64.ZERO;
+ }
+
+ /**
+ * All write methods chain into this write() to use WriteMessageTask.
+ */
+ @Override
+ public void write(final Iterable<OFMessage> msgList) {
+ if (!isConnected()) {
+ if (logger.isDebugEnabled())
+ logger.debug(this.toString() + " : not connected - dropping {} element msglist {} ",
+ Iterables.size(msgList),
+ String.valueOf(msgList).substring(0, 80));
+ return;
+ }
+ for (OFMessage m : msgList) {
+ if (logger.isTraceEnabled())
+ logger.trace("{}: send {}", this, m);
+ counters.updateWriteStats(m);
+ }
+ this.channel.eventLoop().execute(new WriteMessageTask(msgList));
+ }
+
+ @Override
+ public void write(OFMessage m) {
+ this.write(Collections.singletonList(m));
+ }
+
+ @Override
+ public <R extends OFMessage> ListenableFuture<R> writeRequest(OFRequest<R> request) {
+ if (!isConnected()) {
+ return Futures.immediateFailedFuture(new SwitchDisconnectedException(getDatapathId()));
+ }
+
+ DeliverableListenableFuture<R> future = new DeliverableListenableFuture<R>();
+ xidDeliverableMap.put(request.getXid(), future);
+ listener.messageWritten(this, request);
+ this.write(request);
+ return future;
+ }
+
+ @Override
+ public <REPLY extends OFStatsReply> ListenableFuture<List<REPLY>> writeStatsRequest(
+ OFStatsRequest<REPLY> request) {
+ if (!isConnected()) {
+ return Futures.immediateFailedFuture(new SwitchDisconnectedException(getDatapathId()));
+ }
+
+ final DeliverableListenableFuture<List<REPLY>> future =
+ new DeliverableListenableFuture<List<REPLY>>();
+
+ Deliverable<REPLY> deliverable = new Deliverable<REPLY>() {
+ private final List<REPLY> results = Collections
+ .synchronizedList(new ArrayList<REPLY>());
+
+ @Override
+ public void deliver(REPLY reply) {
+ results.add(reply);
+ if (!reply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
+ // done
+ future.deliver(results);
+ }
+ }
+
+ @Override
+ public void deliverError(Throwable cause) {
+ future.deliverError(cause);
+ }
+
+ @Override
+ public boolean isDone() {
+ return future.isDone();
+ }
+
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ return future.cancel(mayInterruptIfRunning);
+ }
+ };
+
+ registerDeliverable(request.getXid(), deliverable);
+ this.write(request);
+ return future;
+ }
+
+ public void disconnected() {
+ SwitchDisconnectedException exception = new SwitchDisconnectedException(getDatapathId());
+ for (Long xid : xidDeliverableMap.keySet()) {
+ // protect against other mechanisms running at the same time
+ // (timeout)
+ Deliverable<?> removed = xidDeliverableMap.remove(xid);
+ if (removed != null) {
+ removed.deliverError(exception);
+ }
+ }
+ }
+
+ @Override
+ public void disconnect() {
+ this.channel.disconnect();
+ this.counters.uninstallCounters();
+ }
+
+ @Override
+ public String toString() {
+ String channelString = (channel != null) ? String.valueOf(channel.remoteAddress()): "?";
+ return "OFConnection [" + getDatapathId() + "(" + getAuxId() + ")" + "@" + channelString + "]";
+ }
+
+ @Override
+ public Date getConnectedSince() {
+ return connectedSince;
+ }
+
+ private void registerDeliverable(long xid, Deliverable<?> deliverable) {
+ this.xidDeliverableMap.put(xid, deliverable);
+ timer.newTimeout(new TimeOutDeliverable(xid), DELIVERABLE_TIME_OUT, DELIVERABLE_TIME_OUT_UNIT);
+ }
+
+ public boolean handleGenericDeliverable(OFMessage reply) {
+ counters.updateReadStats(reply);
+ @SuppressWarnings("unchecked")
+ Deliverable<OFMessage> deliverable =
+ (Deliverable<OFMessage>) this.xidDeliverableMap.get(reply.getXid());
+ if (deliverable != null) {
+ if(reply instanceof OFErrorMsg) {
+ deliverable.deliverError(new OFErrorMsgException((OFErrorMsg) reply));
+ } else {
+ deliverable.deliver(reply);
+ }
+ if (deliverable.isDone())
+ this.xidDeliverableMap.remove(reply.getXid());
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public void cancelAllPendingRequests() {
+ /*
+ * we don't need to be synchronized here. Even if another thread
+ * modifies the map while we're cleaning up the future will eventually
+ * timeout
+ */
+ for (Deliverable<?> d : xidDeliverableMap.values()) {
+ d.cancel(true);
+ }
+ xidDeliverableMap.clear();
+ }
+
+ @Override
+ public boolean isConnected() {
+ return channel.isActive();
+ }
+
+ @Override
+ public SocketAddress getRemoteInetAddress() {
+ return channel.remoteAddress();
+ }
+
+ @Override
+ public SocketAddress getLocalInetAddress() {
+ return channel.localAddress();
+ }
+
+ public boolean deliverResponse(OFMessage m) {
+ if (handleGenericDeliverable(m))
+ return true;
+ else
+ return false;
+ }
+
+ @Override
+ public boolean isWritable() {
+ return channel.isWritable();
+ }
+
+ @Override
+ public DatapathId getDatapathId() {
+ return dpid;
+ }
+
+ @Override
+ public OFAuxId getAuxId() {
+ return auxId;
+ }
+
+ Set<Long> getPendingRequestIds() {
+ return ImmutableSet.copyOf(xidDeliverableMap.keySet());
+ }
+
+ @Override
+ public OFFactory getOFFactory() {
+ return this.factory;
+ }
+
+ /**
+ * Timeout class instantiated for deliverables. Will throw a timeout exception
+ * if proper responses are not received in time.
+ *
+ */
+ private class TimeOutDeliverable implements TimerTask {
+ private final long xid;
+
+ public TimeOutDeliverable(long xid) {
+ this.xid = xid;
+ }
+
+ @Override
+ public void run(Timeout timeout) throws Exception {
+ Deliverable<?> removed = xidDeliverableMap.remove(xid);
+ if (removed != null && !removed.isDone()) {
+ removed.deliverError(new TimeoutException(
+ "timeout - did not receive answer for xid " + xid));
+ }
+
+ }
+ }
+
+ public IOFConnectionListener getListener() {
+ return listener;
+ }
+
+ /** set the connection listener
+ * <p>
+ * Note: this is assumed to be called from the Connection's IO Thread.
+ *
+ * @param listener
+ */
+ @Override
+ public void setListener(IOFConnectionListener listener) {
+ this.listener = listener;
+ }
+
+ public void messageReceived(OFMessage m) {
+ // Check if message was a response for a xid waiting at the switch
+ if(!deliverResponse(m)){
+ listener.messageReceived(this, m);
+ }
+ }
+
+ @Override
+ public U64 getLatency() {
+ return this.latency;
+ }
+
+ @Override
+ public void updateLatency(U64 latency) {
+ if (latency == null) {
logger.error("Latency must be non-null. Ignoring null latency value.");
return;
} else if (this.latency.equals(U64.ZERO)) {
@@ -399,35 +404,35 @@ public class OFConnection implements IOFConnection, IOFConnectionBackend{
this.latency = U64.of((long) (this.latency.getValue() * oldWeight + latency.getValue() * (1 - oldWeight)));
logger.debug("Switch {} latency updated to {}ms", this.getDatapathId(), this.latency.getValue());
}
- }
+ }
- /** A dummy connection listener that just logs warn messages. Saves us a few null checks
- * @author Andreas Wundsam <andreas.wundsam@bigswitch.com>
- */
- private static class NullConnectionListener implements IOFConnectionListener {
- public final static NullConnectionListener INSTANCE = new NullConnectionListener();
+ /** A dummy connection listener that just logs warn messages. Saves us a few null checks
+ * @author Andreas Wundsam <andreas.wundsam@bigswitch.com>
+ */
+ private static class NullConnectionListener implements IOFConnectionListener {
+ public final static NullConnectionListener INSTANCE = new NullConnectionListener();
- private NullConnectionListener() { }
+ private NullConnectionListener() { }
- @Override
- public void connectionClosed(IOFConnectionBackend connection) {
- logger.warn("NullConnectionListener for {} - received connectionClosed", connection);
- }
+ @Override
+ public void connectionClosed(IOFConnectionBackend connection) {
+ logger.warn("NullConnectionListener for {} - received connectionClosed", connection);
+ }
- @Override
- public void messageReceived(IOFConnectionBackend connection, OFMessage m) {
- logger.warn("NullConnectionListener for {} - received messageReceived: {}", connection, m);
- }
+ @Override
+ public void messageReceived(IOFConnectionBackend connection, OFMessage m) {
+ logger.warn("NullConnectionListener for {} - received messageReceived: {}", connection, m);
+ }
- @Override
- public boolean isSwitchHandshakeComplete(IOFConnectionBackend connection) {
- return false;
- }
+ @Override
+ public boolean isSwitchHandshakeComplete(IOFConnectionBackend connection) {
+ return false;
+ }
@Override
public void messageWritten(IOFConnectionBackend connection, OFMessage m) {
// TODO Auto-generated method stub
-
+
}
- }
+ }
}
\ No newline at end of file
diff --git a/src/main/java/net/floodlightcontroller/core/internal/OFSwitch.java b/src/main/java/net/floodlightcontroller/core/internal/OFSwitch.java
index 1c95a2bfe2ff9e84fde91d75daa025d826b095e8..d9f6ea24bce4032c41f4624eaf81adc841ed50e6 100644
--- a/src/main/java/net/floodlightcontroller/core/internal/OFSwitch.java
+++ b/src/main/java/net/floodlightcontroller/core/internal/OFSwitch.java
@@ -39,7 +39,6 @@ import javax.annotation.Nonnull;
import net.floodlightcontroller.core.IOFConnection;
import net.floodlightcontroller.core.IOFConnectionBackend;
-import net.floodlightcontroller.core.IOFSwitch;
import net.floodlightcontroller.core.IOFSwitchBackend;
import net.floodlightcontroller.core.LogicalOFMessageCategory;
import net.floodlightcontroller.core.PortChangeEvent;
@@ -48,7 +47,6 @@ import net.floodlightcontroller.core.SwitchDescription;
import net.floodlightcontroller.core.SwitchDriverSubHandshakeAlreadyStarted;
import net.floodlightcontroller.core.SwitchDriverSubHandshakeCompleted;
import net.floodlightcontroller.core.SwitchDriverSubHandshakeNotStarted;
-import net.floodlightcontroller.core.IOFSwitch.SwitchStatus;
import net.floodlightcontroller.core.util.AppCookie;
import net.floodlightcontroller.core.util.URIUtil;
@@ -730,17 +728,6 @@ public class OFSwitch implements IOFSwitchBackend {
this.connections.remove(connection.getAuxId());
}
- @Override
- public void write(OFMessage m) {
- log.trace("Channel: {}, Connected: {}", connections.get(OFAuxId.MAIN).getRemoteInetAddress(), connections.get(OFAuxId.MAIN).isConnected());
- if (isActive()) {
- connections.get(OFAuxId.MAIN).write(m);
- switchManager.handleOutgoingMessage(this, m);
- } else {
- log.warn("Attempted to write to switch {} that is SLAVE.", this.getId().toString());
- }
- }
-
/**
* Gets a connection specified by aux Id.
* @param auxId the specified aux id for the connection desired.
@@ -764,21 +751,32 @@ public class OFSwitch implements IOFSwitchBackend {
}
@Override
- public void write(OFMessage m, LogicalOFMessageCategory category) {
+ public void write(OFMessage m) {
+ this.write(Collections.singletonList(m));
+ }
+
+ @Override
+ public void write(Iterable<OFMessage> msglist) {
if (isActive()) {
- this.getConnection(category).write(m);
- switchManager.handleOutgoingMessage(this, m);
+ connections.get(OFAuxId.MAIN).write(msglist);
+ for (OFMessage m : msglist) {
+ switchManager.handleOutgoingMessage(this, m);
+ }
} else {
log.warn("Attempted to write to switch {} that is SLAVE.", this.getId().toString());
}
}
+
+ @Override
+ public void write(OFMessage m, LogicalOFMessageCategory category) {
+ this.write(Collections.singletonList(m), category);
+ }
@Override
public void write(Iterable<OFMessage> msglist, LogicalOFMessageCategory category) {
if (isActive()) {
this.getConnection(category).write(msglist);
-
- for(OFMessage m : msglist) {
+ for (OFMessage m : msglist) {
switchManager.handleOutgoingMessage(this, m);
}
} else {
@@ -801,24 +799,10 @@ public class OFSwitch implements IOFSwitchBackend {
return connections.get(OFAuxId.MAIN).writeRequest(request);
}
- @Override
- public void write(Iterable<OFMessage> msglist) {
- if (isActive()) {
- connections.get(OFAuxId.MAIN).write(msglist);
-
- for(OFMessage m : msglist) {
- switchManager.handleOutgoingMessage(this, m);
- }
- } else {
- log.warn("Attempted to write to switch {} that is SLAVE.", this.getId().toString());
- }
- }
-
@Override
public void disconnect() {
-
// Iterate through connections and perform cleanup
- for(Entry<OFAuxId, IOFConnectionBackend> entry : this.connections.entrySet()){
+ for (Entry<OFAuxId, IOFConnectionBackend> entry : this.connections.entrySet()) {
entry.getValue().disconnect();
this.connections.remove(entry.getKey());
}
@@ -953,12 +937,9 @@ public class OFSwitch implements IOFSwitchBackend {
return datapathId;
}
- /* (non-Javadoc)
- * @see java.lang.Object#toString()
- */
@Override
public String toString() {
- return "OFSwitchBase DPID[" + ((datapathId != null) ? datapathId.toString() : "?") + "]";
+ return "OFSwitch DPID[" + ((datapathId != null) ? datapathId.toString() : "?") + "]";
}
@Override
@@ -1067,13 +1048,6 @@ public class OFSwitch implements IOFSwitchBackend {
this.role = role;
}
- @Override
- public void flush() {
- for(Entry<OFAuxId, IOFConnectionBackend> entry : this.connections.entrySet()){
- entry.getValue().flush();
- }
- }
-
/**
* Get the IP Address for the switch
* @return the inet address
diff --git a/src/main/java/net/floodlightcontroller/linkdiscovery/internal/LinkDiscoveryManager.java b/src/main/java/net/floodlightcontroller/linkdiscovery/internal/LinkDiscoveryManager.java
index a26d8d8fd4a6c39539b2c92c347d0848a3d6ecdd..c50930ed6580cb5cf7a499d83b912522fd2c2821 100644
--- a/src/main/java/net/floodlightcontroller/linkdiscovery/internal/LinkDiscoveryManager.java
+++ b/src/main/java/net/floodlightcontroller/linkdiscovery/internal/LinkDiscoveryManager.java
@@ -1219,7 +1219,6 @@ IFloodlightModule, IInfoProvider {
// send
// no more try-catch. switch will silently fail
iofSwitch.write(pob.build());
- iofSwitch.flush();
}
/**
diff --git a/src/main/java/net/floodlightcontroller/staticflowentry/StaticFlowEntryPusher.java b/src/main/java/net/floodlightcontroller/staticflowentry/StaticFlowEntryPusher.java
index d4e95df918bb5792906e633c6bf5cab14c9e3780..7875e70267b68508c23d65ad099c50bb66282829 100644
--- a/src/main/java/net/floodlightcontroller/staticflowentry/StaticFlowEntryPusher.java
+++ b/src/main/java/net/floodlightcontroller/staticflowentry/StaticFlowEntryPusher.java
@@ -610,7 +610,6 @@ implements IOFSwitchListener, IFloodlightModule, IStaticFlowEntryPusherService,
log.debug("Sending {} new entries to {}", messages.size(), dpid);
}
ofswitch.write(messages);
- ofswitch.flush();
}
}
@@ -626,7 +625,6 @@ implements IOFSwitchListener, IFloodlightModule, IStaticFlowEntryPusherService,
log.debug("Sending 1 new entries to {}", dpid.toString());
}
ofswitch.write(message);
- ofswitch.flush();
}
}
@@ -654,7 +652,6 @@ implements IOFSwitchListener, IFloodlightModule, IStaticFlowEntryPusherService,
*/
private void writeFlowModToSwitch(IOFSwitch sw, OFFlowMod flowMod) {
sw.write(flowMod);
- sw.flush();
}
@Override
public String getName() {
diff --git a/src/main/java/net/floodlightcontroller/util/OFMessageDamper.java b/src/main/java/net/floodlightcontroller/util/OFMessageDamper.java
index 296adc157236de7dac02f5f15d544abc66228450..3e1b86f59362d077e513705b76da1c235f956b7b 100644
--- a/src/main/java/net/floodlightcontroller/util/OFMessageDamper.java
+++ b/src/main/java/net/floodlightcontroller/util/OFMessageDamper.java
@@ -114,24 +114,8 @@ public class OFMessageDamper {
* @throws IOException
*/
public boolean write(IOFSwitch sw, OFMessage msg) throws IOException {
- return write(sw, msg, false);
- }
-
- /**
- * write the message to the switch according to our dampening settings
- * @param sw
- * @param msg
- * @param flush true to flush the packet immediately
- * @return true if the message was written to the switch, false if
- * the message was dampened.
- * @throws IOException
- */
- public boolean write(IOFSwitch sw, OFMessage msg, boolean flush) throws IOException {
if (!msgTypesToCache.contains(msg.getType())) {
sw.write(msg);
- if (flush) {
- sw.flush();
- }
return true;
}
@@ -141,9 +125,6 @@ public class OFMessageDamper {
return false;
} else {
sw.write(msg);
- if (flush) {
- sw.flush();
- }
return true;
}
}
diff --git a/src/main/resources/floodlightdefault.properties b/src/main/resources/floodlightdefault.properties
index fe32a37ed46be47de5f6405754885835eff6e6e1..89bf32381355331dd93a011d0e59d2a2581d4d5d 100644
--- a/src/main/resources/floodlightdefault.properties
+++ b/src/main/resources/floodlightdefault.properties
@@ -23,11 +23,11 @@ org.sdnplatform.sync.internal.SyncManager.dbPath=/var/lib/floodlight/
org.sdnplatform.sync.internal.SyncManager.port=6642
net.floodlightcontroller.forwarding.Forwarding.match=vlan, mac, ip, transport
net.floodlightcontroller.forwarding.Forwarding.flood-arp=NO
-net.floodlightcontroller.core.internal.FloodlightProvider.openflowPort=6653
+net.floodlightcontroller.core.internal.FloodlightProvider.openFlowPort=6653
net.floodlightcontroller.core.internal.FloodlightProvider.role=ACTIVE
net.floodlightcontroller.linkdiscovery.internal.LinkDiscoveryManager.latency-history-size=10
net.floodlightcontroller.linkdiscovery.internal.LinkDiscoveryManager.latency-update-threshold=0.5
-net.floodlightcontroller.core.internal.OFSwitchManager.defaultMaxTablesToReceiveTableMissFlow=2
+net.floodlightcontroller.core.internal.OFSwitchManager.defaultMaxTablesToReceiveTableMissFlow=1
net.floodlightcontroller.core.internal.OFSwitchManager.maxTablesToReceiveTableMissFlowPerDpid={"00:00:00:00:00:00:00:01":"1","2":"1"}
net.floodlightcontroller.core.internal.OFSwitchManager.clearTablesOnInitialHandshakeAsMaster=YES
net.floodlightcontroller.core.internal.OFSwitchManager.clearTablesOnEachTransitionToMaster=YES
diff --git a/src/main/resources/logback-test.xml b/src/main/resources/logback-test.xml
index b5e3bfd8093df991c049e5f86e873a9834d12747..8fa1e8f86065f20fa51ea0b42336af21b4e15cdc 100644
--- a/src/main/resources/logback-test.xml
+++ b/src/main/resources/logback-test.xml
@@ -11,6 +11,7 @@
<appender-ref ref="EV_WARN_ERR" />
</root>
<logger name="org" level="WARN"/>
+ <logger name="io" level="INFO"></logger> <!-- Netty logging -->
<logger name="LogService" level="DEBUG"></logger> <!-- Restlet access logging -->
<logger name="net.floodlightcontroller" level="INFO"/>
<logger name="org.sdnplatform" level="INFO"></logger>
diff --git a/src/test/java/net/floodlightcontroller/core/internal/MockOFConnection.java b/src/test/java/net/floodlightcontroller/core/internal/MockOFConnection.java
index c1eab8ba8baedd748aaddeb089504befb4cad6a0..597ab8523b4ce1874d12495d3b4c8c61b5e4c035 100644
--- a/src/test/java/net/floodlightcontroller/core/internal/MockOFConnection.java
+++ b/src/test/java/net/floodlightcontroller/core/internal/MockOFConnection.java
@@ -124,11 +124,6 @@ public class MockOFConnection implements IOFConnectionBackend {
return this.connectedSince;
}
- @Override
- public void flush() {
- // no op
- }
-
@Override
public DatapathId getDatapathId() {
return this.id;
diff --git a/src/test/java/net/floodlightcontroller/core/internal/OFChannelHandlerVer10Test.java b/src/test/java/net/floodlightcontroller/core/internal/OFChannelHandlerVer10Test.java
index 2460b3a7687c93ba64c32af6779ea7517847e031..39ec2714e9aa9170a2f115f46cef7533e6593af8 100644
--- a/src/test/java/net/floodlightcontroller/core/internal/OFChannelHandlerVer10Test.java
+++ b/src/test/java/net/floodlightcontroller/core/internal/OFChannelHandlerVer10Test.java
@@ -23,11 +23,14 @@ import java.util.Set;
import org.easymock.Capture;
import org.easymock.CaptureType;
import org.easymock.EasyMock;
+import org.easymock.IAnswer;
import org.hamcrest.CoreMatchers;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.DefaultChannelPromise;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
@@ -38,6 +41,7 @@ import org.junit.Test;
import net.floodlightcontroller.core.IOFConnectionBackend;
import net.floodlightcontroller.core.internal.OFChannelInitializer.PipelineHandler;
import net.floodlightcontroller.core.internal.OFChannelInitializer.PipelineHandshakeTimeout;
+import net.floodlightcontroller.core.test.TestEventLoop;
import net.floodlightcontroller.debugcounter.DebugCounterServiceImpl;
import net.floodlightcontroller.debugcounter.IDebugCounterService;
@@ -90,6 +94,8 @@ public class OFChannelHandlerVer10Test {
private INewOFConnectionListener newConnectionListener;
private Capture<IOFConnectionBackend> newConnection;
private Capture<OFFeaturesReply> newFeaturesReply;
+
+ private TestEventLoop eventLoop;
public void setUpFeaturesReply() {
portDesc = factory.buildPortDesc()
@@ -115,6 +121,7 @@ public class OFChannelHandlerVer10Test {
newConnectionListener = createMock(INewOFConnectionListener.class);
newConnection = new Capture<IOFConnectionBackend>();
newFeaturesReply = new Capture<OFFeaturesReply>();
+ eventLoop = new TestEventLoop();
ctx = createMock(ChannelHandlerContext.class);
channel = createMock(Channel.class);
@@ -176,6 +183,14 @@ public class OFChannelHandlerVer10Test {
/** Reset the channel mock and set basic method call expectations */
void resetChannel() {
reset(channel);
+ expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
+ @Override
+ public ChannelPromise answer() throws Throwable {
+ return new DefaultChannelPromise(channel);
+ }
+ }).anyTimes();
+ eventLoop = new TestEventLoop();
+ expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
expect(channel.pipeline()).andReturn(pipeline).anyTimes();
expect(channel.remoteAddress()).andReturn(InetSocketAddress.createUnresolved("1.1.1.1", 80)).anyTimes();
}
diff --git a/src/test/java/net/floodlightcontroller/core/internal/OFChannelHandlerVer13Test.java b/src/test/java/net/floodlightcontroller/core/internal/OFChannelHandlerVer13Test.java
index 6764bdef866c9d61dd65628e8ca0eb9413cfc9a7..643bc416ba1df288845efe8ce1debb161e96c13e 100644
--- a/src/test/java/net/floodlightcontroller/core/internal/OFChannelHandlerVer13Test.java
+++ b/src/test/java/net/floodlightcontroller/core/internal/OFChannelHandlerVer13Test.java
@@ -23,11 +23,14 @@ import java.util.Set;
import org.easymock.Capture;
import org.easymock.CaptureType;
import org.easymock.EasyMock;
+import org.easymock.IAnswer;
import org.hamcrest.CoreMatchers;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.DefaultChannelPromise;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
@@ -38,6 +41,7 @@ import org.junit.Test;
import net.floodlightcontroller.core.IOFConnectionBackend;
import net.floodlightcontroller.core.internal.OFChannelInitializer.PipelineHandler;
import net.floodlightcontroller.core.internal.OFChannelInitializer.PipelineHandshakeTimeout;
+import net.floodlightcontroller.core.test.TestEventLoop;
import net.floodlightcontroller.debugcounter.DebugCounterServiceImpl;
import net.floodlightcontroller.debugcounter.IDebugCounterService;
@@ -71,379 +75,391 @@ import com.google.common.collect.ImmutableList;
public class OFChannelHandlerVer13Test {
- private static final DatapathId dpid = DatapathId.of(0x42L);
-
- private IOFSwitchManager switchManager;
- private IOFConnectionListener connectionListener;
- private INewOFConnectionListener newConnectionListener;
- private IDebugCounterService debugCounterService;
- private OFChannelHandler handler;
- private Channel channel;
- private Timer timer;
- private ChannelHandlerContext ctx;
- private ChannelPipeline pipeline;
- private final OFFactory factory = OFFactories.getFactory(OFVersion.OF_13);
-
- private Capture<Throwable> exceptionEventCapture;
- private Capture<List<OFMessage>> writeCapture;
-
- private OFFeaturesReply featuresReply;
- private OFPortDesc portDesc;
-
- private Set<Long> seenXids = null;
-
- private Capture<IOFConnectionBackend> newConnection;
-
- private Capture<OFFeaturesReply> newFeaturesReply;
-
- public void setUpFeaturesReply() {
- portDesc = factory.buildPortDesc()
- .setName("Eth1")
- .setPortNo(OFPort.of(1))
- .build();
- featuresReply = factory.buildFeaturesReply()
- .setDatapathId(dpid)
- .setNBuffers(1)
- .setNTables((short)1)
- .setCapabilities(EnumSet.<OFCapabilities>of(OFCapabilities.FLOW_STATS, OFCapabilities.TABLE_STATS))
- .setAuxiliaryId(OFAuxId.MAIN)
- .build();
- }
-
- @Before
- public void setUp() throws Exception {
- setUpFeaturesReply();
- switchManager = createMock(IOFSwitchManager.class);
- connectionListener = createMock(IOFConnectionListener.class);
- newConnectionListener = createMock(INewOFConnectionListener.class);
- newConnection = new Capture<IOFConnectionBackend>();
- newFeaturesReply = new Capture<OFFeaturesReply>();
-
- ctx = createMock(ChannelHandlerContext.class);
- channel = createMock(Channel.class);
- timer = new HashedWheelTimer();
- exceptionEventCapture = new Capture<Throwable>(CaptureType.ALL);
- pipeline = createMock(ChannelPipeline.class);
- writeCapture = new Capture<List<OFMessage>>(CaptureType.ALL);
- seenXids = null;
-
-
- // TODO: should mock IDebugCounterService and make sure
- // the expected counters are updated.
- debugCounterService = new DebugCounterServiceImpl();
- debugCounterService.registerModule(OFConnectionCounters.COUNTER_MODULE);
- SwitchManagerCounters counters =
- new SwitchManagerCounters(debugCounterService);
- expect(switchManager.getCounters()).andReturn(counters).anyTimes();
- replay(switchManager);
- handler = new OFChannelHandler(switchManager, newConnectionListener,
- pipeline, debugCounterService, /* 62 is OF versions 1.0 thru 1.4 in decimal */
- timer, Collections.singletonList(U32.of(62)), OFFactories.getFactory(OFVersion.OF_14));
-
- verify(switchManager);
- reset(switchManager);
-
- resetChannel();
-
- // replay controller. Reset it if you need more specific behavior
- replay(switchManager);
-
- // Mock ctx and channelStateEvent
- expect(ctx.channel()).andReturn(channel).anyTimes();
- expect(ctx.fireExceptionCaught(capture(exceptionEventCapture))).andReturn(ctx).anyTimes();
- replay(ctx);
-
- /* Setup an exception event capture on the channel. Right now
- * we only expect exception events to be send up the channel.
- * However, it's easy to extend to other events if we need it
- */
- expect(pipeline.get(OFMessageDecoder.class)).andReturn(new OFMessageDecoder()).anyTimes();
- replay(pipeline);
- }
-
- @After
- public void tearDown() {
- /* ensure no exception was thrown */
- if (exceptionEventCapture.hasCaptured()) {
- Throwable ex = exceptionEventCapture.getValue();
- ex.printStackTrace();
- Throwables.propagate(ex);
- }
- assertFalse("Unexpected messages have been captured",
- writeCapture.hasCaptured());
- // verify all mocks.
- verify(channel);
- verify(switchManager);
- verify(ctx);
- verify(pipeline);
- }
-
- /** Reset the channel mock and set basic method call expectations */
- void resetChannel() {
- reset(channel);
- expect(channel.pipeline()).andReturn(pipeline).anyTimes();
- expect(channel.remoteAddress()).andReturn(null).anyTimes();
- }
-
- /** reset, setup, and replay the messageEvent mock for the given
- * messages, mock controller send message to channel handler
- *
- * This method will reset, start replay on controller, and then verify
- */
- void sendMessageToHandlerWithControllerReset(List<OFMessage> messages)
- throws Exception {
- sendMessageToHandlerNoControllerReset(messages);
- }
-
- /** reset, setup, and replay the messageEvent mock for the given
- * messages, mock controller send message to channel handler
- *
- * This method will start replay on controller, and then verify
- */
- void sendMessageToHandlerNoControllerReset(List<OFMessage> messages)
- throws Exception {
- handler.channelRead(ctx, messages);
- }
-
- /**
- * Extract the list of OFMessages that was captured by the Channel.write()
- * capture. Will check that something was actually captured first. We'll
- * collapse the messages from multiple writes into a single list of
- * OFMessages.
- * Resets the channelWriteCapture.
- */
- List<OFMessage> getMessagesFromCapture() {
- List<OFMessage> msgs = new ArrayList<OFMessage>();
-
- assertTrue("No write on channel was captured",
- writeCapture.hasCaptured());
- List<List<OFMessage>> capturedVals = writeCapture.getValues();
-
- for (List<OFMessage> oneWriteList: capturedVals)
- msgs.addAll(oneWriteList);
- writeCapture.reset();
- return msgs;
- }
-
-
- /**
- * Verify that the given exception event capture (as returned by
- * getAndInitExceptionCapture) has thrown an exception of the given
- * expectedExceptionClass.
- * Resets the capture
- */
- void verifyExceptionCaptured(Class<? extends Throwable> expectedExceptionClass) {
- assertTrue("Excpected exception not thrown", exceptionEventCapture.hasCaptured());
- Throwable caughtEx = exceptionEventCapture.getValue();
- assertEquals(expectedExceptionClass, caughtEx.getClass());
- exceptionEventCapture.reset();
- }
-
- /** make sure that the transaction ids in the given messages are
- * not 0 and differ between each other.
- * While it's not a defect per se if the xids are we want to ensure
- * we use different ones for each message we send.
- */
- void verifyUniqueXids(List<OFMessage> msgs) {
- if (seenXids == null)
- seenXids = new HashSet<Long>();
- for (OFMessage m: msgs) {
- long xid = m.getXid();
- assertTrue("Xid in messags is 0", xid != 0);
- assertFalse("Xid " + xid + " has already been used",
- seenXids.contains(xid));
- seenXids.add(xid);
- }
- }
-
- @Test
- public void testNullMsg() throws Exception {
- reset(ctx);
- expect(ctx.fireChannelRead(null)).andReturn(ctx).once();
- replay(ctx, channel);
-
- // null message is not passed to the handler
- handler.channelRead(ctx, null);
- verify(channel, ctx);
- }
-
- @Test
- public void testInitState() throws Exception {
- replay(channel);
-
- // We don't expect to receive /any/ messages in init state since
- // channelConnected moves us to a different state
- OFMessage m = factory.buildHello().build();
- sendMessageToHandlerWithControllerReset(ImmutableList.<OFMessage>of(m));
-
- verifyExceptionCaptured(SwitchStateException.class);
- assertThat(handler.getStateForTesting(), CoreMatchers.instanceOf(OFChannelHandler.InitState.class));
- }
-
- /* Move the channel from scratch to WAIT_HELLO state */
- @Test
- public void moveToWaitHello() throws Exception {
- resetChannel();
- expect(channel.writeAndFlush(capture(writeCapture))).andReturn(null).once();
- replay(channel);
-
- handler.channelActive(ctx);
-
- List<OFMessage> msgs = getMessagesFromCapture();
- assertEquals(1, msgs.size());
- assertEquals(OFType.HELLO, msgs.get(0).getType());
- assertThat(handler.getStateForTesting(), CoreMatchers.instanceOf(OFChannelHandler.WaitHelloState.class));
- verifyUniqueXids(msgs);
- }
-
- /** Move the channel from scratch to WAIT_FEATURES_REPLY state
- * Builds on moveToWaitHello()
- * adds testing for WAIT_HELLO state
- */
- @Test
- public void moveToWaitFeaturesReply() throws Exception {
- moveToWaitHello();
- resetChannel();
- expect(channel.writeAndFlush(capture(writeCapture))).andReturn(null).once();
- replay(channel);
-
- OFMessage hello = factory.buildHello().build();
- sendMessageToHandlerWithControllerReset(ImmutableList.<OFMessage>of(hello));
-
- List<OFMessage> msgs = getMessagesFromCapture();
- assertEquals(1, msgs.size());
- assertEquals(OFType.FEATURES_REQUEST, msgs.get(0).getType());
- verifyUniqueXids(msgs);
-
- assertThat(handler.getStateForTesting(), CoreMatchers.instanceOf(OFChannelHandler.WaitFeaturesReplyState.class));
- }
-
-
- /** Move the channel from scratch to WAIT_FEATURES_REPLY state
- * Builds on moveToWaitHello()
- * adds testing for WAIT_HELLO state
- */
- @Test
- public void moveToComplete() throws Exception {
- moveToWaitFeaturesReply();
-
- reset(pipeline);
- HandshakeTimeoutHandler newHandler = new HandshakeTimeoutHandler(
- handler,
- timer,
- PipelineHandshakeTimeout.SWITCH);
-
- expect(
- pipeline.replace(EasyMock.eq(PipelineHandler.CHANNEL_HANDSHAKE_TIMEOUT),
- EasyMock.eq(PipelineHandler.SWITCH_HANDSHAKE_TIMEOUT),
- EasyMock.anyObject(HandshakeTimeoutHandler.class))).andReturn(newHandler)
- .once();
-
- replay(pipeline);
-
- newConnectionListener.connectionOpened(capture(newConnection), capture(newFeaturesReply));
- expectLastCall().once();
- replay(newConnectionListener);
-
- sendMessageToHandlerWithControllerReset(Collections.<OFMessage>singletonList(featuresReply));
-
- assertThat(handler.getStateForTesting(), CoreMatchers.instanceOf(OFChannelHandler.CompleteState.class));
- assertTrue("A connection has been created and set", handler.getConnectionForTesting() != null);
- verify(newConnectionListener);
- assertTrue(newConnection.hasCaptured());
- assertThat(newFeaturesReply.getValue(), equalTo(featuresReply));
- }
-
- /**
- * Test dispatch of messages while in Complete state
- */
- @Test
- public void testMessageDispatchComplete() throws Exception {
- moveToComplete();
- newConnection.getValue().setListener(connectionListener);
+ private static final DatapathId dpid = DatapathId.of(0x42L);
+
+ private IOFSwitchManager switchManager;
+ private IOFConnectionListener connectionListener;
+ private INewOFConnectionListener newConnectionListener;
+ private IDebugCounterService debugCounterService;
+ private OFChannelHandler handler;
+ private Channel channel;
+ private Timer timer;
+ private ChannelHandlerContext ctx;
+ private ChannelPipeline pipeline;
+ private final OFFactory factory = OFFactories.getFactory(OFVersion.OF_13);
+
+ private Capture<Throwable> exceptionEventCapture;
+ private Capture<List<OFMessage>> writeCapture;
+
+ private OFFeaturesReply featuresReply;
+ private OFPortDesc portDesc;
+
+ private Set<Long> seenXids = null;
+
+ private Capture<IOFConnectionBackend> newConnection;
+
+ private Capture<OFFeaturesReply> newFeaturesReply;
+
+ private TestEventLoop eventLoop;
+
+ public void setUpFeaturesReply() {
+ portDesc = factory.buildPortDesc()
+ .setName("Eth1")
+ .setPortNo(OFPort.of(1))
+ .build();
+ featuresReply = factory.buildFeaturesReply()
+ .setDatapathId(dpid)
+ .setNBuffers(1)
+ .setNTables((short)1)
+ .setCapabilities(EnumSet.<OFCapabilities>of(OFCapabilities.FLOW_STATS, OFCapabilities.TABLE_STATS))
+ .setAuxiliaryId(OFAuxId.MAIN)
+ .build();
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ setUpFeaturesReply();
+ switchManager = createMock(IOFSwitchManager.class);
+ connectionListener = createMock(IOFConnectionListener.class);
+ newConnectionListener = createMock(INewOFConnectionListener.class);
+ newConnection = new Capture<IOFConnectionBackend>();
+ newFeaturesReply = new Capture<OFFeaturesReply>();
+ eventLoop = new TestEventLoop();
+
+ ctx = createMock(ChannelHandlerContext.class);
+ channel = createMock(Channel.class);
+ timer = new HashedWheelTimer();
+ exceptionEventCapture = new Capture<Throwable>(CaptureType.ALL);
+ pipeline = createMock(ChannelPipeline.class);
+ writeCapture = new Capture<List<OFMessage>>(CaptureType.ALL);
+ seenXids = null;
+
+
+ // TODO: should mock IDebugCounterService and make sure
+ // the expected counters are updated.
+ debugCounterService = new DebugCounterServiceImpl();
+ debugCounterService.registerModule(OFConnectionCounters.COUNTER_MODULE);
+ SwitchManagerCounters counters =
+ new SwitchManagerCounters(debugCounterService);
+ expect(switchManager.getCounters()).andReturn(counters).anyTimes();
+ replay(switchManager);
+ handler = new OFChannelHandler(switchManager, newConnectionListener,
+ pipeline, debugCounterService, /* 62 is OF versions 1.0 thru 1.4 in decimal */
+ timer, Collections.singletonList(U32.of(62)), OFFactories.getFactory(OFVersion.OF_14));
+
+ verify(switchManager);
+ reset(switchManager);
+
+ resetChannel();
+
+ // replay controller. Reset it if you need more specific behavior
+ replay(switchManager);
+
+ // Mock ctx and channelStateEvent
+ expect(ctx.channel()).andReturn(channel).anyTimes();
+ expect(ctx.fireExceptionCaught(capture(exceptionEventCapture))).andReturn(ctx).anyTimes();
+ replay(ctx);
+
+ /* Setup an exception event capture on the channel. Right now
+ * we only expect exception events to be send up the channel.
+ * However, it's easy to extend to other events if we need it
+ */
+ expect(pipeline.get(OFMessageDecoder.class)).andReturn(new OFMessageDecoder()).anyTimes();
+ replay(pipeline);
+ }
+
+ @After
+ public void tearDown() {
+ /* ensure no exception was thrown */
+ if (exceptionEventCapture.hasCaptured()) {
+ Throwable ex = exceptionEventCapture.getValue();
+ ex.printStackTrace();
+ Throwables.propagate(ex);
+ }
+ assertFalse("Unexpected messages have been captured",
+ writeCapture.hasCaptured());
+ // verify all mocks.
+ verify(channel);
+ verify(switchManager);
+ verify(ctx);
+ verify(pipeline);
+ }
+
+ /** Reset the channel mock and set basic method call expectations */
+ void resetChannel() {
+ reset(channel);
+ expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
+ @Override
+ public ChannelPromise answer() throws Throwable {
+ return new DefaultChannelPromise(channel);
+ }
+ }).anyTimes();
+ eventLoop = new TestEventLoop();
+ expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
+ expect(channel.pipeline()).andReturn(pipeline).anyTimes();
+ expect(channel.remoteAddress()).andReturn(null).anyTimes();
+ }
+
+ /** reset, setup, and replay the messageEvent mock for the given
+ * messages, mock controller send message to channel handler
+ *
+ * This method will reset, start replay on controller, and then verify
+ */
+ void sendMessageToHandlerWithControllerReset(List<OFMessage> messages)
+ throws Exception {
+ sendMessageToHandlerNoControllerReset(messages);
+ }
+
+ /** reset, setup, and replay the messageEvent mock for the given
+ * messages, mock controller send message to channel handler
+ *
+ * This method will start replay on controller, and then verify
+ */
+ void sendMessageToHandlerNoControllerReset(List<OFMessage> messages)
+ throws Exception {
+ handler.channelRead(ctx, messages);
+ }
+
+ /**
+ * Extract the list of OFMessages that was captured by the Channel.write()
+ * capture. Will check that something was actually captured first. We'll
+ * collapse the messages from multiple writes into a single list of
+ * OFMessages.
+ * Resets the channelWriteCapture.
+ */
+ List<OFMessage> getMessagesFromCapture() {
+ List<OFMessage> msgs = new ArrayList<OFMessage>();
+
+ assertTrue("No write on channel was captured",
+ writeCapture.hasCaptured());
+ List<List<OFMessage>> capturedVals = writeCapture.getValues();
+
+ for (List<OFMessage> oneWriteList: capturedVals)
+ msgs.addAll(oneWriteList);
+ writeCapture.reset();
+ return msgs;
+ }
+
+
+ /**
+ * Verify that the given exception event capture (as returned by
+ * getAndInitExceptionCapture) has thrown an exception of the given
+ * expectedExceptionClass.
+ * Resets the capture
+ */
+ void verifyExceptionCaptured(Class<? extends Throwable> expectedExceptionClass) {
+ assertTrue("Excpected exception not thrown", exceptionEventCapture.hasCaptured());
+ Throwable caughtEx = exceptionEventCapture.getValue();
+ assertEquals(expectedExceptionClass, caughtEx.getClass());
+ exceptionEventCapture.reset();
+ }
+
+ /** make sure that the transaction ids in the given messages are
+ * not 0 and differ between each other.
+ * While it's not a defect per se if the xids are we want to ensure
+ * we use different ones for each message we send.
+ */
+ void verifyUniqueXids(List<OFMessage> msgs) {
+ if (seenXids == null)
+ seenXids = new HashSet<Long>();
+ for (OFMessage m: msgs) {
+ long xid = m.getXid();
+ assertTrue("Xid in messags is 0", xid != 0);
+ assertFalse("Xid " + xid + " has already been used",
+ seenXids.contains(xid));
+ seenXids.add(xid);
+ }
+ }
+
+ @Test
+ public void testNullMsg() throws Exception {
+ reset(ctx);
+ expect(ctx.fireChannelRead(null)).andReturn(ctx).once();
+ replay(ctx, channel);
+
+ // null message is not passed to the handler
+ handler.channelRead(ctx, null);
+ verify(channel, ctx);
+ }
+
+ @Test
+ public void testInitState() throws Exception {
+ replay(channel);
+
+ // We don't expect to receive /any/ messages in init state since
+ // channelConnected moves us to a different state
+ OFMessage m = factory.buildHello().build();
+ sendMessageToHandlerWithControllerReset(ImmutableList.<OFMessage>of(m));
+
+ verifyExceptionCaptured(SwitchStateException.class);
+ assertThat(handler.getStateForTesting(), CoreMatchers.instanceOf(OFChannelHandler.InitState.class));
+ }
+
+ /* Move the channel from scratch to WAIT_HELLO state */
+ @Test
+ public void moveToWaitHello() throws Exception {
+ resetChannel();
+ expect(channel.writeAndFlush(capture(writeCapture))).andReturn(null).once();
+ replay(channel);
+
+ handler.channelActive(ctx);
+ eventLoop.runTasks();
+
+ List<OFMessage> msgs = getMessagesFromCapture();
+ assertEquals(1, msgs.size());
+ assertEquals(OFType.HELLO, msgs.get(0).getType());
+ assertThat(handler.getStateForTesting(), CoreMatchers.instanceOf(OFChannelHandler.WaitHelloState.class));
+ verifyUniqueXids(msgs);
+ }
+
+ /** Move the channel from scratch to WAIT_FEATURES_REPLY state
+ * Builds on moveToWaitHello()
+ * adds testing for WAIT_HELLO state
+ */
+ @Test
+ public void moveToWaitFeaturesReply() throws Exception {
+ moveToWaitHello();
+ resetChannel();
+ expect(channel.writeAndFlush(capture(writeCapture))).andReturn(null).once();
+ replay(channel);
+
+ OFMessage hello = factory.buildHello().build();
+ sendMessageToHandlerWithControllerReset(ImmutableList.<OFMessage>of(hello));
+
+ List<OFMessage> msgs = getMessagesFromCapture();
+ assertEquals(1, msgs.size());
+ assertEquals(OFType.FEATURES_REQUEST, msgs.get(0).getType());
+ verifyUniqueXids(msgs);
+
+ assertThat(handler.getStateForTesting(), CoreMatchers.instanceOf(OFChannelHandler.WaitFeaturesReplyState.class));
+ }
+
+
+ /** Move the channel from scratch to WAIT_FEATURES_REPLY state
+ * Builds on moveToWaitHello()
+ * adds testing for WAIT_HELLO state
+ */
+ @Test
+ public void moveToComplete() throws Exception {
+ moveToWaitFeaturesReply();
+
+ reset(pipeline);
+ HandshakeTimeoutHandler newHandler = new HandshakeTimeoutHandler(
+ handler,
+ timer,
+ PipelineHandshakeTimeout.SWITCH);
+
+ expect(
+ pipeline.replace(EasyMock.eq(PipelineHandler.CHANNEL_HANDSHAKE_TIMEOUT),
+ EasyMock.eq(PipelineHandler.SWITCH_HANDSHAKE_TIMEOUT),
+ EasyMock.anyObject(HandshakeTimeoutHandler.class))).andReturn(newHandler)
+ .once();
+
+ replay(pipeline);
+
+ newConnectionListener.connectionOpened(capture(newConnection), capture(newFeaturesReply));
+ expectLastCall().once();
+ replay(newConnectionListener);
+
+ sendMessageToHandlerWithControllerReset(Collections.<OFMessage>singletonList(featuresReply));
+
+ assertThat(handler.getStateForTesting(), CoreMatchers.instanceOf(OFChannelHandler.CompleteState.class));
+ assertTrue("A connection has been created and set", handler.getConnectionForTesting() != null);
+ verify(newConnectionListener);
+ assertTrue(newConnection.hasCaptured());
+ assertThat(newFeaturesReply.getValue(), equalTo(featuresReply));
+ }
- resetChannel();
- expect(channel.writeAndFlush(capture(writeCapture))).andReturn(null).once();
- replay(channel);
+ /**
+ * Test dispatch of messages while in Complete state
+ */
+ @Test
+ public void testMessageDispatchComplete() throws Exception {
+ moveToComplete();
+ newConnection.getValue().setListener(connectionListener);
- // Send echo request. expect reply
- OFMessage echoRequest = factory.buildEchoRequest().build();
- sendMessageToHandlerWithControllerReset(ImmutableList.<OFMessage>of(echoRequest));
+ resetChannel();
+ expect(channel.writeAndFlush(capture(writeCapture))).andReturn(null).once();
+ replay(channel);
- List<OFMessage> msgs = getMessagesFromCapture();
- assertEquals(1, msgs.size());
- assertEquals(OFType.ECHO_REPLY, msgs.get(0).getType());
+ // Send echo request. expect reply
+ OFMessage echoRequest = factory.buildEchoRequest().build();
+ sendMessageToHandlerWithControllerReset(ImmutableList.<OFMessage>of(echoRequest));
+ List<OFMessage> msgs = getMessagesFromCapture();
+ assertEquals(1, msgs.size());
+ assertEquals(OFType.ECHO_REPLY, msgs.get(0).getType());
- // Send barrier reply. expect dispatch
- OFBarrierReply barrierReply = factory.buildBarrierReply()
- .build();
- resetAndExpectConnectionListener(barrierReply);
+ // Send barrier reply. expect dispatch
+ OFBarrierReply barrierReply = factory.buildBarrierReply()
+ .build();
+ resetAndExpectConnectionListener(barrierReply);
- // Send packet in. expect dispatch
- OFFlowRemoved flowRemoved = factory.buildFlowRemoved()
- .build();
- resetAndExpectConnectionListener(flowRemoved);
+ // Send packet in. expect dispatch
+ OFFlowRemoved flowRemoved = factory.buildFlowRemoved()
+ .build();
- // Send get config reply. expect dispatch
- OFGetConfigReply getConfigReply = factory.buildGetConfigReply()
- .build();
+ resetAndExpectConnectionListener(flowRemoved);
- resetAndExpectConnectionListener(getConfigReply);
+ // Send get config reply. expect dispatch
+ OFGetConfigReply getConfigReply = factory.buildGetConfigReply()
+ .build();
- // Send packet in. expect dispatch
- OFPacketIn pi = factory.buildPacketIn()
- .setReason(OFPacketInReason.NO_MATCH)
- .build();
+ resetAndExpectConnectionListener(getConfigReply);
- resetAndExpectConnectionListener(pi);
+ // Send packet in. expect dispatch
+ OFPacketIn pi = factory.buildPacketIn()
+ .setReason(OFPacketInReason.NO_MATCH)
+ .build();
- // Send port status. expect dispatch
- OFPortStatus portStatus = factory.buildPortStatus()
- .setReason(OFPortReason.DELETE)
- .setDesc(portDesc)
- .build();
+ resetAndExpectConnectionListener(pi);
- resetAndExpectConnectionListener(portStatus);
+ // Send port status. expect dispatch
+ OFPortStatus portStatus = factory.buildPortStatus()
+ .setReason(OFPortReason.DELETE)
+ .setDesc(portDesc)
+ .build();
- // Send queue reply. expect dispatch
- OFQueueGetConfigReply queueReply = factory.buildQueueGetConfigReply()
- .build();
+ resetAndExpectConnectionListener(portStatus);
- resetAndExpectConnectionListener(queueReply);
+ // Send queue reply. expect dispatch
+ OFQueueGetConfigReply queueReply = factory.buildQueueGetConfigReply()
+ .build();
- // Send stat reply. expect dispatch
- OFFlowStatsReply statReply = factory.buildFlowStatsReply()
- .build();
+ resetAndExpectConnectionListener(queueReply);
- resetAndExpectConnectionListener(statReply);
+ // Send stat reply. expect dispatch
+ OFFlowStatsReply statReply = factory.buildFlowStatsReply()
+ .build();
- // Send role reply. expect dispatch
- OFRoleReply roleReply = factory.buildRoleReply()
- .setRole(OFControllerRole.ROLE_MASTER)
- .build();
+ resetAndExpectConnectionListener(statReply);
- resetAndExpectConnectionListener(roleReply);
+ // Send role reply. expect dispatch
+ OFRoleReply roleReply = factory.buildRoleReply()
+ .setRole(OFControllerRole.ROLE_MASTER)
+ .build();
- // Send experimenter. expect dispatch
- OFBsnSetAuxCxnsReply auxReply = factory.buildBsnSetAuxCxnsReply()
- .build();
+ resetAndExpectConnectionListener(roleReply);
- resetAndExpectConnectionListener(auxReply);
+ // Send experimenter. expect dispatch
+ OFBsnSetAuxCxnsReply auxReply = factory.buildBsnSetAuxCxnsReply()
+ .build();
- }
+ resetAndExpectConnectionListener(auxReply);
- public void resetAndExpectConnectionListener(OFMessage m) throws Exception{
- reset(connectionListener);
- connectionListener.messageReceived(handler.getConnectionForTesting(), m);
- expectLastCall().once();
- replay(connectionListener);
+ }
- sendMessageToHandlerWithControllerReset(Collections.<OFMessage>singletonList(m));
+ public void resetAndExpectConnectionListener(OFMessage m) throws Exception{
+ reset(connectionListener);
+ connectionListener.messageReceived(handler.getConnectionForTesting(), m);
+ expectLastCall().once();
+ replay(connectionListener);
- verify(connectionListener);
- }
+ sendMessageToHandlerWithControllerReset(Collections.<OFMessage>singletonList(m));
+
+ verify(connectionListener);
+ }
}
diff --git a/src/test/java/net/floodlightcontroller/core/internal/OFConnectionTest.java b/src/test/java/net/floodlightcontroller/core/internal/OFConnectionTest.java
index 24d39515c6b8dd9d93208959632e7fe1da1249d7..23230074aacb2c69d45801bda2c5bdaa4bddff1c 100644
--- a/src/test/java/net/floodlightcontroller/core/internal/OFConnectionTest.java
+++ b/src/test/java/net/floodlightcontroller/core/internal/OFConnectionTest.java
@@ -18,6 +18,7 @@ import io.netty.channel.Channel;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
+import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -25,6 +26,7 @@ import net.floodlightcontroller.core.SwitchDisconnectedException;
import net.floodlightcontroller.core.internal.OFConnection;
import net.floodlightcontroller.core.internal.OFConnectionCounters;
import net.floodlightcontroller.core.internal.OFErrorMsgException;
+import net.floodlightcontroller.core.test.TestEventLoop;
import net.floodlightcontroller.debugcounter.DebugCounterServiceImpl;
import net.floodlightcontroller.debugcounter.IDebugCounterService;
@@ -68,17 +70,28 @@ public class OFConnectionTest {
private OFConnection conn;
private DatapathId switchId;
private Timer timer;
+ private TestEventLoop eventLoop;
@Before
public void setUp() throws Exception {
factory = OFFactories.getFactory(OFVersion.OF_13);
switchId = DatapathId.of(1);
timer = new HashedWheelTimer();
- channel = EasyMock.createNiceMock(Channel.class);
+ channel = EasyMock.createMock(Channel.class);
IDebugCounterService debugCounterService = new DebugCounterServiceImpl();
debugCounterService.registerModule(OFConnectionCounters.COUNTER_MODULE);
conn = new OFConnection(switchId, factory, channel, OFAuxId.MAIN,
debugCounterService, timer);
+ eventLoop = new TestEventLoop();
+
+ expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (timer != null) {
+ timer.stop();
+ }
}
@Test(timeout = 5000)
@@ -90,6 +103,7 @@ public class OFConnectionTest {
assertThat("Connection should have 1 pending request",
conn.getPendingRequestIds().size(), equalTo(1));
+ eventLoop.runTasks();
assertThat("Should have captured MsgList", cMsgList.getValue(),
Matchers.<OFMessage> contains(echoRequest));
@@ -117,7 +131,8 @@ public class OFConnectionTest {
ListenableFuture<List<OFFlowStatsReply>> future = conn.writeStatsRequest(flowStatsRequest);
assertThat("Connection should have 1 pending request",
conn.getPendingRequestIds().size(), equalTo(1));
-
+
+ eventLoop.runTasks();
assertThat("Should have captured MsgList", cMsgList.getValue(),
Matchers.<OFMessage> contains(flowStatsRequest));
@@ -165,6 +180,7 @@ public class OFConnectionTest {
assertThat("Connection should have 1 pending request",
conn.getPendingRequestIds().size(), equalTo(1));
+ eventLoop.runTasks();
assertThat("Should have captured MsgList", cMsgList.getValue(),
Matchers.<OFMessage> contains(roleRequest));
@@ -222,7 +238,7 @@ public class OFConnectionTest {
conn.getPendingRequestIds().isEmpty(), equalTo(true));
}
- /** write a packetOut, which is buffered */
+ /** write a packetOut, which is not buffered */
@Test(timeout = 5000)
public void testSingleMessageWrite() throws InterruptedException, ExecutionException {
Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();
@@ -233,6 +249,7 @@ public class OFConnectionTest {
.build();
conn.write(packetOut);
+ eventLoop.runTasks();
assertThat("Write should have been flushed", cMsgList.hasCaptured(), equalTo(true));
List<OFMessage> value = cMsgList.getValue();
@@ -253,7 +270,7 @@ public class OFConnectionTest {
.build();
conn.write(ImmutableList.of(hello, packetOut));
-
+ eventLoop.runTasks();
assertThat("Write should have been written", cMsgList.hasCaptured(), equalTo(true));
List<OFMessage> value = cMsgList.getValue();
logger.info("Captured channel write: "+value);
diff --git a/src/test/java/net/floodlightcontroller/core/test/TestEventLoop.java b/src/test/java/net/floodlightcontroller/core/test/TestEventLoop.java
new file mode 100644
index 0000000000000000000000000000000000000000..b31725e6762a47a4b35ce67f059ac27979b46da4
--- /dev/null
+++ b/src/test/java/net/floodlightcontroller/core/test/TestEventLoop.java
@@ -0,0 +1,134 @@
+package net.floodlightcontroller.core.test;
+
+import java.util.ArrayDeque;
+import java.util.Queue;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ImmutableList;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.DefaultChannelPromise;
+import io.netty.channel.EventLoop;
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.concurrent.AbstractScheduledEventExecutor;
+import io.netty.util.concurrent.Future;
+
+public final class TestEventLoop extends AbstractScheduledEventExecutor implements EventLoop {
+
+ private final Queue<Runnable> tasks = new ArrayDeque<Runnable>(2);
+
+ @Override
+ public void execute(Runnable command) {
+ if (command == null) {
+ throw new NullPointerException("command");
+ }
+ tasks.add(command);
+ }
+
+ public void runTasks() {
+ for (;;) {
+ Runnable task = tasks.poll();
+ if (task == null) {
+ break;
+ }
+
+ task.run();
+ }
+ }
+
+ long runScheduledTasks() {
+ long time = AbstractScheduledEventExecutor.nanoTime();
+ for (;;) {
+ Runnable task = pollScheduledTask(time);
+ if (task == null) {
+ return nextScheduledTaskNano();
+ }
+
+ task.run();
+ }
+ }
+
+ long nextScheduledTask() {
+ return nextScheduledTaskNano();
+ }
+
+ @Override
+ protected void cancelScheduledTasks() {
+ super.cancelScheduledTasks();
+ }
+
+ @Override
+ public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Future<?> terminationFuture() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ @Deprecated
+ public void shutdown() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isShuttingDown() {
+ return false;
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return false;
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return false;
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit)
+ throws InterruptedException {
+ Thread.sleep(unit.toMillis(timeout));
+ return false;
+ }
+
+ @Override
+ public ChannelFuture register(Channel channel) {
+ return register(channel, new DefaultChannelPromise(channel, this));
+ }
+
+ @Override
+ public ChannelFuture register(Channel channel, ChannelPromise promise) {
+ channel.unsafe().register(this, promise);
+ return promise;
+ }
+
+ @Override
+ public boolean inEventLoop() {
+ return true;
+ }
+
+ @Override
+ public boolean inEventLoop(Thread thread) {
+ return true;
+ }
+
+ @Override
+ public EventLoop next() {
+ return this;
+ }
+
+ @Override
+ public EventLoopGroup parent() {
+ return this;
+ }
+
+ public ImmutableList<Runnable> getQueuedTasks() {
+ return ImmutableList.copyOf(tasks);
+ }
+}
\ No newline at end of file
diff --git a/src/test/java/net/floodlightcontroller/loadbalancer/LoadBalancerTest.java b/src/test/java/net/floodlightcontroller/loadbalancer/LoadBalancerTest.java
index f19ce3bff93ff051ce38c8853c1b375246e0a1a2..b12b2f5cdc8adca998f20e319c4712164968b248 100644
--- a/src/test/java/net/floodlightcontroller/loadbalancer/LoadBalancerTest.java
+++ b/src/test/java/net/floodlightcontroller/loadbalancer/LoadBalancerTest.java
@@ -448,8 +448,6 @@ public class LoadBalancerTest extends FloodlightTestCase {
expect(sw1.getOFFactory()).andReturn(factory).anyTimes();
sw1.write(capture(wc1));
expectLastCall().anyTimes();
- sw1.flush();
- expectLastCall().anyTimes();
replay(sw1);
sfp.switchAdded(DatapathId.of(1L));
diff --git a/src/test/java/net/floodlightcontroller/staticflowentry/StaticFlowTests.java b/src/test/java/net/floodlightcontroller/staticflowentry/StaticFlowTests.java
index 51b2319f19053890ae402247eaf4b05a83f6bebe..407db1d3c70d6c25ea458460983eec0811329195 100644
--- a/src/test/java/net/floodlightcontroller/staticflowentry/StaticFlowTests.java
+++ b/src/test/java/net/floodlightcontroller/staticflowentry/StaticFlowTests.java
@@ -207,8 +207,6 @@ public class StaticFlowTests extends FloodlightTestCase {
expectLastCall().anyTimes();
mockSwitch.write(capture(writeCaptureList));
expectLastCall().anyTimes();
- mockSwitch.flush();
- expectLastCall().anyTimes();
expect(mockSwitch.getOFFactory()).andReturn(factory).anyTimes();
replay(mockSwitch);
@@ -253,8 +251,6 @@ public class StaticFlowTests extends FloodlightTestCase {
expectLastCall().anyTimes();
mockSwitch.write(capture(writeCaptureList));
expectLastCall().anyTimes();
- mockSwitch.flush();
- expectLastCall().anyTimes();
expect(mockSwitch.getOFFactory()).andReturn(factory).anyTimes();
expect(mockSwitch.getId()).andReturn(DatapathId.of(dpid)).anyTimes();
replay(mockSwitch);
diff --git a/src/test/java/net/floodlightcontroller/util/OFMessageDamperMockSwitch.java b/src/test/java/net/floodlightcontroller/util/OFMessageDamperMockSwitch.java
index 119b7acf8dcba7ca15ca9d5bf89a92ce798c5483..ca5d326569d59c60891e312f46cf878cd28632a0 100644
--- a/src/test/java/net/floodlightcontroller/util/OFMessageDamperMockSwitch.java
+++ b/src/test/java/net/floodlightcontroller/util/OFMessageDamperMockSwitch.java
@@ -160,11 +160,6 @@ public class OFMessageDamperMockSwitch implements IOFSwitch {
return null;
}
- @Override
- public void flush() {
- assertTrue("Unexpected method call", false);
- }
-
@Override
public long getBuffers() {
fail("Unexpected method call");