diff --git a/nativelink-config/src/cas_server.rs b/nativelink-config/src/cas_server.rs index deb1c281b96..bb55746500d 100644 --- a/nativelink-config/src/cas_server.rs +++ b/nativelink-config/src/cas_server.rs @@ -1316,6 +1316,17 @@ pub struct LocalWorkerConfig { )] pub max_download_timeout_s: usize, + /// How many input blobs this worker fetches into its store at once, + /// across every action it is staging. Each fetch is a read from the + /// CAS or the object store, so this bounds what one worker asks of + /// them at a time: several cold actions share this many fetches + /// instead of each opening its own. Blobs already in the store are + /// linked without one. + /// + /// Default: 64 + #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")] + pub max_concurrent_input_downloads: usize, + /// Maximum time to wait for action directory cleanup before timing out. /// Value in seconds. /// diff --git a/nativelink-worker/src/local_worker.rs b/nativelink-worker/src/local_worker.rs index a5f04b0f3fb..07f3566a583 100644 --- a/nativelink-worker/src/local_worker.rs +++ b/nativelink-worker/src/local_worker.rs @@ -1245,6 +1245,7 @@ pub async fn new_local_worker( Duration::from_millis(config.kill_grace_ms) }, set_tmpdir: config.set_tmpdir, + max_concurrent_input_downloads: config.max_concurrent_input_downloads, persistent_workers: persistent_worker_settings( config.persistent_workers.as_ref(), persistent_workers_namespaced, diff --git a/nativelink-worker/src/running_actions_manager.rs b/nativelink-worker/src/running_actions_manager.rs index dece024ae5e..34bd7dccd5a 100644 --- a/nativelink-worker/src/running_actions_manager.rs +++ b/nativelink-worker/src/running_actions_manager.rs @@ -80,7 +80,7 @@ use scopeguard::{ScopeGuard, guard}; use serde::Deserialize; use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; use tokio::process; -use tokio::sync::{Notify, Semaphore, oneshot, watch}; +use tokio::sync::{Notify, Semaphore, SemaphorePermit, oneshot, watch}; use tokio::time::Instant; use tokio_stream::wrappers::ReadDirStream; use tonic::Request; @@ -728,6 +728,16 @@ fn persistent_worker_request_arguments(argv: &[String]) -> Vec { /// overhead. const DOWNLOAD_TO_DIRECTORY_CONCURRENCY: usize = 64; +/// One permit per blob fetched into the worker's store, from the pool the +/// worker shares across every action it is staging. The pool is never +/// closed, so a refusal here is a bug, not a condition. +async fn acquire_download_permit(downloads: &Semaphore) -> Result, Error> { + downloads + .acquire() + .await + .map_err(|e| make_err!(Code::Internal, "Input download pool closed: {e:?}")) +} + /// Keeps every digest used by an action present in locally eviction-managed /// CAS tiers. /// @@ -835,7 +845,39 @@ pub fn download_to_directory<'a>( digest: &'a DigestInfo, current_directory: &'a str, ) -> BoxFuture<'a, Result<(), Error>> { - download_to_directory_with_lease(cas_store, filesystem_store, digest, current_directory, None) + async move { + // A pool of this call's own; the worker's shared one arrives through + // `prepare_action_inputs`. + let downloads = Semaphore::new(DOWNLOAD_TO_DIRECTORY_CONCURRENCY); + download_to_directory_bounded( + cas_store, + filesystem_store, + digest, + current_directory, + &downloads, + ) + .await + } + .boxed() +} + +/// `download_to_directory` drawing its fetches from `downloads`, the pool +/// the worker shares across the actions it stages. +pub fn download_to_directory_bounded<'a>( + cas_store: &'a FastSlowStore, + filesystem_store: Pin<&'a FilesystemStore>, + digest: &'a DigestInfo, + current_directory: &'a str, + downloads: &'a Semaphore, +) -> BoxFuture<'a, Result<(), Error>> { + download_to_directory_with_lease( + cas_store, + filesystem_store, + digest, + current_directory, + None, + downloads, + ) } fn download_to_directory_with_lease<'a>( @@ -844,21 +886,24 @@ fn download_to_directory_with_lease<'a>( digest: &'a DigestInfo, current_directory: &'a str, input_lease: Option>, + downloads: &'a Semaphore, ) -> BoxFuture<'a, Result<(), Error>> { async move { let (dirs, files, inline_nodes) = - collect_download_links(cas_store, digest, current_directory, input_lease).await?; + collect_download_links(cas_store, digest, current_directory, input_lease, downloads) + .await?; let (exec_files, plain_files): (Vec<_>, Vec<_>) = files.into_iter().partition(|file| file.executable); - let mut links = resolve_plain_files(cas_store, filesystem_store, plain_files).await?; - links.extend(resolve_exec_files(cas_store, filesystem_store, exec_files).await?); + let mut links = + resolve_plain_files(cas_store, filesystem_store, plain_files, downloads).await?; + links.extend(resolve_exec_files(cas_store, filesystem_store, exec_files, downloads).await?); // Every directory exists before anything is written into one, so no // later pass has to create a parent it happens to need. fs::create_dir_many(dirs).await?; - materialize_inline_nodes(cas_store, filesystem_store, inline_nodes).await?; - materialize_links(cas_store, filesystem_store, links).await + materialize_inline_nodes(cas_store, filesystem_store, inline_nodes, downloads).await?; + materialize_links(cas_store, filesystem_store, links, downloads).await } .boxed() } @@ -1054,6 +1099,7 @@ async fn materialize_links( cas_store: &FastSlowStore, filesystem_store: Pin<&FilesystemStore>, links: Vec, + downloads: &Semaphore, ) -> Result<(), Error> { let pairs = links .iter() @@ -1081,7 +1127,7 @@ async fn materialize_links( // once, so they are fetched back together, not one round trip each. let repairs: Vec>> = evicted .into_iter() - .map(|link| relink_and_report(cas_store, filesystem_store, link).boxed()) + .map(|link| relink_and_report(cas_store, filesystem_store, link, downloads).boxed()) .collect(); futures::stream::iter(repairs) .buffer_unordered(INPUT_RELINK_CONCURRENCY) @@ -1093,7 +1139,9 @@ async fn relink_and_report( cas_store: &FastSlowStore, filesystem_store: Pin<&FilesystemStore>, link: &PendingLink, + downloads: &Semaphore, ) -> Result<(), Error> { + let _permit = acquire_download_permit(downloads).await?; info!( digest = %link.digest, src_path = ?link.src, @@ -1201,6 +1249,7 @@ async fn resolve_plain_files( cas_store: &FastSlowStore, filesystem_store: Pin<&FilesystemStore>, files: Vec, + downloads: &Semaphore, ) -> Result, Error> { let mut tasks = Vec::with_capacity(files.len()); for file in files { @@ -1211,6 +1260,7 @@ async fn resolve_plain_files( { entry } else { + let _permit = acquire_download_permit(downloads).await?; fetch_and_resolve( file.digest, || cas_store.populate_fast_store(file.digest.into()), @@ -1259,6 +1309,7 @@ async fn resolve_exec_files( cas_store: &FastSlowStore, filesystem_store: Pin<&FilesystemStore>, files: Vec, + downloads: &Semaphore, ) -> Result, Error> { let digests: Vec = files.iter().map(|file| file.digest).collect(); let sources = filesystem_store @@ -1271,6 +1322,7 @@ async fn resolve_exec_files( let src_path = if let Ok(src_path) = source { src_path } else { + let _permit = acquire_download_permit(downloads).await?; fetch_and_resolve( file.digest, || cas_store.populate_fast_store(file.digest.into()), @@ -1302,14 +1354,21 @@ fn collect_download_links<'a>( digest: &'a DigestInfo, current_directory: &'a str, input_lease: Option>, + downloads: &'a Semaphore, ) -> BoxFuture<'a, Result> { async move { if let Some(input_lease) = &input_lease { input_lease.lease_digest(digest); } - let directory = get_and_decode_digest::(cas_store, digest.into()) - .await - .err_tip(|| "Converting digest to Directory")?; + // The permit covers the fetch alone, never the walk below it: a + // parent holding one while its children wait for theirs would + // deadlock a tree deeper than the pool. + let directory = { + let _permit = acquire_download_permit(downloads).await?; + get_and_decode_digest::(cas_store, digest.into()) + .await + .err_tip(|| "Converting digest to Directory")? + }; let mut futures = Vec::new(); let mut pending_files = Vec::new(); let mut inline_nodes = Vec::new(); @@ -1372,6 +1431,7 @@ fn collect_download_links<'a>( &digest, &new_directory_path, input_lease, + downloads, ) .await .err_tip(|| format!("in download_to_directory : {new_directory_path}"))?; @@ -1466,6 +1526,7 @@ async fn materialize_inline_nodes( cas_store: &FastSlowStore, filesystem_store: Pin<&FilesystemStore>, nodes: Vec, + downloads: &Semaphore, ) -> Result<(), Error> { let mut futures = Vec::with_capacity(nodes.len()); for node in nodes { @@ -1482,7 +1543,10 @@ async fn materialize_inline_nodes( mtime, custom_unix_mode, } => { - cas_store.populate_fast_store(digest.into()).await?; + { + let _permit = acquire_download_permit(downloads).await?; + cas_store.populate_fast_store(digest.into()).await?; + } async move { if is_zero_digest(digest) { // Zero-digest files are never persisted by the @@ -1572,6 +1636,7 @@ pub async fn prepare_action_inputs( filesystem_store: Pin<&FilesystemStore>, digest: &DigestInfo, work_directory: &str, + downloads: &Semaphore, ) -> Result<(), Error> { prepare_action_inputs_with_lease( directory_cache, @@ -1580,6 +1645,7 @@ pub async fn prepare_action_inputs( digest, work_directory, None, + downloads, ) .await } @@ -1591,6 +1657,7 @@ async fn prepare_action_inputs_with_lease( digest: &DigestInfo, work_directory: &str, input_lease: Option>, + downloads: &Semaphore, ) -> Result<(), Error> { // Try cache first if available. Cache hits are safe while an action lease // is active because the directory cache pins its own materialized tree @@ -1662,6 +1729,7 @@ async fn prepare_action_inputs_with_lease( digest, work_directory, input_lease, + downloads, ) .await } @@ -2455,6 +2523,7 @@ impl RunningActionImpl { &self.action_info.input_root_digest, &self.work_directory, self.input_lease.clone(), + &self.running_actions_manager.input_downloads, )); let err = match try_join(command_fut, inputs_fut).await { Ok((command, ())) => { @@ -4335,6 +4404,9 @@ pub struct ExecutionConfiguration { pub kill_grace: Duration, /// Point `TMPDIR` at a directory of the action's own, removed with it. pub set_tmpdir: bool, + /// How many input blobs the worker fetches into its store at once, + /// across every action it is staging; 0 takes the default of 64. + pub max_concurrent_input_downloads: usize, /// Buck2-only helper, configured for a dedicated execution container. pub buck2_file_capture: Option, /// If set, will be executed instead of the first argument passed in the @@ -4709,6 +4781,10 @@ pub struct RunningActionsManagerImpl { /// Whether active action inputs are leased against eviction in the local /// CAS tiers (opt-in via `experimental_active_input_leases`). active_input_leases: bool, + /// The pool every input fetch draws a permit from, shared by all the + /// actions this worker stages; see + /// `ExecutionConfiguration::max_concurrent_input_downloads`. + input_downloads: Semaphore, persistent_worker_pool: PersistentWorkerPool, /// Retires idle persistent workers; ends with the manager. _persistent_worker_sweeper: JoinHandleDropGuard<()>, @@ -4716,6 +4792,12 @@ pub struct RunningActionsManagerImpl { } impl RunningActionsManagerImpl { + /// Permits free in the input fetch pool right now; with nothing staging, + /// the pool's size. + pub fn free_input_downloads(&self) -> usize { + self.input_downloads.available_permits() + } + pub fn new_with_callbacks( args: RunningActionsManagerArgs<'_>, callbacks: Callbacks, @@ -4734,9 +4816,16 @@ impl RunningActionsManagerImpl { let persistent_worker_pool = PersistentWorkerPool::new(args.execution_configuration.persistent_workers.pool); let persistent_worker_sweeper = persistent_worker_pool.spawn_sweeper(); + let input_downloads = Semaphore::new( + match args.execution_configuration.max_concurrent_input_downloads { + 0 => DOWNLOAD_TO_DIRECTORY_CONCURRENCY, + n => n, + }, + ); Ok(Self { root_action_directory: args.root_action_directory, execution_configuration: args.execution_configuration, + input_downloads, cas_store: args.cas_store, filesystem_store, upload_action_results: UploadActionResults::new( diff --git a/nativelink-worker/tests/running_actions_manager_test.rs b/nativelink-worker/tests/running_actions_manager_test.rs index 66c721b801e..37e5d017b6f 100644 --- a/nativelink-worker/tests/running_actions_manager_test.rs +++ b/nativelink-worker/tests/running_actions_manager_test.rs @@ -93,6 +93,8 @@ mod tests { #[cfg(target_os = "linux")] use nativelink_worker::running_actions_manager::MemoryLimit; #[cfg(target_family = "unix")] + use nativelink_worker::running_actions_manager::download_to_directory_bounded; + #[cfg(target_family = "unix")] use nativelink_worker::running_actions_manager::free_disk_kb; use nativelink_worker::running_actions_manager::{ Callbacks, ExecutionConfiguration, KillReason, PersistentWorkersSettings, @@ -102,6 +104,8 @@ mod tests { }; use pretty_assertions::assert_eq; use prost::Message; + #[cfg(target_family = "unix")] + use tokio::sync::Semaphore; use tokio::sync::oneshot; #[cfg(target_family = "unix")] use tokio::sync::watch; @@ -2907,6 +2911,7 @@ exit 0 Arc::new(RunningActionsManagerImpl::new(RunningActionsManagerArgs { root_action_directory: root_action_directory.clone(), execution_configuration: ExecutionConfiguration { + max_concurrent_input_downloads: 0, persistent_workers: PersistentWorkersSettings::default(), max_captured_output_bytes: 0, resource_enforcement: None, @@ -3079,6 +3084,7 @@ exit 0 Arc::new(RunningActionsManagerImpl::new(RunningActionsManagerArgs { root_action_directory: root_action_directory.clone(), execution_configuration: ExecutionConfiguration { + max_concurrent_input_downloads: 0, persistent_workers: PersistentWorkersSettings::default(), max_captured_output_bytes: 0, resource_enforcement: None, @@ -3730,6 +3736,322 @@ exit 0 Ok(()) } + /// A memory store that counts the reads in flight against it, standing + /// in for the CAS or the object store a worker fetches inputs from. + /// Unix only, with the one test that drives it. + #[cfg(target_family = "unix")] + #[derive(Debug)] + struct MeteredSlowStore { + inner: Arc, + in_flight: AtomicU64, + max_in_flight: AtomicU64, + } + + #[cfg(target_family = "unix")] + impl MeteredSlowStore { + fn new() -> Arc { + Arc::new(Self { + inner: MemoryStore::new(&MemorySpec::default()), + in_flight: AtomicU64::new(0), + max_in_flight: AtomicU64::new(0), + }) + } + + fn max_in_flight(&self) -> u64 { + self.max_in_flight.load(Ordering::SeqCst) + } + } + + #[cfg(target_family = "unix")] + impl MetricsComponent for MeteredSlowStore { + fn publish( + &self, + _kind: MetricKind, + _field_metadata: MetricFieldData, + ) -> Result { + Ok(MetricPublishKnownKindData::Component) + } + } + + #[cfg(target_family = "unix")] + #[async_trait] + impl StoreDriver for MeteredSlowStore { + async fn post_init(self: Arc) -> Result<(), Error> { + Ok(()) + } + + async fn has_with_results( + self: Pin<&Self>, + keys: &[StoreKey<'_>], + results: &mut [Option], + ) -> Result<(), Error> { + self.inner + .as_store_driver_pin() + .has_with_results(keys, results) + .await + } + + async fn update( + self: Pin<&Self>, + key: StoreKey<'_>, + reader: DropCloserReadHalf, + size_info: UploadSizeInfo, + ) -> Result { + self.inner + .as_store_driver_pin() + .update(key, reader, size_info) + .await + } + + async fn get_part( + self: Pin<&Self>, + key: StoreKey<'_>, + writer: &mut DropCloserWriteHalf, + offset: u64, + length: Option, + ) -> Result<(), Error> { + let now_in_flight = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1; + self.max_in_flight + .fetch_max(now_in_flight, Ordering::SeqCst); + // Let every other read that is ready start before this one + // answers, so an unbounded caller shows as one. + for _ in 0..4 { + tokio::task::yield_now().await; + } + let result = self + .inner + .as_store_driver_pin() + .get_part(key, writer, offset, length) + .await; + self.in_flight.fetch_sub(1, Ordering::SeqCst); + result + } + + fn inner_store(&self, _key: Option) -> &dyn StoreDriver { + self + } + + fn as_any<'a>(&'a self) -> &'a (dyn core::any::Any + Sync + Send + 'static) { + self + } + + fn as_any_arc(self: Arc) -> Arc { + self + } + + fn register_remove_callback( + self: Arc, + _callback: Arc, + ) -> Result<(), Error> { + Ok(()) + } + } + + #[cfg(target_family = "unix")] + default_health_status_indicator!(MeteredSlowStore); + + /// A fast filesystem tier over a metered slow store, as a worker's CAS + /// is laid out. + #[cfg(target_family = "unix")] + async fn metered_cas() -> Result< + ( + Arc, + Arc, + Arc, + ), + Error, + > { + let fast_config = FilesystemSpec { + content_path: make_temp_path("content_path"), + temp_path: make_temp_path("temp_path"), + eviction_policy: None, + ..Default::default() + }; + let fast_store = FilesystemStore::new(&fast_config).await?; + let slow_store = MeteredSlowStore::new(); + let cas_store = FastSlowStore::new( + &FastSlowSpec { + fast: StoreSpec::Filesystem(fast_config), + slow: StoreSpec::Memory(MemorySpec::default()), + fast_direction: StoreDirection::default(), + slow_direction: StoreDirection::default(), + bypass_dedup_threshold_bytes: 0, + }, + Store::new(fast_store.clone()), + Store::new(slow_store.clone()), + ); + Ok((fast_store, slow_store, cas_store)) + } + + #[cfg(target_family = "unix")] + const fn digest_bytes(tree: u8, index: u8) -> [u8; 32] { + let mut bytes = [0u8; 32]; + bytes[0] = tree; + bytes[1] = index; + bytes + } + + /// A flat tree of `files` cold files `f0..` in the slow store, every + /// digest distinct per `tree`, so two trees never share a blob. + #[cfg(target_family = "unix")] + async fn seed_tree( + slow_store: &MeteredSlowStore, + tree: u8, + files: u8, + ) -> Result { + let mut nodes = Vec::new(); + for i in 0..files { + let content = format!("{tree}:{i:030}"); + let digest = DigestInfo::new(digest_bytes(tree, i + 1), content.len() as u64); + slow_store + .inner + .as_ref() + .update_oneshot(digest, content.into()) + .await?; + nodes.push(FileNode { + name: format!("f{i}"), + digest: Some(digest.into()), + is_executable: false, + node_properties: None, + }); + } + let root_digest = DigestInfo::new(digest_bytes(tree, 0xEE), 32); + let root = Directory { + files: nodes, + ..Default::default() + }; + slow_store + .inner + .as_ref() + .update_oneshot(root_digest, root.encode_to_vec().into()) + .await?; + Ok(root_digest) + } + + #[cfg(target_family = "unix")] + async fn assert_tree_downloaded(dir: &str, tree: u8, files: u8) -> Result<(), Error> { + for i in 0..files { + let data = tokio::fs::read(format!("{dir}/f{i}")).await?; + assert_eq!(data, format!("{tree}:{i:030}").into_bytes()); + } + Ok(()) + } + + /// Every blob fetched into the store takes a permit from the one pool, + /// so a tree of many cold files never has more reads open against the + /// slow store than the pool holds. + #[cfg(target_family = "unix")] + #[nativelink_test] + async fn download_to_directory_bounds_fetches_by_the_pool() + -> Result<(), Box> { + const FILES: u8 = 40; + const PERMITS: usize = 3; + + let (fast_store, slow_store, cas_store) = metered_cas().await?; + let root_digest = seed_tree(&slow_store, 1, FILES).await?; + + let download_dir = make_temp_path("download_dir"); + fs::create_dir_all(&download_dir).await?; + let downloads = Semaphore::new(PERMITS); + download_to_directory_bounded( + cas_store.as_ref(), + fast_store.as_pin(), + &root_digest, + &download_dir, + &downloads, + ) + .await?; + + let peak = slow_store.max_in_flight(); + assert!( + peak >= 1 && peak <= PERMITS as u64, + "{peak} reads in flight against a pool of {PERMITS}" + ); + assert_tree_downloaded(&download_dir, 1, FILES).await?; + Ok(()) + } + + /// The pool is the worker's, not the call's: two actions staging cold + /// trees at the same time share its permits, so together they never + /// have more reads open against the slow store than it holds. + #[cfg(target_family = "unix")] + #[nativelink_test] + async fn concurrent_downloads_share_one_pool() -> Result<(), Box> { + const FILES: u8 = 40; + const PERMITS: usize = 3; + + let (fast_store, slow_store, cas_store) = metered_cas().await?; + let root_a = seed_tree(&slow_store, 1, FILES).await?; + let root_b = seed_tree(&slow_store, 2, FILES).await?; + let dir_a = make_temp_path("download_dir_a"); + let dir_b = make_temp_path("download_dir_b"); + fs::create_dir_all(&dir_a).await?; + fs::create_dir_all(&dir_b).await?; + + let downloads = Semaphore::new(PERMITS); + tokio::try_join!( + download_to_directory_bounded( + cas_store.as_ref(), + fast_store.as_pin(), + &root_a, + &dir_a, + &downloads, + ), + download_to_directory_bounded( + cas_store.as_ref(), + fast_store.as_pin(), + &root_b, + &dir_b, + &downloads, + ), + )?; + + let peak = slow_store.max_in_flight(); + assert!( + peak >= 1 && peak <= PERMITS as u64, + "{peak} reads in flight across two downloads sharing a pool of {PERMITS}" + ); + assert_tree_downloaded(&dir_a, 1, FILES).await?; + assert_tree_downloaded(&dir_b, 2, FILES).await?; + Ok(()) + } + + /// `max_concurrent_input_downloads` of 0 takes the default pool of 64; + /// any other value is the pool's size. + #[nativelink_test] + async fn input_download_pool_defaults_to_sixty_four() -> Result<(), Box> + { + let (_, _, cas_store, _) = setup_stores().await?; + for (configured, expected) in [(0, 64), (7, 7)] { + let manager = RunningActionsManagerImpl::new(RunningActionsManagerArgs { + root_action_directory: make_temp_path("root_action_directory"), + execution_configuration: ExecutionConfiguration { + max_concurrent_input_downloads: configured, + ..Default::default() + }, + cas_store: cas_store.clone(), + ac_store: None, + historical_store: Store::new(cas_store.clone()), + upload_action_result_config: &UploadActionResultConfig { + upload_ac_results_strategy: UploadCacheResultsStrategy::Never, + ..Default::default() + }, + max_action_timeout: Duration::from_mins(10), + max_upload_timeout: Duration::from_secs(DEFAULT_MAX_UPLOAD_TIMEOUT), + max_download_timeout: Duration::from_secs(DEFAULT_MAX_DOWNLOAD_TIMEOUT), + max_cleanup_wait: Duration::from_secs(DEFAULT_MAX_CLEANUP_WAIT), + max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), + timeout_handled_externally: false, + active_input_leases: false, + directory_cache: None, + #[cfg(target_os = "linux")] + use_namespaces: use_namespaces(), + })?; + assert_eq!(manager.free_input_downloads(), expected); + } + Ok(()) + } + /// A store whose uploads wait at a gate, counting how many are waiting /// at once. Every upload of an output file arrives here with that file /// open, so the peak is the number of files the worker had open together. @@ -6051,6 +6373,7 @@ exit 1 Arc::new(RunningActionsManagerImpl::new(RunningActionsManagerArgs { root_action_directory: root_action_directory.clone(), execution_configuration: ExecutionConfiguration { + max_concurrent_input_downloads: 0, persistent_workers: PersistentWorkersSettings::default(), max_captured_output_bytes: 0, resource_enforcement: None, @@ -8609,6 +8932,7 @@ done Arc::new(RunningActionsManagerImpl::new(RunningActionsManagerArgs { root_action_directory: root_action_directory.clone(), execution_configuration: ExecutionConfiguration { + max_concurrent_input_downloads: 0, persistent_workers: PersistentWorkersSettings::default(), max_captured_output_bytes: 0, resource_enforcement: None, @@ -8763,6 +9087,7 @@ done Arc::new(RunningActionsManagerImpl::new(RunningActionsManagerArgs { root_action_directory: root_action_directory.clone(), execution_configuration: ExecutionConfiguration { + max_concurrent_input_downloads: 0, persistent_workers: PersistentWorkersSettings::default(), max_captured_output_bytes: 0, resource_enforcement: None,