Add a futures-based include scanner implementation

Enable with --experimental_async_include_scanner

This allows integrating with the async spawn execution API in
SpawnRunner, reducing the need for a large thread pool. Also, in the
legacy implementation, if multiple threads called computeIfAbsent on the
same key, then all except one block. This could lead to all threads in
the thread pool being blocked, which in turn (due to caller-runs policy)
can cause Skyframe evaluator threads to be blocked.

However, note that Skyframe threads are still blocked in
CppCompileAction.findUsedHeaders even after this change. I have a
follow-up change to support an action input discovery API based on
continuations.

Note that this does not change the include scanner to use continuations.
We use continuations elsewhere because we need access to the skyframe
function environment, which requires that such code runs in the skyframe
thread pool. The include scanner does _not_ need the skyframe function
environment, so we can run it in a non-skyframe thread pool.

However, we may change this in the future, e.g., to cache include scan
results in Skyframe. There was a previous attempt to reimplement include
scanning in Skyframe, but that was dropped due to performance overhead.

Progress on #6394.

PiperOrigin-RevId: 246793955

src/main/java/com/google/devtools/build/lib/includescanning/IncludeParser.java

@@ -24,6 +24,9 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Sets;
 import com.google.common.io.CharStreams;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.devtools.build.lib.actions.ActionExecutionContext;
 import com.google.devtools.build.lib.actions.ActionExecutionMetadata;
 import com.google.devtools.build.lib.actions.Artifact;
@@ -886,6 +889,72 @@ Collection<Inclusion> extractInclusions(
     return ImmutableList.copyOf(inclusions);
   }
 
+  /**
+   * Extracts all inclusions from a given source file.
+   *
+   * @param file the file to parse & extract inclusions from
+   * @param actionExecutionContext Services in the scope of the action, like the stream to which
+   *     scanning messages are printed
+   * @return a new set of inclusions, normalized to the cache
+   */
+  ListenableFuture<Collection<Inclusion>> extractInclusionsAsync(
+      Artifact file,
+      ActionExecutionMetadata actionExecutionMetadata,
+      ActionExecutionContext actionExecutionContext,
+      Artifact grepIncludes,
+      @Nullable SpawnIncludeScanner remoteIncludeScanner,
+      boolean isOutputFile)
+      throws IOException {
+    ListenableFuture<Collection<Inclusion>> inclusions;
+    if (remoteIncludeScanner != null
+        && remoteIncludeScanner.shouldParseRemotely(file, actionExecutionContext)) {
+      inclusions =
+          remoteIncludeScanner.extractInclusionsAsync(
+              file,
+              actionExecutionMetadata,
+              actionExecutionContext,
+              grepIncludes,
+              getFileType(),
+              isOutputFile);
+    } else {
+      try (SilentCloseable c =
+          Profiler.instance().profile(ProfilerTask.SCANNER, file.getExecPathString())) {
+        inclusions =
+            Futures.immediateFuture(
+                extractInclusions(
+                    FileSystemUtils.readContent(actionExecutionContext.getInputPath(file))));
+      } catch (IOException e) {
+        if (remoteIncludeScanner != null) {
+          logger.log(
+              Level.WARNING,
+              "Falling back on remote parsing of " + actionExecutionContext.getInputPath(file),
+              e);
+          inclusions =
+              remoteIncludeScanner.extractInclusionsAsync(
+                  file,
+                  actionExecutionMetadata,
+                  actionExecutionContext,
+                  grepIncludes,
+                  getFileType(),
+                  isOutputFile);
+        } else {
+          throw e;
+        }
+      }
+    }
+    if (hints != null) {
+      return Futures.transform(
+          inclusions,
+          (c) -> {
+            // Ugly, but saves doing another copy.
+            c.addAll(hints.getHintedInclusions(file));
+            return c;
+          },
+          MoreExecutors.directExecutor());
+    }
+    return inclusions;
+  }
+
   /**
    * Returns type of the scanned file.
    *

src/main/java/com/google/devtools/build/lib/includescanning/IncludeScannerSupplierImpl.java

@@ -18,6 +18,7 @@
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
+import com.google.common.util.concurrent.ListenableFuture;
 import com.google.devtools.build.lib.actions.Artifact;
 import com.google.devtools.build.lib.actions.ArtifactFactory;
 import com.google.devtools.build.lib.analysis.BlazeDirectories;
@@ -76,14 +77,15 @@ public boolean equals(Object other) {
    * Cache of include scan results mapping source paths to sets of scanned inclusions. Shared by all
    * scanner instances.
    */
-  private final ConcurrentMap<Artifact, Collection<Inclusion>> includeParseCache =
+  private final ConcurrentMap<Artifact, ListenableFuture<Collection<Inclusion>>> includeParseCache =
       new ConcurrentHashMap<>();
 
   /** Map of grepped include files from input (.cc or .h) to a header-grepped file. */
   private final PathExistenceCache pathCache;
 
   private final Supplier<SpawnIncludeScanner> spawnIncludeScannerSupplier;
   private final Path execRoot;
+  private final boolean useAsyncIncludeScanner;
 
   /** Cache of include scanner instances mapped by include-path hashes. */
   private final LoadingCache<IncludeScannerParams, IncludeScanner> scanners =
@@ -102,7 +104,8 @@ public IncludeScanner load(IncludeScannerParams key) {
                       directories.getOutputPath(execRoot.getBaseName()),
                       execRoot,
                       artifactFactory,
-                      spawnIncludeScannerSupplier);
+                      spawnIncludeScannerSupplier,
+                      useAsyncIncludeScanner);
                 }
               });
 
