[Skymeld] De-dup events at BuildDriverFunction.

TopLevelStatusEvents are fired from within BuildDriverFunction, that means there may be duplicated events (e.g. from resets). Up till now, we've been asking the subscribers to do their own de-duplication. This is not ideal since 1) storing the events in deduplicating sets may prevent transitive references from being GCed and 2) Posting the events were observed to be a hot spot for contention [1].

This CL moves the de-duplication into BuildDriverFunction. We exploit the fact that each BuildDriverKey only has 1 effective event per TopLevelStatusEvent type (e.g. a target foo can't be analyzed twice in a build) to prevent sending a 2nd event of the same type. This allows us to store only the type of the sent events, instead of the events themselves.

PiperOrigin-RevId: 522334018
Change-Id: I76aed9fa68f8197d8f0a70158198c7341dc3a884

Author: joeleba

src/main/java/com/google/devtools/build/lib/buildtool/AnalysisAndExecutionPhaseRunner.java

@@ -15,7 +15,6 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Sets;
 import com.google.common.eventbus.Subscribe;
 import com.google.common.flogger.GoogleLogger;
 import com.google.devtools.build.lib.actions.BuildFailedException;
@@ -58,7 +57,6 @@
 import com.google.devtools.common.options.OptionsParsingException;
 import java.util.Collection;
 import java.util.List;
-import java.util.Set;
 
 /**
  * Intended drop-in replacement for AnalysisPhaseRunner after we're done with merging Skyframe's
@@ -322,8 +320,6 @@ private static class TopLevelTargetAnalysisWatcher implements AutoCloseable {
     private final CommandEnvironment env;
     private final BuildRequest buildRequest;
     private final BuildOptions buildOptions;
-    private final Set<TopLevelTargetAnalyzedEvent> processedEvents;
-
     private TopLevelTargetAnalysisWatcher(
         Iterable<BlazeModule> blazeModules,
         CommandEnvironment env,
@@ -333,7 +329,6 @@ private TopLevelTargetAnalysisWatcher(
       this.env = env;
       this.buildRequest = buildRequest;
       this.buildOptions = buildOptions;
-      this.processedEvents = Sets.newConcurrentHashSet();
     }
 
     /** Creates an AnalysisOperationWatcher and registers it with the provided eventBus. */
@@ -351,13 +346,6 @@ public static TopLevelTargetAnalysisWatcher createAndRegisterWithEventBus(
     @Subscribe
     public void handleTopLevelEntityAnalysisConcluded(TopLevelTargetAnalyzedEvent e)
         throws ViewCreationFailedException, InterruptedException {
-      // TopLevelTargetAnalyzedEvent originates from within Skyframe, which means there'll likely
-      // be multiple events fired for the same underlying ConfiguredTarget due to SkyFunction
-      // restarts. We only process them once.
-      if (!processedEvents.add(e)) {
-        return;
-      }
-
       for (BlazeModule blazeModule : blazeModules) {
         blazeModule.afterTopLevelTargetAnalysis(
             env, buildRequest, buildOptions, e.configuredTarget());

src/main/java/com/google/devtools/build/lib/runtime/TargetSummaryPublisher.java

@@ -118,9 +118,6 @@ public void populateTargets(TestFilteringCompleteEvent event) {
   /**
    * Populates the aggregator for a particular top level target, including test targets.
    *
-   * <p>Since the event is fired from within a SkyFunction, it is possible to receive duplicate
-   * events. In case of duplication, simply return without creating any new aggregator.
-   *
    * <p>With skymeld, the corresponding AspectCompleteEvents may arrive before the aggregator is set
    * up. We therefore need to put those events in a queue and resolve them when the aggregator
    * becomes available.

src/main/java/com/google/devtools/build/lib/skyframe/BuildDriverFunction.java

@@ -25,6 +25,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
+import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import com.google.devtools.build.lib.actions.ActionAnalysisMetadata;
 import com.google.devtools.build.lib.actions.ActionLookupKey;
@@ -59,6 +60,7 @@
 import com.google.devtools.build.lib.skyframe.TopLevelStatusEvents.SomeExecutionStartedEvent;
 import com.google.devtools.build.lib.skyframe.TopLevelStatusEvents.TestAnalyzedEvent;
 import com.google.devtools.build.lib.skyframe.TopLevelStatusEvents.TopLevelEntityAnalysisConcludedEvent;
+import com.google.devtools.build.lib.skyframe.TopLevelStatusEvents.TopLevelStatusEventWithType;
 import com.google.devtools.build.lib.skyframe.TopLevelStatusEvents.TopLevelTargetAnalyzedEvent;
 import com.google.devtools.build.lib.skyframe.TopLevelStatusEvents.TopLevelTargetPendingExecutionEvent;
 import com.google.devtools.build.lib.skyframe.TopLevelStatusEvents.TopLevelTargetReadyForSymlinkPlanting;
@@ -72,7 +74,9 @@
 import com.google.devtools.build.skyframe.SkyframeLookupResult;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import javax.annotation.Nullable;
 
@@ -96,6 +100,18 @@ public class BuildDriverFunction implements SkyFunction {
   //   shutting down of the Executors could lead to a RejectedExecutionException.
   private Set<BuildDriverKey> checkedForConflicts = Sets.newConcurrentHashSet();
 
+  // Events coming from Skyframe may contain duplicates (because of resets). It would be better to
+  // de-duplicate at the source to avoid repeated work by each subscriber.
+  //
+  // Each top level key has at most 1 effective status event, e.g. a top level target can't be
+  // analyzed twice in a build. Therefore, to keep track of the posted events, we only need to keep
+  // the sent event types instead of the events themselves.
+  //
+  // We didn't use SkyKeyComputeState since it should only be used as a performance optimization,
+  // whereas in this situation the state determines the behavior of the SkyFunction.
+  private Map<BuildDriverKey, Set<TopLevelStatusEvents.Type>> keyToPostedEvents =
+      Maps.newConcurrentMap();
+
   BuildDriverFunction(
       TransitiveActionLookupValuesHelper transitiveActionLookupValuesHelper,
       Supplier<IncrementalArtifactConflictFinder> incrementalArtifactConflictFinder,
@@ -160,6 +176,9 @@ public SkyValue compute(SkyKey skyKey, Environment env)
       }
     }
 
+    Set<TopLevelStatusEvents.Type> postedEventsTypes =
+        keyToPostedEvents.computeIfAbsent(buildDriverKey, (unused) -> new HashSet<>());
+
     Preconditions.checkState(
         topLevelSkyValue instanceof ConfiguredTargetValue
             || topLevelSkyValue instanceof TopLevelAspectsValue);
@@ -170,7 +189,18 @@ public SkyValue compute(SkyKey skyKey, Environment env)
       // before the TopLevelEntityAnalysisConcludedEvent: when the last of the analysis work is
       // concluded, we need to have the *complete* list of analyzed targets ready in
       // BuildResultListener.
-      postTopLevelTargetAnalyzedEvent(env, configuredTargetValue, configuredTarget);
+      postEventIfNecessary(
+          postedEventsTypes, env, TopLevelTargetAnalyzedEvent.create(configuredTarget));
+      // It's possible that this code path is triggered AFTER the analysis cache clean up and the
+      // transitive packages for package root resolution is already cleared. In such a case, the
+      // symlinks should have already been planted.
+      if (configuredTargetValue.getTransitivePackages() != null) {
+        postEventIfNecessary(
+            postedEventsTypes,
+            env,
+            TopLevelTargetReadyForSymlinkPlanting.create(
+                configuredTargetValue.getTransitivePackages()));
+      }
 
       BuildConfigurationValue buildConfigurationValue =
           configuredTarget.getConfigurationKey() == null
@@ -196,44 +226,56 @@ public SkyValue compute(SkyKey skyKey, Environment env)
 
           state.checkedForCompatibility = true;
           if (!isConfiguredTargetCompatible) {
-            env.getListener().post(TopLevelTargetSkippedEvent.create(configuredTarget));
+            postEventIfNecessary(
+                postedEventsTypes, env, TopLevelTargetSkippedEvent.create(configuredTarget));
             // We still record analyzed but skipped tests, as this information is needed for the
             // result summary.
             if (!NOT_TEST.equals(buildDriverKey.getTestType())) {
-              env.getListener()
-                  .post(
-                      TestAnalyzedEvent.create(
-                          configuredTarget,
-                          Preconditions.checkNotNull(buildConfigurationValue),
-                          /*isSkipped=*/ true));
+              postEventIfNecessary(
+                  postedEventsTypes,
+                  env,
+                  TestAnalyzedEvent.create(
+                      configuredTarget,
+                      Preconditions.checkNotNull(buildConfigurationValue),
+                      /* isSkipped= */ true));
             }
             // Only send the event now to include the compatibility check in the measurement for
             // time spent on analysis work.
-            env.getListener().post(TopLevelEntityAnalysisConcludedEvent.success(buildDriverKey));
+            postEventIfNecessary(
+                postedEventsTypes,
+                env,
+                TopLevelEntityAnalysisConcludedEvent.success(buildDriverKey));
             // We consider the evaluation of this BuildDriverKey successful at this point, even when
             // the target is skipped.
+            removeStatesForKey(buildDriverKey);
             return new BuildDriverValue(topLevelSkyValue, /*skipped=*/ true);
           }
         } catch (TargetCompatibilityCheckException e) {
           throw new BuildDriverFunctionException(e);
         }
       }
 
