// Copyright 2016 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;

import static com.google.common.collect.ImmutableSet.toImmutableSet;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Streams;
import com.google.devtools.build.lib.cmdline.Label;
import com.google.devtools.build.lib.cmdline.PackageIdentifier;
import com.google.devtools.build.lib.collect.compacthashset.CompactHashSet;
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;
import com.google.devtools.build.lib.query2.engine.MinDepthUniquifier;
import com.google.devtools.build.lib.query2.engine.QueryEnvironment.QueryTaskFuture;
import com.google.devtools.build.lib.query2.engine.QueryEnvironment.ThreadSafeMutableSet;
import com.google.devtools.build.lib.query2.engine.QueryException;
import com.google.devtools.build.lib.query2.engine.QueryExpression;
import com.google.devtools.build.lib.query2.engine.QueryUtil;
import com.google.devtools.build.lib.query2.engine.QueryUtil.AggregateAllCallback;
import com.google.devtools.build.lib.query2.engine.QueryUtil.UniquifierImpl;
import com.google.devtools.build.lib.query2.engine.Uniquifier;
import com.google.devtools.build.lib.query2.engine.VariableContext;
import com.google.devtools.build.lib.skyframe.PackageValue;
import com.google.devtools.build.lib.skyframe.SkyFunctions;
import com.google.devtools.build.lib.vfs.PathFragment;
import com.google.devtools.build.skyframe.SkyKey;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import javax.annotation.Nullable;

/**
 * Parallel implementations of various functionality in {@link SkyQueryEnvironment}.
 *
 * <p>Special attention is given to memory usage. Naive parallel implementations of query
 * functionality would lead to memory blowup. Instead of dealing with {@link Target}s, we try to
 * deal with {@link SkyKey}s as much as possible to reduce the number of {@link Package}s forcibly
 * in memory at any given time.
 */
// TODO(bazel-team): Be more deliberate about bounding memory usage here.
public class ParallelSkyQueryUtils {

  /** The maximum number of keys to visit at once. */
  @VisibleForTesting static final int VISIT_BATCH_SIZE = 10000;

  private ParallelSkyQueryUtils() {
  }

  static QueryTaskFuture<Void> getAllRdepsUnboundedParallel(
      SkyQueryEnvironment env,
      QueryExpression expression,
      VariableContext<Target> context,
      Callback<Target> callback,
      MultisetSemaphore<PackageIdentifier> packageSemaphore) {
    return env.eval(
        expression,
        context,
        ParallelVisitor.createParallelVisitorCallback(
            new RdepsUnboundedVisitor.Factory(
                env,
                /*universe=*/ Predicates.alwaysTrue(),
                callback,
                packageSemaphore)));
  }

  static QueryTaskFuture<Void> getAllRdepsBoundedParallel(
      SkyQueryEnvironment env,
      QueryExpression expression,
      int depth,
      VariableContext<Target> context,
      Callback<Target> callback,
      MultisetSemaphore<PackageIdentifier> packageSemaphore) {
    return env.eval(
        expression,
        context,
        ParallelVisitor.createParallelVisitorCallback(
            new RdepsBoundedVisitor.Factory(
                env,
                depth,
                /*universe=*/ Predicates.alwaysTrue(),
                callback,
                packageSemaphore)));
  }

  static QueryTaskFuture<Void> getRdepsInUniverseUnboundedParallel(
      SkyQueryEnvironment env,
      QueryExpression expression,
      Predicate<SkyKey> universe,
      VariableContext<Target> context,
      Callback<Target> callback,
      MultisetSemaphore<PackageIdentifier> packageSemaphore) {
    return env.eval(
        expression,
        context,
        ParallelVisitor.createParallelVisitorCallback(
            new RdepsUnboundedVisitor.Factory(env, universe, callback, packageSemaphore)));
  }

