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