X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fprotocol_plugins%2Fopenflow%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fprotocol_plugin%2Fopenflow%2Finternal%2FTopologyServiceShim.java;h=400e9e869218ad60ea8a0898800e666ac03f7ef4;hp=af502a2177d12c89101ca5b2363f90a9eeabe4da;hb=dfa4383b0b5c9c6de340526a62aef731922fa29f;hpb=44331939251d6c9f800f63076e9d51bd8263b526

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServiceShim.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServiceShim.java
index af502a2177..400e9e8692 100644
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServiceShim.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServiceShim.java
@@ -9,6 +9,9 @@
 package org.opendaylight.controller.protocol_plugin.openflow.internal;
 
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -24,6 +27,7 @@ import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.apache.commons.lang3.tuple.Pair;
 import org.eclipse.osgi.framework.console.CommandInterpreter;
 import org.eclipse.osgi.framework.console.CommandProvider;
+import org.opendaylight.controller.protocol_plugin.openflow.IDiscoveryListener;
 import org.opendaylight.controller.protocol_plugin.openflow.IInventoryShimExternalListener;
 import org.opendaylight.controller.protocol_plugin.openflow.IOFStatisticsManager;
 import org.opendaylight.controller.protocol_plugin.openflow.IRefreshInternalProvider;
@@ -43,7 +47,7 @@ import org.opendaylight.controller.sal.core.NodeConnector;
 import org.opendaylight.controller.sal.core.Property;
 import org.opendaylight.controller.sal.core.State;
 import org.opendaylight.controller.sal.core.UpdateType;
-import org.opendaylight.controller.sal.discovery.IDiscoveryService;
+import org.opendaylight.controller.sal.topology.TopoEdgeUpdate;
 import org.opendaylight.controller.sal.utils.GlobalConstants;
 
 /**
@@ -51,9 +55,9 @@ import org.opendaylight.controller.sal.utils.GlobalConstants;
  * OpenFlow core to various listeners. The notifications are filtered based on
  * container configurations.
  */