  static QueryTaskFuture<Predicate<SkyKey>> getDTCSkyKeyPredicateFuture(
      SkyQueryEnvironment env,
      QueryExpression expression,
      VariableContext<Target> context,
      int processResultsBatchSize,
      int concurrencyLevel) {
    QueryTaskFuture<ThreadSafeMutableSet<Target>> universeValueFuture =
        QueryUtil.evalAll(env, context, expression);

    Function<ThreadSafeMutableSet<Target>, QueryTaskFuture<Predicate<SkyKey>>>
        getTransitiveClosureAsyncFunction =
        universeValue -> {
          ThreadSafeAggregateAllSkyKeysCallback aggregateAllCallback =
              new ThreadSafeAggregateAllSkyKeysCallback(concurrencyLevel);
          return env.executeAsync(
              () -> {
                Callback<Target> visitorCallback =
                    ParallelVisitor.createParallelVisitorCallback(
                        new TransitiveTraversalValueDTCVisitor.Factory(
                            env,
                            env.createSkyKeyUniquifier(),
                            processResultsBatchSize,
                            aggregateAllCallback));
                visitorCallback.process(universeValue);
                return Predicates.in(aggregateAllCallback.getResult());
              });
        };

    return env.transformAsync(universeValueFuture, getTransitiveClosureAsyncFunction);
  }

  static QueryTaskFuture<Void> getRdepsInUniverseBoundedParallel(
      SkyQueryEnvironment env,
      QueryExpression expression,
      int depth,
      Predicate<SkyKey> universe,
      VariableContext<Target> context,
      Callback<Target> callback,
      MultisetSemaphore<PackageIdentifier> packageSemaphore) {
    return env.eval(
        expression,
        context,
        ParallelVisitor.createParallelVisitorCallback(
            new RdepsBoundedVisitor.Factory(
                env,
                depth,
                universe,
                callback,
                packageSemaphore)));
  }

  /** Specialized parallel variant of {@link SkyQueryEnvironment#getRBuildFiles}. */
  static void getRBuildFilesParallel(
      SkyQueryEnvironment env,
      Collection<PathFragment> fileIdentifiers,
      Callback<Target> callback) throws QueryException, InterruptedException {
    Uniquifier<SkyKey> keyUniquifier = env.createSkyKeyUniquifier();
    RBuildFilesVisitor visitor =
        new RBuildFilesVisitor(env, keyUniquifier, callback);
    visitor.visitAndWaitForCompletion(env.getFileStateKeysForFileFragments(fileIdentifiers));
  }

  /** A {@link ParallelVisitor} whose visitations occur on {@link SkyKey}s. */
  public abstract static class AbstractSkyKeyParallelVisitor<T> extends ParallelVisitor<SkyKey, T> {
    private final Uniquifier<SkyKey> uniquifier;

    protected AbstractSkyKeyParallelVisitor(
        Uniquifier<SkyKey> uniquifier,
        Callback<T> callback,
        int visitBatchSize,
        int processResultsBatchSize) {
      super(callback, visitBatchSize, processResultsBatchSize);
      this.uniquifier = uniquifier;
    }

    @Override
    protected ImmutableList<SkyKey> getUniqueValues(Iterable<SkyKey> values) {
      return uniquifier.unique(values);
    }
  }

  /** A helper class that computes 'rbuildfiles(<blah>)' via BFS. */
  private static class RBuildFilesVisitor extends AbstractSkyKeyParallelVisitor<Target> {
    // Each target in the full output of 'rbuildfiles' corresponds to BUILD file InputFile of a
    // unique package. So the processResultsBatchSize we choose to pass to the ParallelVisitor ctor
    // influences how many packages each leaf task doing processPartialResults will have to
    // deal with at once. A value of 100 was chosen experimentally.
    private static final int PROCESS_RESULTS_BATCH_SIZE = 100;
    private final SkyQueryEnvironment env;

    private RBuildFilesVisitor(
        SkyQueryEnvironment env,
        Uniquifier<SkyKey> uniquifier,
        Callback<Target> callback) {
      super(uniquifier, callback, VISIT_BATCH_SIZE, PROCESS_RESULTS_BATCH_SIZE);
      this.env = env;
    }