@@ -111,13 +114,15 @@ public IncludeScannerSupplierImpl(
       ExecutorService includePool,
       ArtifactFactory artifactFactory,
       Supplier<SpawnIncludeScanner> spawnIncludeScannerSupplier,
-      Path execRoot) {
+      Path execRoot,
+      boolean useAsyncIncludeScanner) {
     this.directories = directories;
     this.includePool = includePool;
     this.artifactFactory = artifactFactory;
     this.spawnIncludeScannerSupplier = spawnIncludeScannerSupplier;
     this.execRoot = execRoot;
     this.pathCache = new PathExistenceCache(execRoot, artifactFactory);
+    this.useAsyncIncludeScanner = useAsyncIncludeScanner;
   }
 
   @Override

src/main/java/com/google/devtools/build/lib/includescanning/IncludeScanningModule.java

@@ -19,6 +19,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.devtools.build.lib.actions.ActionContext;
 import com.google.devtools.build.lib.actions.ActionExecutionContext;
@@ -148,15 +149,19 @@ public static final class SwigIncludeScanningContextImpl implements SwigIncludeS
     private final CommandEnvironment env;
     private final Supplier<SpawnIncludeScanner> spawnScannerSupplier;
     private final Supplier<ExecutorService> includePool;
-    private final ConcurrentMap<Artifact, Collection<Inclusion>> cache = new ConcurrentHashMap<>();
+    private final ConcurrentMap<Artifact, ListenableFuture<Collection<Inclusion>>> cache =
+        new ConcurrentHashMap<>();
+    private final boolean useAsyncIncludeScanner;
 
     SwigIncludeScanningContextImpl(
         CommandEnvironment env,
         Supplier<SpawnIncludeScanner> spawnScannerSupplier,
-        Supplier<ExecutorService> includePool) {
+        Supplier<ExecutorService> includePool,
+        boolean useAsyncIncludeScanner) {
       this.env = env;
       this.spawnScannerSupplier = spawnScannerSupplier;
       this.includePool = includePool;
+      this.useAsyncIncludeScanner = useAsyncIncludeScanner;
     }
 
     @Override
@@ -177,7 +182,8 @@ public void extractSwigIncludes(
               swigIncludePaths,
               env.getDirectories(),
               env.getSkyframeBuildView().getArtifactFactory(),
-              env.getExecRoot());
+              env.getExecRoot(),
+              useAsyncIncludeScanner);
       ImmutableMap.Builder<PathFragment, Artifact> pathToLegalOutputArtifact =
           ImmutableMap.builder();
       for (Artifact path : legalOutputPaths) {
@@ -238,7 +244,8 @@ public IncludeScanningActionContextProvider(
       this.strategies =
           ImmutableList.of(
               new CppIncludeExtractionContextImpl(env),
-              new SwigIncludeScanningContextImpl(env, spawnScannerSupplier, () -> includePool),
+              new SwigIncludeScanningContextImpl(
+                  env, spawnScannerSupplier, () -> includePool, options.useAsyncIncludeScanner),
               new CppIncludeScanningContextImpl(() -> includeScannerSupplier));
 
       env.getEventBus().register(this);
@@ -270,8 +277,8 @@ public void executionPhaseStarting(
 
     @Override
     public void executorCreated(Iterable<ActionContext> usedContexts) throws ExecutorInitException {
-      int threads = buildRequest.getOptions(IncludeScanningOptions.class)
-          .includeScanningParallelism;
+      IncludeScanningOptions options = buildRequest.getOptions(IncludeScanningOptions.class);
+      int threads = options.includeScanningParallelism;
       if (threads > 0) {
         log.info(
             String.format("Include scanning configured to use a pool with %d threads", threads));
@@ -286,11 +293,11 @@ public void executorCreated(Iterable<ActionContext> usedContexts) throws Executo
               includePool,
               env.getSkyframeBuildView().getArtifactFactory(),
               spawnScannerSupplier,
-              env.getExecRoot());
+              env.getExecRoot(),
+              options.useAsyncIncludeScanner);
 
       spawnScannerSupplier.get().setOutputService(env.getOutputService());
-      spawnScannerSupplier.get().setInMemoryOutput(
-          buildRequest.getOptions(IncludeScanningOptions.class).inMemoryIncludesFiles);
+      spawnScannerSupplier.get().setInMemoryOutput(options.inMemoryIncludesFiles);
     }
   }
 }

src/main/java/com/google/devtools/build/lib/includescanning/IncludeScanningOptions.java

@@ -100,4 +100,16 @@ public ParallelismConverter() throws OptionsParsingException {
               + " \"auto\" means to use a reasonable value derived from the machine's hardware"
               + " profile (e.g. the number of processors).")
   public int includeScanningParallelism;
+
+  @Option(
+      name = "experimental_async_include_scanner",
+      documentationCategory = OptionDocumentationCategory.UNDOCUMENTED,
+      effectTags = {
+        OptionEffectTag.BAZEL_INTERNAL_CONFIGURATION,
+        OptionEffectTag.EXECUTION,
+        OptionEffectTag.HOST_MACHINE_RESOURCE_OPTIMIZATIONS
+      },
+      defaultValue = "false",
+      help = "Switches to a new implementation of include scanning based on futures.")
+  public boolean useAsyncIncludeScanner;
 }