Decouple remote dep file cache checking from action cache checking

Summary:
Previously, we just had one `cache_checker`, and if the action cache lookup failed, and remote dep file lookup was enabled, then we'd do a remote dep file lookup from within the action cache checker.

I want to change the order of the lookups that we do so that we check the local dep file cache before we do a remote dep file lookup, so this diff decouples the remote dep file lookup from the action cache lookup (but doesn't yet change the order). Now, we do two explicit calls in `run.rs`, first to the action cache and then to the remote dep file cache.

Reviewed By: blackm00n

Differential Revision: D74064170

fbshipit-source-id: 10c561ebe06dc87131a3eb113fa43cfdad78e051

Author: Ian Childs

app/buck2_action_impl/src/actions/impls/run.rs

@@ -674,19 +674,27 @@ impl RunAction {
 
         let action_cache_result = ctx.action_cache(manager, &req, &prepared_action).await;
 
-        // If the result was served by the remote dep file cache, we can't use the result just yet. We need to verify that
-        // the inputs tracked by a depfile that was actually used in the cache hit are indentical to the inputs we have for this action.
-        let result = if let ControlFlow::Break(res) = action_cache_result {
-            self.check_cache_result_is_useable(
-                ctx,
-                &req,
-                &prepared_action.action_and_blobs.action,
-                res,
-                &dep_file_bundle,
-            )
-            .await?
-        } else {
-            action_cache_result
+        let result = match action_cache_result {
+            ControlFlow::Break(_) => action_cache_result,
+            ControlFlow::Continue(manager) => {
+                let remote_dep_file_result = ctx
+                    .remote_dep_file_cache(manager, &req, &prepared_action)
+                    .await;
+                if let ControlFlow::Break(res) = remote_dep_file_result {
+                    // If the result was served by the remote dep file cache, we can't use the result just yet. We need to verify that
+                    // the inputs tracked by a depfile that was actually used in the cache hit are identical to the inputs we have for this action.
+                    self.check_cache_result_is_useable(
+                        ctx,
+                        &req,
+                        &prepared_action.action_and_blobs.action,
+                        res,
+                        &dep_file_bundle,
+                    )
+                    .await?
+                } else {
+                    remote_dep_file_result
+                }
+            }
         };
 
         // If the cache queries did not yield to a result, fallback to local dep file query (continuation), then execution.

app/buck2_build_api/src/actions.rs

@@ -213,6 +213,13 @@ pub trait ActionExecutionCtx: Send + Sync {
         prepared_action: &PreparedAction,
     ) -> ControlFlow<CommandExecutionResult, CommandExecutionManager>;
 
+    async fn remote_dep_file_cache(
+        &mut self,
+        manager: CommandExecutionManager,
+        request: &CommandExecutionRequest,
+        prepared_action: &PreparedAction,
+    ) -> ControlFlow<CommandExecutionResult, CommandExecutionManager>;
+
     async fn cache_upload(
         &mut self,
         action: &ActionDigestAndBlobs,

app/buck2_build_api/src/actions/execute/action_executor.rs

@@ -241,7 +241,8 @@ impl HasActionExecutor for DiceComputations<'_> {
         let CommandExecutorResponse {
             executor,
             platform,
-            cache_checker,
+            action_cache_checker,
+            remote_dep_file_cache_checker,
             cache_uploader,
         } = self.get_command_executor_from_dice(executor_config).await?;
         let blocking_executor = self.get_blocking_executor();
@@ -257,7 +258,8 @@ impl HasActionExecutor for DiceComputations<'_> {
         Ok(Arc::new(BuckActionExecutor::new(
             CommandExecutor::new(
                 executor,
-                cache_checker,
+                action_cache_checker,
+                remote_dep_file_cache_checker,
                 cache_uploader,
                 artifact_fs,
                 executor_config.options,
@@ -423,6 +425,28 @@ impl ActionExecutionCtx for BuckActionExecutionContext<'_> {
             .await
     }
 
+    async fn remote_dep_file_cache(
+        &mut self,
+        manager: CommandExecutionManager,
+        request: &CommandExecutionRequest,
+        prepared_action: &PreparedAction,
+    ) -> ControlFlow<CommandExecutionResult, CommandExecutionManager> {
+        let action = self.target();
+        self.executor
+            .command_executor
+            .remote_dep_file_cache(
+                manager,
+                &PreparedCommand {
+                    target: &action as _,
+                    request,
+                    prepared_action,
+                    digest_config: self.digest_config(),
+                },
+                self.cancellations,
+            )
+            .await
+    }
+
     fn unpack_command_execution_result(
         &mut self,
         executor_preference: ExecutorPreference,
@@ -774,6 +798,7 @@ mod tests {
             CommandExecutor::new(
                 Arc::new(DryRunExecutor::new(tracker, artifact_fs.clone())),
                 Arc::new(NoOpCommandOptionalExecutor {}),
+                Arc::new(NoOpCommandOptionalExecutor {}),
                 Arc::new(NoOpCacheUploader {}),
                 artifact_fs,
                 CommandGenerationOptions {

app/buck2_build_api/src/actions/execute/dice_data.rs

@@ -32,7 +32,8 @@ use crate::actions::artifact::get_artifact_fs::GetArtifactFs;
 pub struct CommandExecutorResponse {
     pub executor: Arc<dyn PreparedCommandExecutor>,
     pub platform: RE::Platform,
-    pub cache_checker: Arc<dyn PreparedCommandOptionalExecutor>,
+    pub action_cache_checker: Arc<dyn PreparedCommandOptionalExecutor>,
+    pub remote_dep_file_cache_checker: Arc<dyn PreparedCommandOptionalExecutor>,
     pub cache_uploader: Arc<dyn UploadCache>,
 }
 

app/buck2_build_api_tests/src/actions/calculation.rs

@@ -220,7 +220,8 @@ async fn make_default_dice_state(
             ));
             Ok(CommandExecutorResponse {
                 executor,
-                cache_checker: Arc::new(NoOpCommandOptionalExecutor {}),
+                action_cache_checker: Arc::new(NoOpCommandOptionalExecutor {}),
+                remote_dep_file_cache_checker: Arc::new(NoOpCommandOptionalExecutor {}),
                 platform: Default::default(),
                 cache_uploader: Arc::new(NoOpCacheUploader {}),
             })

app/buck2_execute/src/execute/command_executor.rs

@@ -77,7 +77,8 @@ pub struct CommandExecutor(Arc<CommandExecutorData>);
 
 struct CommandExecutorData {
     inner: Arc<dyn PreparedCommandExecutor>,
-    cache_checker: Arc<dyn PreparedCommandOptionalExecutor>,
+    action_cache_checker: Arc<dyn PreparedCommandOptionalExecutor>,
+    remote_dep_file_cache_checker: Arc<dyn PreparedCommandOptionalExecutor>,
     artifact_fs: ArtifactFs,
     options: CommandGenerationOptions,
     re_platform: RE::Platform,
@@ -87,15 +88,17 @@ struct CommandExecutorData {
 impl CommandExecutor {
     pub fn new(
         inner: Arc<dyn PreparedCommandExecutor>,
-        cache_checker: Arc<dyn PreparedCommandOptionalExecutor>,
+        action_cache_checker: Arc<dyn PreparedCommandOptionalExecutor>,
+        remote_dep_file_cache_checker: Arc<dyn PreparedCommandOptionalExecutor>,
         cache_uploader: Arc<dyn UploadCache>,
         artifact_fs: ArtifactFs,
         options: CommandGenerationOptions,
         re_platform: RE::Platform,
     ) -> Self {
         Self(Arc::new(CommandExecutorData {
             inner,
-            cache_checker,
+            action_cache_checker,
+            remote_dep_file_cache_checker,
             artifact_fs,
             options,
             re_platform,
@@ -123,7 +126,19 @@ impl CommandExecutor {
         cancellations: &CancellationContext,
     ) -> ControlFlow<CommandExecutionResult, CommandExecutionManager> {
         self.0
-            .cache_checker
+            .action_cache_checker
+            .maybe_execute(prepared_command, manager, cancellations)
+            .await
+    }
+
+    pub async fn remote_dep_file_cache(
+        &self,
+        manager: CommandExecutionManager,
+        prepared_command: &PreparedCommand<'_, '_>,
+        cancellations: &CancellationContext,
+    ) -> ControlFlow<CommandExecutionResult, CommandExecutionManager> {
+        self.0
+            .remote_dep_file_cache_checker
             .maybe_execute(prepared_command, manager, cancellations)
             .await
     }

app/buck2_execute_impl/src/executors/action_cache.rs

@@ -47,7 +47,6 @@ pub struct ActionCacheChecker {
     pub upload_all_actions: bool,
     pub knobs: ExecutorGlobalKnobs,
     pub paranoid: Option<ParanoidDownloader>,
-    pub remote_dep_file_checker: Arc<dyn PreparedCommandOptionalExecutor>,
 }
 
 enum CacheType {
@@ -234,7 +233,7 @@ impl PreparedCommandOptionalExecutor for ActionCacheChecker {
             &cache_type,
             details.clone(),
         ));
-        let result = query_action_cache_and_download_result(
+        query_action_cache_and_download_result(
             cache_type,
             &self.artifact_fs,
             &self.materializer,
@@ -249,17 +248,7 @@ impl PreparedCommandOptionalExecutor for ActionCacheChecker {
             self.knobs.log_action_keys,
             details,
         )
-        .await;
-
-        // If continue (not a cache hit), invoke the remote dep file cache checker
-        match result {
-            ControlFlow::Continue(manager) => {
-                self.remote_dep_file_checker
-                    .maybe_execute(command, manager, cancellations)
-                    .await
-            }
-            ControlFlow::Break(result) => ControlFlow::Break(result),
-        }
+        .await
     }
 }
 

app/buck2_execute_impl/src/executors/stacked.rs

@@ -17,7 +17,8 @@ use buck2_execute::execute::result::CommandExecutionResult;
 use buck2_futures::cancellation::CancellationContext;
 
 pub struct StackedExecutor<O, F> {
-    pub optional: O,
+    pub optional1: O,
+    pub optional2: O,
     pub fallback: F,
 }
 
@@ -34,7 +35,11 @@ where
         cancellations: &CancellationContext,
     ) -> CommandExecutionResult {
         let manager = self
-            .optional
+            .optional1
+            .maybe_execute(command, manager, cancellations)
+            .await?; // This actually returns if we get a response.
+        let manager = self
+            .optional2
             .maybe_execute(command, manager, cancellations)
             .await?; // This actually returns if we get a response.
 

app/buck2_server/src/daemon/common.rs

@@ -206,7 +206,8 @@ impl HasCommandExecutor for CommandExecutorFactory {
             return Ok(CommandExecutorResponse {
                 executor: Arc::new(local_executor_new(&LocalExecutorOptions::default())),
                 platform: Default::default(),
-                cache_checker: Arc::new(NoOpCommandOptionalExecutor {}),
+                action_cache_checker: Arc::new(NoOpCommandOptionalExecutor {}),
+                remote_dep_file_cache_checker: Arc::new(NoOpCommandOptionalExecutor {}),
                 cache_uploader: Arc::new(NoOpCacheUploader {}),
             });
         }
@@ -243,7 +244,8 @@ impl HasCommandExecutor for CommandExecutorFactory {
                     Some(CommandExecutorResponse {
                         executor: Arc::new(local_executor_new(local)),
                         platform: Default::default(),
-                        cache_checker: Arc::new(NoOpCommandOptionalExecutor {}),
+                        action_cache_checker: Arc::new(NoOpCommandOptionalExecutor {}),
+                        remote_dep_file_cache_checker: Arc::new(NoOpCommandOptionalExecutor {}),
                         cache_uploader: Arc::new(NoOpCacheUploader {}),
                     })
                 }
@@ -268,12 +270,15 @@ impl HasCommandExecutor for CommandExecutorFactory {
                     applicability = testing
                 )?;
 
-                let cache_checker_new = || -> Arc<dyn PreparedCommandOptionalExecutor> {
+                let cache_checker_new = || -> (Arc<dyn PreparedCommandOptionalExecutor>, Arc<dyn PreparedCommandOptionalExecutor>) {
                     if disable_caching {
-                        return Arc::new(NoOpCommandOptionalExecutor {}) as _;
+                        return (
+                            Arc::new(NoOpCommandOptionalExecutor {}) as _,
+                            Arc::new(NoOpCommandOptionalExecutor {}) as _,
+                        );
                     }
 
-                    let remote_dep_file_checker: Arc<dyn PreparedCommandOptionalExecutor> =
+                    let remote_dep_file_cache_checker: Arc<dyn PreparedCommandOptionalExecutor> =
                         if remote_options.remote_dep_file_cache_enabled {
                             Arc::new(RemoteDepFileCacheChecker {
                                 artifact_fs: artifact_fs.clone(),
@@ -288,20 +293,22 @@ impl HasCommandExecutor for CommandExecutorFactory {
                             Arc::new(NoOpCommandOptionalExecutor {}) as _
                         };
 
-                    if only_remote_dep_file_cache {
-                        remote_dep_file_checker
-                    } else {
-                        Arc::new(ActionCacheChecker {
-                            artifact_fs: artifact_fs.clone(),
-                            materializer: self.materializer.dupe(),
-                            re_client: self.get_prepared_re_client(remote_options.re_use_case),
-                            re_action_key: remote_options.re_action_key.clone(),
-                            upload_all_actions: self.upload_all_actions,
-                            knobs: self.executor_global_knobs.dupe(),
-                            paranoid: self.paranoid.dupe(),
-                            remote_dep_file_checker,
-                        }) as _
-                    }
+                    let action_cache_checker: Arc<dyn PreparedCommandOptionalExecutor> =
+                        if only_remote_dep_file_cache {
+                            Arc::new(NoOpCommandOptionalExecutor {}) as _
+                        } else {
+                            Arc::new(ActionCacheChecker {
+                                artifact_fs: artifact_fs.clone(),
+                                materializer: self.materializer.dupe(),
+                                re_client: self.get_prepared_re_client(remote_options.re_use_case),
+                                re_action_key: remote_options.re_action_key.clone(),
+                                upload_all_actions: self.upload_all_actions,
+                                knobs: self.executor_global_knobs.dupe(),
+                                paranoid: self.paranoid.dupe(),
+                            }) as _
+                        };
+
+                    (action_cache_checker, remote_dep_file_cache_checker)
                 };
 
                 let executor: Option<Arc<dyn PreparedCommandExecutor>> =
@@ -343,10 +350,13 @@ impl HasCommandExecutor for CommandExecutorFactory {
                                 let executor_preference = executor_preference
                                     .and(ExecutorPreference::DefaultErasePreferences)?;
 
+                                let (action_cache_checker, remote_dep_file_cache_checker) =
+                                    cache_checker_new();
                                 Some(Arc::new(HybridExecutor {
                                     local,
                                     remote: StackedExecutor {
-                                        optional: cache_checker_new(),
+                                        optional1: action_cache_checker,
+                                        optional2: remote_dep_file_cache_checker,
                                         fallback: remote,
                                     },
                                     level: HybridExecutionLevel::Full {
@@ -375,11 +385,15 @@ impl HasCommandExecutor for CommandExecutorFactory {
                         _ => None,
                     };
 
-                let cache_checker = if self.paranoid.is_some() {
-                    Arc::new(NoOpCommandOptionalExecutor {}) as _
-                } else {
-                    cache_checker_new()
-                };
+                let (action_cache_checker, remote_dep_file_cache_checker) =
+                    if self.paranoid.is_some() {
+                        (
+                            Arc::new(NoOpCommandOptionalExecutor {}) as _,
+                            Arc::new(NoOpCommandOptionalExecutor {}) as _,
+                        )
+                    } else {
+                        cache_checker_new()
+                    };
 
                 let cache_uploader = if force_cache_upload()? {
                     Arc::new(CacheUploader::new(
@@ -410,7 +424,8 @@ impl HasCommandExecutor for CommandExecutorFactory {
                 executor.map(|executor| CommandExecutorResponse {
                     executor,
                     platform: remote_options.re_properties.to_re_platform(),
-                    cache_checker,
+                    action_cache_checker,
+                    remote_dep_file_cache_checker,
                     cache_uploader,
                 })
             }