opendaylight/sal/implementation/src/main/java/org/opendaylight/controller/sal/implementation/internal/DataPacketService.java

   1
   2 /*
   3  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   4  *
   5  * This program and the accompanying materials are made available under the
   6  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   7  * and is available at http://www.eclipse.org/legal/epl-v10.html
   8  */
   9
  10 /**
  11  * @file   DataPacketService.java
  12  *
  13  * @brief  Implementation of Data Packet services in SAL
  14  *
  15  * Implementation of Data Packet services in SAL
  16  */
  17
  18 package org.opendaylight.controller.sal.implementation.internal;
  19
  20 import java.util.Collections;
  21 import java.util.HashMap;
  22 import java.util.HashSet;
  23 import java.util.List;
  24 import java.util.Map;
  25 import java.util.Set;
  26 import java.util.concurrent.ConcurrentHashMap;
  27 import java.util.concurrent.CopyOnWriteArrayList;
  28 import java.util.concurrent.LinkedBlockingQueue;
  29 import java.util.concurrent.atomic.AtomicInteger;
  30
  31 import org.opendaylight.controller.sal.core.ConstructionException;
  32 import org.opendaylight.controller.sal.core.NodeConnector;
  33 import org.opendaylight.controller.sal.match.Match;
  34 import org.opendaylight.controller.sal.packet.Ethernet;
  35 import org.opendaylight.controller.sal.packet.IDataPacketService;
  36 import org.opendaylight.controller.sal.packet.IListenDataPacket;
  37 import org.opendaylight.controller.sal.packet.IPluginInDataPacketService;
  38 import org.opendaylight.controller.sal.packet.IPluginOutDataPacketService;
  39 import org.opendaylight.controller.sal.packet.LinkEncap;
  40 import org.opendaylight.controller.sal.packet.Packet;
  41 import org.opendaylight.controller.sal.packet.PacketResult;
  42 import org.opendaylight.controller.sal.packet.RawPacket;
  43 import org.opendaylight.controller.sal.utils.GlobalConstants;
  44 import org.opendaylight.controller.sal.utils.NetUtils;
  45 import org.slf4j.Logger;
  46 import org.slf4j.LoggerFactory;
  47
  48 public class DataPacketService implements IPluginOutDataPacketService,
  49         IDataPacketService {
  50     private int TXMAXQUEUESIZE = 1000;
  51     protected static final Logger logger = LoggerFactory
  52             .getLogger(DataPacketService.class);
  53     /**
  54      * Database that associates one NodeIDType to the
  55      * IPluginDataPacketService, in fact we expect that there will be
  56      * one instance of IPluginDataPacketService for each southbound
  57      * plugin.
  58      * Using the ConcurrentHashMap because the threads that will be
  59      * adding a new service, removing a service, going through all of
  60      * them maybe different.
  61      */
  62     private ConcurrentHashMap<String, IPluginInDataPacketService>
  63         pluginInDataService =
  64         new ConcurrentHashMap<String, IPluginInDataPacketService>();
  65     private Map<String, AtomicInteger> statistics = new HashMap<String, AtomicInteger>();
  66
  67     /**
  68      * Queue for packets that need to be transmitted to Data Path
  69      */
  70     private LinkedBlockingQueue<RawPacket> txQueue = new LinkedBlockingQueue<RawPacket>(
  71             TXMAXQUEUESIZE);
  72     /**
  73      * Transmission thread
  74      */
  75     private Thread txThread = new Thread(new TxLoop(),
  76             "DataPacketService TX thread");
  77
  78     /**
  79      * Representation of a Data Packet Listener including of its
  80      * properties
  81      */
  82     private class DataPacketListener {
  83         // Key fields
  84         private String listenerName;
  85         // Attribute fields
  86         private IListenDataPacket listener;
  87         private String dependency;
  88         private Match match;
  89
  90         DataPacketListener(String name, IListenDataPacket s, String dependency,
  91                 Match match) {
  92             this.listenerName = name;
  93             this.listener = s;
  94             this.dependency = dependency;
  95             this.match = match;
  96         }
  97
  98         @Override
  99         public boolean equals(Object obj) {
 100             if (this == obj)
 101                 return true;
 102             if (obj == null)
 103                 return false;
 104             if (getClass() != obj.getClass())
 105                 return false;
 106             DataPacketListener other = (DataPacketListener) obj;
 107             if (!getOuterType().equals(other.getOuterType()))
 108                 return false;
 109             if (listenerName == null) {
 110                 if (other.listenerName != null)
 111                     return false;
 112             } else if (!listenerName.equals(other.listenerName))
 113                 return false;
 114             return true;
 115         }
 116
 117         @Override
 118         public int hashCode() {
 119             final int prime = 31;
 120             int result = 1;
 121             result = prime * result + getOuterType().hashCode();
 122             result = prime * result
 123                     + ((listenerName == null) ? 0 : listenerName.hashCode());
 124             return result;
 125         }
 126
 127         private DataPacketService getOuterType() {
 128             return DataPacketService.this;
 129         }
 130     }
 131
 132     /**
 133      * This very expensive version of List is being used because it
 134      * work well in concurrent situation, as we expect new service
 135      * addition, service removal and walk of the service will happen
 136      * from different places
 137      */
 138     private List<List<DataPacketListener>> listenDataPacket = new CopyOnWriteArrayList<List<DataPacketListener>>();
 139     // Quick index to make sure there are no duplicate elements
 140     private Set<DataPacketListener> indexDataPacket = Collections
 141             .synchronizedSet(new HashSet<DataPacketListener>());
 142
 143     /**
 144      * Loop for processing Received packets
 145      */
 146     private void dispatchPacket(RawPacket pkt) {
 147
 148         // for now we treat all listeners as serial listeners
 149         for (List<DataPacketListener> serialListeners : listenDataPacket) {
 150             for (DataPacketListener l : serialListeners) {
 151
 152                 // TODO: possibly deal with read-only and read-write packet
 153                 // copies
 154                 IListenDataPacket s = (l == null ? null : l.listener);
 155                 if (s != null) {
 156                     try {
 157                         // TODO Make sure to filter based on the match too,
 158                         // later on
 159                         PacketResult res = s.receiveDataPacket(pkt);
 160                         increaseStat("RXPacketSuccess");
 161                         if (res.equals(PacketResult.CONSUME)) {
 162                             increaseStat("RXPacketSerialExit");
 163                             break;
 164                         }
 165                     } catch (Exception e) {
 166                         increaseStat("RXPacketFailedForException");
 167                     }
 168                 }
 169             }
 170         }
 171     }
 172
 173     /**
 174      * Loop for processing packets to be transmitted
 175      *
 176      */
 177     private class TxLoop implements Runnable {
 178         public void run() {
 179             RawPacket pkt;
 180             try {
 181                 for (pkt = txQueue.take(); pkt != null; pkt = txQueue.take()) {
 182                     // Retrieve outgoing node connector so to send out
 183                     // the packet to corresponding node
 184                     NodeConnector p = pkt.getOutgoingNodeConnector();
 185                     if (p != null) {
 186                         String t = p.getNode()
 187                                 .getType();
 188                         // Now locate the TX dispatcher
 189                         IPluginInDataPacketService s = pluginInDataService
 190                                 .get(t);
 191                         if (s != null) {
 192                             try {
 193                                 s.transmitDataPacket(pkt);
 194                                 increaseStat("TXPacketSuccess");
 195                             } catch (Exception e) {
 196                                 increaseStat("TXPacketFailedForException");
 197                             }
 198                         } else {
 199                             increaseStat("TXpluginNotFound");
 200                         }
 201                     }
 202                 }
 203             } catch (InterruptedException e) {
 204                 // Not a big deal
 205             }
 206         }
 207     }
 208
 209     void setPluginInDataService(Map props, IPluginInDataPacketService s) {
 210         if (this.pluginInDataService == null) {
 211             logger.error("pluginInDataService store null");
 212             return;
 213         }
 214         String type = null;
 215         logger.trace("Received setPluginInDataService request");
 216         for (Object e : props.entrySet()) {
 217             Map.Entry entry = (Map.Entry) e;
 218             logger.trace("Prop key:({}) value:({})",entry.getKey(), entry.getValue());
 219         }
 220
 221         Object value = props.get(GlobalConstants.PROTOCOLPLUGINTYPE.toString());
 222         if (value instanceof String) {
 223             type = (String) value;
 224         }
 225         if (type == null) {
 226             logger.error("Received a PluginInDataService without any "
 227                     + "protocolPluginType provided");
 228         } else {
 229             this.pluginInDataService.put(type, s);
 230             logger.debug("Stored the PluginInDataService for type: {}", type);
 231         }
 232     }
 233
 234     void unsetPluginInDataService(Map props, IPluginInDataPacketService s) {
 235         if (this.pluginInDataService == null) {
 236             logger.error("pluginInDataService store null");
 237             return;
 238         }
 239
 240         String type = null;
 241         logger.trace("Received unsetPluginInDataService request");
 242         for (Object e : props.entrySet()) {
 243             Map.Entry entry = (Map.Entry) e;
 244             logger.trace("Prop key:({}) value:({})",entry.getKey(), entry.getValue());
 245         }
 246
 247         Object value = props.get(GlobalConstants.PROTOCOLPLUGINTYPE.toString());
 248         if (value instanceof String) {
 249             type = (String) value;
 250         }
 251         if (type == null) {
 252             logger.error("Received a PluginInDataService without any "
 253                     + "protocolPluginType provided");
 254         } else if (this.pluginInDataService.get(type).equals(s)) {
 255             this.pluginInDataService.remove(type);
 256             logger.debug("Removed the PluginInDataService for type: {}", type);
 257         }
 258     }
 259
 260     void setListenDataPacket(Map props, IListenDataPacket s) {
 261         if (this.listenDataPacket == null || this.indexDataPacket == null) {
 262             logger.error("data structure to store data is NULL");
 263             return;
 264         }
 265         logger.trace("Received setListenDataPacket request");
 266         for (Object e : props.entrySet()) {
 267             Map.Entry entry = (Map.Entry) e;
 268             logger.trace("Prop key:({}) value:({})",entry.getKey(), entry.getValue());
 269         }
 270
 271         String listenerName = null;
 272         String listenerDependency = null;
 273         Match filter = null;
 274         Object value;
 275         // Read the listenerName
 276         value = props.get("salListenerName");
 277         if (value instanceof String) {
 278             listenerName = (String) value;
 279         }
 280
 281         if (listenerName == null) {
 282             logger.error("Trying to set a listener without a Name");
 283             return;
 284         }
 285
 286         //Read the dependency
 287         value = props.get("salListenerDependency");
 288         if (value instanceof String) {
 289             listenerDependency = (String) value;
 290         }
 291
 292         //Read match filter if any
 293         value = props.get("salListenerFilter");
 294         if (value instanceof Match) {
 295             filter = (Match) value;
 296         }
 297
 298         DataPacketListener l = new DataPacketListener(listenerName, s,
 299                 listenerDependency, filter);
 300
 301         DataPacketListener lDependency = new DataPacketListener(
 302                 listenerDependency, null, null, null);
 303
 304         // Now let see if there is any dependency
 305         if (listenerDependency == null) {
 306             logger.debug("listener without any dependency");
 307             if (this.indexDataPacket.contains(l)) {
 308                 logger.error("trying to add an existing element");
 309             } else {
 310                 logger.debug("adding listener: {}", listenerName);
 311                 CopyOnWriteArrayList<DataPacketListener> serialListeners = new CopyOnWriteArrayList<DataPacketListener>();
 312                 serialListeners.add(l);
 313                 this.listenDataPacket.add(serialListeners);
 314                 this.indexDataPacket.add(l);
 315             }
 316         } else {
 317             logger.debug("listener with dependency");
 318             // Now search for the dependency and put things in order
 319             if (this.indexDataPacket.contains(l)) {
 320                 logger.error("trying to add an existing element");
 321             } else {
 322                 logger.debug("adding listener: {}", listenerName);
 323                 // Lets find the set with the dependency in it, if we
 324                 // find it lets just add our dependency at the end of
 325                 // the list.
 326                 for (List<DataPacketListener> serialListeners : this.listenDataPacket) {
 327                     boolean done = false;
 328                     if (serialListeners.contains(lDependency)) {
 329                         serialListeners.add(l);
 330                         done = true;
 331                     }
 332                     // If we did fine the element, lets break early
 333                     if (done) {
 334                         break;
 335                     }
 336                 }
 337
 338                 this.indexDataPacket.add(l);
 339             }
 340         }
 341     }
 342
 343     void unsetListenDataPacket(Map props, IListenDataPacket s) {
 344         if (this.listenDataPacket == null || this.indexDataPacket == null) {
 345             logger.error("data structure to store data is NULL");
 346             return;
 347         }
 348         logger.trace("Received UNsetListenDataPacket request");
 349         for (Object e : props.entrySet()) {
 350             Map.Entry entry = (Map.Entry) e;
 351             logger.trace("Prop key:({}) value:({})",entry.getKey(), entry.getValue());
 352         }
 353
 354         String listenerName = null;
 355         Object value;
 356         // Read the listenerName
 357         value = props.get("salListenerName");
 358         if (value instanceof String) {
 359             listenerName = (String) value;
 360         }
 361
 362         if (listenerName == null) {
 363             logger.error("Trying to set a listener without a Name");
 364             return;
 365         }
 366
 367         DataPacketListener l = new DataPacketListener(listenerName, s, null,
 368                 null);
 369         if (!this.indexDataPacket.contains(l)) {
 370             logger.error("trying to remove a non-existing element");
 371         } else {
 372             logger.debug("removing listener: {}", listenerName);
 373             for (List<DataPacketListener> serialListeners : this.listenDataPacket) {
 374                 int i = 0;
 375                 boolean done = false;
 376                 for (i = 0; i < serialListeners.size(); i++) {
 377                     if (serialListeners.get(i).equals(l)) {
 378                         serialListeners.remove(i);
 379                         done = true;
 380                         break;
 381                     }
 382                 }
 383                 // Now remove a serialListener that maybe empty
 384                 if (serialListeners.isEmpty()) {
 385                     this.listenDataPacket.remove(serialListeners);
 386                 }
 387                 // If we did fine the element, lets break early
 388                 if (done) {
 389                     break;
 390                 }
 391             }
 392
 393             this.indexDataPacket.remove(l);
 394         }
 395     }
 396
 397     /**
 398      * Function called by the dependency manager when all the required
 399      * dependencies are satisfied
 400      *
 401      */
 402     void init() {
 403         this.txThread.start();
 404     }
 405
 406     /**
 407      * Function called by the dependency manager when at least one
 408      * dependency become unsatisfied or when the component is shutting
 409      * down because for example bundle is being stopped.
 410      *
 411      */
 412     void destroy() {
 413         // Make sure to cleanup the data structure we use to track
 414         // services
 415         this.listenDataPacket.clear();
 416         this.indexDataPacket.clear();
 417         this.pluginInDataService.clear();
 418         this.statistics.clear();
 419         this.txQueue.clear();
 420         this.txThread.interrupt();
 421         // Wait for them to be done
 422         try {
 423             this.txThread.join();
 424         } catch (InterruptedException ex) {
 425             // Not a big deal
 426         }
 427     }
 428
 429     private void increaseStat(String name) {
 430         if (this.statistics == null) {
 431             return;
 432         }
 433
 434         AtomicInteger currValue = null;
 435         synchronized (this.statistics) {
 436             currValue = this.statistics.get(name);
 437
 438             if (currValue == null) {
 439                 this.statistics.put(name, new AtomicInteger(0));
 440                 return;
 441             }
 442         }
 443         currValue.incrementAndGet();
 444     }
 445
 446     @Override
 447     public PacketResult receiveDataPacket(RawPacket inPkt) {
 448         if (inPkt.getIncomingNodeConnector() == null) {
 449             increaseStat("nullIncomingNodeConnector");
 450             return PacketResult.IGNORED;
 451         }
 452
 453         // send the packet off to be processed by listeners
 454         this.dispatchPacket(inPkt);
 455
 456         // Walk the chain of listener going first throw all the
 457         // parallel ones and for each parallel in serial
 458         return PacketResult.IGNORED;
 459     }
 460
 461     @Override
 462     public void transmitDataPacket(RawPacket outPkt) {
 463         if (outPkt.getOutgoingNodeConnector() == null) {
 464             increaseStat("nullOutgoingNodeConnector");
 465             return;
 466         }
 467
 468         if (!this.txQueue.offer(outPkt)) {
 469             increaseStat("fullTXQueue");
 470             return;
 471         }
 472     }
 473
 474     @Override
 475     public Packet decodeDataPacket(RawPacket pkt) {
 476         // Sanity checks
 477         if (pkt == null) {
 478             return null;
 479         }
 480         byte[] data = pkt.getPacketData();
 481         if (data.length <= 0) {
 482             return null;
 483         }
 484         if (pkt.getEncap().equals(LinkEncap.ETHERNET)) {
 485             Ethernet res = new Ethernet();
 486             try {
 487                 res.deserialize(data, 0, data.length * NetUtils.NumBitsInAByte);
 488             } catch (Exception e) {
 489                 logger.warn("Failed to decode packet: {}", e.getMessage());
 490             }
 491             return res;
 492         }
 493         return null;
 494     }
 495
 496     @Override
 497     public RawPacket encodeDataPacket(Packet pkt) {
 498         // Sanity checks
 499         if (pkt == null) {
 500             return null;
 501         }
 502         byte[] data;
 503         try {
 504             data = pkt.serialize();
 505         } catch (Exception e) {
 506             logger.error("",e);
 507             return null;
 508         }
 509         if (data.length <= 0) {
 510             return null;
 511         }
 512         try {
 513             RawPacket res = new RawPacket(data);
 514             return res;
 515         } catch (ConstructionException cex) {
 516         }
 517         // If something goes wrong then we have to return null
 518         return null;
 519     }
 520 }