Use DJC to handle subnet addition/removal from VPN

In process the each subnet in parallel mannger
introduced Job Coordinator

Signed-off-by: manjunathpayala <manjupayala@gmail.com>
Change-Id: I7b2a7e382fc1983016134e8a76addebe836891ee
Signed-off-by: manjunathpayala <manjupayala@gmail.com>

diff --git a/vpnmanager/impl/src/main/java/org/opendaylight/netvirt/vpnmanager/SubnetmapChangeListener.java b/vpnmanager/impl/src/main/java/org/opendaylight/netvirt/vpnmanager/SubnetmapChangeListener.java
index 9adba97b22091719ed381db4d6510b1e7edfe9fb..930c360fffb430b43c3f70e3a1acde7581fd8d13 100644 (file)
--- a/vpnmanager/impl/src/main/java/org/opendaylight/netvirt/vpnmanager/SubnetmapChangeListener.java
+++ b/vpnmanager/impl/src/main/java/org/opendaylight/netvirt/vpnmanager/SubnetmapChangeListener.java
@@ -8,7 +8,9 @@
 
 package org.opendaylight.netvirt.vpnmanager;
 
+import com.google.common.util.concurrent.ListenableFuture;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
@@ -17,6 +19,7 @@ import javax.annotation.PreDestroy;
 import javax.inject.Inject;
 import javax.inject.Singleton;
 import org.opendaylight.genius.utils.JvmGlobalLocks;
+import org.opendaylight.infrautils.jobcoordinator.JobCoordinator;
 import org.opendaylight.infrautils.utils.concurrent.Executors;
 import org.opendaylight.infrautils.utils.concurrent.LoggingFutures;
 import org.opendaylight.mdsal.binding.api.DataBroker;
@@ -47,10 +50,11 @@ public class SubnetmapChangeListener extends AbstractAsyncDataTreeChangeListener
     private final VpnUtil vpnUtil;
     private final IVpnManager vpnManager;
     private final ManagedNewTransactionRunner txRunner;
+    private final JobCoordinator jobCoordinator;
 
     @Inject
     public SubnetmapChangeListener(final DataBroker dataBroker, final VpnSubnetRouteHandler vpnSubnetRouteHandler,
-                                   VpnUtil vpnUtil, IVpnManager vpnManager) {
+                                   VpnUtil vpnUtil, IVpnManager vpnManager,JobCoordinator jobCoordinator) {
         super(dataBroker, LogicalDatastoreType.CONFIGURATION,
                 InstanceIdentifier.create(Subnetmaps.class).child(Subnetmap.class),
                 Executors.newListeningSingleThreadExecutor("SubnetmapChangeListener", LOG));
@@ -59,6 +63,7 @@ public class SubnetmapChangeListener extends AbstractAsyncDataTreeChangeListener
         this.vpnUtil = vpnUtil;
         this.vpnManager = vpnManager;
         this.txRunner = new ManagedNewTransactionRunnerImpl(dataBroker);
+        this.jobCoordinator = jobCoordinator;
         start();
     }
 
@@ -93,39 +98,47 @@ public class SubnetmapChangeListener extends AbstractAsyncDataTreeChangeListener
                       + "ExternalSubnetVpnInstanceListener", subnetId.getValue());
             return;
         }
