Materialize "large" files in a new store location and hardlink them in sandboxes

src/rust/engine/fs/fs_util/src/main.rs

@@ -524,7 +524,6 @@ async fn execute(top_match: &clap::ArgMatches) -> Result<(), ExitError> {
               destination,
               output_digest,
               &BTreeSet::new(),
-              None,
               Permissions::Writable,
             )
             .await?,
@@ -544,13 +543,7 @@ async fn execute(top_match: &clap::ArgMatches) -> Result<(), ExitError> {
         let digest = DirectoryDigest::from_persisted_digest(Digest::new(fingerprint, size_bytes));
         Ok(
           store
-            .materialize_directory(
-              destination,
-              digest,
-              &BTreeSet::new(),
-              None,
-              Permissions::Writable,
-            )
+            .materialize_directory(destination, digest, &BTreeSet::new(), Permissions::Writable)
             .await?,
         )
       }

src/rust/engine/fs/store/src/immutable_inputs.rs

@@ -51,69 +51,6 @@ impl ImmutableInputs {
     self.0.workdir.path()
   }
 
-  /// Returns an absolute Path to immutably consume the given Digest from.
-  pub(crate) async fn path_for_file(
-    &self,
-    digest: Digest,
-    is_executable: bool,
-  ) -> Result<PathBuf, StoreError> {
-    let cell = self.0.contents.lock().entry(digest).or_default().clone();
-
-    // We (might) need to initialize the value.
-    //
-    // Because this code executes a side-effect which could be observed elsewhere within this
-    // process (other threads can observe the contents of the temporary directory), we need to
-    // ensure that if this method is cancelled (via async Drop), whether the cell has been
-    // initialized or not stays in sync with whether the side-effect is visible.
-    //
-    // Making the initialization "cancellation safe", involves either:
-    //
-    //   1. Adding a Drop guard to "undo" the side-effect if we're dropped before we fully
-    //      initialize the cell.
-    //       * This is challenging to do correctly in this case, because the `Drop` guard cannot
-    //         be created until after initialization begins, but cannot be cleared until after the
-    //         cell has been initialized (i.e., after `get_or_try_init` returns).
-    //   2. Shielding ourselves from cancellation by `spawn`ing a new Task to guarantee that the
-    //      cell initialization always runs to completion.
-    //       * This would work, but would mean that we would finish initializing cells even when
-    //         work was cancelled. Cancellation usually means that the work is no longer necessary,
-    //         and so that could result in a lot of spurious IO (in e.g. warm cache cases which
-    //         never end up actually needing any inputs).
-    //       * An advanced variant of this approach would be to _pause_ work on materializing a
-    //         Digest when demand for it disappeared, and resume the work if another caller
-    //         requested that Digest.
-    //   3. Using anonymous destination paths, such that multiple attempts to initialize cannot
-    //      collide.
-    //       * This means that although the side-effect is visible, it can never collide.
-    //
-    // We take the final approach here currently (for simplicity's sake), but the advanced variant
-    // of approach 2 might eventually be worthwhile.
-    cell
-      .get_or_try_init(async {
-        let chroot = TempDir::new_in(self.0.workdir.path()).map_err(|e| {
-          format!(
-            "Failed to create a temporary directory for materialization of immutable input \
-          digest {digest:?}: {e}"
-          )
-        })?;
-
-        let dest = chroot.path().join(digest.hash.to_hex());
-        self
-          .0
-          .store
-          .materialize_file(dest.clone(), digest, Permissions::ReadOnly, is_executable)
-          .await?;
-
-        // Now that we've successfully initialized the destination, forget the TempDir so that it
-        // is not cleaned up.
-        let _ = chroot.into_path();
-
-        Ok(dest)
-      })
-      .await
-      .cloned()
-  }
-
   /// Returns an absolute Path to immutably consume the given Digest from.
   pub(crate) async fn path_for_dir(
     &self,
@@ -168,7 +105,6 @@ impl ImmutableInputs {
             dest.clone(),
             directory_digest,
             &BTreeSet::new(),
-            Some(self),
             Permissions::ReadOnly,
           )
           .await?;

src/rust/engine/fs/store/src/lib.rs

@@ -39,10 +39,9 @@ pub use crate::snapshot_ops::{SnapshotOps, SubsetParams};
 
 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::fmt::{self, Debug, Display};
-use std::fs::hard_link;
 use std::fs::OpenOptions;
 use std::future::Future;
-use std::io::{self, Read, Write};
+use std::io::{self, Write};
 use std::os::unix::fs::{symlink, OpenOptionsExt, PermissionsExt};
 use std::path::{Path, PathBuf};
 use std::sync::{Arc, Weak};
@@ -59,6 +58,7 @@ use fs::{
 use futures::future::{self, BoxFuture, Either, FutureExt, TryFutureExt};
 use grpc_util::prost::MessageExt;
 use hashing::Digest;
+use local::ByteStore;
 use parking_lot::Mutex;
 use prost::Message;
 use protos::gen::build::bazel::remote::execution::v2 as remexec;
@@ -73,12 +73,6 @@ const KILOBYTES: usize = 1024;
 const MEGABYTES: usize = 1024 * KILOBYTES;
 const GIGABYTES: usize = 1024 * MEGABYTES;
 
-/// How big a file must be to become an immutable input symlink.
-// NB: These numbers were chosen after micro-benchmarking the code on one machine at the time of
-// writing. They were chosen using a rough equation from the microbenchmarks that are optimized
-// for somewhere between 2 and 3 uses of the corresponding entry to "break even".
-const IMMUTABLE_FILE_SIZE_LIMIT: usize = 512 * KILOBYTES;
-
 mod local;
 #[cfg(test)]
 pub mod local_tests;
@@ -440,24 +434,15 @@ impl Store {
   ///
   /// Store a file locally by streaming its contents.
   ///
-  pub async fn store_file<F, R>(
+  pub async fn store_file(
     &self,
     initial_lease: bool,
     data_is_immutable: bool,
-    data_provider: F,
-  ) -> Result<Digest, String>
-  where
-    R: Read + Debug,
-    F: Fn() -> Result<R, io::Error> + Send + 'static,
-  {
+    src: PathBuf,
+  ) -> Result<Digest, String> {
     self
       .local
-      .store(
-        EntryType::File,
-        initial_lease,
-        data_is_immutable,
-        data_provider,
-      )
+      .store(EntryType::File, initial_lease, data_is_immutable, src)
       .await
   }
 
@@ -1203,7 +1188,6 @@ impl Store {
     destination: PathBuf,
     digest: DirectoryDigest,
     mutable_paths: &BTreeSet<RelativePath>,
-    immutable_inputs: Option<&ImmutableInputs>,
     perms: Permissions,
   ) -> Result<(), StoreError> {
     // Load the DigestTrie for the digest, and convert it into a mapping between a fully qualified
@@ -1231,7 +1215,6 @@ impl Store {
         false,
         &parent_to_child,
         &mutable_path_ancestors,
-        immutable_inputs,
         perms,
       )
       .await
@@ -1244,7 +1227,6 @@ impl Store {
     force_mutable: bool,
     parent_to_child: &'a HashMap<PathBuf, Vec<directory::Entry>>,
     mutable_paths: &'a BTreeSet<PathBuf>,
-    immutable_inputs: Option<&'a ImmutableInputs>,
     perms: Permissions,
   ) -> BoxFuture<'a, Result<(), StoreError>> {
     let store = self.clone();
@@ -1278,19 +1260,17 @@ impl Store {
           let path = destination.join(child.name().as_ref());
           let store = store.clone();
           child_futures.push(async move {
-            let can_be_immutable =
-              immutable_inputs.is_some() && !mutable_paths.contains(&path) && !force_mutable;
+            let can_be_immutable = !mutable_paths.contains(&path) && !force_mutable;
 
             match child {
               directory::Entry::File(f) => {
                 store
-                  .materialize_file_maybe_hardlink(
+                  .materialize_file_maybe_symlink(
                     path,
                     f.digest(),
                     perms,
                     f.is_executable(),
                     can_be_immutable,
-                    immutable_inputs,
                   )
                   .await
               }
@@ -1307,7 +1287,6 @@ impl Store {
                     mutable_paths.contains(&path),
                     parent_to_child,
                     mutable_paths,
-                    immutable_inputs,
                     perms,
                   )
                   .await
@@ -1334,22 +1313,25 @@ impl Store {
     .boxed()
   }
 
-  async fn materialize_file_maybe_hardlink(
+  async fn materialize_file_maybe_symlink(
     &self,
     destination: PathBuf,
     digest: Digest,
     perms: Permissions,
     is_executable: bool,
     can_be_immutable: bool,
-    immutable_inputs: Option<&ImmutableInputs>,
   ) -> Result<(), StoreError> {
-    if can_be_immutable && digest.size_bytes > IMMUTABLE_FILE_SIZE_LIMIT {
-      let dest_path = immutable_inputs
-        .unwrap()
-        .path_for_file(digest, is_executable)
-        .await?;
+    if can_be_immutable && ByteStore::uses_large_file_store(digest.size_bytes) {
       self
-        .materialize_hardlink(destination, dest_path.to_str().unwrap().to_string())
+        .materialize_symlink(
+          destination,
+          self
+            .local
+            .large_file_path(digest)
+            .to_str()
+            .unwrap()
+            .to_string(),
+        )
         .await
     } else {
       self
@@ -1401,15 +1383,6 @@ impl Store {
     Ok(())
   }
 
-  pub async fn materialize_hardlink(
-    &self,
-    destination: PathBuf,
-    target: String,
-  ) -> Result<(), StoreError> {
-    hard_link(target, destination)?;
-    Ok(())
-  }
-
   ///
   /// Returns files sorted by their path.
   ///