bdp: remove queues for sending BGP messages (#6226)

Re-apply #6217 with a fix to PrefixTracer.

* use deltas instead of queues for outgoing route advertisements
(cherry picked from commit edaa1aa)
* make PrefixTracer thread-safe

projects/batfish/src/main/java/org/batfish/dataplane/ibdp/BgpRoutingProcess.java

@@ -122,6 +122,12 @@ final class BgpRoutingProcess implements RoutingProcess<BgpTopology, BgpRoute<?,
   @Nonnull Bgpv4Rib _ebgpv4StagingRib;
   /** RIB containing paths obtained with iBGP, for IPv4 unicast */
   @Nonnull Bgpv4Rib _ibgpv4Rib;
+
+  // outgoing RIB deltas for the current round.
+  RibDelta<Bgpv4Route> _ebgpDelta = null;
+  RibDelta<AnnotatedRoute<AbstractRoute>> _mainRibBgpv4RouteDelta = null;
+  RibDelta<Bgpv4Route> _bgpv4Delta = null;
+
   /**
    * Helper RIB containing paths obtained with iBGP during current iteration. An Adj-RIB-in of sorts
    * for IPv4 unicast
@@ -1067,6 +1073,10 @@ int iterationHashCode() {
             // RIBs
             _bgpv4Rib.getTypedRoutes(),
             _evpnRib.getTypedRoutes(),
+            // Outgoing RIB deltas
+            _ebgpDelta,
+            _bgpv4Delta,
+            _mainRibBgpv4RouteDelta,
             // Message queues
             _bgpv4IncomingRoutes,
             _evpnType3IncomingRoutes,

projects/batfish/src/main/java/org/batfish/dataplane/ibdp/IncrementalBdpEngine.java

@@ -19,6 +19,7 @@
 import io.opentracing.Span;
 import io.opentracing.util.GlobalTracer;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
@@ -27,6 +28,7 @@
 import java.util.SortedSet;
 import java.util.TreeSet;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.batfish.common.BatfishLogger;
@@ -427,7 +429,6 @@ private static void computeDependentRoutesIteration(
       }
 
       computeIterationOfBgpRoutes(
-          nodes,
           iterationLabel,
           allNodes,
           topologyContext.getBgpTopology(),
@@ -454,7 +455,6 @@ private static void computeDependentRoutesIteration(
   }
 
   private static void computeIterationOfBgpRoutes(
-      Map<String, Node> nodes,
       String iterationLabel,
       Map<String, Node> allNodes,
       BgpTopology bgpTopology,
@@ -465,7 +465,7 @@ private static void computeIterationOfBgpRoutes(
     LOGGER.info("{}: Init for new BGP iteration", iterationLabel);
     try (Scope scope = GlobalTracer.get().scopeManager().activate(span)) {
       assert scope != null; // avoid unused warning
-      nodes
+      allNodes
           .values()
           .parallelStream()
           .forEach(
@@ -487,7 +487,7 @@ private static void computeIterationOfBgpRoutes(
     try (Scope innerScope = GlobalTracer.get().scopeManager().activate(genSpan)) {
       assert innerScope != null; // avoid unused warning
       // first let's initialize nodes-level generated/aggregate routes
-      nodes
+      allNodes
           .values()
           .parallelStream()
           .forEach(
@@ -499,29 +499,51 @@ private static void computeIterationOfBgpRoutes(
     Span propSpan =
         GlobalTracer.get().buildSpan(iterationLabel + ": Propagate BGP v4 routes").start();
     LOGGER.info("{}: Propagate BGP v4 routes", iterationLabel);
+
     try (Scope innerScope = GlobalTracer.get().scopeManager().activate(propSpan)) {
       assert innerScope != null; // avoid unused warning
-      nodes
-          .values()
-          .parallelStream()
-          .flatMap(n -> n.getVirtualRouters().values().stream())
-          .forEach(
-              vr -> {
-                Map<Bgpv4Rib, RibDelta<Bgpv4Route>> deltas =
-                    vr.processBgpMessages(bgpTopology, networkConfigurations, nodes);
-                vr.finalizeBgpRoutesAndQueueOutgoingMessages(
-                    deltas, allNodes, bgpTopology, networkConfigurations);
-              });
+
+      /*
+      In a round, each VR will process BGP messages from its neighbors, adding them to the right
+      RIBs and building a RibDelta containing messages that will be sent to neighbors in the next
+      round (the "outgoing delta"). Processing messages from a neighbor requires reading from
+      their outgoing delta. We do this for all VRs in parallel, so we need to be careful not to
+      read from and write to the same delta simultaneously. We separate reading and writing into
+      two stages. In the first stage, all VRs read from their neighbors' deltas, and build
+      Runnables (thunks) to clear and then write to their own deltas. In the second stage, we
+      simply run these Runnables (also in parallel).
+      */
+      List<Runnable> finalizeBgpRoutesAndQueueOutgoingMessages =
+          allNodes
+              .values()
+              .parallelStream()
+              .flatMap(n -> n.getVirtualRouters().values().stream())
+              .map(
+                  vr -> {
+                    // Stage 1: read from all eighbors' queues.
+                    Map<Bgpv4Rib, RibDelta<Bgpv4Route>> deltas =
+                        vr.processBgpMessages(bgpTopology, networkConfigurations, allNodes);
+                    // return a runnable for stage 2.
+                    return (Runnable)
+                        () -> {
+                          vr.clearBgpOutgoingDeltas();
+                          vr.finalizeBgpRoutesAndBuildOutgoingDeltas(deltas);
+                        };
+                  })
+              .collect(Collectors.toList());
+
+      // Stage 2: update all the outgoing queues
+      finalizeBgpRoutesAndQueueOutgoingMessages.parallelStream().forEach(Runnable::run);
 
       // Merge BGP routes from BGP process into the main RIB
-      nodes
+      allNodes
           .values()
           .parallelStream()
           .flatMap(n -> n.getVirtualRouters().values().stream())
           .forEach(VirtualRouter::mergeBgpRoutesToMainRib);
 
       // Multi-VRF redistribution of BGP routes:
-      nodes
+      allNodes
           .values()
           .parallelStream()
           .forEach(
@@ -738,8 +760,7 @@ private boolean computeNonMonotonicPortionOfDataPlane(
             .flatMap(n -> n.getVirtualRouters().values().stream())
             .forEach(
                 vr -> {
-                  vr.processExternalBgpAdvertisements(
-                      externalAdverts, ipVrfOwners, nodes, bgpTopology, networkConfigurations);
+                  vr.processExternalBgpAdvertisements(externalAdverts, ipVrfOwners);
                   vr.queueInitialBgpMessages(bgpTopology, nodes, networkConfigurations);
                 });
       } finally {

projects/batfish/src/main/java/org/batfish/dataplane/ibdp/PrefixTracer.java

@@ -2,13 +2,15 @@
 
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableSortedSet;
+import com.google.common.collect.Sets;
 import java.io.Serializable;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.SortedSet;
+import java.util.concurrent.ConcurrentHashMap;
 import javax.annotation.Nullable;
 import org.batfish.datamodel.Ip;
 import org.batfish.datamodel.Prefix;
@@ -104,11 +106,11 @@ public String toString() {
   private final Map<Prefix, Set<Neighbor>> _sent;
 
   PrefixTracer() {
-    _originated = new HashSet<>();
-    _installed = new HashMap<>();
-    _sent = new HashMap<>();
-    _filteredOnImport = new HashMap<>();
-    _filteredOnExport = new HashMap<>();
+    _originated = Sets.newConcurrentHashSet();
+    _installed = new ConcurrentHashMap<>();
+    _sent = new ConcurrentHashMap<>();
+    _filteredOnImport = new ConcurrentHashMap<>();
+    _filteredOnExport = new ConcurrentHashMap<>();
   }
 
   /** Note that we considered given prefix for origination */
@@ -123,7 +125,7 @@ void sentTo(
       Ip neighborIp,
       String neighborVrf,
       @Nullable String exportPolicy) {
-    Set<Neighbor> set = _sent.computeIfAbsent(prefix, p -> new HashSet<>());
+    Set<Neighbor> set = _sent.computeIfAbsent(prefix, p -> Sets.newConcurrentHashSet());
     set.add(new Neighbor(neighborHostname, neighborIp, neighborVrf, exportPolicy));
   }
 
@@ -134,7 +136,7 @@ void installed(
       Ip neighborIp,
       String neighborVrf,
       @Nullable String importPolicy) {
-    Set<Neighbor> set = _installed.computeIfAbsent(prefix, p -> new HashSet<>());
+    Set<Neighbor> set = _installed.computeIfAbsent(prefix, p -> Sets.newConcurrentHashSet());
     set.add(new Neighbor(neighborHostname, neighborIp, neighborVrf, importPolicy));
   }
 
@@ -152,10 +154,12 @@ void filtered(
       @Nullable String policyName,
       Direction direction) {
     if (direction == Direction.IN) {
-      Set<Neighbor> set = _filteredOnImport.computeIfAbsent(prefix, p -> new HashSet<>());
+      Set<Neighbor> set =
+          _filteredOnImport.computeIfAbsent(prefix, p -> Sets.newConcurrentHashSet());
       set.add(new Neighbor(neighborHostname, neighborIp, neighborVrf, policyName));
     } else if (direction == Direction.OUT) {
-      Set<Neighbor> set = _filteredOnExport.computeIfAbsent(prefix, p -> new HashSet<>());
+      Set<Neighbor> set =
+          _filteredOnExport.computeIfAbsent(prefix, p -> Sets.newConcurrentHashSet());
       set.add(new Neighbor(neighborHostname, neighborIp, neighborVrf, policyName));
     } else {
       throw new UnsupportedOperationException("Unknown filtering direction");