opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketMuxDemux.java

   1 /*
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.protocol_plugin.openflow.internal;
  10
  11 import java.util.Collections;
  12 import java.util.List;
  13 import java.util.Map;
  14 import java.util.Set;
  15 import java.util.concurrent.ConcurrentHashMap;
  16 import java.util.concurrent.ConcurrentMap;
  17 import java.util.concurrent.CopyOnWriteArrayList;
  18
  19 import org.opendaylight.controller.protocol_plugin.openflow.IDataPacketListen;
  20 import org.opendaylight.controller.protocol_plugin.openflow.IDataPacketMux;
  21 import org.opendaylight.controller.protocol_plugin.openflow.IInventoryShimExternalListener;
  22 import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
  23 import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageListener;
  24 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
  25 import org.openflow.protocol.OFMessage;
  26 import org.openflow.protocol.OFPacketIn;
  27 import org.openflow.protocol.OFPacketOut;
  28 import org.openflow.protocol.OFPort;
  29 import org.openflow.protocol.OFType;
  30 import org.openflow.protocol.action.OFAction;
  31 import org.openflow.protocol.action.OFActionOutput;
  32 import org.slf4j.Logger;
  33 import org.slf4j.LoggerFactory;
  34
  35 import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
  36 import org.opendaylight.controller.sal.core.ConstructionException;
  37 import org.opendaylight.controller.sal.core.ContainerFlow;
  38 import org.opendaylight.controller.sal.core.IContainerListener;
  39 import org.opendaylight.controller.sal.core.Node;
  40 import org.opendaylight.controller.sal.core.NodeConnector;
  41 import org.opendaylight.controller.sal.core.Property;
  42 import org.opendaylight.controller.sal.core.UpdateType;
  43 import org.opendaylight.controller.sal.packet.IPluginOutDataPacketService;
  44 import org.opendaylight.controller.sal.packet.PacketResult;
  45 import org.opendaylight.controller.sal.packet.RawPacket;
  46 import org.opendaylight.controller.sal.utils.GlobalConstants;
  47 import org.opendaylight.controller.sal.utils.HexEncode;
  48
  49 public class DataPacketMuxDemux implements IContainerListener,
  50         IMessageListener, IDataPacketMux, IInventoryShimExternalListener {
  51     protected static final Logger logger = LoggerFactory
  52             .getLogger(DataPacketMuxDemux.class);
  53     private IController controller = null;
  54     private ConcurrentMap<Long, ISwitch> swID2ISwitch = new ConcurrentHashMap<Long, ISwitch>();
  55     // Gives a map between a Container and all the DataPacket listeners on SAL
  56     private ConcurrentMap<String, IPluginOutDataPacketService> pluginOutDataPacketServices = new ConcurrentHashMap<String, IPluginOutDataPacketService>();
  57     // Gives a map between a NodeConnector and the containers to which it
  58     // belongs
  59     private ConcurrentMap<NodeConnector, List<String>> nc2Container = new ConcurrentHashMap<NodeConnector, List<String>>();
  60     // Gives a map between a Container and the FlowSpecs on it
  61     private ConcurrentMap<String, List<ContainerFlow>> container2FlowSpecs = new ConcurrentHashMap<String, List<ContainerFlow>>();
  62     // Track local data packet listener
  63     private List<IDataPacketListen> iDataPacketListen = new CopyOnWriteArrayList<IDataPacketListen>();
  64     private IPluginOutConnectionService connectionOutService;
  65
  66     void setIDataPacketListen(IDataPacketListen s) {
  67         if (this.iDataPacketListen != null) {
  68             if (!this.iDataPacketListen.contains(s)) {
  69                 logger.debug("Added new IDataPacketListen");
  70                 this.iDataPacketListen.add(s);
  71             }
  72         }
  73     }
  74
  75     void unsetIDataPacketListen(IDataPacketListen s) {
  76         if (this.iDataPacketListen != null) {
  77             if (this.iDataPacketListen.contains(s)) {
  78                 logger.debug("Removed IDataPacketListen");
  79                 this.iDataPacketListen.remove(s);
  80             }
  81         }
  82     }
  83
  84     void setPluginOutDataPacketService(Map<String, Object> props,
  85             IPluginOutDataPacketService s) {
  86         if (props == null) {
  87             logger.error("Didn't receive the service properties");
  88             return;
  89         }
  90         String containerName = (String) props.get("containerName");
  91         if (containerName == null) {
  92             logger.error("containerName not supplied");
  93             return;
  94         }
  95         if (this.pluginOutDataPacketServices != null) {
  96             // It's expected only one SAL per container as long as the
  97             // replication is done in the SAL implementation toward
  98             // the different APPS
  99             this.pluginOutDataPacketServices.put(containerName, s);
 100             logger.debug("New outService for container: {}", containerName);
 101         }
 102     }
 103
 104     void unsetPluginOutDataPacketService(Map<String, Object> props,
 105             IPluginOutDataPacketService s) {
 106         if (props == null) {
 107             logger.error("Didn't receive the service properties");
 108             return;
 109         }
 110         String containerName = (String) props.get("containerName");
 111         if (containerName == null) {
 112             logger.error("containerName not supplied");
 113             return;
 114         }
 115         if (this.pluginOutDataPacketServices != null) {
 116             this.pluginOutDataPacketServices.remove(containerName);
 117             logger.debug("Removed outService for container: {}", containerName);
 118         }
 119     }
 120
 121     void setController(IController s) {
 122         logger.debug("Controller provider set in DATAPACKET SERVICES");
 123         this.controller = s;
 124     }
 125
 126     void unsetController(IController s) {
 127         if (this.controller == s) {
 128             logger.debug("Controller provider UNset in DATAPACKET SERVICES");
 129             this.controller = null;
 130         }
 131     }
 132
 133     void setIPluginOutConnectionService(IPluginOutConnectionService s) {
 134         connectionOutService = s;
 135     }
 136
 137     void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
 138         if (connectionOutService == s) {
 139             connectionOutService = null;
 140         }
 141     }
 142
 143     /**
 144      * Function called by the dependency manager when all the required
 145      * dependencies are satisfied
 146      *
 147      */
 148     void init() {
 149         this.controller.addMessageListener(OFType.PACKET_IN, this);
 150     }
 151
 152     /**
 153      * Function called by the dependency manager when at least one dependency
 154      * become unsatisfied or when the component is shutting down because for
 155      * example bundle is being stopped.
 156      *
 157      */
 158     void destroy() {
 159         this.controller.removeMessageListener(OFType.PACKET_IN, this);
 160
 161         // Clear state that may need to be reused on component restart
 162         this.pluginOutDataPacketServices.clear();
 163         this.nc2Container.clear();
 164         this.container2FlowSpecs.clear();
 165         this.controller = null;
 166         this.swID2ISwitch.clear();
 167     }
 168
 169     @Override
 170     public void receive(ISwitch sw, OFMessage msg) {
 171         if (sw == null || msg == null
 172                 || this.pluginOutDataPacketServices == null) {
 173             // Something fishy, we cannot do anything
 174             logger.debug(
 175                     "sw: {} and/or msg: {} and/or pluginOutDataPacketServices: {} is null!",
 176                     new Object[] { sw, msg, this.pluginOutDataPacketServices });
 177             return;
 178         }
 179
 180         Long ofSwitchID = Long.valueOf(sw.getId());
 181         try {
 182             Node n = new Node(Node.NodeIDType.OPENFLOW, ofSwitchID);
 183             if (!connectionOutService.isLocal(n)) {
 184                 logger.debug("Connection service refused DataPacketMuxDemux receive {} {}", sw, msg);
 185                 return;
 186             }
 187         }
 188         catch (Exception e) {
 189             return;
 190         }
 191
 192         if (msg instanceof OFPacketIn) {
 193             OFPacketIn ofPacket = (OFPacketIn) msg;
 194             Short ofPortID = Short.valueOf(ofPacket.getInPort());
 195
 196             try {
 197                 Node n = new Node(Node.NodeIDType.OPENFLOW, ofSwitchID);
 198                 NodeConnector p = PortConverter.toNodeConnector(ofPortID, n);
 199                 RawPacket dataPacket = new RawPacket(ofPacket.getPacketData());
 200                 dataPacket.setIncomingNodeConnector(p);
 201
 202                 // Try to dispatch the packet locally, in here we will
 203                 // pass the parsed packet simply because once the
 204                 // packet infra is settled all the packets will passed
 205                 // around as parsed and read-only
 206                 for (int i = 0; i < this.iDataPacketListen.size(); i++) {
 207                     IDataPacketListen s = this.iDataPacketListen.get(i);
 208                     if (s.receiveDataPacket(dataPacket).equals(
 209                             PacketResult.CONSUME)) {
 210                         logger.trace("Consumed locally data packet");
 211                         return;
 212                     }
 213                 }
 214
 215                 // Now dispatch the packet toward SAL at least for
 216                 // default container, we need to revisit this in a general
 217                 // slicing architecture refresh
 218                 IPluginOutDataPacketService defaultOutService = this.pluginOutDataPacketServices
 219                         .get(GlobalConstants.DEFAULT.toString());
 220                 if (defaultOutService != null) {
 221                     defaultOutService.receiveDataPacket(dataPacket);
 222                     if (logger.isTraceEnabled()) {
 223                       logger.trace(
 224                               "Dispatched to apps a frame of size: {} on " +
 225                               "container: {}: {}", new Object[] {
 226                                     ofPacket.getPacketData().length,
 227                                     GlobalConstants.DEFAULT.toString(),
 228                                     HexEncode.bytesToHexString(dataPacket
 229                                             .getPacketData()) });
 230                     }
 231                 }
 232                 // Now check the mapping between nodeConnector and
 233                 // Container and later on optimally filter based on
 234                 // flowSpec
 235                 List<String> containersRX = this.nc2Container.get(p);
 236                 if (containersRX != null) {
 237                     for (int i = 0; i < containersRX.size(); i++) {
 238                         String container = containersRX.get(i);
 239                         IPluginOutDataPacketService s = this.pluginOutDataPacketServices
 240                                 .get(container);
 241                         if (s != null) {
 242                             // TODO add filtering on a per-flowSpec base
 243                             s.receiveDataPacket(dataPacket);
 244                             if (logger.isTraceEnabled()) {
 245                               logger.trace(
 246                                       "Dispatched to apps a frame of size: {}" +
 247                                       " on container: {}: {}", new Object[] {
 248                                             ofPacket.getPacketData().length,
 249                                             container,
 250                                             HexEncode.bytesToHexString(dataPacket
 251                                                     .getPacketData()) });
 252                             }
 253                         }
 254                     }
 255                 }
 256
 257                 // This is supposed to be the catch all for all the
 258                 // DataPacket hence we will assume it has been handled
 259                 return;
 260             } catch (ConstructionException cex) {
 261             }
 262
 263             // If we reach this point something went wrong.
 264             return;
 265         } else {
 266             // We don't care about non-data packets
 267             return;
 268         }
 269     }
 270
 271     @Override
 272     public void transmitDataPacket(RawPacket outPkt) {
 273         // Sanity check area
 274         if (outPkt == null) {
 275             logger.debug("outPkt is null!");
 276             return;
 277         }
 278
 279         NodeConnector outPort = outPkt.getOutgoingNodeConnector();
 280         if (outPort == null) {
 281             logger.debug("outPort is null! outPkt: {}", outPkt);
 282             return;
 283         }
 284
 285         if (!connectionOutService.isLocal(outPort.getNode())) {
 286             logger.debug("data packets will not be sent to {} in a non-master controller", outPort.toString());
 287             return;
 288         }
 289
 290
 291         if (!outPort.getType().equals(
 292                 NodeConnector.NodeConnectorIDType.OPENFLOW)) {
 293             // The output Port is not of type OpenFlow
 294             logger.debug("outPort is not OF Type! outPort: {}", outPort);
 295             return;
 296         }
 297
 298         Short port = (Short) outPort.getID();
 299         Long swID = (Long) outPort.getNode().getID();
 300         ISwitch sw = this.swID2ISwitch.get(swID);
 301
 302         if (sw == null) {
 303             // If we cannot get the controller descriptor we cannot even
 304             // send out the frame
 305             logger.debug("swID: {} - sw is null!", swID);
 306             return;
 307         }
 308
 309         byte[] data = outPkt.getPacketData();
 310         // build action
 311         OFActionOutput action = new OFActionOutput().setPort(port);
 312         // build packet out
 313         OFPacketOut po = new OFPacketOut()
 314                 .setBufferId(OFPacketOut.BUFFER_ID_NONE)
 315                 .setInPort(OFPort.OFPP_NONE)
 316                 .setActions(Collections.singletonList((OFAction) action))
 317                 .setActionsLength((short) OFActionOutput.MINIMUM_LENGTH);
 318
 319         po.setLengthU(OFPacketOut.MINIMUM_LENGTH + po.getActionsLength()
 320                 + data.length);
 321         po.setPacketData(data);
 322
 323         // send PACKET_OUT at high priority
 324         sw.asyncFastSend(po);
 325         logger.trace("Transmitted a frame of size: {}", data.length);
 326     }
 327
 328     public void addNode(Node node, Set<Property> props) {
 329         if (node == null) {
 330             logger.debug("node is null!");
 331             return;
 332         }
 333
 334         long sid = (Long) node.getID();
 335         ISwitch sw = controller.getSwitches().get(sid);
 336         if (sw == null) {
 337             logger.debug("sid: {} - sw is null!", sid);
 338             return;
 339         }
 340         this.swID2ISwitch.put(sw.getId(), sw);
 341     }
 342
 343     public void removeNode(Node node) {
 344         if (node == null) {
 345             logger.debug("node is null!");
 346             return;
 347         }
 348
 349         long sid = (Long) node.getID();
 350         ISwitch sw = controller.getSwitches().get(sid);
 351         if (sw == null) {
 352             logger.debug("sid: {} - sw is null!", sid);
 353             return;
 354         }
 355         this.swID2ISwitch.remove(sw.getId());
 356     }
 357
 358     @Override
 359     public void tagUpdated(String containerName, Node n, short oldTag,
 360             short newTag, UpdateType t) {
 361         // Do nothing
 362     }
 363
 364     @Override
 365     public void containerFlowUpdated(String containerName,
 366             ContainerFlow previousFlow, ContainerFlow currentFlow, UpdateType t) {
 367         if (this.container2FlowSpecs == null) {
 368             logger.error("container2FlowSpecs is NULL");
 369             return;
 370         }
 371         List<ContainerFlow> fSpecs = this.container2FlowSpecs
 372                 .get(containerName);
 373         if (fSpecs == null) {
 374             fSpecs = new CopyOnWriteArrayList<ContainerFlow>();
 375         }
 376         switch (t) {
 377         case ADDED:
 378             if (!fSpecs.contains(previousFlow)) {
 379                 fSpecs.add(previousFlow);
 380             }
 381             break;
 382         case REMOVED:
 383             if (fSpecs.contains(previousFlow)) {
 384                 fSpecs.remove(previousFlow);
 385             }
 386             break;
 387         case CHANGED:
 388             break;
 389         }
 390     }
 391
 392     @Override
 393     public void nodeConnectorUpdated(String containerName, NodeConnector p,
 394             UpdateType t) {
 395         if (this.nc2Container == null) {
 396             logger.error("nc2Container is NULL");
 397             return;
 398         }
 399         List<String> containers = this.nc2Container.get(p);
 400         if (containers == null) {
 401             containers = new CopyOnWriteArrayList<String>();
 402         }
 403         boolean updateMap = false;
 404         switch (t) {
 405         case ADDED:
 406             if (!containers.contains(containerName)) {
 407                 containers.add(containerName);
 408                 updateMap = true;
 409             }
 410             break;
 411         case REMOVED:
 412             if (containers.contains(containerName)) {
 413                 containers.remove(containerName);
 414                 updateMap = true;
 415             }
 416             break;
 417         case CHANGED:
 418             break;
 419         }
 420         if (updateMap) {
 421             if (containers.isEmpty()) {
 422                 // Do cleanup to reduce memory footprint if no
 423                 // elements to be tracked
 424                 this.nc2Container.remove(p);
 425             } else {
 426                 this.nc2Container.put(p, containers);
 427             }
 428         }
 429     }
 430
 431     @Override
 432     public void containerModeUpdated(UpdateType t) {
 433         // do nothing
 434     }
 435
 436     @Override
 437     public void updateNode(Node node, UpdateType type, Set<Property> props) {
 438         switch (type) {
 439         case ADDED:
 440             addNode(node, props);
 441             break;
 442         case REMOVED:
 443             removeNode(node);
 444             break;
 445         default:
 446             break;
 447         }
 448     }
 449
 450     @Override
 451     public void updateNodeConnector(NodeConnector nodeConnector,
 452             UpdateType type, Set<Property> props) {
 453         // do nothing
 454     }
 455 }