-      env.getListener().post(TopLevelEntityAnalysisConcludedEvent.success(buildDriverKey));
-      env.getListener()
-          .post(
-              TopLevelTargetPendingExecutionEvent.create(
-                  configuredTarget, buildDriverKey.isTest()));
+      postEventIfNecessary(
+          postedEventsTypes, env, TopLevelEntityAnalysisConcludedEvent.success(buildDriverKey));
+      postEventIfNecessary(
+          postedEventsTypes,
+          env,
+          TopLevelTargetPendingExecutionEvent.create(configuredTarget, buildDriverKey.isTest()));
       requestConfiguredTargetExecution(
           configuredTarget,
           buildDriverKey,
           actionLookupKey,
           buildConfigurationValue,
           env,
-          topLevelArtifactContext);
+          topLevelArtifactContext,
+          postedEventsTypes);
     } else {
       announceAspectAnalysisDoneAndRequestExecution(
-          buildDriverKey, (TopLevelAspectsValue) topLevelSkyValue, env, topLevelArtifactContext);
+          buildDriverKey,
+          (TopLevelAspectsValue) topLevelSkyValue,
+          env,
+          topLevelArtifactContext,
+          postedEventsTypes);
     }
 
     if (env.valuesMissing()) {
@@ -245,33 +287,35 @@ public SkyValue compute(SkyKey skyKey, Environment env)
     if (EXCLUSIVE.equals(buildDriverKey.getTestType())
         || EXCLUSIVE_IF_LOCAL.equals(buildDriverKey.getTestType())) {
       Preconditions.checkState(topLevelSkyValue instanceof ConfiguredTargetValue);
+      removeStatesForKey(buildDriverKey);
       return new ExclusiveTestBuildDriverValue(
           topLevelSkyValue, ((ConfiguredTargetValue) topLevelSkyValue).getConfiguredTarget());
     }
 
+    removeStatesForKey(buildDriverKey);
     return new BuildDriverValue(topLevelSkyValue, /*skipped=*/ false);
   }
 
