Add a mechanism for bounding the number of Packages SkyQueryEnvironment's expensive parallel operations can operate on at once.
--
MOS_MIGRATED_REVID=138779172
diff --git a/src/main/java/com/google/devtools/build/lib/query2/ParallelSkyQueryUtils.java b/src/main/java/com/google/devtools/build/lib/query2/ParallelSkyQueryUtils.java
index ac3e2a1..5f7447b 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/ParallelSkyQueryUtils.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/ParallelSkyQueryUtils.java
@@ -18,12 +18,15 @@
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Collections2;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
+import com.google.common.collect.Multimap;
import com.google.devtools.build.lib.cmdline.Label;
import com.google.devtools.build.lib.cmdline.PackageIdentifier;
import com.google.devtools.build.lib.collect.CompactHashSet;
import com.google.devtools.build.lib.concurrent.MoreFutures;
+import com.google.devtools.build.lib.concurrent.MultisetSemaphore;
import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadSafe;
import com.google.devtools.build.lib.packages.Target;
import com.google.devtools.build.lib.query2.engine.Callback;
@@ -69,13 +72,14 @@
QueryExpression expression,
VariableContext<Target> context,
ThreadSafeCallback<Target> callback,
- ForkJoinPool forkJoinPool)
+ ForkJoinPool forkJoinPool,
+ MultisetSemaphore<PackageIdentifier> packageSemaphore)
throws QueryException, InterruptedException {
env.eval(
expression,
context,
new SkyKeyBFSVisitorCallback(
- new AllRdepsUnboundedVisitor.Factory(env, callback, forkJoinPool)));
+ new AllRdepsUnboundedVisitor.Factory(env, callback, forkJoinPool, packageSemaphore)));
}
/** Specialized parallel variant of {@link SkyQueryEnvironment#getRBuildFiles}. */
@@ -83,24 +87,29 @@
SkyQueryEnvironment env,
Collection<PathFragment> fileIdentifiers,
ThreadSafeCallback<Target> callback,
- ForkJoinPool forkJoinPool)
+ ForkJoinPool forkJoinPool,
+ MultisetSemaphore<PackageIdentifier> packageSemaphore)
throws QueryException, InterruptedException {
ThreadSafeUniquifier<SkyKey> keyUniquifier = env.createSkyKeyUniquifier();
- RBuildFilesVisitor visitor = new RBuildFilesVisitor(env, forkJoinPool, keyUniquifier, callback);
+ RBuildFilesVisitor visitor =
+ new RBuildFilesVisitor(env, forkJoinPool, keyUniquifier, callback, packageSemaphore);
visitor.visitAndWaitForCompletion(env.getSkyKeysForFileFragments(fileIdentifiers));
}
/** A helper class that computes 'rbuildfiles(<blah>)' via BFS. */
private static class RBuildFilesVisitor extends AbstractSkyKeyBFSVisitor<SkyKey> {
private final SkyQueryEnvironment env;
+ private final MultisetSemaphore<PackageIdentifier> packageSemaphore;
private RBuildFilesVisitor(
SkyQueryEnvironment env,
ForkJoinPool forkJoinPool,
ThreadSafeUniquifier<SkyKey> uniquifier,
- Callback<Target> callback) {
+ Callback<Target> callback,
+ MultisetSemaphore<PackageIdentifier> packageSemaphore) {
super(forkJoinPool, uniquifier, callback);
this.env = env;
+ this.packageSemaphore = packageSemaphore;
}
@Override
@@ -125,10 +134,21 @@
}
@Override
- protected Iterable<Target> getTargetsToAddToResult(Iterable<SkyKey> keysToUseForResult)
- throws InterruptedException {
- return SkyQueryEnvironment.getBuildFilesForPackageValues(
- env.graph.getSuccessfulValues(keysToUseForResult).values());
+ protected void processResultantTargets(
+ Iterable<SkyKey> keysToUseForResult, Callback<Target> callback)
+ throws QueryException, InterruptedException {
+ Set<PackageIdentifier> pkgIdsNeededForResult =
+ ImmutableSet.copyOf(
+ Iterables.transform(
+ keysToUseForResult,
+ SkyQueryEnvironment.PACKAGE_SKYKEY_TO_PACKAGE_IDENTIFIER));
+ packageSemaphore.acquireAll(pkgIdsNeededForResult);
+ try {
+ callback.process(SkyQueryEnvironment.getBuildFilesForPackageValues(
+ env.graph.getSuccessfulValues(keysToUseForResult).values()));
+ } finally {
+ packageSemaphore.releaseAll(pkgIdsNeededForResult);
+ }
}
@Override
@@ -152,14 +172,17 @@
private static class AllRdepsUnboundedVisitor
extends AbstractSkyKeyBFSVisitor<Pair<SkyKey, SkyKey>> {
private final SkyQueryEnvironment env;
+ private final MultisetSemaphore<PackageIdentifier> packageSemaphore;
private AllRdepsUnboundedVisitor(
SkyQueryEnvironment env,
ForkJoinPool forkJoinPool,
ThreadSafeUniquifier<Pair<SkyKey, SkyKey>> uniquifier,
- ThreadSafeCallback<Target> callback) {
+ ThreadSafeCallback<Target> callback,
+ MultisetSemaphore<PackageIdentifier> packageSemaphore) {
super(forkJoinPool, uniquifier, callback);
this.env = env;
+ this.packageSemaphore = packageSemaphore;
}
/**
@@ -174,20 +197,24 @@
private final ForkJoinPool forkJoinPool;
private final ThreadSafeUniquifier<Pair<SkyKey, SkyKey>> uniquifier;
private final ThreadSafeCallback<Target> callback;
+ private final MultisetSemaphore<PackageIdentifier> packageSemaphore;
private Factory(
SkyQueryEnvironment env,
ThreadSafeCallback<Target> callback,
- ForkJoinPool forkJoinPool) {
+ ForkJoinPool forkJoinPool,
+ MultisetSemaphore<PackageIdentifier> packageSemaphore) {
this.env = env;
this.forkJoinPool = forkJoinPool;
this.uniquifier = env.createReverseDepSkyKeyUniquifier();
this.callback = callback;
+ this.packageSemaphore = packageSemaphore;
}
@Override
public AbstractSkyKeyBFSVisitor<Pair<SkyKey, SkyKey>> create() {
- return new AllRdepsUnboundedVisitor(env, forkJoinPool, uniquifier, callback);
+ return new AllRdepsUnboundedVisitor(
+ env, forkJoinPool, uniquifier, callback, packageSemaphore);
}
}
@@ -213,13 +240,27 @@
reverseDepsMap.get(reverseDepPair.first).add(reverseDepPair.second);
}
- // Filter out disallowed deps. We cannot defer the targetification any further as we do not
- // want to retrieve the rdeps of unwanted nodes (targets).
- if (!reverseDepsMap.isEmpty()) {
- Collection<Target> filteredTargets =
- env.filterRawReverseDepsOfTransitiveTraversalKeys(reverseDepsMap);
- filteredKeys.addAll(
- Collections2.transform(filteredTargets, SkyQueryEnvironment.TARGET_TO_SKY_KEY));
+ Multimap<SkyKey, SkyKey> packageKeyToTargetKeyMap =
+ env.makePackageKeyToTargetKeyMap(Iterables.concat(reverseDepsMap.values()));
+ Set<PackageIdentifier> pkgIdsNeededForTargetification =
+ ImmutableSet.copyOf(
+ Iterables.transform(
+ packageKeyToTargetKeyMap.keySet(),
+ SkyQueryEnvironment.PACKAGE_SKYKEY_TO_PACKAGE_IDENTIFIER));
+ packageSemaphore.acquireAll(pkgIdsNeededForTargetification);
+
+ try {
+ // Filter out disallowed deps. We cannot defer the targetification any further as we do not
+ // want to retrieve the rdeps of unwanted nodes (targets).
+ if (!reverseDepsMap.isEmpty()) {
+ Collection<Target> filteredTargets =
+ env.filterRawReverseDepsOfTransitiveTraversalKeys(
+ reverseDepsMap, packageKeyToTargetKeyMap);
+ filteredKeys.addAll(
+ Collections2.transform(filteredTargets, SkyQueryEnvironment.TARGET_TO_SKY_KEY));
+ }
+ } finally {
+ packageSemaphore.releaseAll(pkgIdsNeededForTargetification);
}
// Retrieve the reverse deps as SkyKeys and defer the targetification and filtering to next
@@ -252,9 +293,23 @@
}
@Override
- protected Iterable<Target> getTargetsToAddToResult(Iterable<SkyKey> keysToUseForResult)
- throws InterruptedException {
- return env.makeTargetsFromSkyKeys(keysToUseForResult).values();
+ protected void processResultantTargets(
+ Iterable<SkyKey> keysToUseForResult, Callback<Target> callback)
+ throws QueryException, InterruptedException {
+ Multimap<SkyKey, SkyKey> packageKeyToTargetKeyMap =
+ env.makePackageKeyToTargetKeyMap(keysToUseForResult);
+ Set<PackageIdentifier> pkgIdsNeededForResult =
+ ImmutableSet.copyOf(
+ Iterables.transform(
+ packageKeyToTargetKeyMap.keySet(),
+ SkyQueryEnvironment.PACKAGE_SKYKEY_TO_PACKAGE_IDENTIFIER));
+ packageSemaphore.acquireAll(pkgIdsNeededForResult);
+ try {
+ callback.process(
+ env.makeTargetsFromPackageKeyToTargetKeyMap(packageKeyToTargetKeyMap).values());
+ } finally {
+ packageSemaphore.releaseAll(pkgIdsNeededForResult);
+ }
}
@Override
@@ -294,7 +349,7 @@
/**
* A helper class for performing a custom BFS visitation on the Skyframe graph, using {@link
- * ForkJoinQuiescingExecutor}.
+ * ForkJoinPool}.
*
* <p>The choice of {@link ForkJoinPool} over, say, AbstractQueueVisitor backed by a
* ThreadPoolExecutor, is very deliberate. {@link SkyKeyBFSVisitorCallback#process} kicks off a
@@ -310,7 +365,9 @@
private static final int VISIT_BATCH_SIZE = 10000;
private AbstractSkyKeyBFSVisitor(
- ForkJoinPool forkJoinPool, ThreadSafeUniquifier<T> uniquifier, Callback<Target> callback) {
+ ForkJoinPool forkJoinPool,
+ ThreadSafeUniquifier<T> uniquifier,
+ Callback<Target> callback) {
this.forkJoinPool = forkJoinPool;
this.uniquifier = uniquifier;
this.callback = callback;
@@ -402,7 +459,7 @@
@Override
protected void computeImpl() throws QueryException, InterruptedException {
- callback.process(getTargetsToAddToResult(keysToUseForResult));
+ processResultantTargets(keysToUseForResult, callback);
}
}
@@ -425,11 +482,12 @@
}
/**
- * Gets the given {@code keysToUseForResult}'s contribution to the set of {@link Target}s in the
- * full visitation.
+ * Forwards the given {@code keysToUseForResult}'s contribution to the set of {@link Target}s
+ * in the full visitation to the given {@link Callback}.
*/
- protected abstract Iterable<Target> getTargetsToAddToResult(
- Iterable<SkyKey> keysToUseForResult) throws InterruptedException;
+ protected abstract void processResultantTargets(
+ Iterable<SkyKey> keysToUseForResult, Callback<Target> callback)
+ throws QueryException, InterruptedException;
/** Gets the {@link Visit} representing the local visitation of the given {@code values}. */
protected abstract Visit getVisitResult(Iterable<T> values) throws InterruptedException;
diff --git a/src/main/java/com/google/devtools/build/lib/query2/SkyQueryEnvironment.java b/src/main/java/com/google/devtools/build/lib/query2/SkyQueryEnvironment.java
index c5efe91a..a3f7c6d 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/SkyQueryEnvironment.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/SkyQueryEnvironment.java
@@ -34,6 +34,7 @@
import com.google.devtools.build.lib.cmdline.TargetParsingException;
import com.google.devtools.build.lib.cmdline.TargetPattern;
import com.google.devtools.build.lib.collect.CompactHashSet;
+import com.google.devtools.build.lib.concurrent.MultisetSemaphore;
import com.google.devtools.build.lib.concurrent.NamedForkJoinPool;
import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadSafe;
import com.google.devtools.build.lib.events.Event;
@@ -132,6 +133,7 @@
private final int queryEvaluationParallelismLevel;
// The following fields are set in the #beforeEvaluateQuery method.
+ private MultisetSemaphore<PackageIdentifier> packageSemaphore;
protected WalkableGraph graph;
private InterruptibleSupplier<ImmutableSet<PathFragment>> blacklistPatternsSupplier;
private ForkJoinPool forkJoinPool;
@@ -205,6 +207,7 @@
}
checkEvaluationResult(result);
+ packageSemaphore = makeFreshPackageMultisetSemaphore();
graph = result.getWalkableGraph();
blacklistPatternsSupplier = InterruptibleSupplier.Memoize.of(new BlacklistSupplier(graph));
@@ -220,7 +223,17 @@
graphBackedRecursivePackageProvider,
eventHandler,
TargetPatternEvaluator.DEFAULT_FILTERING_POLICY,
- forkJoinPool);
+ forkJoinPool,
+ packageSemaphore);
+ }
+
+ protected MultisetSemaphore<PackageIdentifier> makeFreshPackageMultisetSemaphore() {
+ return MultisetSemaphore.unbounded();
+ }
+
+ @ThreadSafe
+ public MultisetSemaphore<PackageIdentifier> getPackageMultisetSemaphore() {
+ return packageSemaphore;
}
/**
@@ -342,10 +355,18 @@
private Map<SkyKey, Collection<Target>> targetifyValues(
Map<SkyKey, ? extends Iterable<SkyKey>> input) throws InterruptedException {
+ return targetifyValues(
+ input,
+ makePackageKeyToTargetKeyMap(ImmutableSet.copyOf(Iterables.concat(input.values()))));
+ }
+
+ private Map<SkyKey, Collection<Target>> targetifyValues(
+ Map<SkyKey, ? extends Iterable<SkyKey>> input,
+ Multimap<SkyKey, SkyKey> packageKeyToTargetKeyMap) throws InterruptedException {
ImmutableMap.Builder<SkyKey, Collection<Target>> result = ImmutableMap.builder();
Map<SkyKey, Target> allTargets =
- makeTargetsFromSkyKeys(Sets.newHashSet(Iterables.concat(input.values())));
+ makeTargetsFromPackageKeyToTargetKeyMap(packageKeyToTargetKeyMap);
for (Map.Entry<SkyKey, ? extends Iterable<SkyKey>> entry : input.entrySet()) {
Iterable<SkyKey> skyKeys = entry.getValue();
@@ -446,8 +467,9 @@
/** Targetify SkyKeys of reverse deps and filter out targets whose deps are not allowed. */
Collection<Target> filterRawReverseDepsOfTransitiveTraversalKeys(
- Map<SkyKey, ? extends Iterable<SkyKey>> rawReverseDeps) throws InterruptedException {
- return processRawReverseDeps(targetifyValues(rawReverseDeps));
+ Map<SkyKey, ? extends Iterable<SkyKey>> rawReverseDeps,
+ Multimap<SkyKey, SkyKey> packageKeyToTargetKeyMap) throws InterruptedException {
+ return processRawReverseDeps(targetifyValues(rawReverseDeps, packageKeyToTargetKeyMap));
}
private Collection<Target> processRawReverseDeps(Map<SkyKey, Collection<Target>> rawReverseDeps)
@@ -752,9 +774,16 @@
}
};
+ static final Function<SkyKey, PackageIdentifier> PACKAGE_SKYKEY_TO_PACKAGE_IDENTIFIER =
+ new Function<SkyKey, PackageIdentifier>() {
+ @Override
+ public PackageIdentifier apply(SkyKey skyKey) {
+ return (PackageIdentifier) skyKey.argument();
+ }
+ };
+
@ThreadSafe
- public Map<SkyKey, Target> makeTargetsFromSkyKeys(Iterable<SkyKey> keys)
- throws InterruptedException {
+ Multimap<SkyKey, SkyKey> makePackageKeyToTargetKeyMap(Iterable<SkyKey> keys) {
Multimap<SkyKey, SkyKey> packageKeyToTargetKeyMap = ArrayListMultimap.create();
for (SkyKey key : keys) {
Label label = SKYKEY_TO_LABEL.apply(key);
@@ -763,6 +792,18 @@
}
packageKeyToTargetKeyMap.put(PackageValue.key(label.getPackageIdentifier()), key);
}
+ return packageKeyToTargetKeyMap;
+ }
+
+ @ThreadSafe
+ public Map<SkyKey, Target> makeTargetsFromSkyKeys(Iterable<SkyKey> keys)
+ throws InterruptedException {
+ return makeTargetsFromPackageKeyToTargetKeyMap(makePackageKeyToTargetKeyMap(keys));
+ }
+
+ @ThreadSafe
+ public Map<SkyKey, Target> makeTargetsFromPackageKeyToTargetKeyMap(
+ Multimap<SkyKey, SkyKey> packageKeyToTargetKeyMap) throws InterruptedException {
ImmutableMap.Builder<SkyKey, Target> result = ImmutableMap.builder();
Map<SkyKey, SkyValue> packageMap = graph.getSuccessfulValues(packageKeyToTargetKeyMap.keySet());
for (Map.Entry<SkyKey, SkyValue> entry : packageMap.entrySet()) {
@@ -924,7 +965,8 @@
ThreadSafeCallback<Target> callback,
ForkJoinPool forkJoinPool)
throws QueryException, InterruptedException {
- ParallelSkyQueryUtils.getRBuildFilesParallel(this, fileIdentifiers, callback, forkJoinPool);
+ ParallelSkyQueryUtils.getRBuildFilesParallel(
+ this, fileIdentifiers, callback, forkJoinPool, packageSemaphore);
}
/**
@@ -1107,7 +1149,7 @@
ForkJoinPool forkJoinPool)
throws QueryException, InterruptedException {
ParallelSkyQueryUtils.getAllRdepsUnboundedParallel(
- this, expression, context, callback, forkJoinPool);
+ this, expression, context, callback, forkJoinPool, packageSemaphore);
}
@ThreadSafe
diff --git a/src/main/java/com/google/devtools/build/lib/skyframe/RecursivePackageProviderBackedTargetPatternResolver.java b/src/main/java/com/google/devtools/build/lib/skyframe/RecursivePackageProviderBackedTargetPatternResolver.java
index 62926c8..0071098 100644
--- a/src/main/java/com/google/devtools/build/lib/skyframe/RecursivePackageProviderBackedTargetPatternResolver.java
+++ b/src/main/java/com/google/devtools/build/lib/skyframe/RecursivePackageProviderBackedTargetPatternResolver.java
@@ -29,6 +29,7 @@
import com.google.devtools.build.lib.cmdline.TargetParsingException;
import com.google.devtools.build.lib.cmdline.TargetPatternResolver;
import com.google.devtools.build.lib.concurrent.MoreFutures;
+import com.google.devtools.build.lib.concurrent.MultisetSemaphore;
import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadCompatible;
import com.google.devtools.build.lib.events.Event;
import com.google.devtools.build.lib.events.EventHandler;
@@ -65,16 +66,19 @@
private final EventHandler eventHandler;
private final FilteringPolicy policy;
private final ExecutorService executor;
+ private final MultisetSemaphore<PackageIdentifier> packageSemaphore;
public RecursivePackageProviderBackedTargetPatternResolver(
RecursivePackageProvider recursivePackageProvider,
EventHandler eventHandler,
FilteringPolicy policy,
- ExecutorService executor) {
+ ExecutorService executor,
+ MultisetSemaphore<PackageIdentifier> packageSemaphore) {
this.recursivePackageProvider = recursivePackageProvider;
this.eventHandler = eventHandler;
this.policy = policy;
this.executor = executor;
+ this.packageSemaphore = packageSemaphore;
}
@Override
@@ -215,22 +219,28 @@
tasks.add(executor.submit(new Callable<Void>() {
@Override
public Void call() throws E, TargetParsingException, InterruptedException {
- Iterable<ResolvedTargets<Target>> resolvedTargets =
- bulkGetTargetsInPackage(originalPattern, pkgIdBatch, NO_FILTER).values();
- List<Target> filteredTargets = new ArrayList<>(calculateSize(resolvedTargets));
- for (ResolvedTargets<Target> targets : resolvedTargets) {
- for (Target target : targets.getTargets()) {
- // Perform the no-targets-found check before applying the filtering policy
- // so we only return the error if the input directory's subtree really
- // contains no targets.
- foundTarget.set(true);
- if (actualPolicy.shouldRetain(target, false)) {
- filteredTargets.add(target);
+ ImmutableSet<PackageIdentifier> pkgIdBatchSet = ImmutableSet.copyOf(pkgIdBatch);
+ packageSemaphore.acquireAll(pkgIdBatchSet);
+ try {
+ Iterable<ResolvedTargets<Target>> resolvedTargets =
+ bulkGetTargetsInPackage(originalPattern, pkgIdBatch, NO_FILTER).values();
+ List<Target> filteredTargets = new ArrayList<>(calculateSize(resolvedTargets));
+ for (ResolvedTargets<Target> targets : resolvedTargets) {
+ for (Target target : targets.getTargets()) {
+ // Perform the no-targets-found check before applying the filtering policy
+ // so we only return the error if the input directory's subtree really
+ // contains no targets.
+ foundTarget.set(true);
+ if (actualPolicy.shouldRetain(target, false)) {
+ filteredTargets.add(target);
+ }
}
}
- }
- synchronized (callbackLock) {
- callback.process(filteredTargets);
+ synchronized (callbackLock) {
+ callback.process(filteredTargets);
+ }
+ } finally {
+ packageSemaphore.releaseAll(pkgIdBatchSet);
}
return null;
}
diff --git a/src/main/java/com/google/devtools/build/lib/skyframe/TargetPatternFunction.java b/src/main/java/com/google/devtools/build/lib/skyframe/TargetPatternFunction.java
index 32f64e8..ac75806 100644
--- a/src/main/java/com/google/devtools/build/lib/skyframe/TargetPatternFunction.java
+++ b/src/main/java/com/google/devtools/build/lib/skyframe/TargetPatternFunction.java
@@ -17,10 +17,12 @@
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.devtools.build.lib.cmdline.Label;
+import com.google.devtools.build.lib.cmdline.PackageIdentifier;
import com.google.devtools.build.lib.cmdline.ResolvedTargets;
import com.google.devtools.build.lib.cmdline.TargetParsingException;
import com.google.devtools.build.lib.cmdline.TargetPattern;
import com.google.devtools.build.lib.collect.CompactHashSet;
+import com.google.devtools.build.lib.concurrent.MultisetSemaphore;
import com.google.devtools.build.lib.packages.Target;
import com.google.devtools.build.lib.skyframe.EnvironmentBackedRecursivePackageProvider.MissingDepException;
import com.google.devtools.build.lib.util.BatchCallback;
@@ -30,9 +32,7 @@
import com.google.devtools.build.skyframe.SkyFunctionException;
import com.google.devtools.build.skyframe.SkyKey;
import com.google.devtools.build.skyframe.SkyValue;
-
import java.util.Set;
-
import javax.annotation.Nullable;
/**
@@ -54,8 +54,12 @@
EnvironmentBackedRecursivePackageProvider provider =
new EnvironmentBackedRecursivePackageProvider(env);
RecursivePackageProviderBackedTargetPatternResolver resolver =
- new RecursivePackageProviderBackedTargetPatternResolver(provider, env.getListener(),
- patternKey.getPolicy(), MoreExecutors.newDirectExecutorService());
+ new RecursivePackageProviderBackedTargetPatternResolver(
+ provider,
+ env.getListener(),
+ patternKey.getPolicy(),
+ MoreExecutors.newDirectExecutorService(),
+ MultisetSemaphore.<PackageIdentifier>unbounded());
TargetPattern parsedPattern = patternKey.getParsedPattern();
ImmutableSet<PathFragment> excludedSubdirectories = patternKey.getExcludedSubdirectories();
final Set<Target> results = CompactHashSet.create();