    @Override
    protected Visit getVisitResult(Iterable<SkyKey> values) throws InterruptedException {
      Collection<Iterable<SkyKey>> reverseDeps = env.graph.getReverseDeps(values).values();
      Set<SkyKey> keysToUseForResult = CompactHashSet.create();
      Set<SkyKey> keysToVisitNext = CompactHashSet.create();
      for (SkyKey rdep : Iterables.concat(reverseDeps)) {
        if (rdep.functionName().equals(SkyFunctions.PACKAGE)) {
          keysToUseForResult.add(rdep);
          // Every package has a dep on the external package, so we need to include those edges too.
          if (rdep.equals(PackageValue.key(Label.EXTERNAL_PACKAGE_IDENTIFIER))) {
            keysToVisitNext.add(rdep);
          }
        } else if (!rdep.functionName().equals(SkyFunctions.PACKAGE_LOOKUP)
            && !rdep.functionName().equals(SkyFunctions.GLOB)) {
          // Packages may depend on the existence of subpackages, but these edges aren't relevant to
          // rbuildfiles. They may also depend on files transitively through globs, but these cannot
          // be included in load statements and so we don't traverse through these either.
          keysToVisitNext.add(rdep);
        }
      }
      return new Visit(keysToUseForResult, keysToVisitNext);
    }

    @Override
    protected void processPartialResults(
        Iterable<SkyKey> keysToUseForResult, Callback<Target> callback)
        throws QueryException, InterruptedException {
      env.getBuildFileTargetsForPackageKeysAndProcessViaCallback(keysToUseForResult, callback);
    }

    @Override
    protected Iterable<SkyKey> preprocessInitialVisit(Iterable<SkyKey> keys) {
      return keys;
    }
  }

  private static class DepAndRdep {
    @Nullable
    private final SkyKey dep;
    private final SkyKey rdep;

    private DepAndRdep(@Nullable SkyKey dep, SkyKey rdep) {
      this.dep = dep;
      this.rdep = rdep;
    }

    @Override
    public boolean equals(Object obj) {
      if (!(obj instanceof DepAndRdep)) {
        return false;
      }
      DepAndRdep other = (DepAndRdep) obj;
      return Objects.equals(dep, other.dep) && rdep.equals(other.rdep);
    }

    @Override
    public int hashCode() {
      // N.B. - We deliberately use a garbage-free hashCode implementation (rather than e.g.
      // Objects#hash). Depending on the structure of the graph being traversed, this method can
      // be very hot.
      return 31 * Objects.hashCode(dep) + rdep.hashCode();
    }
  }

  private abstract static class AbstractRdepsVisitor<T> extends ParallelVisitor<T, Target> {
    private static final int PROCESS_RESULTS_BATCH_SIZE = SkyQueryEnvironment.BATCH_CALLBACK_SIZE;

    protected final SkyQueryEnvironment env;
    protected final MultisetSemaphore<PackageIdentifier> packageSemaphore;

    protected AbstractRdepsVisitor(
        SkyQueryEnvironment env,
        Callback<Target> callback,
        MultisetSemaphore<PackageIdentifier> packageSemaphore) {
      super(callback, VISIT_BATCH_SIZE, PROCESS_RESULTS_BATCH_SIZE);
      this.env = env;
      this.packageSemaphore = packageSemaphore;
    }

    @Override
    protected void processPartialResults(
        Iterable<SkyKey> keysToUseForResult, Callback<Target> callback)
        throws QueryException, InterruptedException {
      Multimap<SkyKey, SkyKey> packageKeyToTargetKeyMap =
          env.makePackageKeyToTargetKeyMap(keysToUseForResult);
      Set<PackageIdentifier> pkgIdsNeededForResult =
          packageKeyToTargetKeyMap
              .keySet()
              .stream()
              .map(SkyQueryEnvironment.PACKAGE_SKYKEY_TO_PACKAGE_IDENTIFIER)
              .collect(toImmutableSet());
      packageSemaphore.acquireAll(pkgIdsNeededForResult);
      try {
        callback.process(
            env.makeTargetsFromPackageKeyToTargetKeyMap(packageKeyToTargetKeyMap).values());
      } finally {
        packageSemaphore.releaseAll(pkgIdsNeededForResult);
      }
    }

    protected abstract SkyKey getRdepOfVisit(T visit);

