import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
+import java.util.Comparator;
import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
-import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import org.eclipse.osgi.framework.console.CommandInterpreter;
import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageListener;
import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitchStateListener;
+import org.opendaylight.controller.sal.connection.ConnectionConstants;
+import org.opendaylight.controller.sal.connection.IPluginInConnectionService;
+import org.opendaylight.controller.sal.core.Node;
+import org.opendaylight.controller.sal.utils.Status;
+import org.opendaylight.controller.sal.utils.StatusCode;
import org.openflow.protocol.OFMessage;
import org.openflow.protocol.OFType;
import org.openflow.util.HexString;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class Controller implements IController, CommandProvider {
+public class Controller implements IController, CommandProvider, IPluginInConnectionService {
private static final Logger logger = LoggerFactory
.getLogger(Controller.class);
private ControllerIO controllerIO;
private Thread switchEventThread;
private ConcurrentHashMap<Long, ISwitch> switches;
- private BlockingQueue<SwitchEvent> switchEvents;
+ private PriorityBlockingQueue<SwitchEvent> switchEvents;
// only 1 message listener per OFType
private ConcurrentMap<OFType, IMessageListener> messageListeners;
// only 1 switch state listener
private ISwitchStateListener switchStateListener;
private AtomicInteger switchInstanceNumber;
+ private int MAXQUEUESIZE = 50000;
+
+ private static enum SwitchEventPriority { LOW, NORMAL, HIGH }
/*
* this thread monitors the switchEvents queue for new incoming events from
public void init() {
logger.debug("Initializing!");
this.switches = new ConcurrentHashMap<Long, ISwitch>();
- this.switchEvents = new LinkedBlockingQueue<SwitchEvent>();
+ this.switchEvents = new PriorityBlockingQueue<SwitchEvent>(MAXQUEUESIZE, new Comparator<SwitchEvent>() {
+ @Override
+ public int compare(SwitchEvent p1, SwitchEvent p2) {
+ return p2.getPriority() - p1.getPriority();
+ }
+ });
this.messageListeners = new ConcurrentHashMap<OFType, IMessageListener>();
this.switchStateListener = null;
this.switchInstanceNumber = new AtomicInteger(0);
// create new switch
int i = this.switchInstanceNumber.addAndGet(1);
String instanceName = "SwitchHandler-" + i;
- SwitchHandler switchHandler = new SwitchHandler(this, sc,
- instanceName);
+ SwitchHandler switchHandler = new SwitchHandler(this, sc, instanceName);
switchHandler.start();
if (sc.isConnected()) {
logger.info("Switch:{} is connected to the Controller",
if (((SwitchHandler) sw).isOperational()) {
Long sid = sw.getId();
if (this.switches.remove(sid, sw)) {
- logger.warn("{} is Disconnected", sw);
+ logger.info("{} is removed", sw);
notifySwitchDeleted(sw);
}
}
}
private synchronized void addSwitchEvent(SwitchEvent event) {
- try {
- this.switchEvents.put(event);
- } catch (InterruptedException e) {
- logger.debug("SwitchEvent caught Interrupt Exception");
- }
+ this.switchEvents.put(event);
}
public void takeSwitchEventAdd(ISwitch sw) {
- SwitchEvent ev = new SwitchEvent(
- SwitchEvent.SwitchEventType.SWITCH_ADD, sw, null);
+ SwitchEvent ev = new SwitchEvent(SwitchEvent.SwitchEventType.SWITCH_ADD, sw, null,
+ SwitchEventPriority.HIGH.ordinal());
addSwitchEvent(ev);
}
public void takeSwitchEventDelete(ISwitch sw) {
- SwitchEvent ev = new SwitchEvent(
- SwitchEvent.SwitchEventType.SWITCH_DELETE, sw, null);
+ SwitchEvent ev = new SwitchEvent(SwitchEvent.SwitchEventType.SWITCH_DELETE, sw, null,
+ SwitchEventPriority.HIGH.ordinal());
addSwitchEvent(ev);
}
public void takeSwitchEventError(ISwitch sw) {
- SwitchEvent ev = new SwitchEvent(
- SwitchEvent.SwitchEventType.SWITCH_ERROR, sw, null);
+ SwitchEvent ev = new SwitchEvent(SwitchEvent.SwitchEventType.SWITCH_ERROR, sw, null,
+ SwitchEventPriority.NORMAL.ordinal());
addSwitchEvent(ev);
}
public void takeSwitchEventMsg(ISwitch sw, OFMessage msg) {
if (messageListeners.get(msg.getType()) != null) {
- SwitchEvent ev = new SwitchEvent(
- SwitchEvent.SwitchEventType.SWITCH_MESSAGE, sw, msg);
+ SwitchEvent ev = new SwitchEvent(SwitchEvent.SwitchEventType.SWITCH_MESSAGE, sw, msg,
+ SwitchEventPriority.LOW.ordinal());
addSwitchEvent(ev);
}
}
return this.switches.get(switchId);
}
+ public void _controllerShowQueueSize(CommandInterpreter ci) {
+ ci.print("switchEvents queue size: " + switchEvents.size() + "\n");
+ }
+
public void _controllerShowSwitches(CommandInterpreter ci) {
Set<Long> sids = switches.keySet();
StringBuffer s = new StringBuffer();
help.append("\t controllerShowSwitches\n");
help.append("\t controllerReset\n");
help.append("\t controllerShowConnConfig\n");
+ help.append("\t controllerShowQueueSize\n");
return help.toString();
}
+
+ @Override
+ public Status disconnect(Node node) {
+ ISwitch sw = getSwitch((Long) node.getID());
+ if (sw != null) disconnectSwitch(sw);
+ return new Status(StatusCode.SUCCESS);
+ }
+
+ @Override
+ public Node connect(String connectionIdentifier, Map<ConnectionConstants, String> params) {
+ return null;
+ }
+
+ /**
+ * View Change notification
+ */
+ public void notifyClusterViewChanged() {
+ for (ISwitch sw : switches.values()) {
+ notifySwitchAdded(sw);
+ }
+ }
+
+ /**
+ * Node Disconnected from the node's master controller.
+ */
+ @Override
+ public void notifyNodeDisconnectFromMaster(Node node) {
+ ISwitch sw = switches.get((Long)node.getID());
+ if (sw != null) notifySwitchAdded(sw);
+ }
}
Code Review / controller.git / blobdiff