-public class TopologyServiceShim implements IDiscoveryService,
-		IContainerListener, CommandProvider, IRefreshInternalProvider,
-		IInventoryShimExternalListener {
+public class TopologyServiceShim implements IDiscoveryListener,
+        IContainerListener, CommandProvider, IRefreshInternalProvider,
+        IInventoryShimExternalListener {
     protected static final Logger logger = LoggerFactory
             .getLogger(TopologyServiceShim.class);
     private ConcurrentMap<String, ITopologyServiceShimListener> topologyServiceShimListeners = new ConcurrentHashMap<String, ITopologyServiceShimListener>();
@@ -76,43 +80,71 @@ public class TopologyServiceShim implements IDiscoveryService,
 
     class NotifyEntry {
         String container;
-        Pair<Edge, Set<Property>> edgeProps;
-        UpdateType type;
+        List<TopoEdgeUpdate> teuList;
 
-        NotifyEntry(String container, Pair<Edge, Set<Property>> edgeProps,
-                UpdateType type) {
+        public NotifyEntry(String container, TopoEdgeUpdate teu) {
             this.container = container;
-            this.edgeProps = edgeProps;
-            this.type = type;
+            this.teuList = new ArrayList<TopoEdgeUpdate>();
+            if (teu != null) {
+                this.teuList.add(teu);
+            }
+        }
+
+        public NotifyEntry(String container, List<TopoEdgeUpdate> teuList) {
+            this.container = container;
+            this.teuList = new ArrayList<TopoEdgeUpdate>();
+            if (teuList != null) {
+                this.teuList.addAll(teuList);
+            }
         }
     }
 
     class TopologyNotify implements Runnable {
         private final BlockingQueue<NotifyEntry> notifyQ;
+        private NotifyEntry entry;
+        private Map<String, List<TopoEdgeUpdate>> teuMap = new HashMap<String, List<TopoEdgeUpdate>>();
+        private List<TopoEdgeUpdate> teuList;
+        private boolean notifyListeners;
 
         TopologyNotify(BlockingQueue<NotifyEntry> notifyQ) {
             this.notifyQ = notifyQ;
         }
 
+        @Override
         public void run() {
             while (true) {
                 try {
-                    NotifyEntry entry = notifyQ.take();
+                    teuMap.clear();
+                    notifyListeners = false;
+                    while (!notifyQ.isEmpty()) {
+                        entry = notifyQ.take();
+                        teuList = teuMap.get(entry.container);
+                        if (teuList == null) {
+                            teuList = new ArrayList<TopoEdgeUpdate>();
+                        }
+                        // group all the updates together
+                        teuList.addAll(entry.teuList);
+                        teuMap.put(entry.container, teuList);
+                        notifyListeners = true;
+                    }
 
-                    ITopologyServiceShimListener topologServiceShimListener = topologyServiceShimListeners
-                            .get(entry.container);
-                    topologServiceShimListener.edgeUpdate(
-                            entry.edgeProps.getLeft(), entry.type,
-                            entry.edgeProps.getRight());
+                    if (notifyListeners) {
+                        for (String container : teuMap.keySet()) {
+                            // notify the listener
+                            topologyServiceShimListeners.get(container)
+                                    .edgeUpdate(teuMap.get(container));
+                        }
+                    }
 
-                    entry = null;
+                    Thread.sleep(100);
                 } catch (InterruptedException e1) {
-                    logger.warn("TopologyNotify interrupted {}", e1.getMessage());
+                    logger.warn("TopologyNotify interrupted {}",
+                            e1.getMessage());
                     if (shuttingDown) {
                         return;
                     }
                 } catch (Exception e2) {
-                    logger.error("",e2);
+                    logger.error("", e2);
                 }
             }
         }
@@ -135,6 +167,7 @@ public class TopologyServiceShim implements IDiscoveryService,
             this.notifyQ = notifyQ;
         }
 
+        @Override
         public void run() {
             while (true) {
                 try {
@@ -165,7 +198,7 @@ public class TopologyServiceShim implements IDiscoveryService,
                         return;
                     }
                 } catch (Exception e2) {
-                    logger.error("",e2);
+                    logger.error("", e2);
                 }
             }
         }
@@ -174,7 +207,7 @@ public class TopologyServiceShim implements IDiscoveryService,
     /**
      * Function called by the dependency manager when all the required
      * dependencies are satisfied
-     * 
+     *
      */
     void init() {
         logger.trace("Init called");
@@ -284,7 +317,7 @@ public class TopologyServiceShim implements IDiscoveryService,
      * 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() {
         logger.trace("DESTROY called!");
@@ -295,7 +328,7 @@ public class TopologyServiceShim implements IDiscoveryService,
     /**
      * 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() {
         logger.trace("START called!");
@@ -309,7 +342,7 @@ public class TopologyServiceShim implements IDiscoveryService,
      * 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() {
         logger.trace("STOP called!");
@@ -353,7 +386,8 @@ public class TopologyServiceShim implements IDiscoveryService,
                 && this.topologyServiceShimListeners.get(containerName).equals(
                         s)) {
             this.topologyServiceShimListeners.remove(containerName);
-            logger.trace("Removed topologyServiceShimListener for container: {}",
+            logger.trace(
+                    "Removed topologyServiceShimListener for container: {}",
                     containerName);
         }
     }
@@ -368,8 +402,95 @@ public class TopologyServiceShim implements IDiscoveryService,
         }
     }
 
+    private void updateContainerMap(List<String> containers, NodeConnector p) {
+        if (containers.isEmpty()) {
+            // Do cleanup to reduce memory footprint if no
+            // elements to be tracked
+            this.containerMap.remove(p);
+        } else {
+            this.containerMap.put(p, containers);
+        }
+    }
+
+    /**
+     * From a given edge map, retrieve the edge sourced by the port and update
+     * the local cache in the container
+     *
+     * @param container
+     *            the container name
+     * @param nodeConnector
+     *            the node connector
+     * @param edges
+     *            the given edge map
+     * @return the found edge
+     */
+    private Edge addEdge(String container, NodeConnector nodeConnector,
+            Map<NodeConnector, Pair<Edge, Set<Property>>> edges) {
+        logger.debug("Search edge sourced by port {} in container {}", nodeConnector, container);
+
+        // Retrieve the associated edge
+        Pair<Edge, Set<Property>> edgeProps = edges.get(nodeConnector);
+        if (edgeProps == null) {
+            logger.debug("edgePros is null for port {} in container {}", nodeConnector, container);
+            return null;
+        }
+
+        Edge edge = edgeProps.getLeft();
+        if (edge == null) {
+            logger.debug("edge is null for port {} in container {}", nodeConnector, container);
+            return null;
+        }
+
+        // Make sure the peer port is in the same container
+        NodeConnector peerConnector = edge.getHeadNodeConnector();
+        List<String> containers = this.containerMap.get(peerConnector);
+        if ((containers == null) || !containers.contains(container)) {
+            logger.debug("peer port {} of edge {} is not part of the container {}", new Object[] { peerConnector, edge,
+                    container });
+            return null;
+        }
+
+        // Update the local cache
+        updateLocalEdgeMap(container, edge, UpdateType.ADDED, edgeProps.getRight());
+        logger.debug("Added edge {} to local cache in container {}", edge, container);
+
+        return edge;
+    }
+
+    private void addNodeConnector(String container,
+            NodeConnector nodeConnector) {
+        // Use the global edge map for the newly added port in a container
+        Map<NodeConnector, Pair<Edge, Set<Property>>> globalEdgeMap = edgeMap.get(GlobalConstants.DEFAULT
+                .toString());
+        if (globalEdgeMap == null) {
+            return;
+        }
+
+        // Get the edge and update local cache in the container
+        Edge edge1, edge2;
+        edge1 = addEdge(container, nodeConnector, globalEdgeMap);
+        if (edge1 == null) {
+            return;
+        }
+
+        // Get the edge in reverse direction and update local cache in the container
+        NodeConnector peerConnector = edge1.getHeadNodeConnector();
+        edge2 = addEdge(container, peerConnector, globalEdgeMap);
+
+        // Send notification upwards in one shot
+        List<TopoEdgeUpdate> teuList = new ArrayList<TopoEdgeUpdate>();
+        teuList.add(new TopoEdgeUpdate(edge1, null, UpdateType.ADDED));
+        logger.debug("Notify edge1: {} in container {}", edge1, container);
+        if (edge2 != null) {
+            teuList.add(new TopoEdgeUpdate(edge2, null, UpdateType.ADDED));
+            logger.debug("Notify edge2: {} in container {}", edge2, container);
+        }
+        notifyEdge(container, teuList);
+    }
+
     private void removeNodeConnector(String container,
             NodeConnector nodeConnector) {
+        List<TopoEdgeUpdate> teuList = new ArrayList<TopoEdgeUpdate>();
         Map<NodeConnector, Pair<Edge, Set<Property>>> edgePropsMap = edgeMap
                 .get(container);
         if (edgePropsMap == null) {
@@ -381,7 +502,8 @@ public class TopologyServiceShim implements IDiscoveryService,
         if (edgeProps == null) {
             return;
         }
-        notifyEdge(container, edgeProps.getLeft(), UpdateType.REMOVED, null);
+        teuList.add(new TopoEdgeUpdate(edgeProps.getLeft(), null,
+                UpdateType.REMOVED));
 
         // Remove edge in another direction
         edgeProps = edgePropsMap
@@ -389,52 +511,130 @@ public class TopologyServiceShim implements IDiscoveryService,
         if (edgeProps == null) {
             return;
         }
-        notifyEdge(container, edgeProps.getLeft(), UpdateType.REMOVED, null);
+        teuList.add(new TopoEdgeUpdate(edgeProps.getLeft(), null,
+                UpdateType.REMOVED));
+
+        // Update in one shot
+        notifyEdge(container, teuList);
     }
 
-    private void notifyEdge(String container, Edge edge, UpdateType type,
-            Set<Property> props) {
+    /**
+     * Update local cache and return true if it needs to notify upper layer
+     * Topology listeners.
+     *
+     * @param container
+     *            The network container
+     * @param edge
+     *            The edge
+     * @param type
+     *            The update type
+     * @param props
+     *            The edge properties
+     * @return true if it needs to notify upper layer Topology listeners
+     */
+    private boolean updateLocalEdgeMap(String container, Edge edge,
+            UpdateType type, Set<Property> props) {
         ConcurrentMap<NodeConnector, Pair<Edge, Set<Property>>> edgePropsMap = edgeMap
                 .get(container);
         NodeConnector src = edge.getTailNodeConnector();
         Pair<Edge, Set<Property>> edgeProps = new ImmutablePair<Edge, Set<Property>>(
                 edge, props);
+        boolean rv = false;
 
         switch (type) {
         case ADDED:
         case CHANGED:
             if (edgePropsMap == null) {
                 edgePropsMap = new ConcurrentHashMap<NodeConnector, Pair<Edge, Set<Property>>>();
+                rv = true;
             } else {
                 if (edgePropsMap.containsKey(src)
                         && edgePropsMap.get(src).equals(edgeProps)) {
-                    // Entry already exists. Return here.
-                    return;
+                    // Entry already exists. No update.
+                    rv = false;
+                } else {
+                    rv = true;
                 }
             }
-            edgePropsMap.put(src, edgeProps);
-            edgeMap.put(container, edgePropsMap);
+            if (rv) {
+                edgePropsMap.put(src, edgeProps);
+                edgeMap.put(container, edgePropsMap);
+            }
             break;
         case REMOVED:
-            if (edgePropsMap != null) {
+            if ((edgePropsMap != null) && edgePropsMap.containsKey(src)) {
                 edgePropsMap.remove(src);
                 if (edgePropsMap.isEmpty()) {
                     edgeMap.remove(container);
                 } else {
                     edgeMap.put(container, edgePropsMap);
                 }
+                rv = true;
             }
             break;
         default:
-            logger.debug("notifyEdge: invalid {} for Edge {} in container {}",
-                    type, edge, container);
+            logger.debug(
+                    "notifyLocalEdgeMap: invalid {} for Edge {} in container {}",
+                    new Object[] { type.getName(), edge, container });
+        }
+
+        if (rv) {
+            logger.debug(
+                    "notifyLocalEdgeMap: {} for Edge {} in container {}",
+                    new Object[] { type.getName(), edge, container });
+        }
+
+        return rv;
+    }
+
+    private void notifyEdge(String container, Edge edge, UpdateType type,
+            Set<Property> props) {
+        boolean notifyListeners;
+
+        // Update local cache
+        notifyListeners = updateLocalEdgeMap(container, edge, type, props);
+
+        // Prepare to update TopologyService
+        if (notifyListeners) {
+            notifyQ.add(new NotifyEntry(container, new TopoEdgeUpdate(edge, props,
+                    type)));
+            logger.debug("notifyEdge: {} Edge {} in container {}",
+                    new Object[] { type.getName(), edge, container });
+        }
+    }
+
+    private void notifyEdge(String container, List<TopoEdgeUpdate> etuList) {
+        if (etuList == null) {
             return;
         }
 
-        notifyQ.add(new NotifyEntry(container, edgeProps, type));
+        Edge edge;
+        UpdateType type;
+        List<TopoEdgeUpdate> etuNotifyList = new ArrayList<TopoEdgeUpdate>();
+        boolean notifyListeners = false, rv;
+
+        for (TopoEdgeUpdate etu : etuList) {
+            edge = etu.getEdge();
+            type = etu.getUpdateType();
+
+            // Update local cache
+            rv = updateLocalEdgeMap(container, edge, type, etu.getProperty());
+            if (rv) {
+                if (!notifyListeners) {
+                    notifyListeners = true;
+                }
+                etuNotifyList.add(etu);
+                logger.debug(
+                        "notifyEdge(TopoEdgeUpdate): {} Edge {} in container {}",
+                        new Object[] { type.getName(), edge, container });
+            }
+        }
 
-        logger.debug("notifyEdge: {} Edge {} in container {}",
-                type, edge, container);
+        // Prepare to update TopologyService
+        if (notifyListeners) {
+            notifyQ.add(new NotifyEntry(container, etuNotifyList));
+            logger.debug("notifyEdge(TopoEdgeUpdate): add notifyQ");
+        }
     }
 
     @Override
@@ -501,33 +701,24 @@ public class TopologyServiceShim implements IDiscoveryService,
         if (containers == null) {
             containers = new CopyOnWriteArrayList<String>();
         }
-        boolean updateMap = false;
         switch (t) {
         case ADDED:
             if (!containers.contains(containerName)) {
                 containers.add(containerName);
-                updateMap = true;
+                updateContainerMap(containers, p);
+                addNodeConnector(containerName, p);
             }
             break;
         case REMOVED:
             if (containers.contains(containerName)) {
                 containers.remove(containerName);
-                updateMap = true;
+                updateContainerMap(containers, p);
                 removeNodeConnector(containerName, p);
             }
             break;
         case CHANGED:
             break;
         }
-        if (updateMap) {
-            if (containers.isEmpty()) {
-                // Do cleanup to reduce memory footprint if no
-                // elements to be tracked
-                this.containerMap.remove(p);
-            } else {
-                this.containerMap.put(p, containers);
-            }
-        }
     }
 
     @Override
@@ -572,9 +763,12 @@ public class TopologyServiceShim implements IDiscoveryService,
             }
 
             long bw = 0;
-            for (Property prop : edgeProps.getRight()) {
-                if (prop.getName().equals(Bandwidth.BandwidthPropName)) {
-                    bw = ((Bandwidth) prop).getValue();
+            Set<Property> props = edgeProps.getRight();
+            if (props != null) {
+                for (Property prop : props) {
+                    if (prop.getName().equals(Bandwidth.BandwidthPropName)) {
+                        bw = ((Bandwidth) prop).getValue();
+                    }
                 }
             }
             count++;
@@ -599,11 +793,11 @@ public class TopologyServiceShim implements IDiscoveryService,
      * pushes the updates to ALL the applications that have registered as
      * listeners for this service. SAL has no way of knowing which application
      * requested for the refresh.
-     * 
+     *
      * As an example of this case, is stopping and starting the Topology
      * Manager. When the topology Manager is stopped, and restarted, it will no
      * longer have the latest topology. Hence, a request is sent here.
-     * 
+     *
      * @param containerName
      * @return void
      */
@@ -614,19 +808,36 @@ public class TopologyServiceShim implements IDiscoveryService,
         bulkNotifyQ.add(containerName);
     }
 
+    /**
+     * Retrieve the edges for a given container
+     *
+     * @param containerName
+     *            the container name
+     * @return the edges and their properties
+     */
+    private Collection<Pair<Edge, Set<Property>>> getEdgeProps(String containerName) {
+        Map<NodeConnector, Pair<Edge, Set<Property>>> edgePropMap = null;
+        edgePropMap = edgeMap.get(containerName);
+        if (edgePropMap == null) {
+            return null;
+        }
+        return edgePropMap.values();
+    }
+
     /**
      * Reading the current topology database, the method will replay all the
      * edge updates for the ITopologyServiceShimListener instance in the given
      * container, which will in turn publish them toward SAL.
-     * 
+     *
      * @param containerName
+     *            the container name
      */
     private void TopologyBulkUpdate(String containerName) {
-        Map<NodeConnector, Pair<Edge, Set<Property>>> edgePropMap = null;
+        Collection<Pair<Edge, Set<Property>>> edgeProps = null;
 
         logger.debug("Try bulk update for container:{}", containerName);
-        edgePropMap = edgeMap.get(containerName);
-        if (edgePropMap == null) {
+        edgeProps = getEdgeProps(containerName);
+        if (edgeProps == null) {
             logger.debug("No edges known for container:{}", containerName);
             return;
         }
@@ -638,14 +849,18 @@ public class TopologyServiceShim implements IDiscoveryService,
             return;
         }
         int i = 0;
-        for (Pair<Edge, Set<Property>> edgeProps : edgePropMap.values()) {
-            if (edgeProps != null) {
+        List<TopoEdgeUpdate> teuList = new ArrayList<TopoEdgeUpdate>();
+        for (Pair<Edge, Set<Property>> edgeProp : edgeProps) {
+            if (edgeProp != null) {
                 i++;
-                logger.trace("Add edge {}", edgeProps.getLeft());
-                topologServiceShimListener.edgeUpdate(edgeProps.getLeft(),
-                        UpdateType.ADDED, edgeProps.getRight());
+                teuList.add(new TopoEdgeUpdate(edgeProp.getLeft(), edgeProp
+                        .getRight(), UpdateType.ADDED));
+                logger.trace("Add edge {}", edgeProp.getLeft());
             }
         }
+        if (i > 0) {
+            topologServiceShimListener.edgeUpdate(teuList);
+        }
         logger.debug("Sent {} updates", i);
     }
 
@@ -663,7 +878,7 @@ public class TopologyServiceShim implements IDiscoveryService,
         if (conList != null) {
             containers.addAll(conList);
         }
-        
+
         switch (type) {
         case ADDED:
             break;