-  public void resetActionConflictCheckingStatus() {
+  public void resetStates() {
     checkedForConflicts = Sets.newConcurrentHashSet();
+    keyToPostedEvents = Maps.newConcurrentMap();
   }
 
-  private static void postTopLevelTargetAnalyzedEvent(
+  private void removeStatesForKey(BuildDriverKey key) {
+    checkedForConflicts.remove(key);
+    keyToPostedEvents.remove(key);
+  }
+
+  private static void postEventIfNecessary(
+      Set<TopLevelStatusEvents.Type> postedEventsTypes,
       Environment env,
-      ConfiguredTargetValue configuredTargetValue,
-      ConfiguredTarget configuredTarget) {
-    env.getListener().post(TopLevelTargetAnalyzedEvent.create(configuredTarget));
-    // It's possible that this code path is triggered AFTER the analysis cache clean up and the
-    // transitive packages for package root resolution is already cleared. In such a case, the
-    // symlinks should have already been planted.
-    if (configuredTargetValue.getTransitivePackages() != null) {
-      env.getListener()
-          .post(
-              TopLevelTargetReadyForSymlinkPlanting.create(
-                  configuredTargetValue.getTransitivePackages()));
+      TopLevelStatusEventWithType event) {
+    if (postedEventsTypes.add(event.getType())) {
+      env.getListener().post(event);
     }
   }
 
+
   /**
    * Checks if a ConfiguredTarget is compatible with the platform/environment. See {@link
    * TopLevelConstraintSemantics}.
@@ -352,12 +396,13 @@ private void requestConfiguredTargetExecution(
       ActionLookupKey actionLookupKey,
       BuildConfigurationValue buildConfigurationValue,
       Environment env,
-      TopLevelArtifactContext topLevelArtifactContext)
+      TopLevelArtifactContext topLevelArtifactContext,
+      Set<TopLevelStatusEvents.Type> postedEventsTypes)
       throws InterruptedException {
     ImmutableSet.Builder<Artifact> artifactsToBuild = ImmutableSet.builder();
     addExtraActionsIfRequested(
         configuredTarget.getProvider(ExtraActionArtifactsProvider.class), artifactsToBuild);
-    env.getListener().post(SomeExecutionStartedEvent.create());
+    postEventIfNecessary(postedEventsTypes, env, SomeExecutionStartedEvent.create());
     if (NOT_TEST.equals(buildDriverKey.getTestType())) {
       declareDependenciesAndCheckValues(
           env,
@@ -369,12 +414,13 @@ private void requestConfiguredTargetExecution(
       return;
     }
 
-    env.getListener()
-        .post(
-            TestAnalyzedEvent.create(
-                configuredTarget,
-                Preconditions.checkNotNull(buildConfigurationValue),
-                /*isSkipped=*/ false));
+    postEventIfNecessary(
+        postedEventsTypes,
+        env,
+        TestAnalyzedEvent.create(
+            configuredTarget,
+            Preconditions.checkNotNull(buildConfigurationValue),
+            /* isSkipped= */ false));
 
     if (PARALLEL.equals(buildDriverKey.getTestType())) {
       // Only run non-exclusive tests here. Exclusive tests need to be run sequentially later.
@@ -398,12 +444,19 @@ private void announceAspectAnalysisDoneAndRequestExecution(
       BuildDriverKey buildDriverKey,
       TopLevelAspectsValue topLevelAspectsValue,
       Environment env,
-      TopLevelArtifactContext topLevelArtifactContext)
+      TopLevelArtifactContext topLevelArtifactContext,
+      Set<TopLevelStatusEvents.Type> postedEventsTypes)
       throws InterruptedException {
 
-    env.getListener().post(SomeExecutionStartedEvent.create());
+    postEventIfNecessary(postedEventsTypes, env, SomeExecutionStartedEvent.create());
     ImmutableSet.Builder<Artifact> artifactsToBuild = ImmutableSet.builder();
     List<SkyKey> aspectCompletionKeys = new ArrayList<>();
+
+    boolean symlinkPlantingEventsSent =
+        !postedEventsTypes.add(
+            TopLevelStatusEvents.Type.TOP_LEVEL_TARGET_READY_FOR_SYMLINK_PLANTING);
+    boolean aspectAnalyzedEventsSent =
+        !postedEventsTypes.add(TopLevelStatusEvents.Type.ASPECT_ANALYZED);
     for (AspectValue aspectValue : topLevelAspectsValue.getTopLevelAspectsValues()) {
       AspectKey aspectKey = aspectValue.getKey();
       ConfiguredAspect configuredAspect = aspectValue.getConfiguredAspect();
@@ -413,21 +466,25 @@ private void announceAspectAnalysisDoneAndRequestExecution(
       // It's possible that this code path is triggered AFTER the analysis cache clean up and the
       // transitive packages for package root resolution is already cleared. In such a case, the
       // symlinks should have already been planted.
-      if (aspectValue.getTransitivePackages() != null) {
+      if (aspectValue.getTransitivePackages() != null && !symlinkPlantingEventsSent) {
         env.getListener()
             .post(
                 TopLevelTargetReadyForSymlinkPlanting.create(aspectValue.getTransitivePackages()));
       }
-      env.getListener().post(AspectAnalyzedEvent.create(aspectKey, configuredAspect));
+      if (!aspectAnalyzedEventsSent) {
+        env.getListener().post(AspectAnalyzedEvent.create(aspectKey, configuredAspect));
+      }
 
       aspectCompletionKeys.add(AspectCompletionKey.create(aspectKey, topLevelArtifactContext));
     }
+
     // Send the AspectAnalyzedEvents first to make sure the BuildResultListener is up-to-date before
     // signaling that the analysis of this top level aspect has concluded.
-    env.getListener().post(TopLevelEntityAnalysisConcludedEvent.success(buildDriverKey));
-
+    postEventIfNecessary(
+        postedEventsTypes, env, TopLevelEntityAnalysisConcludedEvent.success(buildDriverKey));
     declareDependenciesAndCheckValues(
         env, Iterables.concat(Artifact.keys(artifactsToBuild.build()), aspectCompletionKeys));
+
   }
 
   /**

src/main/java/com/google/devtools/build/lib/skyframe/SkyframeExecutor.java

@@ -2446,7 +2446,7 @@ void resetActionConflictsStoredInSkyframe() {
   }
 
   public void resetBuildDriverFunction() {
-    buildDriverFunction.resetActionConflictCheckingStatus();
+    buildDriverFunction.resetStates();
   }
 
   /** Resets the incremental artifact conflict finder to ensure incremental correctness. */