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.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                 // Now dispatch the packet toward SAL for default container
 216                 IPluginOutDataPacketService defaultOutService = this.pluginOutDataPacketServices
 217                         .get(GlobalConstants.DEFAULT.toString());
 218                 if (defaultOutService != null) {
 219                     defaultOutService.receiveDataPacket(dataPacket);
 220                     if (logger.isTraceEnabled()) {
 221                         logger.trace("Dispatched to apps a frame of size: {} on " + "container: {}: {}",
 222                                 new Object[] { ofPacket.getPacketData().length, GlobalConstants.DEFAULT.toString(),
 223                                         HexEncode.bytesToHexString(dataPacket.getPacketData()) });
 224                     }
 225                 }
 226                 // Now check the mapping between nodeConnector and
 227                 // Container and later on optimally filter based on
 228                 // flowSpec
 229                 List<String> containersRX = this.nc2Container.get(p);
 230                 if (containersRX != null) {
 231                     Ethernet frame = new Ethernet();
 232                     byte data[] = dataPacket.getPacketData();
 233                     frame.deserialize(data, 0, data.length * NetUtils.NumBitsInAByte);
 234                     Match packetMatch = frame.getMatch();
 235                     for (String container : containersRX) {
 236                         boolean notify = true;
 237                         List<ContainerFlow> containerFlows = this.container2FlowSpecs.get(container);
 238                         if (containerFlows != null) {
 239                             notify = false;
 240                             for (ContainerFlow cFlow : containerFlows) {
 241                                 if (cFlow.allowsMatch(packetMatch)) {
 242                                     notify = true;
 243                                     break;
 244                                 }
 245                             }
 246                             if (notify) {
 247                                 IPluginOutDataPacketService s = this.pluginOutDataPacketServices.get(container);
 248                                 if (s != null) {
 249                                     s.receiveDataPacket(dataPacket);
 250                                     if (logger.isTraceEnabled()) {
 251                                         logger.trace(
 252                                                 "Dispatched to apps a frame of size: {}" + " on container: {}: {}",
 253                                                 new Object[] { ofPacket.getPacketData().length, container,
 254                                                         HexEncode.bytesToHexString(dataPacket.getPacketData()) });
 255                                     }
 256                                 }
 257                             }
 258                         }
 259                     }
 260                 }
 261                 // This is supposed to be the catch all for all the
 262                 // DataPacket hence we will assume it has been handled
 263                 return;
 264             } catch (ConstructionException cex) {
 265             } catch (PacketException e) {
 266                 logger.debug("Failed to deserialize raw packet: ", e.getMessage());
 267             }
 268             // If we reach this point something went wrong.
 269             return;
 270         } else {
 271             // We don't care about non-data packets
 272             return;
 273         }
 274     }
 275
 276     @Override
 277     public void transmitDataPacket(RawPacket outPkt) {
 278         // Sanity check area
 279         if (outPkt == null) {
 280             logger.debug("outPkt is null!");
 281             return;
 282         }
 283
 284         NodeConnector outPort = outPkt.getOutgoingNodeConnector();
 285         if (outPort == null) {
 286             logger.debug("outPort is null! outPkt: {}", outPkt);
 287             return;
 288         }
 289
 290         if (!connectionOutService.isLocal(outPort.getNode())) {
 291             logger.debug("data packets will not be sent to {} in a non-master controller", outPort.toString());
 292             return;
 293         }
 294
 295
 296         if (!outPort.getType().equals(
 297                 NodeConnector.NodeConnectorIDType.OPENFLOW)) {
 298             // The output Port is not of type OpenFlow
 299             logger.debug("outPort is not OF Type! outPort: {}", outPort);
 300             return;
 301         }
 302
 303         Short port = (Short) outPort.getID();
 304         Long swID = (Long) outPort.getNode().getID();
 305         ISwitch sw = this.swID2ISwitch.get(swID);
 306
 307         if (sw == null) {
 308             // If we cannot get the controller descriptor we cannot even
 309             // send out the frame
 310             logger.debug("swID: {} - sw is null!", swID);
 311             return;
 312         }
 313
 314         byte[] data = outPkt.getPacketData();
 315         // build action
 316         OFActionOutput action = new OFActionOutput().setPort(port);
 317         // build packet out
 318         OFPacketOut po = new OFPacketOut()
 319                 .setBufferId(OFPacketOut.BUFFER_ID_NONE)
 320                 .setActions(Collections.singletonList((OFAction) action))
 321                 .setActionsLength((short) OFActionOutput.MINIMUM_LENGTH);
 322         if(outPkt.getIncomingNodeConnector() != null) {
 323             po.setInPort((Short)outPkt.getIncomingNodeConnector().getID());
 324         } else {
 325             po.setInPort(OFPort.OFPP_NONE);
 326         }
 327
 328         po.setLengthU(OFPacketOut.MINIMUM_LENGTH + po.getActionsLength()
 329                 + data.length);
 330         po.setPacketData(data);
 331
 332         // send PACKET_OUT at high priority
 333         sw.asyncFastSend(po);
 334         logger.trace("Transmitted a frame of size: {}", data.length);
 335     }
 336
 337     public void addNode(Node node, Set<Property> props) {
 338         if (node == null) {
 339             logger.debug("node is null!");
 340             return;
 341         }
 342
 343         long sid = (Long) node.getID();
 344         ISwitch sw = controller.getSwitches().get(sid);
 345         if (sw == null) {
 346             logger.debug("sid: {} - sw is null!", sid);
 347             return;
 348         }
 349         this.swID2ISwitch.put(sw.getId(), sw);
 350     }
 351
 352     public void removeNode(Node node) {
 353         if (node == null) {
 354             logger.debug("node is null!");
 355             return;
 356         }
 357
 358         long sid = (Long) node.getID();
 359         ISwitch sw = controller.getSwitches().get(sid);
 360         if (sw == null) {
 361             logger.debug("sid: {} - sw is null!", sid);
 362             return;
 363         }
 364         this.swID2ISwitch.remove(sw.getId());
 365     }
 366
 367     @Override
 368     public void tagUpdated(String containerName, Node n, short oldTag,
 369             short newTag, UpdateType t) {
 370         // Do nothing
 371     }
 372
 373     @Override
 374     public void containerFlowUpdated(String containerName,
 375             ContainerFlow previousFlow, ContainerFlow currentFlow, UpdateType t) {
 376         if (this.container2FlowSpecs == null) {
 377             logger.error("container2FlowSpecs is NULL");
 378             return;
 379         }
 380         List<ContainerFlow> fSpecs = this.container2FlowSpecs
 381                 .get(containerName);
 382         if (fSpecs == null) {
 383             fSpecs = new CopyOnWriteArrayList<ContainerFlow>();
 384         }
 385         switch (t) {
 386         case ADDED:
 387             if (!fSpecs.contains(currentFlow)) {
 388                 fSpecs.add(currentFlow);
 389             }
 390             container2FlowSpecs.put(containerName, fSpecs);
 391             break;
 392         case REMOVED:
 393             fSpecs.remove(currentFlow);
 394             break;
 395         case CHANGED:
 396             break;
 397         }
 398     }
 399
 400     @Override
 401     public void nodeConnectorUpdated(String containerName, NodeConnector p,
 402             UpdateType t) {
 403         if (this.nc2Container == null) {
 404             logger.error("nc2Container is NULL");
 405             return;
 406         }
 407         List<String> containers = this.nc2Container.get(p);
 408         if (containers == null) {
 409             containers = new CopyOnWriteArrayList<String>();
 410         }
 411         boolean updateMap = false;
 412         switch (t) {
 413         case ADDED:
 414             if (!containers.contains(containerName)) {
 415                 containers.add(containerName);
 416                 updateMap = true;
 417             }
 418             break;
 419         case REMOVED:
 420             if (containers.contains(containerName)) {
 421                 containers.remove(containerName);
 422                 updateMap = true;
 423             }
 424             break;
 425         case CHANGED:
 426             break;
 427         }
 428         if (updateMap) {
 429             if (containers.isEmpty()) {
 430                 // Do cleanup to reduce memory footprint if no
 431                 // elements to be tracked
 432                 this.nc2Container.remove(p);
 433             } else {
 434                 this.nc2Container.put(p, containers);
 435             }
 436         }
 437     }
 438
 439     @Override
 440     public void containerModeUpdated(UpdateType t) {
 441         // do nothing
 442     }
 443
 444     @Override
 445     public void updateNode(Node node, UpdateType type, Set<Property> props) {
 446         switch (type) {
 447         case ADDED:
 448             addNode(node, props);
 449             break;
 450         case REMOVED:
 451             removeNode(node);
 452             break;
 453         default:
 454             break;
 455         }
 456     }
 457
 458     @Override
 459     public void updateNodeConnector(NodeConnector nodeConnector,
 460             UpdateType type, Set<Property> props) {
 461         // do nothing
 462     }
 463
 464     @Override
 465     public void containerCreate(String containerName) {
 466         // do nothing
 467     }
 468
 469     @Override
 470     public void containerDestroy(String containerName) {
 471         Set<NodeConnector> removeNodeConnectorSet = new HashSet<NodeConnector>();
 472         for (Map.Entry<NodeConnector, List<String>> entry : nc2Container.entrySet()) {
 473             List<String> ncContainers = entry.getValue();
 474             if (ncContainers.contains(containerName)) {
 475                 NodeConnector nodeConnector = entry.getKey();
 476                 removeNodeConnectorSet.add(nodeConnector);
 477             }
 478         }
 479         for (NodeConnector nodeConnector : removeNodeConnectorSet) {
 480             List<String> ncContainers = nc2Container.get(nodeConnector);
 481             ncContainers.remove(containerName);
 482             if (ncContainers.isEmpty()) {
 483                 nc2Container.remove(nodeConnector);
 484             }
 485         }
 486         container2FlowSpecs.remove(containerName);
 487     }
 488 }