    @Override
    protected Iterable<Task> getVisitTasks(Collection<T> pendingVisits) {
      // Group pending visitation by the package of the rdep, since we'll be targetfying the
      // rdep during the visitation.
      ListMultimap<PackageIdentifier, T> visitsByPackage = ArrayListMultimap.create();
      for (T visit : pendingVisits) {
        Label label = SkyQueryEnvironment.SKYKEY_TO_LABEL.apply(getRdepOfVisit(visit));
        if (label != null) {
          visitsByPackage.put(label.getPackageIdentifier(), visit);
        }
      }

      ImmutableList.Builder<Task> builder = ImmutableList.builder();

      // A couple notes here:
      // (i)  ArrayListMultimap#values returns the values grouped by key, which is exactly what we
      //      want.
      // (ii) ArrayListMultimap#values returns a Collection view, so we make a copy to avoid
      //      accidentally retaining the entire ArrayListMultimap object.
      for (Iterable<T> visitBatch :
          Iterables.partition(ImmutableList.copyOf(visitsByPackage.values()), VISIT_BATCH_SIZE)) {
        builder.add(new VisitTask(visitBatch));
      }

      return builder.build();
    }
  }

  /**
   * A helper class that computes unbounded 'allrdeps(<expr>)' or
   * 'rdeps(<precomputed-universe>, <expr>)' via BFS.
   *
   * <p>The visitor uses {@link DepAndRdep} to keep track the nodes to visit and avoid dealing with
   * targetification of reverse deps until they are needed. The rdep node itself is needed to filter
   * out disallowed deps later. Compared against the approach using a single SkyKey, it consumes 16
   * more bytes in a 64-bit environment for each edge. However it defers the need to load all the
   * packages which have at least a target as a rdep of the current batch, thus greatly reduces the
   * risk of OOMs. The additional memory usage should not be a large concern here, as even with 10M
   * edges, the memory overhead is around 160M, and the memory can be reclaimed by regular GC.
   */
  private static class RdepsUnboundedVisitor extends AbstractRdepsVisitor<DepAndRdep> {
    /**
     * A {@link Uniquifier} for visitations. Solely used for {@link #getUniqueValues}, which
     * actually isn't that useful. See the method javadoc.
     */
    private final Uniquifier<DepAndRdep> depAndRdepUniquifier;
    /**
     * A {@link Uniquifier} for *valid* visitations of rdeps. {@code env}'s dependency filter might
     * mean that some rdep edges are invalid, meaning that any individual {@link DepAndRdep}
     * visitation may actually be invalid. Because the same rdep can be reached through more than
     * one reverse edge, it'd be incorrect to naively dedupe visitations solely based on the rdep.
     */
    private final Uniquifier<SkyKey> validRdepUniquifier;
    private final Predicate<SkyKey> universe;

    private RdepsUnboundedVisitor(
        SkyQueryEnvironment env,
        Uniquifier<DepAndRdep> depAndRdepUniquifier,
        Uniquifier<SkyKey> validRdepUniquifier,
        Predicate<SkyKey> universe,
        Callback<Target> callback,
        MultisetSemaphore<PackageIdentifier> packageSemaphore) {
      super(env, callback, packageSemaphore);
      this.depAndRdepUniquifier = depAndRdepUniquifier;
      this.validRdepUniquifier = validRdepUniquifier;
      this.universe = universe;
    }

    /**
     * A {@link Factory} for {@link RdepsUnboundedVisitor} instances, each of which will be used
     * to perform visitation of the reverse transitive closure of the {@link Target}s passed in a
     * single {@link Callback#process} call. Note that all the created instances share the same
     * {@link Uniquifier} so that we don't visit the same Skyframe node more than once.
     */
    private static class Factory implements ParallelVisitor.Factory {
      private final SkyQueryEnvironment env;
      private final Uniquifier<DepAndRdep> depAndRdepUniquifier;
      private final Uniquifier<SkyKey> validRdepUniquifier;
      private final Predicate<SkyKey> universe;
      private final Callback<Target> callback;
      private final MultisetSemaphore<PackageIdentifier> packageSemaphore;

      private Factory(
        SkyQueryEnvironment env,
        Predicate<SkyKey> universe,
        Callback<Target> callback,
        MultisetSemaphore<PackageIdentifier> packageSemaphore) {
        this.env = env;
        this.universe = universe;
        this.depAndRdepUniquifier = new UniquifierImpl<>(depAndRdep -> depAndRdep);
        this.validRdepUniquifier = env.createSkyKeyUniquifier();
        this.callback = callback;
        this.packageSemaphore = packageSemaphore;
      }