-        String elanInstanceName = subnetmap.getNetworkId().getValue();
-        long elanTag = getElanTag(elanInstanceName);
-        if (elanTag == 0L) {
-            LOG.error("add: unable to fetch elantag from ElanInstance {} for subnet {}",
-                      elanInstanceName, subnetId.getValue());
-            return;
-        }
-        Uuid vpnId = subnetmap.getVpnId();
-        if (vpnId != null) {
-            boolean isBgpVpn = !vpnId.equals(subnetmap.getRouterId());
-            LOG.info("add: subnetmap {} with elanTag {} to VPN {}", subnetmap, elanTag,
-                     vpnId);
-            vpnSubnetRouteHandler.onSubnetAddedToVpn(subnetmap, isBgpVpn, elanTag);
-            if (isBgpVpn && subnetmap.getRouterId() == null) {
-                Set<VpnTarget> routeTargets = vpnManager.getRtListForVpn(vpnId.getValue());
-                if (!routeTargets.isEmpty()) {
-                    // FIXME: separate this out somehow?
-                    final ReentrantLock lock = JvmGlobalLocks.getLockForString(subnetmap.getSubnetIp());
-                    lock.lock();
-                    try {
-                        vpnManager.updateRouteTargetsToSubnetAssociation(routeTargets, subnetmap.getSubnetIp(),
-                                vpnId.getValue());
-                    } finally {
-                        lock.unlock();
+        jobCoordinator.enqueueJob("SUBNETROUTE-" + subnetId, () -> {
+            String elanInstanceName = subnetmap.getNetworkId().getValue();
+            long elanTag = getElanTag(elanInstanceName);
+            if (elanTag == 0L) {
+                LOG.error("add: unable to fetch elantag from ElanInstance {} for subnet {}",
+                        elanInstanceName, subnetId.getValue());
+                return Collections.emptyList();
+            }
+            Uuid vpnId = subnetmap.getVpnId();
+            if (vpnId != null) {
+                boolean isBgpVpn = !vpnId.equals(subnetmap.getRouterId());
+                LOG.info("add: subnetmap {} with elanTag {} to VPN {}", subnetmap, elanTag,
+                        vpnId);
+                vpnSubnetRouteHandler.onSubnetAddedToVpn(subnetmap, isBgpVpn, elanTag);
+                if (isBgpVpn && subnetmap.getRouterId() == null) {
+                    Set<VpnTarget> routeTargets = vpnManager.getRtListForVpn(vpnId.getValue());
+                    if (!routeTargets.isEmpty()) {
+                        // FIXME: separate this out somehow?
+                        final ReentrantLock lock = JvmGlobalLocks.getLockForString(subnetmap.getSubnetIp());
+                        lock.lock();
+                        try {
+                            vpnManager.updateRouteTargetsToSubnetAssociation(routeTargets, subnetmap.getSubnetIp(),
+                                    vpnId.getValue());
+                        } finally {
+                            lock.unlock();
+                        }
                     }
                 }
             }
-        }
+            return Collections.emptyList();
+        });
     }
 
     @Override
     public void remove(InstanceIdentifier<Subnetmap> identifier, Subnetmap subnetmap) {
         LOG.trace("remove: subnetmap method - key: {}, value: {}", identifier, subnetmap);
+        jobCoordinator.enqueueJob("SUBNETROUTE-" + subnetmap.getId(), () -> {
+            java.util.Optional.ofNullable(subnetmap.getPortList()).ifPresent(portList ->
+                    portList.forEach(port -> vpnSubnetRouteHandler.onPortRemovedFromSubnet(subnetmap, port)));
+            return Collections.emptyList();
+        });
     }
 
     @Override
@@ -141,61 +154,63 @@ public class SubnetmapChangeListener extends AbstractAsyncDataTreeChangeListener
             LOG.error("update: network was not found for subnetId {}", subnetId.getValue());
             return;
         }
-        String elanInstanceName = subnetmapUpdate.getNetworkId().getValue();
-        long elanTag = getElanTag(elanInstanceName);
-        if (elanTag == 0L) {
-            LOG.error("update: unable to fetch elantag from ElanInstance {} for subnetId {}",
-                      elanInstanceName, subnetId);
-            return;
-        }
-        updateVlanDataEntry(subnetmapOriginal.getVpnId(), subnetmapUpdate.getVpnId(), subnetmapUpdate,
-                subnetmapOriginal, elanInstanceName);
-        if (VpnUtil.getIsExternal(network)) {
-            LOG.debug("update: provider subnetwork {} is handling in "
-                      + "ExternalSubnetVpnInstanceListener", subnetId.getValue());
-            return;
-        }
-        // update on BGPVPN or InternalVPN change
-        Uuid vpnIdOld = subnetmapOriginal.getVpnId();
-        Uuid vpnIdNew = subnetmapUpdate.getVpnId();
-        if (!Objects.equals(vpnIdOld, vpnIdNew)) {
-            LOG.info("update: update subnetOpDataEntry for subnet {} imported in VPN",
-                     subnetmapUpdate.getId().getValue());
-            updateSubnetmapOpDataEntry(subnetmapOriginal.getVpnId(), subnetmapUpdate.getVpnId(), subnetmapUpdate,
-                                       subnetmapOriginal, elanTag);
-        }
-        // update on Internet VPN Id change
-        Uuid inetVpnIdOld = subnetmapOriginal.getInternetVpnId();
-        Uuid inetVpnIdNew = subnetmapUpdate.getInternetVpnId();
-        if (!Objects.equals(inetVpnIdOld, inetVpnIdNew)) {
-            LOG.info("update: update subnetOpDataEntry for subnet {} imported in InternetVPN",
-                     subnetmapUpdate.getId().getValue());
-            updateSubnetmapOpDataEntry(inetVpnIdOld, inetVpnIdNew, subnetmapUpdate, subnetmapOriginal, elanTag);
-        }
-        // update on PortList change
-        List<Uuid> oldPortList;
-        List<Uuid> newPortList;
-        newPortList = subnetmapUpdate.getPortList() != null ? subnetmapUpdate.getPortList() : new ArrayList<>();
-        oldPortList = subnetmapOriginal.getPortList() != null ? subnetmapOriginal.getPortList() : new ArrayList<>();
-        if (newPortList.size() == oldPortList.size()) {
-            return;
-        }
-        LOG.info("update: update port list for subnet {}", subnetmapUpdate.getId().getValue());
-        if (newPortList.size() > oldPortList.size()) {
-            for (Uuid portId : newPortList) {
-                if (! oldPortList.contains(portId)) {
-                    vpnSubnetRouteHandler.onPortAddedToSubnet(subnetmapUpdate, portId);
-                    return;
-                }
+        jobCoordinator.enqueueJob("SUBNETROUTE-" + subnetId, () -> {
+            List<ListenableFuture<Void>> futures = Collections.emptyList();
+            String elanInstanceName = subnetmapUpdate.getNetworkId().getValue();
+            long elanTag = getElanTag(elanInstanceName);
+            if (elanTag == 0L) {
+                LOG.error("update: unable to fetch elantag from ElanInstance {} for subnetId {}",
+                        elanInstanceName, subnetId);
+                return futures;
+            }
+            updateVlanDataEntry(subnetmapOriginal.getVpnId(), subnetmapUpdate.getVpnId(), subnetmapUpdate,
+                    subnetmapOriginal, elanInstanceName);
+            if (VpnUtil.getIsExternal(network)) {
+                LOG.debug("update: provider subnetwork {} is handling in "
+                        + "ExternalSubnetVpnInstanceListener", subnetId.getValue());
+                return futures;
+            }
+            // update on BGPVPN or InternalVPN change
+            Uuid vpnIdOld = subnetmapOriginal.getVpnId();
+            Uuid vpnIdNew = subnetmapUpdate.getVpnId();
+            if (!Objects.equals(vpnIdOld, vpnIdNew)) {
+                LOG.info("update: update subnetOpDataEntry for subnet {} imported in VPN",
+                        subnetmapUpdate.getId().getValue());
+                updateSubnetmapOpDataEntry(subnetmapOriginal.getVpnId(), subnetmapUpdate.getVpnId(), subnetmapUpdate,
+                        subnetmapOriginal, elanTag);
             }
-        } else {
-            for (Uuid portId : oldPortList) {
-                if (! newPortList.contains(portId)) {
-                    vpnSubnetRouteHandler.onPortRemovedFromSubnet(subnetmapUpdate, portId);
-                    return;
+            // update on Internet VPN Id change
+            Uuid inetVpnIdOld = subnetmapOriginal.getInternetVpnId();
+            Uuid inetVpnIdNew = subnetmapUpdate.getInternetVpnId();
+            if (!Objects.equals(inetVpnIdOld, inetVpnIdNew)) {
+                LOG.info("update: update subnetOpDataEntry for subnet {} imported in InternetVPN",
+                        subnetmapUpdate.getId().getValue());
+                updateSubnetmapOpDataEntry(inetVpnIdOld, inetVpnIdNew, subnetmapUpdate, subnetmapOriginal, elanTag);
+            }
+            // update on PortList change
+            List<Uuid> oldPortList;
+            List<Uuid> newPortList;
+            newPortList = subnetmapUpdate.getPortList() != null ? subnetmapUpdate.getPortList() : new ArrayList<>();
+            oldPortList = subnetmapOriginal.getPortList() != null ? subnetmapOriginal.getPortList() : new ArrayList<>();
+            if (newPortList.size() == oldPortList.size()) {
+                return futures;
+            }
+            LOG.info("update: update port list for subnet {}", subnetmapUpdate.getId().getValue());
+            if (newPortList.size() > oldPortList.size()) {
+                for (Uuid portId : newPortList) {
+                    if (!oldPortList.contains(portId)) {
+                        vpnSubnetRouteHandler.onPortAddedToSubnet(subnetmapUpdate, portId);
+                    }
+                }
+            } else {
+                for (Uuid portId : oldPortList) {
+                    if (!newPortList.contains(portId)) {
+                        vpnSubnetRouteHandler.onPortRemovedFromSubnet(subnetmapUpdate, portId);
+                    }
                 }
             }
-        }
+            return futures;
+        });
     }
 
     private void updateSubnetmapOpDataEntry(Uuid vpnIdOld, Uuid vpnIdNew, Subnetmap subnetmapUpdate,