Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

bdp: remove queues for sending BGP messages #6226

Merged
merged 2 commits into from
Sep 22, 2020
Merged

bdp: remove queues for sending BGP messages #6226

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bd88892
bdp: remove queues for sending BGP messages (#6217)
anothermattbrown Sep 19, 2020
d5ac678
make PrefixTracer thread-safe
anothermattbrown Sep 21, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
10 changes: 10 additions & 0 deletions projects/batfish/src/main/java/org/batfish/dataplane/ibdp/BgpRoutingProcess.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -1067,6 +1073,10 @@ int iterationHashCode() {
// RIBs
_bgpv4Rib.getTypedRoutes(),
_evpnRib.getTypedRoutes(),
// Outgoing RIB deltas
_ebgpDelta,
_bgpv4Delta,
_mainRibBgpv4RouteDelta,
// Message queues
_bgpv4IncomingRoutes,
_evpnType3IncomingRoutes,
Expand Down
59 changes: 40 additions & 19 deletions projects/batfish/src/main/java/org/batfish/dataplane/ibdp/IncrementalBdpEngine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -427,7 +429,6 @@ private static void computeDependentRoutesIteration(
}

computeIterationOfBgpRoutes(
nodes,
iterationLabel,
allNodes,
topologyContext.getBgpTopology(),
Expand All @@ -454,7 +455,6 @@ private static void computeDependentRoutesIteration(
}

private static void computeIterationOfBgpRoutes(
Map<String, Node> nodes,
String iterationLabel,
Map<String, Node> allNodes,
BgpTopology bgpTopology,
Expand All @@ -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(
Expand All @@ -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(
Expand All @@ -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(
Expand Down Expand Up @@ -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 {
Expand Down
22 changes: 13 additions & 9 deletions projects/batfish/src/main/java/org/batfish/dataplane/ibdp/PrefixTracer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 */
Expand All @@ -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));
}

Expand All @@ -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));
}

Expand All @@ -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");
Expand Down