      @Override
      public ParallelVisitor<DepAndRdep, Target> create() {
        return new RdepsUnboundedVisitor(
            env, depAndRdepUniquifier, validRdepUniquifier, universe, callback, packageSemaphore);
      }
    }

    @Override
    protected Visit getVisitResult(Iterable<DepAndRdep> depAndRdeps) throws InterruptedException {
      Collection<SkyKey> validRdeps = new ArrayList<>();

      // Multimap of dep to all the reverse deps in this visitation. Used to filter out the
      // disallowed deps.
      Multimap<SkyKey, SkyKey> reverseDepMultimap = ArrayListMultimap.create();
      for (DepAndRdep depAndRdep : depAndRdeps) {
        // The "roots" of our visitation (see #preprocessInitialVisit) have a null 'dep' field.
        if (depAndRdep.dep == null) {
          validRdeps.add(depAndRdep.rdep);
        } else {
          reverseDepMultimap.put(depAndRdep.dep, depAndRdep.rdep);
        }
      }

      Multimap<SkyKey, SkyKey> packageKeyToTargetKeyMap =
          env.makePackageKeyToTargetKeyMap(Iterables.concat(reverseDepMultimap.values()));
      Set<PackageIdentifier> pkgIdsNeededForTargetification =
          packageKeyToTargetKeyMap
              .keySet()
              .stream()
              .map(SkyQueryEnvironment.PACKAGE_SKYKEY_TO_PACKAGE_IDENTIFIER)
              .collect(toImmutableSet());
      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 (!reverseDepMultimap.isEmpty()) {
          Collection<Target> filteredTargets =
              env.filterRawReverseDepsOfTransitiveTraversalKeys(
                  reverseDepMultimap.asMap(), packageKeyToTargetKeyMap);
          filteredTargets
              .stream()
              .map(SkyQueryEnvironment.TARGET_TO_SKY_KEY)
              .forEachOrdered(validRdeps::add);
        }
      } finally {
        packageSemaphore.releaseAll(pkgIdsNeededForTargetification);
      }

      ImmutableList<SkyKey> uniqueValidRdeps = validRdeps.stream()
          .filter(validRdepUniquifier::unique)
          .collect(ImmutableList.toImmutableList());

      // Retrieve the reverse deps as SkyKeys and defer the targetification and filtering to next
      // recursive visitation.
      ImmutableList.Builder<DepAndRdep> depAndRdepsToVisitBuilder = ImmutableList.builder();
      env.graph.getReverseDeps(uniqueValidRdeps).entrySet()
          .forEach(reverseDepsEntry -> depAndRdepsToVisitBuilder.addAll(
              Iterables.transform(
                  Iterables.filter(
                      reverseDepsEntry.getValue(),
                      Predicates.and(SkyQueryEnvironment.IS_TTV, universe)),
                  rdep -> new DepAndRdep(reverseDepsEntry.getKey(), rdep))));

      return new Visit(
          /*keysToUseForResult=*/ uniqueValidRdeps,
          /*keysToVisit=*/ depAndRdepsToVisitBuilder.build());
    }

    @Override
    protected Iterable<DepAndRdep> preprocessInitialVisit(Iterable<SkyKey> keys) {
      return Iterables.transform(
          Iterables.filter(keys, k -> universe.apply(k)),
          key -> new DepAndRdep(null, key));
    }

    @Override
    protected SkyKey getRdepOfVisit(DepAndRdep visit) {
      return visit.rdep;
    }

