opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServiceShim.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.ArrayList;
  12 import java.util.HashMap;
  13 import java.util.List;
  14 import java.util.Map;
  15 import java.util.Set;
  16 import java.util.Timer;
  17 import java.util.TimerTask;
  18 import java.util.concurrent.BlockingQueue;
  19 import java.util.concurrent.ConcurrentHashMap;
  20 import java.util.concurrent.ConcurrentMap;
  21 import java.util.concurrent.CopyOnWriteArrayList;
  22 import java.util.concurrent.LinkedBlockingQueue;
  23
  24 import org.apache.commons.lang3.tuple.ImmutablePair;
  25 import org.apache.commons.lang3.tuple.Pair;
  26 import org.eclipse.osgi.framework.console.CommandInterpreter;
  27 import org.eclipse.osgi.framework.console.CommandProvider;
  28 import org.opendaylight.controller.protocol_plugin.openflow.IDiscoveryListener;
  29 import org.opendaylight.controller.protocol_plugin.openflow.IInventoryShimExternalListener;
  30 import org.opendaylight.controller.protocol_plugin.openflow.IOFStatisticsManager;
  31 import org.opendaylight.controller.protocol_plugin.openflow.IRefreshInternalProvider;
  32 import org.opendaylight.controller.protocol_plugin.openflow.ITopologyServiceShimListener;
  33 import org.osgi.framework.BundleContext;
  34 import org.osgi.framework.FrameworkUtil;
  35 import org.slf4j.Logger;
  36 import org.slf4j.LoggerFactory;
  37
  38 import org.opendaylight.controller.sal.core.Bandwidth;
  39 import org.opendaylight.controller.sal.core.Config;
  40 import org.opendaylight.controller.sal.core.ContainerFlow;
  41 import org.opendaylight.controller.sal.core.Edge;
  42 import org.opendaylight.controller.sal.core.IContainerListener;
  43 import org.opendaylight.controller.sal.core.Node;
  44 import org.opendaylight.controller.sal.core.NodeConnector;
  45 import org.opendaylight.controller.sal.core.Property;
  46 import org.opendaylight.controller.sal.core.State;
  47 import org.opendaylight.controller.sal.core.UpdateType;
  48 import org.opendaylight.controller.sal.topology.TopoEdgeUpdate;
  49 import org.opendaylight.controller.sal.utils.GlobalConstants;
  50
  51 /**
  52  * The class describes a shim layer that relays the topology events from
  53  * OpenFlow core to various listeners. The notifications are filtered based on
  54  * container configurations.
  55  */
  56 public class TopologyServiceShim implements IDiscoveryListener,
  57         IContainerListener, CommandProvider, IRefreshInternalProvider,
  58         IInventoryShimExternalListener {
  59     protected static final Logger logger = LoggerFactory
  60             .getLogger(TopologyServiceShim.class);
  61     private ConcurrentMap<String, ITopologyServiceShimListener> topologyServiceShimListeners = new ConcurrentHashMap<String, ITopologyServiceShimListener>();
  62     private ConcurrentMap<NodeConnector, List<String>> containerMap = new ConcurrentHashMap<NodeConnector, List<String>>();
  63     private ConcurrentMap<String, ConcurrentMap<NodeConnector, Pair<Edge, Set<Property>>>> edgeMap = new ConcurrentHashMap<String, ConcurrentMap<NodeConnector, Pair<Edge, Set<Property>>>>();
  64
  65     private BlockingQueue<NotifyEntry> notifyQ;
  66     private Thread notifyThread;
  67     private BlockingQueue<String> bulkNotifyQ;
  68     private Thread ofPluginTopoBulkUpdate;
  69     private volatile Boolean shuttingDown = false;
  70     private IOFStatisticsManager statsMgr;
  71     private Timer pollTimer;
  72     private TimerTask txRatePoller;
  73     private Thread bwUtilNotifyThread;
  74     private BlockingQueue<UtilizationUpdate> bwUtilNotifyQ;
  75     private List<NodeConnector> connectorsOverUtilized;
  76     private float bwThresholdFactor = (float) 0.8; // Threshold = 80% of link
  77                                                    // bandwidth
  78
  79     class NotifyEntry {
  80         String container;
  81         List<TopoEdgeUpdate> teuList;
  82
  83         public NotifyEntry(String container, TopoEdgeUpdate teu) {
  84             this.container = container;
  85             this.teuList = new ArrayList<TopoEdgeUpdate>();
  86             if (teu != null) {
  87                 this.teuList.add(teu);
  88             }
  89         }
  90
  91         public NotifyEntry(String container, List<TopoEdgeUpdate> teuList) {
  92             this.container = container;
  93             this.teuList = new ArrayList<TopoEdgeUpdate>();
  94             if (teuList != null) {
  95                 this.teuList.addAll(teuList);
  96             }
  97         }
  98     }
  99
 100     class TopologyNotify implements Runnable {
 101         private final BlockingQueue<NotifyEntry> notifyQ;
 102         private NotifyEntry entry;
 103         private Map<String, List<TopoEdgeUpdate>> teuMap = new HashMap<String, List<TopoEdgeUpdate>>();
 104         private List<TopoEdgeUpdate> teuList;
 105         private boolean notifyListeners;
 106
 107         TopologyNotify(BlockingQueue<NotifyEntry> notifyQ) {
 108             this.notifyQ = notifyQ;
 109         }
 110
 111         @Override
 112         public void run() {
 113             while (true) {
 114                 try {
 115                     teuMap.clear();
 116                     notifyListeners = false;
 117                     while (!notifyQ.isEmpty()) {
 118                         entry = notifyQ.take();
 119                         teuList = teuMap.get(entry.container);
 120                         if (teuList == null) {
 121                             teuList = new ArrayList<TopoEdgeUpdate>();
 122                         }
 123                         // group all the updates together
 124                         teuList.addAll(entry.teuList);
 125                         teuMap.put(entry.container, teuList);
 126                         notifyListeners = true;
 127                     }
 128
 129                     if (notifyListeners) {
 130                         for (String container : teuMap.keySet()) {
 131                             // notify the listener
 132                             topologyServiceShimListeners.get(container)
 133                                     .edgeUpdate(teuMap.get(container));
 134                         }
 135                     }
 136
 137                     Thread.sleep(100);
 138                 } catch (InterruptedException e1) {
 139                     logger.warn("TopologyNotify interrupted {}",
 140                             e1.getMessage());
 141                     if (shuttingDown) {
 142                         return;
 143                     }
 144                 } catch (Exception e2) {
 145                     logger.error("", e2);
 146                 }
 147             }
 148         }
 149     }
 150
 151     class UtilizationUpdate {
 152         NodeConnector connector;
 153         UpdateType type;
 154
 155         UtilizationUpdate(NodeConnector connector, UpdateType type) {
 156             this.connector = connector;
 157             this.type = type;
 158         }
 159     }
 160
 161     class BwUtilizationNotify implements Runnable {
 162         private final BlockingQueue<UtilizationUpdate> notifyQ;
 163
 164         BwUtilizationNotify(BlockingQueue<UtilizationUpdate> notifyQ) {
 165             this.notifyQ = notifyQ;
 166         }
 167
 168         @Override
 169         public void run() {
 170             while (true) {
 171                 try {
 172                     UtilizationUpdate update = notifyQ.take();
 173                     NodeConnector connector = update.connector;
 174                     Set<String> containerList = edgeMap.keySet();
 175                     for (String container : containerList) {
 176                         Map<NodeConnector, Pair<Edge, Set<Property>>> edgePropsMap = edgeMap
 177                                 .get(container);
 178                         Edge edge = edgePropsMap.get(connector).getLeft();
 179                         if (edge.getTailNodeConnector().equals(connector)) {
 180                             ITopologyServiceShimListener topologServiceShimListener = topologyServiceShimListeners
 181                                     .get(container);
 182                             if (update.type == UpdateType.ADDED) {
 183                                 topologServiceShimListener
 184                                         .edgeOverUtilized(edge);
 185                             } else {
 186                                 topologServiceShimListener
 187                                         .edgeUtilBackToNormal(edge);
 188                             }
 189                         }
 190                     }
 191                 } catch (InterruptedException e1) {
 192                     logger.warn(
 193                             "Edge Bandwidth Utilization Notify Thread interrupted {}",
 194                             e1.getMessage());
 195                     if (shuttingDown) {
 196                         return;
 197                     }
 198                 } catch (Exception e2) {
 199                     logger.error("", e2);
 200                 }
 201             }
 202         }
 203     }
 204
 205     /**
 206      * Function called by the dependency manager when all the required
 207      * dependencies are satisfied
 208      *
 209      */
 210     void init() {
 211         logger.trace("Init called");
 212         connectorsOverUtilized = new ArrayList<NodeConnector>();
 213         notifyQ = new LinkedBlockingQueue<NotifyEntry>();
 214         notifyThread = new Thread(new TopologyNotify(notifyQ));
 215         bwUtilNotifyQ = new LinkedBlockingQueue<UtilizationUpdate>();
 216         bwUtilNotifyThread = new Thread(new BwUtilizationNotify(bwUtilNotifyQ));
 217         bulkNotifyQ = new LinkedBlockingQueue<String>();
 218         ofPluginTopoBulkUpdate = new Thread(new Runnable() {
 219             @Override
 220             public void run() {
 221                 while (true) {
 222                     try {
 223                         String containerName = bulkNotifyQ.take();
 224                         logger.debug("Bulk Notify container:{}", containerName);
 225                         TopologyBulkUpdate(containerName);
 226                     } catch (InterruptedException e) {
 227                         logger.warn("Topology Bulk update thread interrupted");
 228                         if (shuttingDown) {
 229                             return;
 230                         }
 231                     }
 232                 }
 233             }
 234         }, "Topology Bulk Update");
 235
 236         // Initialize node connector tx bit rate poller timer
 237         pollTimer = new Timer();
 238         txRatePoller = new TimerTask() {
 239             @Override
 240             public void run() {
 241                 pollTxBitRates();
 242             }
 243         };
 244
 245         registerWithOSGIConsole();
 246     }
 247
 248     /**
 249      * Continuously polls the transmit bit rate for all the node connectors from
 250      * statistics manager and trigger the warning notification upward when the
 251      * transmit rate is above a threshold which is a percentage of the edge
 252      * bandwidth
 253      */
 254     protected void pollTxBitRates() {
 255         Map<NodeConnector, Pair<Edge, Set<Property>>> globalContainerEdges = edgeMap
 256                 .get(GlobalConstants.DEFAULT.toString());
 257         if (globalContainerEdges == null) {
 258             return;
 259         }
 260
 261         for (NodeConnector connector : globalContainerEdges.keySet()) {
 262             // Skip if node connector belongs to production switch
 263             if (connector.getType().equals(
 264                     NodeConnector.NodeConnectorIDType.PRODUCTION)) {
 265                 continue;
 266             }
 267
 268             // Get edge for which this node connector is head
 269             Pair<Edge, Set<Property>> props = this.edgeMap.get(
 270                     GlobalConstants.DEFAULT.toString()).get(connector);
 271             // On switch mgr restart the props get reset
 272             if (props == null) {
 273                 continue;
 274             }
 275             Set<Property> propSet = props.getRight();
 276             if (propSet == null) {
 277                 continue;
 278             }
 279
 280             float bw = 0;
 281             for (Property prop : propSet) {
 282                 if (prop instanceof Bandwidth) {
 283                     bw = ((Bandwidth) prop).getValue();
 284                     break;
 285                 }
 286             }
 287
 288             // Skip if agent did not provide a bandwidth info for the edge
 289             if (bw == 0) {
 290                 continue;
 291             }
 292
 293             // Compare bandwidth usage
 294             Long switchId = (Long) connector.getNode().getID();
 295             Short port = (Short) connector.getID();
 296             float rate = statsMgr.getTransmitRate(switchId, port);
 297             if (rate > bwThresholdFactor * bw) {
 298                 if (!connectorsOverUtilized.contains(connector)) {
 299                     connectorsOverUtilized.add(connector);
 300                     this.bwUtilNotifyQ.add(new UtilizationUpdate(connector,
 301                             UpdateType.ADDED));
 302                 }
 303             } else {
 304                 if (connectorsOverUtilized.contains(connector)) {
 305                     connectorsOverUtilized.remove(connector);
 306                     this.bwUtilNotifyQ.add(new UtilizationUpdate(connector,
 307                             UpdateType.REMOVED));
 308                 }
 309             }
 310         }
 311
 312     }
 313
 314     /**
 315      * Function called by the dependency manager when at least one dependency
 316      * become unsatisfied or when the component is shutting down because for
 317      * example bundle is being stopped.
 318      *
 319      */
 320     void destroy() {
 321         logger.trace("DESTROY called!");
 322         notifyQ = null;
 323         notifyThread = null;
 324     }
 325
 326     /**
 327      * Function called by dependency manager after "init ()" is called and after
 328      * the services provided by the class are registered in the service registry
 329      *
 330      */
 331     void start() {
 332         logger.trace("START called!");
 333         notifyThread.start();
 334         bwUtilNotifyThread.start();
 335         ofPluginTopoBulkUpdate.start();
 336         pollTimer.scheduleAtFixedRate(txRatePoller, 10000, 5000);
 337     }
 338
 339     /**
 340      * Function called by the dependency manager before the services exported by
 341      * the component are unregistered, this will be followed by a "destroy ()"
 342      * calls
 343      *
 344      */
 345     void stop() {
 346         logger.trace("STOP called!");
 347         shuttingDown = true;
 348         notifyThread.interrupt();
 349     }
 350
 351     void setTopologyServiceShimListener(Map<?, ?> props,
 352             ITopologyServiceShimListener s) {
 353         if (props == null) {
 354             logger.error("Didn't receive the service properties");
 355             return;
 356         }
 357         String containerName = (String) props.get("containerName");
 358         if (containerName == null) {
 359             logger.error("containerName not supplied");
 360             return;
 361         }
 362         if ((this.topologyServiceShimListeners != null)
 363                 && !this.topologyServiceShimListeners
 364                         .containsKey(containerName)) {
 365             this.topologyServiceShimListeners.put(containerName, s);
 366             logger.trace("Added topologyServiceShimListener for container: {}",
 367                     containerName);
 368         }
 369     }
 370
 371     void unsetTopologyServiceShimListener(Map<?, ?> props,
 372             ITopologyServiceShimListener s) {
 373         if (props == null) {
 374             logger.error("Didn't receive the service properties");
 375             return;
 376         }
 377         String containerName = (String) props.get("containerName");
 378         if (containerName == null) {
 379             logger.error("containerName not supplied");
 380             return;
 381         }
 382         if ((this.topologyServiceShimListeners != null)
 383                 && this.topologyServiceShimListeners.containsKey(containerName)
 384                 && this.topologyServiceShimListeners.get(containerName).equals(
 385                         s)) {
 386             this.topologyServiceShimListeners.remove(containerName);
 387             logger.trace(
 388                     "Removed topologyServiceShimListener for container: {}",
 389                     containerName);
 390         }
 391     }
 392
 393     void setStatisticsManager(IOFStatisticsManager s) {
 394         this.statsMgr = s;
 395     }
 396
 397     void unsetStatisticsManager(IOFStatisticsManager s) {
 398         if (this.statsMgr == s) {
 399             this.statsMgr = null;
 400         }
 401     }
 402
 403     private void removeNodeConnector(String container,
 404             NodeConnector nodeConnector) {
 405         List<TopoEdgeUpdate> teuList = new ArrayList<TopoEdgeUpdate>();
 406         Map<NodeConnector, Pair<Edge, Set<Property>>> edgePropsMap = edgeMap
 407                 .get(container);
 408         if (edgePropsMap == null) {
 409             return;
 410         }
 411
 412         // Remove edge in one direction
 413         Pair<Edge, Set<Property>> edgeProps = edgePropsMap.get(nodeConnector);
 414         if (edgeProps == null) {
 415             return;
 416         }
 417         teuList.add(new TopoEdgeUpdate(edgeProps.getLeft(), null,
 418                 UpdateType.REMOVED));
 419
 420         // Remove edge in another direction
 421         edgeProps = edgePropsMap
 422                 .get(edgeProps.getLeft().getHeadNodeConnector());
 423         if (edgeProps == null) {
 424             return;
 425         }
 426         teuList.add(new TopoEdgeUpdate(edgeProps.getLeft(), null,
 427                 UpdateType.REMOVED));
 428
 429         // Update in one shot
 430         notifyEdge(container, teuList);
 431     }
 432
 433     /**
 434      * Update local cache and return true if it needs to notify upper layer
 435      * Topology listeners.
 436      *
 437      * @param container
 438      *            The network container
 439      * @param edge
 440      *            The edge
 441      * @param type
 442      *            The update type
 443      * @param props
 444      *            The edge properties
 445      * @return true if it needs to notify upper layer Topology listeners
 446      */
 447     private boolean updateLocalEdgeMap(String container, Edge edge,
 448             UpdateType type, Set<Property> props) {
 449         ConcurrentMap<NodeConnector, Pair<Edge, Set<Property>>> edgePropsMap = edgeMap
 450                 .get(container);
 451         NodeConnector src = edge.getTailNodeConnector();
 452         Pair<Edge, Set<Property>> edgeProps = new ImmutablePair<Edge, Set<Property>>(
 453                 edge, props);
 454         boolean rv = false;
 455
 456         switch (type) {
 457         case ADDED:
 458         case CHANGED:
 459             if (edgePropsMap == null) {
 460                 edgePropsMap = new ConcurrentHashMap<NodeConnector, Pair<Edge, Set<Property>>>();
 461                 rv = true;
 462             } else {
 463                 if (edgePropsMap.containsKey(src)
 464                         && edgePropsMap.get(src).equals(edgeProps)) {
 465                     // Entry already exists. No update.
 466                     rv = false;
 467                 } else {
 468                     rv = true;
 469                 }
 470             }
 471             if (rv) {
 472                 edgePropsMap.put(src, edgeProps);
 473                 edgeMap.put(container, edgePropsMap);
 474             }
 475             break;
 476         case REMOVED:
 477             if ((edgePropsMap != null) && edgePropsMap.containsKey(src)) {
 478                 edgePropsMap.remove(src);
 479                 if (edgePropsMap.isEmpty()) {
 480                     edgeMap.remove(container);
 481                 } else {
 482                     edgeMap.put(container, edgePropsMap);
 483                 }
 484                 rv = true;
 485             }
 486             break;
 487         default:
 488             logger.debug(
 489                     "notifyLocalEdgeMap: invalid {} for Edge {} in container {}",
 490                     new Object[] { type.getName(), edge, container });
 491         }
 492
 493         if (rv) {
 494             logger.debug(
 495                     "notifyLocalEdgeMap: {} for Edge {} in container {}",
 496                     new Object[] { type.getName(), edge, container });
 497         }
 498
 499         return rv;
 500     }
 501
 502     private void notifyEdge(String container, Edge edge, UpdateType type,
 503             Set<Property> props) {
 504         boolean notifyListeners;
 505
 506         // Update local cache
 507         notifyListeners = updateLocalEdgeMap(container, edge, type, props);
 508
 509         // Prepare to update TopologyService
 510         if (notifyListeners) {
 511             notifyQ.add(new NotifyEntry(container, new TopoEdgeUpdate(edge, props,
 512                     type)));
 513             logger.debug("notifyEdge: {} Edge {} in container {}",
 514                     new Object[] { type.getName(), edge, container });
 515         }
 516     }
 517
 518     private void notifyEdge(String container, List<TopoEdgeUpdate> etuList) {
 519         if (etuList == null) {
 520             return;
 521         }
 522
 523         Edge edge;
 524         UpdateType type;
 525         List<TopoEdgeUpdate> etuNotifyList = new ArrayList<TopoEdgeUpdate>();
 526         boolean notifyListeners = false, rv;
 527
 528         for (TopoEdgeUpdate etu : etuList) {
 529             edge = etu.getEdge();
 530             type = etu.getUpdateType();
 531
 532             // Update local cache
 533             rv = updateLocalEdgeMap(container, edge, type, etu.getProperty());
 534             if (rv) {
 535                 if (!notifyListeners) {
 536                     notifyListeners = true;
 537                 }
 538                 etuNotifyList.add(etu);
 539                 logger.debug(
 540                         "notifyEdge(TopoEdgeUpdate): {} Edge {} in container {}",
 541                         new Object[] { type.getName(), edge, container });
 542             }
 543         }
 544
 545         // Prepare to update TopologyService
 546         if (notifyListeners) {
 547             notifyQ.add(new NotifyEntry(container, etuNotifyList));
 548             logger.debug("notifyEdge(TopoEdgeUpdate): add notifyQ");
 549         }
 550     }
 551
 552     @Override
 553     public void notifyEdge(Edge edge, UpdateType type, Set<Property> props) {
 554         if ((edge == null) || (type == null)) {
 555             return;
 556         }
 557
 558         // Notify default container
 559         notifyEdge(GlobalConstants.DEFAULT.toString(), edge, type, props);
 560
 561         // Notify the corresponding containers
 562         List<String> containers = getEdgeContainers(edge);
 563         if (containers != null) {
 564             for (String container : containers) {
 565                 notifyEdge(container, edge, type, props);
 566             }
 567         }
 568     }
 569
 570     /*
 571      * Return a list of containers the edge associated with
 572      */
 573     private List<String> getEdgeContainers(Edge edge) {
 574         NodeConnector src = edge.getTailNodeConnector(), dst = edge
 575                 .getHeadNodeConnector();
 576
 577         if (!src.getType().equals(NodeConnector.NodeConnectorIDType.PRODUCTION)) {
 578             /* Find the common containers for both ends */
 579             List<String> srcContainers = this.containerMap.get(src), dstContainers = this.containerMap
 580                     .get(dst), cmnContainers = null;
 581             if ((srcContainers != null) && (dstContainers != null)) {
 582                 cmnContainers = new ArrayList<String>(srcContainers);
 583                 cmnContainers.retainAll(dstContainers);
 584             }
 585             return cmnContainers;
 586         } else {
 587             /*
 588              * If the neighbor is part of a monitored production network, get
 589              * the containers that the edge port belongs to
 590              */
 591             return this.containerMap.get(dst);
 592         }
 593     }
 594
 595     @Override
 596     public void tagUpdated(String containerName, Node n, short oldTag,
 597             short newTag, UpdateType t) {
 598     }
 599
 600     @Override
 601     public void containerFlowUpdated(String containerName,
 602             ContainerFlow previousFlow, ContainerFlow currentFlow, UpdateType t) {
 603     }
 604
 605     @Override
 606     public void nodeConnectorUpdated(String containerName, NodeConnector p,
 607             UpdateType t) {
 608         if (this.containerMap == null) {
 609             logger.error("containerMap is NULL");
 610             return;
 611         }
 612         List<String> containers = this.containerMap.get(p);
 613         if (containers == null) {
 614             containers = new CopyOnWriteArrayList<String>();
 615         }
 616         boolean updateMap = false;
 617         switch (t) {
 618         case ADDED:
 619             if (!containers.contains(containerName)) {
 620                 containers.add(containerName);
 621                 updateMap = true;
 622             }
 623             break;
 624         case REMOVED:
 625             if (containers.contains(containerName)) {
 626                 containers.remove(containerName);
 627                 updateMap = true;
 628                 removeNodeConnector(containerName, p);
 629             }
 630             break;
 631         case CHANGED:
 632             break;
 633         }
 634         if (updateMap) {
 635             if (containers.isEmpty()) {
 636                 // Do cleanup to reduce memory footprint if no
 637                 // elements to be tracked
 638                 this.containerMap.remove(p);
 639             } else {
 640                 this.containerMap.put(p, containers);
 641             }
 642         }
 643     }
 644
 645     @Override
 646     public void containerModeUpdated(UpdateType t) {
 647         // do nothing
 648     }
 649
 650     private void registerWithOSGIConsole() {
 651         BundleContext bundleContext = FrameworkUtil.getBundle(this.getClass())
 652                 .getBundleContext();
 653         bundleContext.registerService(CommandProvider.class.getName(), this,
 654                 null);
 655     }
 656
 657     @Override
 658     public String getHelp() {
 659         StringBuffer help = new StringBuffer();
 660         help.append("---Topology Service Shim---\n");
 661         help.append("\t pem [container]               - Print edgeMap entries");
 662         help.append(" for a given container\n");
 663         return help.toString();
 664     }
 665
 666     public void _pem(CommandInterpreter ci) {
 667         String container = ci.nextArgument();
 668         if (container == null) {
 669             container = GlobalConstants.DEFAULT.toString();
 670         }
 671
 672         ci.println("Container: " + container);
 673         ci.println("                             Edge                                          Bandwidth");
 674
 675         Map<NodeConnector, Pair<Edge, Set<Property>>> edgePropsMap = edgeMap
 676                 .get(container);
 677         if (edgePropsMap == null) {
 678             return;
 679         }
 680         int count = 0;
 681         for (Pair<Edge, Set<Property>> edgeProps : edgePropsMap.values()) {
 682             if (edgeProps == null) {
 683                 continue;
 684             }
 685
 686             long bw = 0;
 687             Set<Property> props = edgeProps.getRight();
 688             if (props != null) {
 689                 for (Property prop : props) {
 690                     if (prop.getName().equals(Bandwidth.BandwidthPropName)) {
 691                         bw = ((Bandwidth) prop).getValue();
 692                     }
 693                 }
 694             }
 695             count++;
 696             ci.println(edgeProps.getLeft() + "          " + bw);
 697         }
 698         ci.println("Total number of Edges: " + count);
 699     }
 700
 701     public void _bwfactor(CommandInterpreter ci) {
 702         String factorString = ci.nextArgument();
 703         if (factorString == null) {
 704             ci.println("Bw threshold: " + this.bwThresholdFactor);
 705             ci.println("Insert a non null bw threshold");
 706             return;
 707         }
 708         bwThresholdFactor = Float.parseFloat(factorString);
 709         ci.println("New Bw threshold: " + this.bwThresholdFactor);
 710     }
 711
 712     /**
 713      * This method will trigger topology updates to be sent toward SAL. SAL then
 714      * pushes the updates to ALL the applications that have registered as
 715      * listeners for this service. SAL has no way of knowing which application
 716      * requested for the refresh.
 717      *
 718      * As an example of this case, is stopping and starting the Topology
 719      * Manager. When the topology Manager is stopped, and restarted, it will no
 720      * longer have the latest topology. Hence, a request is sent here.
 721      *
 722      * @param containerName
 723      * @return void
 724      */
 725     @Override
 726     public void requestRefresh(String containerName) {
 727         // wake up a bulk update thread and exit
 728         // the thread will execute the bulkUpdate()
 729         bulkNotifyQ.add(containerName);
 730     }
 731
 732     /**
 733      * Retrieve/construct edge map for a given container
 734      *
 735      * @param containerName
 736      *            the container name
 737      * @return the edges and their properties
 738      */
 739     private Map<NodeConnector, Pair<Edge, Set<Property>>> getEdgeMap(String containerName) {
 740         Map<NodeConnector, Pair<Edge, Set<Property>>> edgePropMap = null;
 741
 742         /*
 743          * When container is freshly created, need to construct map based on global map.
 744          */
 745         edgePropMap = edgeMap.get(containerName);
 746
 747         return edgePropMap;
 748     }
 749
 750     /**
 751      * Reading the current topology database, the method will replay all the
 752      * edge updates for the ITopologyServiceShimListener instance in the given
 753      * container, which will in turn publish them toward SAL.
 754      *
 755      * @param containerName
 756      *            the container name
 757      */
 758     private void TopologyBulkUpdate(String containerName) {
 759         Map<NodeConnector, Pair<Edge, Set<Property>>> edgePropMap = null;
 760
 761         logger.debug("Try bulk update for container:{}", containerName);
 762         edgePropMap = getEdgeMap(containerName);
 763         if (edgePropMap == null) {
 764             logger.debug("No edges known for container:{}", containerName);
 765             return;
 766         }
 767         ITopologyServiceShimListener topologServiceShimListener = topologyServiceShimListeners
 768                 .get(containerName);
 769         if (topologServiceShimListener == null) {
 770             logger.debug("No topology service shim listener for container:{}",
 771                     containerName);
 772             return;
 773         }
 774         int i = 0;
 775         List<TopoEdgeUpdate> teuList = new ArrayList<TopoEdgeUpdate>();
 776         for (Pair<Edge, Set<Property>> edgeProps : edgePropMap.values()) {
 777             if (edgeProps != null) {
 778                 i++;
 779                 teuList.add(new TopoEdgeUpdate(edgeProps.getLeft(), edgeProps
 780                         .getRight(), UpdateType.ADDED));
 781                 logger.trace("Add edge {}", edgeProps.getLeft());
 782             }
 783         }
 784         if (i > 0) {
 785             topologServiceShimListener.edgeUpdate(teuList);
 786         }
 787         logger.debug("Sent {} updates", i);
 788     }
 789
 790     @Override
 791     public void updateNode(Node node, UpdateType type, Set<Property> props) {
 792     }
 793
 794     @Override
 795     public void updateNodeConnector(NodeConnector nodeConnector,
 796             UpdateType type, Set<Property> props) {
 797         List<String> containers = new ArrayList<String>();
 798         List<String> conList = this.containerMap.get(nodeConnector);
 799
 800         containers.add(GlobalConstants.DEFAULT.toString());
 801         if (conList != null) {
 802             containers.addAll(conList);
 803         }
 804
 805         switch (type) {
 806         case ADDED:
 807             break;
 808         case CHANGED:
 809             if (props == null) {
 810                 break;
 811             }
 812
 813             boolean rmEdge = false;
 814             for (Property prop : props) {
 815                 if (((prop instanceof Config) && (((Config) prop).getValue() != Config.ADMIN_UP))
 816                         || ((prop instanceof State) && (((State) prop)
 817                                 .getValue() != State.EDGE_UP))) {
 818                     /*
 819                      * If port admin down or link down, remove the edges
 820                      * associated with the port
 821                      */
 822                     rmEdge = true;
 823                     break;
 824                 }
 825             }
 826
 827             if (rmEdge) {
 828                 for (String cName : containers) {
 829                     removeNodeConnector(cName, nodeConnector);
 830                 }
 831             }
 832             break;
 833         case REMOVED:
 834             for (String cName : containers) {
 835                 removeNodeConnector(cName, nodeConnector);
 836             }
 837             break;
 838         default:
 839             break;
 840         }
 841     }
 842 }