gt:                         && false == pendingStatsRequests.offer(request)) {
context: /*
   * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   *
   * This program and the accompanying materials are made available under the
   * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   * and is available at http://www.eclipse.org/legal/epl-v10.html
   */
  
  package org.opendaylight.controller.protocol_plugin.openflow.internal;
  
  import java.nio.ByteBuffer;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.Deque;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.util.Set;
  import java.util.Timer;
  import java.util.TimerTask;
  import java.util.concurrent.BlockingQueue;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ConcurrentMap;
  import java.util.concurrent.CopyOnWriteArraySet;
  import java.util.concurrent.LinkedBlockingDeque;
  import java.util.concurrent.LinkedBlockingQueue;
  
  import org.eclipse.osgi.framework.console.CommandInterpreter;
  import org.eclipse.osgi.framework.console.CommandProvider;
  import org.opendaylight.controller.protocol_plugin.openflow.IInventoryShimExternalListener;
  import org.opendaylight.controller.protocol_plugin.openflow.IOFStatisticsListener;
  import org.opendaylight.controller.protocol_plugin.openflow.IOFStatisticsManager;
  import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
  import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
  import org.opendaylight.controller.protocol_plugin.openflow.vendorextension.v6extension.V6Match;
  import org.opendaylight.controller.protocol_plugin.openflow.vendorextension.v6extension.V6StatsReply;
  import org.opendaylight.controller.protocol_plugin.openflow.vendorextension.v6extension.V6StatsRequest;
  import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
  import org.opendaylight.controller.sal.core.Node;
  import org.opendaylight.controller.sal.core.NodeConnector;
  import org.opendaylight.controller.sal.core.Property;
  import org.opendaylight.controller.sal.core.UpdateType;
  import org.opendaylight.controller.sal.utils.HexEncode;
  import org.openflow.protocol.OFError;
  import org.openflow.protocol.OFMatch;
  import org.openflow.protocol.OFPort;
  import org.openflow.protocol.OFStatisticsRequest;
  import org.openflow.protocol.statistics.OFAggregateStatisticsRequest;
  import org.openflow.protocol.statistics.OFFlowStatisticsReply;
  import org.openflow.protocol.statistics.OFFlowStatisticsRequest;
  import org.openflow.protocol.statistics.OFPortStatisticsReply;
  import org.openflow.protocol.statistics.OFPortStatisticsRequest;
  import org.openflow.protocol.statistics.OFQueueStatisticsRequest;
  import org.openflow.protocol.statistics.OFStatistics;
  import org.openflow.protocol.statistics.OFStatisticsType;
  import org.openflow.protocol.statistics.OFTableStatistics;
  import org.openflow.protocol.statistics.OFVendorStatistics;
  import org.openflow.util.HexString;
  import org.osgi.framework.BundleContext;
  import org.osgi.framework.FrameworkUtil;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Periodically polls the different OF statistics from the OF switches, caches
   * them, and publishes results towards SAL. It also provides an API to directly
   * query the switch for any specific statistics.
   */
  public class OFStatisticsManager implements IOFStatisticsManager, IInventoryShimExternalListener, CommandProvider {
      private static final Logger log = LoggerFactory.getLogger(OFStatisticsManager.class);
      private static final int INITIAL_SIZE = 64;
      private static final long FLOW_STATS_PERIOD = 10000;
      private static final long DESC_STATS_PERIOD = 60000;
      private static final long PORT_STATS_PERIOD = 5000;
      private static final long TABLE_STATS_PERIOD = 10000;
      private static final long TICK = 1000;
      private static short statisticsTickNumber = (short) (FLOW_STATS_PERIOD / TICK);
      private static short descriptionTickNumber = (short) (DESC_STATS_PERIOD / TICK);
      private static short portTickNumber = (short) (PORT_STATS_PERIOD / TICK);
      private static short tableTickNumber = (short) (TABLE_STATS_PERIOD / TICK);
      private static short factoredSamples = (short) 2;
      private static short counter = 1;
      private IController controller = null;
      private ConcurrentMap<Long, List<OFStatistics>> flowStatistics;
      private ConcurrentMap<Long, List<OFStatistics>> descStatistics;
      private ConcurrentMap<Long, List<OFStatistics>> portStatistics;
      private ConcurrentMap<Long, List<OFStatistics>> tableStatistics;
      private ConcurrentMap<Long, StatisticsTicks> statisticsTimerTicks;
      protected BlockingQueue<StatsRequest> pendingStatsRequests;
      protected BlockingQueue<Long> switchPortStatsUpdated;
      private Thread statisticsCollector;
      private Thread txRatesUpdater;
      private Timer statisticsTimer;
      private TimerTask statisticsTimerTask;
      private ConcurrentMap<Long, Boolean> switchSupportsVendorExtStats;
      // Per port sampled (every portStatsPeriod) transmit rate
      private Map<Long, Map<Short, TxRates>> txRates;
      private Set<IOFStatisticsListener> statisticsListeners = new CopyOnWriteArraySet<IOFStatisticsListener>();
  
      /**
       * The object containing the latest factoredSamples tx rate samples for a
       * given switch port
       */
      protected class TxRates {
          // contains the latest factoredSamples sampled transmitted bytes
          Deque<Long> sampledTxBytes;
  
          public TxRates() {
              sampledTxBytes = new LinkedBlockingDeque<Long>();
          }
  
          public void update(Long txBytes) {
              /*
               * Based on how many samples our average works on, we might have to
               * remove the oldest sample
               */
              if (sampledTxBytes.size() == factoredSamples) {
                  sampledTxBytes.removeLast();
              }
  
              // Add the latest sample to the top of the queue
              sampledTxBytes.addFirst(txBytes);
          }
  
          /**
           * Returns the average transmit rate in bps
           *
           * @return the average transmit rate [bps]
           */
          public long getAverageTxRate() {
              long average = 0;
              /*
               * If we cannot provide the value for the time window length set
               */
              if (sampledTxBytes.size() < factoredSamples) {
                  return average;
              }
              long increment = sampledTxBytes.getFirst() - sampledTxBytes
                      .getLast();
              long timePeriod = factoredSamples * PORT_STATS_PERIOD / TICK;
              average = (8L * increment) / timePeriod;
              return average;
          }
      }
  
      public void setController(IController core) {
          this.controller = core;
      }
  
      public void unsetController(IController core) {
          if (this.controller == core) {
              this.controller = null;
          }
      }
  
      private short getStatsQueueSize() {
          String statsQueueSizeStr = System.getProperty("of.statsQueueSize");
          short statsQueueSize = INITIAL_SIZE;
          if (statsQueueSizeStr != null) {
              try {
                  statsQueueSize = Short.parseShort(statsQueueSizeStr);
                  if (statsQueueSize <= 0) {
                      statsQueueSize = INITIAL_SIZE;
                  }
              } catch (Exception e) {
              }
          }
          return statsQueueSize;
      }
  
      IPluginOutConnectionService connectionPluginOutService;
      void setIPluginOutConnectionService(IPluginOutConnectionService s) {
          connectionPluginOutService = s;
      }
  
      void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
          if (connectionPluginOutService == s) {
              connectionPluginOutService = null;
          }
      }
  
      /**
       * Function called by the dependency manager when all the required
       * dependencies are satisfied
       *
       */
      void init() {
          flowStatistics = new ConcurrentHashMap<Long, List<OFStatistics>>();
          descStatistics = new ConcurrentHashMap<Long, List<OFStatistics>>();
          portStatistics = new ConcurrentHashMap<Long, List<OFStatistics>>();
          tableStatistics = new ConcurrentHashMap<Long, List<OFStatistics>>();
          pendingStatsRequests = new LinkedBlockingQueue<StatsRequest>(getStatsQueueSize());
          statisticsTimerTicks = new ConcurrentHashMap<Long, StatisticsTicks>(INITIAL_SIZE);
          switchPortStatsUpdated = new LinkedBlockingQueue<Long>(INITIAL_SIZE);
          switchSupportsVendorExtStats = new ConcurrentHashMap<Long, Boolean>(INITIAL_SIZE);
          txRates = new HashMap<Long, Map<Short, TxRates>>(INITIAL_SIZE);
  
          configStatsPollIntervals();
  
          // Initialize managed timers
          statisticsTimer = new Timer("Statistics Timer Ticks");
          statisticsTimerTask = new TimerTask() {
              @Override
              public void run() {
                  decrementTicks();
              }
          };
  
          // Initialize Statistics collector thread
          statisticsCollector = new Thread(new Runnable() {
              @Override
              public void run() {
                  while (true) {
                      try {
                          StatsRequest req = pendingStatsRequests.take();
                          queryStatisticsInternal(req.switchId, req.type);
                      } catch (InterruptedException e) {
                          log.warn("Flow Statistics Collector thread "
                                  + "interrupted", e);
                          return;
                      }
                  }
              }
          }, "Statistics Collector");
  
          // Initialize Tx Rate Updater thread
          txRatesUpdater = new Thread(new Runnable() {
              @Override
              public void run() {
                  while (true) {
                      try {
                          long switchId = switchPortStatsUpdated.take();
                          updatePortsTxRate(switchId);
                      } catch (InterruptedException e) {
                          log.warn("TX Rate Updater thread interrupted", e);
                          return;
                      }
                  }
              }
          }, "TX Rate Updater");
      }
  
      /**
       * Function called by the dependency manager when at least one dependency
       * become unsatisfied or when the component is shutting down because for
       * example bundle is being stopped.
       *
       */
      void destroy() {
          statisticsListeners.clear();
      }
  
      /**
       * Function called by dependency manager after "init ()" is called and after
       * the services provided by the class are registered in the service registry
       *
       */
      void start() {
          // Start managed timers
          statisticsTimer.scheduleAtFixedRate(statisticsTimerTask, 0, TICK);
  
          // Start statistics collector thread
          statisticsCollector.start();
  
          // Start bandwidth utilization computer thread
          txRatesUpdater.start();
  
          // OSGI console
          registerWithOSGIConsole();
      }
  
      /**
       * Function called by the dependency manager before the services exported by
       * the component are unregistered, this will be followed by a "destroy ()"
       * calls
       *
       */
      void stop() {
          // Stop managed timers
          statisticsTimer.cancel();
      }
  
      public void setStatisticsListener(IOFStatisticsListener s) {
          this.statisticsListeners.add(s);
      }
  
      public void unsetStatisticsListener(IOFStatisticsListener s) {
          if (s != null) {
              this.statisticsListeners.remove(s);
          }
      }
  
      private void registerWithOSGIConsole() {
          BundleContext bundleContext = FrameworkUtil.getBundle(this.getClass()).getBundleContext();
          bundleContext.registerService(CommandProvider.class.getName(), this, null);
      }
  
      private static class StatsRequest {
          protected Long switchId;
          protected OFStatisticsType type;
  
          public StatsRequest(Long d, OFStatisticsType t) {
              switchId = d;
              type = t;
          }
  
          @Override
          public String toString() {
              return "SReq = {switchId=" + switchId + ", type=" + type + "}";
          }
  
          @Override
          public int hashCode() {
              final int prime = 31;
              int result = 1;
              result = prime * result
                      + ((switchId == null) ? 0 : switchId.hashCode());
              result = prime * result + ((type == null) ? 0 : type.ordinal());
              return result;
          }
  
          @Override
          public boolean equals(Object obj) {
              if (this == obj) {
                  return true;
              }
              if (obj == null) {
                  return false;
              }
              if (getClass() != obj.getClass()) {
                  return false;
              }
              StatsRequest other = (StatsRequest) obj;
              if (switchId == null) {
                  if (other.switchId != null) {
                      return false;
                  }
              } else if (!switchId.equals(other.switchId)) {
                  return false;
              }
              if (type != other.type) {
                  return false;
              }
              return true;
          }
      }
  
      private void addStatisticsTicks(Long switchId) {
          switchSupportsVendorExtStats.put(switchId, Boolean.TRUE); // Assume
                                                                    // switch
                                                                    // supports
                                                                    // Vendor
                                                                    // extension
                                                                    // stats
          statisticsTimerTicks.put(switchId, new StatisticsTicks(true));
          log.debug("Added Switch {} to target pool",
                  HexString.toHexString(switchId.longValue()));
      }
  
      protected static class StatisticsTicks {
          private short flowStatisticsTicks;
          private short descriptionTicks;
          private short portStatisticsTicks;
          private short tableStatisticsTicks;
  
          public StatisticsTicks(boolean scattered) {
              if (scattered) {
                  // scatter bursts by statisticsTickPeriod
                  if (++counter < 0) {
                      counter = 0;
                  } // being paranoid here
                  flowStatisticsTicks = (short) (1 + counter
                          % statisticsTickNumber);
                  descriptionTicks = (short) (1 + counter % descriptionTickNumber);
                  portStatisticsTicks = (short) (1 + counter % portTickNumber);
                  tableStatisticsTicks = (short) (1 + counter % tableTickNumber);
              } else {
                  flowStatisticsTicks = statisticsTickNumber;
                  descriptionTicks = descriptionTickNumber;
                  portStatisticsTicks = portTickNumber;
                  tableStatisticsTicks = tableTickNumber;
              }
          }
  
          public boolean decrementFlowTicksIsZero() {
              // Please ensure no code is inserted between the if check and the
              // flowStatisticsTicks reset
              if (--flowStatisticsTicks == 0) {
                  flowStatisticsTicks = statisticsTickNumber;
                  return true;
              }
              return false;
          }
  
          public boolean decrementDescTicksIsZero() {
              // Please ensure no code is inserted between the if check and the
              // descriptionTicks reset
              if (--descriptionTicks == 0) {
                  descriptionTicks = descriptionTickNumber;
                  return true;
              }
              return false;
          }
  
          public boolean decrementPortTicksIsZero() {
              // Please ensure no code is inserted between the if check and the
              // descriptionTicks reset
              if (--portStatisticsTicks == 0) {
                  portStatisticsTicks = portTickNumber;
                  return true;
              }
              return false;
          }
  
          public boolean decrementTableTicksIsZero() {
              // Please ensure no code is inserted between the if check and the
              // descriptionTicks reset
              if(--tableStatisticsTicks == 0) {
                  tableStatisticsTicks = tableTickNumber;
                  return true;
              }
              return false;
          }
  
          @Override
          public String toString() {
              return "{fT=" + flowStatisticsTicks + ",dT=" + descriptionTicks
                      + ",pT=" + portStatisticsTicks + ",tT=" + tableStatisticsTicks + "}";
          }
      }
  
      private void printInfoMessage(String type, StatsRequest request) {
          log.trace("{} stats request not inserted for switch: {}. Queue size: {}. Collector state: {}.",
                  new Object[] {type, HexString.toHexString(request.switchId), pendingStatsRequests.size(),
                  statisticsCollector.getState().toString() });
      }
  
      protected void decrementTicks() {
          StatsRequest request = null;
          for (Map.Entry<Long, StatisticsTicks> entry : statisticsTimerTicks
                  .entrySet()) {
              StatisticsTicks clock = entry.getValue();
              Long switchId = entry.getKey();
              if (clock.decrementFlowTicksIsZero()) {
                  request = (switchSupportsVendorExtStats.get(switchId) == Boolean.TRUE) ?
                          new StatsRequest(switchId, OFStatisticsType.VENDOR) :
                          new StatsRequest(switchId, OFStatisticsType.FLOW);
                  // If a request for this switch is already in the queue, skip to
                  // add this new request
                  if (!pendingStatsRequests.contains(request)
                          && false == pendingStatsRequests.offer(request)) {
                      printInfoMessage("Flow", request);
                  }
              }
  
              if (clock.decrementDescTicksIsZero()) {
                  request = new StatsRequest(switchId, OFStatisticsType.DESC);
                  // If a request for this switch is already in the queue, skip to
                  // add this new request
                  if (!pendingStatsRequests.contains(request)
                          && false == pendingStatsRequests.offer(request)) {
                      printInfoMessage("Description", request);
                  }
              }
  
              if (clock.decrementPortTicksIsZero()) {
                  request = new StatsRequest(switchId, OFStatisticsType.PORT);
                  // If a request for this switch is already in the queue, skip to
                  // add this new request
                  if (!pendingStatsRequests.contains(request)
  