    @Override
    protected ImmutableList<DepAndRdep> getUniqueValues(Iterable<DepAndRdep> depAndRdeps) {
      // See the javadoc for 'validRdepUniquifier'.
      //
      // N.B. - Except for the visitation roots, 'depAndRdepUniquifier' is actually completely
      // unneeded in practice for ensuring literal unique {@link DepAndRdep} visitations. Valid rdep
      // visitations are deduped in 'getVisitResult' using 'validRdepUniquifier', so there's
      // actually no way the same DepAndRdep visitation can ever be returned from 'getVisitResult'.
      // Still, we include an implementation of 'getUniqueValues' that is correct in isolation so as
      // to not be depending on implementation details of 'ParallelVisitor'.
      //
      // Even so, there's value in not visiting a rdep if it's already been visited *validly*
      // before. We use the intentionally racy {@link Uniquifier#uniquePure} to attempt to do this.
      return depAndRdepUniquifier.unique(
          Iterables.filter(
              depAndRdeps,
              depAndRdep -> validRdepUniquifier.uniquePure(depAndRdep.rdep)));
    }
  }

  private static class DepAndRdepAtDepth {
    private final DepAndRdep depAndRdep;
    private final int rdepDepth;

    private DepAndRdepAtDepth(DepAndRdep depAndRdep, int rdepDepth) {
      this.depAndRdep = depAndRdep;
      this.rdepDepth = rdepDepth;
    }
  }

  /**
   * A helper class that computes bounded 'allrdeps(<expr>, <depth>)' or
   * 'rdeps(<precomputed-universe>, <expr>, <depth>)' via BFS.
   *
   * <p>This is very similar to {@link RdepsUnboundedVisitor}. A lot of the same concerns apply here
   * but there are additional subtle concerns about the correctness of the bounded traversal: just
   * like for the sequential implementation of bounded allrdeps, we use {@link MinDepthUniquifier}.
   */
  private static class RdepsBoundedVisitor extends AbstractRdepsVisitor<DepAndRdepAtDepth> {
    private final int depth;
    private final Uniquifier<DepAndRdepAtDepth> depAndRdepAtDepthUniquifier;
    private final MinDepthUniquifier<SkyKey> validRdepMinDepthUniquifier;
    private final Predicate<SkyKey> universe;

    private RdepsBoundedVisitor(
        SkyQueryEnvironment env,
        int depth,
        Uniquifier<DepAndRdepAtDepth> depAndRdepAtDepthUniquifier,
        MinDepthUniquifier<SkyKey> validRdepMinDepthUniquifier,
        Predicate<SkyKey> universe,
        Callback<Target> callback,
        MultisetSemaphore<PackageIdentifier> packageSemaphore) {
      super(env, callback, packageSemaphore);
      this.depth = depth;
      this.depAndRdepAtDepthUniquifier = depAndRdepAtDepthUniquifier;
      this.validRdepMinDepthUniquifier = validRdepMinDepthUniquifier;
      this.universe = universe;
    }

    private static class Factory implements ParallelVisitor.Factory {
      private final SkyQueryEnvironment env;
      private final int depth;
      private final Uniquifier<DepAndRdepAtDepth> depAndRdepAtDepthUniquifier;
      private final MinDepthUniquifier<SkyKey> validRdepMinDepthUniquifier;
      private final Predicate<SkyKey> universe;
      private final Callback<Target> callback;
      private final MultisetSemaphore<PackageIdentifier> packageSemaphore;

      private Factory(
          SkyQueryEnvironment env,
          int depth,
          Predicate<SkyKey> universe,
          Callback<Target> callback,
          MultisetSemaphore<PackageIdentifier> packageSemaphore) {
        this.env = env;
        this.depth = depth;
        this.universe = universe;
        this.depAndRdepAtDepthUniquifier =
            new UniquifierImpl<>(depAndRdepAtDepth -> depAndRdepAtDepth);
        this.validRdepMinDepthUniquifier = env.createMinDepthSkyKeyUniquifier();
        this.callback = callback;
        this.packageSemaphore = packageSemaphore;
      }

      @Override
      public ParallelVisitor<DepAndRdepAtDepth, Target> create() {
        return new RdepsBoundedVisitor(
            env,
            depth,
            depAndRdepAtDepthUniquifier,
            validRdepMinDepthUniquifier,
            universe,
            callback,
            packageSemaphore);
      }
    }

    @Override
    protected Visit getVisitResult(Iterable<DepAndRdepAtDepth> depAndRdepAtDepths)
        throws InterruptedException {
      Map<SkyKey, Integer> shallowestRdepDepthMap = new HashMap<>();
      depAndRdepAtDepths.forEach(
          depAndRdepAtDepth -> shallowestRdepDepthMap.merge(
              depAndRdepAtDepth.depAndRdep.rdep, depAndRdepAtDepth.rdepDepth, Integer::min));

      Collection<SkyKey> validRdeps = new ArrayList<>();

      // Multimap of dep to all the reverse deps in this visitation. Used to filter out the
      // disallowed deps.
      Multimap<SkyKey, SkyKey> reverseDepMultimap = ArrayListMultimap.create();
      for (DepAndRdepAtDepth depAndRdepAtDepth : depAndRdepAtDepths) {
        // The "roots" of our visitation (see #preprocessInitialVisit) have a null 'dep' field.
        if (depAndRdepAtDepth.depAndRdep.dep == null) {
          validRdeps.add(depAndRdepAtDepth.depAndRdep.rdep);
        } else {
          reverseDepMultimap.put(
              depAndRdepAtDepth.depAndRdep.dep, depAndRdepAtDepth.depAndRdep.rdep);
        }
      }

      Multimap<SkyKey, SkyKey> packageKeyToTargetKeyMap =
          env.makePackageKeyToTargetKeyMap(Iterables.concat(reverseDepMultimap.values()));
      Set<PackageIdentifier> pkgIdsNeededForTargetification =
          packageKeyToTargetKeyMap
              .keySet()
              .stream()
              .map(SkyQueryEnvironment.PACKAGE_SKYKEY_TO_PACKAGE_IDENTIFIER)
              .collect(toImmutableSet());
      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 (!reverseDepMultimap.isEmpty()) {
          Collection<Target> filteredTargets =
              env.filterRawReverseDepsOfTransitiveTraversalKeys(
                  reverseDepMultimap.asMap(), packageKeyToTargetKeyMap);
          filteredTargets
              .stream()
              .map(SkyQueryEnvironment.TARGET_TO_SKY_KEY)
              .forEachOrdered(validRdeps::add);
        }
      } finally {
        packageSemaphore.releaseAll(pkgIdsNeededForTargetification);
      }

      ImmutableList<SkyKey> uniqueValidRdeps = validRdeps.stream()
          .filter(validRdep -> validRdepMinDepthUniquifier.uniqueAtDepthLessThanOrEqualTo(
              validRdep, shallowestRdepDepthMap.get(validRdep)))
          .collect(ImmutableList.toImmutableList());

      // Don't bother getting the rdeps of the rdeps that are already at the depth bound.
      Iterable<SkyKey> uniqueValidRdepsBelowDepthBound = Iterables.filter(
          uniqueValidRdeps,
          uniqueValidRdep -> shallowestRdepDepthMap.get(uniqueValidRdep) < depth);

      // Retrieve the reverse deps as SkyKeys and defer the targetification and filtering to next
      // recursive visitation.
      Map<SkyKey, Iterable<SkyKey>> unfilteredRdepsOfRdeps =
          env.graph.getReverseDeps(uniqueValidRdepsBelowDepthBound);

      ImmutableList.Builder<DepAndRdepAtDepth> depAndRdepAtDepthsToVisitBuilder =
          ImmutableList.builder();
      unfilteredRdepsOfRdeps.entrySet().forEach(entry -> {
        SkyKey rdep = entry.getKey();
        int depthOfRdepOfRdep = shallowestRdepDepthMap.get(rdep) + 1;
        Streams.stream(entry.getValue())
            .filter(Predicates.and(SkyQueryEnvironment.IS_TTV, universe))
            .forEachOrdered(rdepOfRdep -> {
          depAndRdepAtDepthsToVisitBuilder.add(
              new DepAndRdepAtDepth(new DepAndRdep(rdep, rdepOfRdep), depthOfRdepOfRdep));
        });
      });

      return new Visit(
          /*keysToUseForResult=*/ uniqueValidRdeps,
          /*keysToVisit=*/ depAndRdepAtDepthsToVisitBuilder.build());
    }

    @Override
    protected Iterable<DepAndRdepAtDepth> preprocessInitialVisit(Iterable<SkyKey> keys) {
      return Iterables.transform(
          Iterables.filter(keys, k -> universe.apply(k)),
          key -> new DepAndRdepAtDepth(new DepAndRdep(null, key), 0));
    }

    @Override
    protected SkyKey getRdepOfVisit(DepAndRdepAtDepth visit) {
      return visit.depAndRdep.rdep;
    }

    @Override
    protected ImmutableList<DepAndRdepAtDepth> getUniqueValues(
        Iterable<DepAndRdepAtDepth> depAndRdepAtDepths) {
      // See the comment in RdepsUnboundedVisitor#getUniqueValues.
      return depAndRdepAtDepthUniquifier.unique(
          Iterables.filter(
              depAndRdepAtDepths,
              depAndRdepAtDepth -> validRdepMinDepthUniquifier.uniqueAtDepthLessThanOrEqualToPure(
                  depAndRdepAtDepth.depAndRdep.rdep, depAndRdepAtDepth.rdepDepth)));
    }
  }

  /**
   * Helper class that computes the TTV-only DTC of some given TTV keys, via BFS following all
   * TTV->TTV dep edges.
   */
  private static class TransitiveTraversalValueDTCVisitor extends ParallelVisitor<SkyKey, SkyKey> {
    private final SkyQueryEnvironment env;
    private final Uniquifier<SkyKey> uniquifier;

    private TransitiveTraversalValueDTCVisitor(
        SkyQueryEnvironment env,
        Uniquifier<SkyKey> uniquifier,
        int processResultsBatchSize,
        AggregateAllCallback<SkyKey, ImmutableSet<SkyKey>> aggregateAllCallback) {
      super(aggregateAllCallback, VISIT_BATCH_SIZE, processResultsBatchSize);
      this.env = env;
      this.uniquifier = uniquifier;
    }

    private static class Factory implements ParallelVisitor.Factory {
      private final SkyQueryEnvironment env;
      private final Uniquifier<SkyKey> uniquifier;
      private final AggregateAllCallback<SkyKey, ImmutableSet<SkyKey>> aggregateAllCallback;
      private final int processResultsBatchSize;

      private Factory(
          SkyQueryEnvironment env,
          Uniquifier<SkyKey> uniquifier,
          int processResultsBatchSize,
          AggregateAllCallback<SkyKey, ImmutableSet<SkyKey>> aggregateAllCallback) {
        this.env = env;
        this.uniquifier = uniquifier;
        this.processResultsBatchSize = processResultsBatchSize;
        this.aggregateAllCallback = aggregateAllCallback;
      }

      @Override
      public ParallelVisitor<SkyKey, SkyKey> create() {
        return new TransitiveTraversalValueDTCVisitor(
            env, uniquifier, processResultsBatchSize, aggregateAllCallback);
      }
    }

    @Override
    protected void processPartialResults(
        Iterable<SkyKey> keysToUseForResult, Callback<SkyKey> callback)
        throws QueryException, InterruptedException {
      callback.process(keysToUseForResult);
    }

    @Override
    protected Visit getVisitResult(Iterable<SkyKey> ttvKeys) throws InterruptedException {
      Multimap<SkyKey, SkyKey> deps = env.getDirectDepsOfSkyKeys(ttvKeys);
      return new Visit(
          /*keysToUseForResult=*/ deps.keySet(),
          /*keysToVisit=*/ deps.values().stream()
          .filter(SkyQueryEnvironment.IS_TTV)
          .collect(ImmutableList.toImmutableList()));
    }

    @Override
    protected Iterable<SkyKey> preprocessInitialVisit(Iterable<SkyKey> keys) {
      // ParallelVisitorCallback passes in TTV keys.
      Preconditions.checkState(Iterables.all(keys, SkyQueryEnvironment.IS_TTV), keys);
      return keys;
    }

    @Override
    protected ImmutableList<SkyKey> getUniqueValues(Iterable<SkyKey> values) {
      return uniquifier.unique(values);
    }
  }

  /** Thread-safe {@link AggregateAllCallback} backed by a concurrent {@link Set}. */
  @ThreadSafe
  private static class ThreadSafeAggregateAllSkyKeysCallback
      implements AggregateAllCallback<SkyKey, ImmutableSet<SkyKey>> {

    private final Set<SkyKey> results;

    private ThreadSafeAggregateAllSkyKeysCallback(int concurrencyLevel) {
      this.results =
          Collections.newSetFromMap(
              new ConcurrentHashMap<>(
                  /*initialCapacity=*/ concurrencyLevel, /*loadFactor=*/ 0.75f));
    }

    @Override
    public void process(Iterable<SkyKey> partialResult)
        throws QueryException, InterruptedException {
      Iterables.addAll(results, partialResult);
    }

    @Override
    public ImmutableSet<SkyKey> getResult() {
      return ImmutableSet.copyOf(results);
    }
  }
}