- TLS configuration is specified in file ./configuration/config.ini. To enable the TLS feature, set secureChannelEnabled=true and specify the location of controller Java KeyStore and TrustStore files. The Java KeyStore contains controller's private key and certificate. The Java TrustStore contains the trusted certificate entries, including switches' Certification Authority (CA) certificates. Here is the sample configuration,
secureChannelEnabled=true
controllerKeyStore=./configuration/ctlKeyStore
controllerKeyStorePassword=xxxxx (this password should match the password used for KeyStore generation)
controllerTrustStore=./configuration/ctlTrustStore
controllerTrustStorePassword=xxxxx (this password should match the password used for TrustStore generation)
- Added two message read/write services, one for clear text, one for secure communication.
- Added priority queue for message transmission. The system critical messages, such as Hello, Echo Reply etc will be treated as high priority and will be served ahead of other messages, like statistics request, discovery packets etc.
Signed-off-by: Jason Ye <yisye@cisco.com>
# of.listenPort=6633
# The time (in milliseconds) the controller will wait for a response after sending a Barrier Request or a Statistic Request message (default 2000 msec)
# of.messageResponseTimer=2000
+
+# TLS configuration
+# To enable TLS, set secureChannelEnabled=true and specify the location of controller Java KeyStore and TrustStore files.
+# The Java KeyStore contains controller's private key and certificate. The Java TrustStore contains the trusted certificate
+# entries, including switches' Certification Authority (CA) certificates. For example,
+# secureChannelEnabled=true
+# controllerKeyStore=./configuration/ctlKeyStore
+# controllerKeyStorePassword=xxxxx (this password should match the password used for KeyStore generation)
+# controllerTrustStore=./configuration/ctlTrustStore
+# controllerTrustStorePassword=xxxxx (this password should match the password used for TrustStore generation)
+
+secureChannelEnabled=false
+controllerKeyStore=
+controllerKeyStorePassword=
+controllerTrustStore=
+controllerTrustStorePassword=
<logger name="org.opendaylight.controller.protocol_plugin.openflow.internal.InventoryServiceShim" level="INFO"/>
<logger name="org.opendaylight.controller.protocol_plugin.openflow.internal.TopologyServices" level="INFO"/>
<logger name="org.opendaylight.controller.protocol_plugin.openflow.internal.TopologyServiceShim" level="INFO"/>
+ <logger name="org.opendaylight.controller.protocol_plugin.openflow.core.internal.Controller" level="INFO"/>
+ <logger name="org.opendaylight.controller.protocol_plugin.openflow.core.internal.SwitchHandler" level="INFO"/>
+ <logger name="org.opendaylight.controller.protocol_plugin.openflow.core.internal.SwitchIOSecureService" level="INFO"/>
<!-- SAL -->
<logger name="org.opendaylight.controller.sal" level="INFO"/>
<logger name="org.opendaylight.controller.sal.implementation" level="INFO"/>
org.apache.felix.dm,
org.slf4j,
org.eclipse.osgi.framework.console,
- org.osgi.framework
+ org.osgi.framework,
+ javax.net.ssl
</Import-Package>
<Export-Package>
org.opendaylight.controller.protocol_plugin.openflow.internal
--- /dev/null
+
+/*
+ * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.protocol_plugin.openflow.core;
+
+import java.util.List;
+
+import org.openflow.protocol.OFMessage;
+
+/**
+ * This interface defines low level routines to read/write messages on an open
+ * socket channel. If secure communication is desired, these methods also perform
+ * encryption and decryption of the network data.
+ */
+public interface IMessageReadWrite {
+ /**
+ * Sends the OF message out over the socket channel. For secure
+ * communication, the data will be encrypted.
+ *
+ * @param msg OF message to be sent
+ * @throws Exception
+ */
+ public void asyncSend(OFMessage msg) throws Exception;
+
+ /**
+ * Resumes sending the remaining messages in the outgoing buffer
+ * @throws Exception
+ */
+ public void resumeSend() throws Exception;
+
+ /**
+ * Reads the incoming network data from the socket and retrieves the OF
+ * messages. For secure communication, the data will be decrypted first.
+ *
+ * @return list of OF messages
+ * @throws Exception
+ */
+ public List<OFMessage> readMessages() throws Exception;
+}
public Date getConnectedDate();
/**
- * Sends the message. A unique XID is generated automatically and inserted into the message.
- * @param msg TheOF message to be sent
+ * This method puts the message in an outgoing priority queue with normal
+ * priority. It will be served after high priority messages. The method
+ * should be used for non-critical messages such as statistics request,
+ * discovery packets, etc. An unique XID is generated automatically and
+ * inserted into the message.
+ *
+ * @param msg The OF message to be sent
* @return The XID used
*/
public Integer asyncSend(OFMessage msg);
/**
- * Sends the message with the specified XID.
+ * This method puts the message in an outgoing priority queue with normal
+ * priority. It will be served after high priority messages. The method
+ * should be used for non-critical messages such as statistics request,
+ * discovery packets, etc. The specified XID is inserted into the message.
+ *
* @param msg The OF message to be Sent
* @param xid The XID to be used in the message
* @return The XID used
*/
public Integer asyncSend(OFMessage msg, int xid);
+ /**
+ * This method puts the message in an outgoing priority queue with high
+ * priority. It will be served first before normal priority messages. The
+ * method should be used for critical messages such as hello, echo reply
+ * etc. An unique XID is generated automatically and inserted into the
+ * message.
+ *
+ * @param msg The OF message to be sent
+ * @return The XID used
+ */
+ public Integer asyncFastSend(OFMessage msg);
+
+ /**
+ * This method puts the message in an outgoing priority queue with high
+ * priority. It will be served first before normal priority messages. The
+ * method should be used for critical messages such as hello, echo reply
+ * etc. The specified XID is inserted into the message.
+ *
+ * @param msg The OF message to be sent
+ * @return The XID used
+ */
+ public Integer asyncFastSend(OFMessage msg, int xid);
+
/**
* Sends the OF message followed by a Barrier Request with a unique XID which is automatically generated,
* and waits for a result from the switch.
--- /dev/null
+
+/*
+ * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.protocol_plugin.openflow.core.internal;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousCloseException;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.SocketChannel;
+import java.util.List;
+
+import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageReadWrite;
+import org.openflow.protocol.OFMessage;
+import org.openflow.protocol.factory.BasicFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class implements methods to read/write messages over an established
+ * socket channel. The data exchange is in clear text format.
+ */
+public class MessageReadWriteService implements IMessageReadWrite {
+ private static final Logger logger = LoggerFactory
+ .getLogger(MessageReadWriteService.class);
+ private static final int bufferSize = 1024 * 1024;
+
+ private Selector selector;
+ private SelectionKey clientSelectionKey;
+ private SocketChannel socket;
+ private ByteBuffer inBuffer;
+ private ByteBuffer outBuffer;
+ private BasicFactory factory;
+
+ public MessageReadWriteService(SocketChannel socket, Selector selector) throws ClosedChannelException {
+ this.socket = socket;
+ this.selector = selector;
+ this.factory = new BasicFactory();
+ this.inBuffer = ByteBuffer.allocateDirect(bufferSize);
+ this.outBuffer = ByteBuffer.allocateDirect(bufferSize);
+ this.clientSelectionKey = this.socket.register(this.selector,
+ SelectionKey.OP_READ);
+ }
+
+ /**
+ * Sends the OF message out over the socket channel.
+ *
+ * @param msg OF message to be sent
+ * @throws Exception
+ */
+ @Override
+ public void asyncSend(OFMessage msg) throws IOException {
+ synchronized (outBuffer) {
+ int msgLen = msg.getLengthU();
+ if (outBuffer.remaining() < msgLen) {
+ // increase the buffer size so that it can contain this message
+ ByteBuffer newBuffer = ByteBuffer.allocateDirect(outBuffer
+ .capacity()
+ + msgLen);
+ outBuffer.flip();
+ newBuffer.put(outBuffer);
+ outBuffer = newBuffer;
+ }
+ msg.writeTo(outBuffer);
+
+ if (!socket.isOpen()) {
+ return;
+ }
+
+ outBuffer.flip();
+ socket.write(outBuffer);
+ outBuffer.compact();
+ if (outBuffer.position() > 0) {
+ this.clientSelectionKey = this.socket.register(
+ this.selector, SelectionKey.OP_WRITE, this);
+ }
+ logger.trace("Message sent: {}", msg.toString());
+ }
+ }
+
+ /**
+ * Resumes sending the remaining messages in the outgoing buffer
+ * @throws Exception
+ */
+ @Override
+ public void resumeSend() throws IOException {
+ synchronized (outBuffer) {
+ if (!socket.isOpen()) {
+ return;
+ }
+
+ outBuffer.flip();
+ socket.write(outBuffer);
+ outBuffer.compact();
+ if (outBuffer.position() > 0) {
+ this.clientSelectionKey = this.socket.register(
+ this.selector, SelectionKey.OP_WRITE, this);
+ } else {
+ this.clientSelectionKey = this.socket.register(
+ this.selector, SelectionKey.OP_READ, this);
+ }
+ }
+ }
+
+ /**
+ * Reads the incoming network data from the socket and retrieves the OF
+ * messages.
+ *
+ * @return list of OF messages
+ * @throws Exception
+ */
+ @Override
+ public List<OFMessage> readMessages() throws IOException {
+ if (!socket.isOpen()) {
+ return null;
+ }
+
+ List<OFMessage> msgs = null;
+ int bytesRead = -1;
+ bytesRead = socket.read(inBuffer);
+ if (bytesRead < 0) {
+ throw new AsynchronousCloseException();
+ }
+
+ inBuffer.flip();
+ msgs = factory.parseMessages(inBuffer);
+ if (inBuffer.hasRemaining()) {
+ inBuffer.compact();
+ } else {
+ inBuffer.clear();
+ }
+ return msgs;
+ }
+}
--- /dev/null
+
+/*
+ * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.protocol_plugin.openflow.core.internal;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.commons.lang3.builder.ReflectionToStringBuilder;
+import org.openflow.protocol.OFMessage;
+
+/**
+ * This class describes an OpenFlow message with priority
+ */
+class PriorityMessage {
+ OFMessage msg;
+ int priority;
+
+ public PriorityMessage(OFMessage msg, int priority) {
+ this.msg = msg;
+ this.priority = priority;
+ }
+
+ public OFMessage getMsg() {
+ return msg;
+ }
+
+ public void setMsg(OFMessage msg) {
+ this.msg = msg;
+ }
+
+ public int getPriority() {
+ return priority;
+ }
+
+ public void setPriority(int priority) {
+ this.priority = priority;
+ }
+
+ @Override
+ public int hashCode() {
+ return HashCodeBuilder.reflectionHashCode(this);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return EqualsBuilder.reflectionEquals(this, obj);
+ }
+
+ @Override
+ public String toString() {
+ return "PriorityMessage[" + ReflectionToStringBuilder.toString(this) + "]";
+ }
+}
--- /dev/null
+
+/*
+ * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.protocol_plugin.openflow.core.internal;
+
+import java.io.FileInputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousCloseException;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.SocketChannel;
+import java.security.KeyStore;
+import java.security.SecureRandom;
+import java.util.List;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLEngineResult;
+import javax.net.ssl.SSLSession;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.SSLEngineResult.HandshakeStatus;
+import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageReadWrite;
+import org.openflow.protocol.OFMessage;
+import org.openflow.protocol.factory.BasicFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class implements methods to read/write messages over an established
+ * socket channel. The data exchange is encrypted/decrypted by SSLEngine.
+ */
+public class SecureMessageReadWriteService implements IMessageReadWrite {
+ private static final Logger logger = LoggerFactory
+ .getLogger(SecureMessageReadWriteService.class);
+
+ private Selector selector;
+ private SelectionKey clientSelectionKey;
+ private SocketChannel socket;
+ private BasicFactory factory;
+
+ private SSLEngine sslEngine;
+ private SSLEngineResult sslEngineResult; // results from sslEngine last operation
+ private ByteBuffer myAppData; // clear text message to be sent
+ private ByteBuffer myNetData; // encrypted message to be sent
+ private ByteBuffer peerAppData; // clear text message received from the switch
+ private ByteBuffer peerNetData; // encrypted message from the switch
+
+ public SecureMessageReadWriteService(SocketChannel socket, Selector selector) throws Exception {
+ this.socket = socket;
+ this.selector = selector;
+ this.factory = new BasicFactory();
+
+ createSecureChannel(socket);
+ createBuffers(sslEngine);
+ }
+
+ /**
+ * Bring up secure channel using SSL Engine
+ *
+ * @param socket TCP socket channel
+ * @throws Exception
+ */
+ private void createSecureChannel(SocketChannel socket) throws Exception {
+ String keyStoreFile = System.getProperty("controllerKeyStore");
+ String keyStorePassword = System.getProperty("controllerKeyStorePassword");
+ String trustStoreFile = System.getProperty("controllerTrustStore");;
+ String trustStorePassword = System.getProperty("controllerTrustStorePassword");;
+
+ KeyStore ks = KeyStore.getInstance("JKS");
+ KeyStore ts = KeyStore.getInstance("JKS");
+ KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509");
+ TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509");
+ ks.load(new FileInputStream(keyStoreFile), keyStorePassword.toCharArray());
+ ts.load(new FileInputStream(trustStoreFile), trustStorePassword.toCharArray());
+ kmf.init(ks, keyStorePassword.toCharArray());
+ tmf.init(ts);
+
+ SecureRandom random = new SecureRandom();
+ random.nextInt();
+
+ SSLContext sslContext = SSLContext.getInstance("TLS");
+ sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), random);
+ sslEngine = sslContext.createSSLEngine();
+ sslEngine.setUseClientMode(false);
+ sslEngine.setNeedClientAuth(true);
+
+ // Do initial handshake
+ doHandshake(socket, sslEngine);
+
+ this.clientSelectionKey = this.socket.register(this.selector,
+ SelectionKey.OP_READ);
+ }
+
+ /**
+ * Sends the OF message out over the socket channel. The message is
+ * encrypted by SSL Engine.
+ *
+ * @param msg OF message to be sent
+ * @throws Exception
+ */
+ @Override
+ public void asyncSend(OFMessage msg) throws Exception {
+ synchronized (myAppData) {
+ int msgLen = msg.getLengthU();
+ if (myAppData.remaining() < msgLen) {
+ // increase the buffer size so that it can contain this message
+ ByteBuffer newBuffer = ByteBuffer.allocateDirect(myAppData
+ .capacity()
+ + msgLen);
+ myAppData.flip();
+ newBuffer.put(myAppData);
+ myAppData = newBuffer;
+ }
+ msg.writeTo(myAppData);
+ myAppData.flip();
+ sslEngineResult = sslEngine.wrap(myAppData, myNetData);
+ logger.trace("asyncSend sslEngine wrap: {}", sslEngineResult);
+ runDelegatedTasks(sslEngineResult, sslEngine);
+
+ if (!socket.isOpen()) {
+ return;
+ }
+
+ myNetData.flip();
+ socket.write(myNetData);
+ if (myNetData.hasRemaining()) {
+ myNetData.compact();
+ } else {
+ myNetData.clear();
+ }
+
+ if (myAppData.hasRemaining()) {
+ myAppData.compact();
+ this.clientSelectionKey = this.socket.register(
+ this.selector, SelectionKey.OP_WRITE, this);
+ } else {
+ myAppData.clear();
+ this.clientSelectionKey = this.socket.register(
+ this.selector, SelectionKey.OP_READ, this);
+ }
+
+ logger.trace("Message sent: {}", msg.toString());
+ }
+ }
+
+ /**
+ * Resumes sending the remaining messages in the outgoing buffer
+ * @throws Exception
+ */
+ @Override
+ public void resumeSend() throws Exception {
+ synchronized (myAppData) {
+ myAppData.flip();
+ sslEngineResult = sslEngine.wrap(myAppData, myNetData);
+ logger.trace("resumeSend sslEngine wrap: {}", sslEngineResult);
+ runDelegatedTasks(sslEngineResult, sslEngine);
+
+ if (!socket.isOpen()) {
+ return;
+ }
+
+ myNetData.flip();
+ socket.write(myNetData);
+ if (myNetData.hasRemaining()) {
+ myNetData.compact();
+ } else {
+ myNetData.clear();
+ }
+
+ if (myAppData.hasRemaining()) {
+ myAppData.compact();
+ this.clientSelectionKey = this.socket.register(this.selector,
+ SelectionKey.OP_WRITE, this);
+ } else {
+ myAppData.clear();
+ this.clientSelectionKey = this.socket.register(this.selector,
+ SelectionKey.OP_READ, this);
+ }
+ }
+ }
+
+ /**
+ * Reads the incoming network data from the socket, decryptes them and then
+ * retrieves the OF messages.
+ *
+ * @return list of OF messages
+ * @throws Exception
+ */
+ @Override
+ public List<OFMessage> readMessages() throws Exception {
+ if (!socket.isOpen()) {
+ return null;
+ }
+
+ List<OFMessage> msgs = null;
+ int bytesRead = -1;
+ int countDown = 50;
+
+ bytesRead = socket.read(peerNetData);
+ if (bytesRead < 0) {
+ throw new AsynchronousCloseException();
+ }
+
+ do {
+ peerNetData.flip();
+ sslEngineResult = sslEngine.unwrap(peerNetData, peerAppData);
+ if (peerNetData.hasRemaining()) {
+ peerNetData.compact();
+ } else {
+ peerNetData.clear();
+ }
+ logger.trace("sslEngine unwrap result: {}", sslEngineResult);
+ runDelegatedTasks(sslEngineResult, sslEngine);
+ } while ((sslEngineResult.getStatus() == SSLEngineResult.Status.OK) &&
+ peerNetData.hasRemaining() && (--countDown > 0));
+
+ if (countDown == 0) {
+ logger.trace("countDown reaches 0. peerNetData pos {} lim {}", peerNetData.position(), peerNetData.limit());
+ }
+
+ peerAppData.flip();
+ msgs = factory.parseMessages(peerAppData);
+ if (peerAppData.hasRemaining()) {
+ peerAppData.compact();
+ } else {
+ peerAppData.clear();
+ }
+
+ this.clientSelectionKey = this.socket.register(
+ this.selector, SelectionKey.OP_READ, this);
+
+ return msgs;
+ }
+
+ /**
+ * If the result indicates that we have outstanding tasks to do,
+ * go ahead and run them in this thread.
+ */
+ private void runDelegatedTasks(SSLEngineResult result,
+ SSLEngine engine) throws Exception {
+
+ if (result.getHandshakeStatus() == HandshakeStatus.NEED_TASK) {
+ Runnable runnable;
+ while ((runnable = engine.getDelegatedTask()) != null) {
+ logger.debug("\trunning delegated task...");
+ runnable.run();
+ }
+ HandshakeStatus hsStatus = engine.getHandshakeStatus();
+ if (hsStatus == HandshakeStatus.NEED_TASK) {
+ throw new Exception(
+ "handshake shouldn't need additional tasks");
+ }
+ logger.debug("\tnew HandshakeStatus: {}", hsStatus);
+ }
+ }
+
+ private void doHandshake(SocketChannel socket, SSLEngine engine) throws Exception {
+ SSLSession session = engine.getSession();
+ ByteBuffer myAppData = ByteBuffer.allocate(session.getApplicationBufferSize());
+ ByteBuffer peerAppData = ByteBuffer.allocate(session.getApplicationBufferSize());
+ ByteBuffer myNetData = ByteBuffer.allocate(session.getPacketBufferSize());
+ ByteBuffer peerNetData = ByteBuffer.allocate(session.getPacketBufferSize());
+
+ // Begin handshake
+ engine.beginHandshake();
+ SSLEngineResult.HandshakeStatus hs = engine.getHandshakeStatus();
+
+ // Process handshaking message
+ while (hs != SSLEngineResult.HandshakeStatus.FINISHED &&
+ hs != SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING) {
+ switch (hs) {
+ case NEED_UNWRAP:
+ // Receive handshaking data from peer
+ if (socket.read(peerNetData) < 0) {
+ throw new AsynchronousCloseException();
+ }
+
+ // Process incoming handshaking data
+ peerNetData.flip();
+ SSLEngineResult res = engine.unwrap(peerNetData, peerAppData);
+ peerNetData.compact();
+ hs = res.getHandshakeStatus();
+
+ // Check status
+ switch (res.getStatus()) {
+ case OK :
+ // Handle OK status
+ break;
+ }
+ break;
+
+ case NEED_WRAP :
+ // Empty the local network packet buffer.
+ myNetData.clear();
+
+ // Generate handshaking data
+ res = engine.wrap(myAppData, myNetData);
+ hs = res.getHandshakeStatus();
+
+ // Check status
+ switch (res.getStatus()) {
+ case OK :
+ myNetData.flip();
+
+ // Send the handshaking data to peer
+ while (myNetData.hasRemaining()) {
+ if (socket.write(myNetData) < 0) {
+ throw new AsynchronousCloseException();
+ }
+ }
+ break;
+ }
+ break;
+
+ case NEED_TASK :
+ // Handle blocking tasks
+ Runnable runnable;
+ while ((runnable = engine.getDelegatedTask()) != null) {
+ logger.debug("\trunning delegated task...");
+ runnable.run();
+ }
+ hs = engine.getHandshakeStatus();
+ if (hs == HandshakeStatus.NEED_TASK) {
+ throw new Exception(
+ "handshake shouldn't need additional tasks");
+ }
+ logger.debug("\tnew HandshakeStatus: {}", hs);
+ break;
+ }
+ }
+ }
+
+ private void createBuffers(SSLEngine engine) {
+ SSLSession session = engine.getSession();
+ this.myAppData = ByteBuffer.allocate(session.getApplicationBufferSize());
+ this.peerAppData = ByteBuffer.allocate(session.getApplicationBufferSize());
+ this.myNetData = ByteBuffer.allocate(session.getPacketBufferSize());
+ this.peerNetData = ByteBuffer.allocate(session.getPacketBufferSize());
+ }
+}
package org.opendaylight.controller.protocol_plugin.openflow.core.internal;
-import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.ArrayList;
+import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
+import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageReadWrite;
import org.openflow.protocol.OFBarrierReply;
import org.openflow.protocol.OFEchoReply;
import org.openflow.protocol.OFError;
private static final int SWITCH_LIVENESS_TIMER = 5000;
private static final int SWITCH_LIVENESS_TIMEOUT = 2 * SWITCH_LIVENESS_TIMER + 500;
private int MESSAGE_RESPONSE_TIMER = 2000;
- private static final int bufferSize = 1024 * 1024;
private String instanceName;
private ISwitch thisISwitch;
private Byte tables;
private Integer actions;
private Selector selector;
- private SelectionKey clientSelectionKey;
private SocketChannel socket;
- private ByteBuffer inBuffer;
- private ByteBuffer outBuffer;
private BasicFactory factory;
private AtomicInteger xid;
private SwitchState state;
private ExecutorService executor;
private ConcurrentHashMap<Integer, Callable<Object>> messageWaitingDone;
private boolean running;
+ private IMessageReadWrite msgReadWriteService;
private Thread switchHandlerThread;
private Integer responseTimerValue;
-
+ private PriorityBlockingQueue<PriorityMessage> transmitQ;
+ private Thread transmitThread;
+
private enum SwitchState {
NON_OPERATIONAL(0), WAIT_FEATURES_REPLY(1), WAIT_CONFIG_REPLY(2), OPERATIONAL(
3);
this.periodicTimer = null;
this.executor = Executors.newFixedThreadPool(4);
this.messageWaitingDone = new ConcurrentHashMap<Integer, Callable<Object>>();
- this.inBuffer = ByteBuffer.allocateDirect(bufferSize);
- this.outBuffer = ByteBuffer.allocateDirect(bufferSize);
this.responseTimerValue = MESSAGE_RESPONSE_TIMER;
String rTimer = System.getProperty("of.messageResponseTimer");
if (rTimer != null) {
- try {
- responseTimerValue = Integer.decode(rTimer);
- } catch (NumberFormatException e) {
- logger.warn("Invalid of.messageResponseTimer:" + rTimer + ", use default("
- + MESSAGE_RESPONSE_TIMER+ ")");
- }
+ try {
+ responseTimerValue = Integer.decode(rTimer);
+ } catch (NumberFormatException e) {
+ logger.warn("Invalid of.messageResponseTimer: {} use default({})",
+ rTimer, MESSAGE_RESPONSE_TIMER);
+ }
}
- }
+ }
public void start() {
try {
- this.selector = SelectorProvider.provider().openSelector();
- this.socket.configureBlocking(false);
- this.socket.socket().setTcpNoDelay(true);
- this.clientSelectionKey = this.socket.register(this.selector,
- SelectionKey.OP_READ);
+ startTransmitThread();
+ setupCommChannel();
+ sendFirstHello();
startHandlerThread();
} catch (Exception e) {
reportError(e);
- return;
}
}
private void startHandlerThread() {
- OFMessage msg = factory.getMessage(OFType.HELLO);
- asyncSend(msg);
switchHandlerThread = new Thread(new Runnable() {
@Override
public void run() {
running = true;
while (running) {
try {
- // wait for an incoming connection
+ // wait for an incoming connection
selector.select(0);
Iterator<SelectionKey> selectedKeys = selector
.selectedKeys().iterator();
running = false;
selector.wakeup();
cancelSwitchTimer();
- this.clientSelectionKey.cancel();
+ this.selector.close();
this.socket.close();
executor.shutdown();
} catch (Exception e) {
return this.xid.incrementAndGet();
}
+ /**
+ * This method puts the message in an outgoing priority queue with normal
+ * priority. It will be served after high priority messages. The method
+ * should be used for non-critical messages such as statistics request,
+ * discovery packets, etc. An unique XID is generated automatically and
+ * inserted into the message.
+ *
+ * @param msg The OF message to be sent
+ * @return The XID used
+ */
@Override
public Integer asyncSend(OFMessage msg) {
- return asyncSend(msg, getNextXid());
+ return asyncSend(msg, getNextXid());
+ }
+
+ /**
+ * This method puts the message in an outgoing priority queue with normal
+ * priority. It will be served after high priority messages. The method
+ * should be used for non-critical messages such as statistics request,
+ * discovery packets, etc. The specified XID is inserted into the message.
+ *
+ * @param msg The OF message to be Sent
+ * @param xid The XID to be used in the message
+ * @return The XID used
+ */
+ @Override
+ public Integer asyncSend(OFMessage msg, int xid) {
+ msg.setXid(xid);
+ transmitQ.add(new PriorityMessage(msg, 0));
+ return xid;
}
+ /**
+ * This method puts the message in an outgoing priority queue with high
+ * priority. It will be served first before normal priority messages. The
+ * method should be used for critical messages such as hello, echo reply
+ * etc. An unique XID is generated automatically and inserted into the
+ * message.
+ *
+ * @param msg The OF message to be sent
+ * @return The XID used
+ */
@Override
- public Integer asyncSend(OFMessage msg, int xid) {
- synchronized (outBuffer) {
- /*
- if ((msg.getType() != OFType.ECHO_REQUEST) &&
- (msg.getType() != OFType.ECHO_REPLY)) {
- logger.debug("sending " + msg.getType().toString() + " to " + toString());
- }
- */
- msg.setXid(xid);
- int msgLen = msg.getLengthU();
- if (outBuffer.remaining() < msgLen) {
- // increase the buffer size so that it can contain this message
- ByteBuffer newBuffer = ByteBuffer.allocateDirect(outBuffer
- .capacity()
- + msgLen);
- outBuffer.flip();
- newBuffer.put(outBuffer);
- outBuffer = newBuffer;
- }
- msg.writeTo(outBuffer);
- outBuffer.flip();
- try {
- socket.write(outBuffer);
- outBuffer.compact();
- if (outBuffer.position() > 0) {
- this.clientSelectionKey = this.socket.register(
- this.selector, SelectionKey.OP_WRITE, this);
- }
- logger.trace("Message sent: " + msg.toString());
- } catch (Exception e) {
- reportError(e);
- }
- }
+ public Integer asyncFastSend(OFMessage msg) {
+ return asyncFastSend(msg, getNextXid());
+ }
+
+ /**
+ * This method puts the message in an outgoing priority queue with high
+ * priority. It will be served first before normal priority messages. The
+ * method should be used for critical messages such as hello, echo reply
+ * etc. The specified XID is inserted into the message.
+ *
+ * @param msg The OF message to be sent
+ * @return The XID used
+ */
+ @Override
+ public Integer asyncFastSend(OFMessage msg, int xid) {
+ msg.setXid(xid);
+ transmitQ.add(new PriorityMessage(msg, 1));
return xid;
}
- public void resumeSend() {
- synchronized (outBuffer) {
- try {
- outBuffer.flip();
- socket.write(outBuffer);
- outBuffer.compact();
- if (outBuffer.position() > 0) {
- this.clientSelectionKey = this.socket.register(
- this.selector, SelectionKey.OP_WRITE, this);
- } else {
- this.clientSelectionKey = this.socket.register(
- this.selector, SelectionKey.OP_READ, this);
- }
- } catch (Exception e) {
- reportError(e);
- }
- }
+ public void resumeSend() {
+ try {
+ msgReadWriteService.resumeSend();
+ } catch (Exception e) {
+ reportError(e);
+ }
}
public void handleMessages() {
- List<OFMessage> msgs = readMessages();
+ List<OFMessage> msgs = null;
+
+ try {
+ msgs = msgReadWriteService.readMessages();
+ } catch (Exception e) {
+ reportError(e);
+ }
+
if (msgs == null) {
- logger.debug(toString() + " is down");
+ logger.debug("{} is down", toString());
// the connection is down, inform core
reportSwitchStateChange(false);
return;
}
for (OFMessage msg : msgs) {
- logger.trace("Message received: " + msg.toString());
+ logger.trace("Message received: {}", msg.toString());
/*
if ((msg.getType() != OFType.ECHO_REQUEST) &&
(msg.getType() != OFType.ECHO_REPLY)) {
// send feature request
OFMessage featureRequest = factory
.getMessage(OFType.FEATURES_REQUEST);
- asyncSend(featureRequest);
+ asyncFastSend(featureRequest);
// delete all pre-existing flows
OFMatch match = new OFMatch().setWildcards(OFMatch.OFPFW_ALL);
OFFlowMod flowMod = (OFFlowMod) factory
flowMod.setMatch(match).setCommand(OFFlowMod.OFPFC_DELETE)
.setOutPort(OFPort.OFPP_NONE).setLength(
(short) OFFlowMod.MINIMUM_LENGTH);
- asyncSend(flowMod);
+ asyncFastSend(flowMod);
this.state = SwitchState.WAIT_FEATURES_REPLY;
startSwitchTimer();
break;
case ECHO_REQUEST:
OFEchoReply echoReply = (OFEchoReply) factory
.getMessage(OFType.ECHO_REPLY);
- asyncSend(echoReply);
+ asyncFastSend(echoReply);
break;
case ECHO_REPLY:
this.probeSent = false;
}
- private List<OFMessage> readMessages() {
- List<OFMessage> msgs = null;
- int bytesRead;
- try {
- bytesRead = socket.read(inBuffer);
- } catch (Exception e) {
- reportError(e);
- return null;
- }
- if (bytesRead == -1) {
- return null;
- }
- inBuffer.flip();
- msgs = factory.parseMessages(inBuffer);
- if (inBuffer.hasRemaining()) {
- inBuffer.compact();
- } else {
- inBuffer.clear();
- }
- return msgs;
- }
-
private void startSwitchTimer() {
this.periodicTimer = new Timer();
this.periodicTimer.scheduleAtFixedRate(new TimerTask() {
if ((now - lastMsgReceivedTimeStamp) > SWITCH_LIVENESS_TIMEOUT) {
if (probeSent) {
// switch failed to respond to our probe, consider it down
- logger.warn(toString()
- + " is idle for too long, disconnect");
+ logger.warn("{} is idle for too long, disconnect", toString());
reportSwitchStateChange(false);
} else {
// send a probe to see if the switch is still alive
probeSent = true;
OFMessage echo = factory
.getMessage(OFType.ECHO_REQUEST);
- asyncSend(echo);
+ asyncFastSend(echo);
}
} else {
if (state == SwitchState.WAIT_FEATURES_REPLY) {
// send another features request
OFMessage request = factory
.getMessage(OFType.FEATURES_REQUEST);
- asyncSend(request);
+ asyncFastSend(request);
} else {
if (state == SwitchState.WAIT_CONFIG_REPLY) {
// send another config request
.getMessage(OFType.SET_CONFIG);
config.setMissSendLength((short) 0xffff)
.setLengthU(OFSetConfig.MINIMUM_LENGTH);
- asyncSend(config);
+ asyncFastSend(config);
OFMessage getConfig = factory
.getMessage(OFType.GET_CONFIG_REQUEST);
- asyncSend(getConfig);
+ asyncFastSend(getConfig);
}
}
}
}
private void reportError(Exception e) {
- //logger.error(toString() + " caught Error " + e.toString());
- // notify core of this error event
+ logger.debug("Caught exception ", e);
+ // notify core of this error event and disconnect the switch
((Controller) core).takeSwitchEventError(this);
}
.getMessage(OFType.SET_CONFIG);
config.setMissSendLength((short) 0xffff).setLengthU(
OFSetConfig.MINIMUM_LENGTH);
- asyncSend(config);
+ asyncFastSend(config);
// send config request to make sure the switch can handle the set config
OFMessage getConfig = factory.getMessage(OFType.GET_CONFIG_REQUEST);
- asyncSend(getConfig);
+ asyncFastSend(getConfig);
this.state = SwitchState.WAIT_CONFIG_REPLY;
// inform core that a new switch is now operational
reportSwitchStateChange(true);
.get(MESSAGE_RESPONSE_TIMER, TimeUnit.MILLISECONDS);
return result;
} catch (Exception e) {
- logger.warn("Timeout while waiting for " + req.getType()
- + " replies");
+ logger.warn("Timeout while waiting for {} replies", req.getType());
result = null; // to indicate timeout has occurred
return result;
}
} else {
// if result is not null, this means the switch can't handle this message
// the result if OFError already
- logger.debug("Send " + msg.getType().toString()
- + " failed --> " + ((OFError) result).toString());
+ logger.debug("Send {} failed --> {}",
+ msg.getType().toString(), ((OFError) result).toString());
}
return result;
} catch (Exception e) {
- logger.warn("Timeout while waiting for " + msg.getType().toString()
- + " reply");
+ logger.warn("Timeout while waiting for {} reply", msg.getType().toString());
// convert the result into a Boolean with value false
status = false;
result = status;
worker.wakeup();
}
}
-
+
@Override
public Map<Short, OFPhysicalPort> getPhysicalPorts() {
return this.physicalPorts;
}
return result;
}
+
+ /*
+ * Transmit thread polls the message out of the priority queue and invokes
+ * messaging service to transmit it over the socket channel
+ */
+ class PriorityMessageTransmit implements Runnable {
+ public void run() {
+ while (true) {
+ try {
+ if (!transmitQ.isEmpty()) {
+ PriorityMessage pmsg = transmitQ.poll();
+ msgReadWriteService.asyncSend(pmsg.msg);
+ logger.trace("Message sent: {}", pmsg.toString());
+ }
+ Thread.sleep(10);
+ } catch (Exception e) {
+ reportError(e);
+ }
+ }
+ }
+ }
+
+ /*
+ * Setup and start the transmit thread
+ */
+ private void startTransmitThread() {
+ this.transmitQ = new PriorityBlockingQueue<PriorityMessage>(11,
+ new Comparator<PriorityMessage>() {
+ public int compare(PriorityMessage p1, PriorityMessage p2) {
+ return p2.priority - p1.priority;
+ }
+ });
+ this.transmitThread = new Thread(new PriorityMessageTransmit());
+ this.transmitThread.start();
+ }
+
+ /*
+ * Setup communication services
+ */
+ private void setupCommChannel() throws Exception {
+ this.selector = SelectorProvider.provider().openSelector();
+ this.socket.configureBlocking(false);
+ this.socket.socket().setTcpNoDelay(true);
+ this.msgReadWriteService = getMessageReadWriteService();
+ }
+
+ private void sendFirstHello() {
+ try {
+ OFMessage msg = factory.getMessage(OFType.HELLO);
+ asyncFastSend(msg);
+ } catch (Exception e) {
+ reportError(e);
+ }
+ }
+
+ private IMessageReadWrite getMessageReadWriteService() throws Exception {
+ String str = System.getProperty("secureChannelEnabled");
+ return ((str != null) && (str.equalsIgnoreCase("true"))) ?
+ new SecureMessageReadWriteService(socket, selector) :
+ new MessageReadWriteService(socket, selector);
+ }
}