| // Copyright 2014 The Bazel Authors. All rights reserved. |
| // |
| // Licensed under the Apache License, Version 2.0 (the "License"); |
| // you may not use this file except in compliance with the License. |
| // You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, software |
| // distributed under the License is distributed on an "AS IS" BASIS, |
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| // See the License for the specific language governing permissions and |
| // limitations under the License. |
| |
| package com.google.devtools.build.lib.query2.query; |
| |
| import com.google.common.base.Throwables; |
| import com.google.common.collect.ImmutableList; |
| import com.google.common.collect.ImmutableSet; |
| import com.google.common.util.concurrent.ThreadFactoryBuilder; |
| import com.google.devtools.build.lib.cmdline.Label; |
| import com.google.devtools.build.lib.concurrent.AbstractQueueVisitor; |
| import com.google.devtools.build.lib.concurrent.ErrorClassifier; |
| import com.google.devtools.build.lib.concurrent.NamedForkJoinPool; |
| import com.google.devtools.build.lib.concurrent.QuiescingExecutor; |
| import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadSafe; |
| import com.google.devtools.build.lib.events.ErrorSensingEventHandler; |
| import com.google.devtools.build.lib.events.ExtendedEventHandler; |
| import com.google.devtools.build.lib.packages.Aspect; |
| import com.google.devtools.build.lib.packages.AspectDefinition; |
| import com.google.devtools.build.lib.packages.Attribute; |
| import com.google.devtools.build.lib.packages.DependencyFilter; |
| import com.google.devtools.build.lib.packages.LabelVisitationUtils; |
| import com.google.devtools.build.lib.packages.NoSuchThingException; |
| import com.google.devtools.build.lib.packages.OutputFile; |
| import com.google.devtools.build.lib.packages.Rule; |
| import com.google.devtools.build.lib.packages.Target; |
| import com.google.devtools.build.lib.pkgcache.TargetEdgeObserver; |
| import com.google.devtools.build.lib.pkgcache.TargetProvider; |
| import java.util.Set; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| |
| /** |
| * Visit the transitive closure of a label. Primarily used to "fault in" packages to the |
| * packageProvider and ensure the necessary targets exists, in advance of the configuration step, |
| * which is intolerant of missing packages/targets. |
| * |
| * <p>LabelVisitor loads packages concurrently where possible, to increase I/O parallelism. However, |
| * the public interface is not thread-safe: calls to public methods should not be made concurrently. |
| * |
| * <p>LabelVisitor is stateful: It remembers the previous visitation and can check its validity on |
| * subsequent calls to sync() instead of doing the normal visitation. |
| * |
| * <p>TODO(bazel-team): (2009) a small further optimization could be achieved if we create tasks at |
| * the package (not individual label) level, since package loading is the expensive step. This would |
| * require additional bookkeeping to maintain the list of labels that we need to visit once a |
| * package becomes available. Profiling suggests that there is still a potential benefit to be |
| * gained: when the set of packages is known a-priori, loading a set of packages that took 20 |
| * seconds can be done under 5 in the sequential case or 7 in the current (parallel) case. |
| * |
| * <h4>Concurrency</h4> |
| * |
| * <p>The sync() methods of this class is thread-compatible. The accessor ({@link #hasVisited} and |
| * similar must not be called until the concurrent phase is over, i.e. all external calls to visit() |
| * methods have completed. |
| */ |
| final class LabelVisitor { |
| private static final VisitationAttributes NONE = new VisitationAttributes(ImmutableSet.of(), 0); |
| |
| /** Attributes of a visitation which determine whether it is up-to-date or not. */ |
| private static class VisitationAttributes { |
| private final Set<Target> targetsToVisit; |
| private final int maxDepth; |
| private boolean success; |
| |
| VisitationAttributes(Set<Target> targetsToVisit, int maxDepth) { |
| this.targetsToVisit = targetsToVisit; |
| this.maxDepth = maxDepth; |
| } |
| |
| /** Returns true if and only if this visitation attribute is still up-to-date. */ |
| boolean current(VisitationAttributes lastVisitation) { |
| return targetsToVisit.equals(lastVisitation.targetsToVisit) |
| && maxDepth <= lastVisitation.maxDepth; |
| } |
| } |
| |
| /* |
| * Interrupts during the loading phase =================================== |
| * |
| * Bazel can be interrupted in the middle of the loading phase. The mechanics |
| * of this are far from trivial, so there is an explanation of how they are |
| * supposed to work. For a description how the same thing works in the |
| * execution phase, see ParallelBuilder.java . |
| * |
| * The sequence of events that happen when the user presses Ctrl-C is the |
| * following: |
| * |
| * 1. A SIGINT gets delivered to the Bazel client process. |
| * |
| * 2. The client process delivers the SIGINT to the server process. |
| * |
| * 3. The interruption state of the main thread is set to true. |
| * |
| * 4. Sooner or later, this results in an InterruptedException being thrown. |
| * Usually this takes place because the main thread is interrupted during |
| * AbstractQueueVisitor.awaitTermination(). The only exception to this is when |
| * the interruption occurs during the loading of a package of a label |
| * specified on the command line; in this case, the InterruptedException is |
| * thrown during the loading of an individual package (see below where this |
| * can occur) |
| * |
| * 5. The main thread calls ThreadPoolExecutor.shutdown(), which in turn |
| * interrupts every worker thread. Then the main thread waits for their |
| * termination. |
| * |
| * 6. An InterruptedException is thrown during the loading of an individual |
| * package in the worker threads. |
| * |
| * 7. All worker threads terminate. |
| * |
| * 8. An InterruptedException is thrown from |
| * AbstractQueueVisitor.awaitTermination() |
| * |
| * 9. This exception causes the execution of the currently running command to |
| * terminate prematurely. |
| * |
| * The interruption of the loading of an individual package happens as follow: |
| * |
| * 1. We periodically check the interruption state of the thread in |
| * UnixGlob.reallyGlob(). If it is interrupted, an InterruptedException is |
| * thrown. |
| * |
| * 2. The stack is unwound until we are out of the part of the call stack |
| * responsible for package loading. This either means that the worker thread |
| * terminates or that the label parsing terminates if the package that is |
| * being loaded was specified on the command line. |
| */ |
| private final TargetProvider targetProvider; |
| private final DependencyFilter edgeFilter; |
| private final ConcurrentMap<Label, Integer> visitedTargets = new ConcurrentHashMap<>(); |
| |
| private VisitationAttributes lastVisitation; |
| |
| /** Constant for limiting the permitted depth of recursion. */ |
| private static final int RECURSION_LIMIT = 100; |
| |
| /** |
| * Construct a LabelVisitor. |
| * |
| * @param targetProvider how to resolve labels to targets |
| * @param edgeFilter which edges may be traversed |
| */ |
| public LabelVisitor(TargetProvider targetProvider, DependencyFilter edgeFilter) { |
| this.targetProvider = targetProvider; |
| this.lastVisitation = NONE; |
| this.edgeFilter = edgeFilter; |
| } |
| |
| public void syncWithVisitor( |
| ExtendedEventHandler eventHandler, |
| Set<Target> targetsToVisit, |
| boolean keepGoing, |
| int parallelThreads, |
| int maxDepth, |
| TargetEdgeObserver... observers) |
| throws InterruptedException { |
| VisitationAttributes nextVisitation = new VisitationAttributes(targetsToVisit, maxDepth); |
| if (!lastVisitation.success || !nextVisitation.current(lastVisitation)) { |
| lastVisitation = nextVisitation; |
| lastVisitation.success = |
| redoVisitation( |
| eventHandler, targetsToVisit, keepGoing, parallelThreads, maxDepth, observers); |
| } |
| } |
| |
| /** |
| * Performs a bounded transitive closure visitation, similar to syncWithVisitor, but does not |
| * cache the result. |
| */ |
| public void syncUncached( |
| ErrorSensingEventHandler eventHandler, |
| Iterable<Target> targetsToVisit, |
| boolean keepGoing, |
| int parallelThreads, |
| int maxDepth, |
| TargetEdgeObserver... observers) |
| throws InterruptedException { |
| lastVisitation = NONE; |
| redoVisitation(eventHandler, targetsToVisit, keepGoing, parallelThreads, maxDepth, observers); |
| } |
| |
| // Does a bounded transitive visitation starting at the given top-level targets. |
| private boolean redoVisitation( |
| ExtendedEventHandler eventHandler, |
| Iterable<Target> targetsToVisit, |
| boolean keepGoing, |
| int parallelThreads, |
| int maxDepth, |
| TargetEdgeObserver... observers) |
| throws InterruptedException { |
| visitedTargets.clear(); |
| |
| Visitor visitor = new Visitor(eventHandler, keepGoing, parallelThreads, maxDepth, observers); |
| |
| Throwable uncaught = null; |
| boolean result; |
| try { |
| visitor.visitTargets(targetsToVisit); |
| } catch (Throwable t) { |
| visitor.stopNewActions(); |
| uncaught = t; |
| } finally { |
| // Run finish() in finally block to ensure we don't leak threads on exceptions. |
| result = visitor.finish(); |
| } |
| Throwables.propagateIfPossible(uncaught); |
| return result; |
| } |
| |
| public boolean hasVisited(Label target) { |
| return visitedTargets.containsKey(target); |
| } |
| |
| private class Visitor { |
| private static final String THREAD_NAME = "LabelVisitor"; |
| |
| private final ExecutorService executorService; |
| private final QuiescingExecutor executor; |
| private final ExtendedEventHandler eventHandler; |
| private final int maxDepth; |
| private final Iterable<TargetEdgeObserver> observers; |
| private final TargetEdgeErrorObserver errorObserver; |
| |
| Visitor( |
| ExtendedEventHandler eventHandler, |
| boolean keepGoing, |
| int parallelThreads, |
| int maxDepth, |
| TargetEdgeObserver... observers) { |
| if (parallelThreads > 1) { |
| this.executorService = NamedForkJoinPool.newNamedPool(THREAD_NAME, parallelThreads); |
| } else { |
| // ForkJoinPool has a bug where it deadlocks with parallelism=1, so use a |
| // SingleThreadExecutor instead. |
| this.executorService = |
| Executors.newSingleThreadExecutor( |
| new ThreadFactoryBuilder().setNameFormat(THREAD_NAME + " %d").build()); |
| } |
| this.executor = |
| AbstractQueueVisitor.createWithExecutorService( |
| executorService, /*failFastOnException=*/ !keepGoing, ErrorClassifier.DEFAULT); |
| this.eventHandler = eventHandler; |
| this.maxDepth = maxDepth; |
| this.errorObserver = new TargetEdgeErrorObserver(); |
| ImmutableList.Builder<TargetEdgeObserver> builder = ImmutableList.builder(); |
| builder.add(observers); |
| builder.add(errorObserver); |
| this.observers = builder.build(); |
| } |
| |
| /** |
| * Visit the specified labels and follow the transitive closure of their outbound dependencies. |
| * |
| * @param targets the targets to visit |
| */ |
| @ThreadSafe |
| public void visitTargets(Iterable<Target> targets) throws InterruptedException { |
| for (Target target : targets) { |
| visit(null, null, target, 0, 0); |
| } |
| } |
| |
| @ThreadSafe |
| public boolean finish() throws InterruptedException { |
| executor.awaitQuiescence(/*interruptWorkers=*/ true); |
| return !errorObserver.hasErrors(); |
| } |
| |
| public void stopNewActions() { |
| executorService.shutdownNow(); |
| } |
| |
| private void enqueueTarget( |
| final Target from, |
| final Attribute attr, |
| final Label label, |
| final int depth, |
| final int count) { |
| // Don't perform the targetProvider lookup if at the maximum depth already. |
| if (depth >= maxDepth) { |
| return; |
| } |
| |
| // Avoid thread-related overhead when not crossing packages. |
| // Can start a new thread when count reaches 100, to prevent infinite recursion. |
| if (from != null |
| && from.getLabel().getPackageFragment().equals(label.getPackageFragment()) |
| && count < RECURSION_LIMIT) { |
| newVisitRunnable(from, attr, label, depth, count + 1).run(); |
| } else { |
| executor.execute(newVisitRunnable(from, attr, label, depth, 0)); |
| } |
| } |
| |
| private Runnable newVisitRunnable( |
| final Target from, |
| final Attribute attr, |
| final Label label, |
| final int depth, |
| final int count) { |
| return () -> { |
| try { |
| try { |
| visit(from, attr, targetProvider.getTarget(eventHandler, label), depth + 1, count); |
| } catch (NoSuchThingException e) { |
| observeError(from, label, e); |
| } |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| }; |
| } |
| |
| /** |
| * Visits the target and its package. |
| * |
| * <p>Potentially blocking invocations into the package cache are enqueued in the worker pool if |
| * CONCURRENT. |
| */ |
| private void visit(Target from, Attribute attribute, final Target target, int depth, int count) |
| throws InterruptedException { |
| if (target == null) { |
| throw new NullPointerException( |
| String.format( |
| "'%s' attribute '%s'", |
| from == null ? "(null)" : from.getLabel().toString(), |
| attribute == null ? "(null)" : attribute.getName())); |
| } |
| if (depth > maxDepth) { |
| return; |
| } |
| |
| if (from != null) { |
| observeEdge(from, attribute, target); |
| visitAspectsIfRequired(from, attribute, target, depth, count); |
| } |
| visitTargetNode(target, depth, count); |
| } |
| |
| private void visitAspectsIfRequired( |
| Target from, Attribute attribute, final Target to, int depth, int count) { |
| // TODO(bazel-team): The getAspects call below is duplicate work for each direct dep entailed |
| // by an attribute's value. Additionally, we might end up enqueueing the same exact visitation |
| // multiple times: consider the case where the same direct dependency is entailed by aspects |
| // of *different* attributes. These visitations get culled later, but we still have to pay the |
| // overhead for all that. |
| |
| if (!(from instanceof Rule) || !(to instanceof Rule)) { |
| return; |
| } |
| Rule fromRule = (Rule) from; |
| Rule toRule = (Rule) to; |
| for (Aspect aspect : attribute.getAspects(fromRule)) { |
| if (AspectDefinition.satisfies( |
| aspect, toRule.getRuleClassObject().getAdvertisedProviders())) { |
| AspectDefinition.forEachLabelDepFromAllAttributesOfAspect( |
| fromRule, |
| aspect, |
| edgeFilter, |
| (aspectAttribute, aspectLabel) -> |
| enqueueTarget(from, aspectAttribute, aspectLabel, depth, count)); |
| } |
| } |
| } |
| |
| /** |
| * Visit the specified target. Called in a worker thread if CONCURRENT. |
| * |
| * @param target the target to visit |
| */ |
| private void visitTargetNode(Target target, int depth, int count) throws InterruptedException { |
| Integer minTargetDepth = visitedTargets.putIfAbsent(target.getLabel(), depth); |
| if (minTargetDepth != null) { |
| // The target was already visited at a greater depth. |
| // The closure we are about to build is therefore a subset of what |
| // has already been built, and we can skip it. |
| // Also special case MAX_VALUE, where we never want to revisit targets. |
| // (This avoids loading phase overhead outside of queries). |
| if (maxDepth == Integer.MAX_VALUE || minTargetDepth <= depth) { |
| return; |
| } |
| // Check again in case it was overwritten by another thread. |
| synchronized (visitedTargets) { |
| if (visitedTargets.get(target.getLabel()) <= depth) { |
| return; |
| } |
| visitedTargets.put(target.getLabel(), depth); |
| } |
| } |
| |
| observeNode(target); |
| |
| // LabelVisitor has some legacy special handling of OutputFiles. |
| if (target instanceof OutputFile) { |
| Rule rule = ((OutputFile) target).getGeneratingRule(); |
| observeEdge(target, null, rule); |
| visit(null, null, rule, depth + 1, count + 1); |
| } |
| |
| LabelVisitationUtils.visitTargetExceptionally( |
| target, |
| edgeFilter, |
| (fromTarget, attribute, toLabel) -> |
| enqueueTarget(target, attribute, toLabel, depth, count)); |
| } |
| |
| private void observeEdge(Target from, Attribute attribute, Target to) { |
| for (TargetEdgeObserver observer : observers) { |
| observer.edge(from, attribute, to); |
| } |
| } |
| |
| private void observeNode(Target target) { |
| for (TargetEdgeObserver observer : observers) { |
| observer.node(target); |
| } |
| } |
| |
| private void observeError(Target from, Label label, NoSuchThingException e) |
| throws InterruptedException { |
| for (TargetEdgeObserver observer : observers) { |
| observer.missingEdge(from, label, e); |
| } |
| } |
| } |
| } |