diff options
| -rw-r--r-- | Cargo.lock | 29 | ||||
| -rw-r--r-- | Cargo.toml | 1 | ||||
| -rw-r--r-- | crates/shirabe/Cargo.toml | 1 | ||||
| -rw-r--r-- | crates/shirabe/src/util/loop.rs | 21 |
4 files changed, 42 insertions, 10 deletions
@@ -563,6 +563,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" [[package]] +name = "futures" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + +[[package]] name = "futures-channel" version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -596,6 +611,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" [[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] name = "futures-sink" version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -613,8 +639,10 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ + "futures-channel", "futures-core", "futures-io", + "futures-macro", "futures-sink", "futures-task", "memchr", @@ -1998,6 +2026,7 @@ dependencies = [ "async-trait", "base64", "chrono", + "futures", "indexmap", "jsonschema", "md5", @@ -20,6 +20,7 @@ async-trait = "0.1.89" base64 = "0.22.1" chrono = { version = "0.4.44", features = ["serde"] } fastrand = "2.4.1" +futures = "0.3.32" indexmap = { version = "2.14.0", features = ["serde"] } jsonschema = { version = "0.46.6", default-features = false } md5 = "0.7.0" diff --git a/crates/shirabe/Cargo.toml b/crates/shirabe/Cargo.toml index 7ec9db82..00ce5211 100644 --- a/crates/shirabe/Cargo.toml +++ b/crates/shirabe/Cargo.toml @@ -15,6 +15,7 @@ anyhow.workspace = true async-trait.workspace = true base64.workspace = true chrono.workspace = true +futures.workspace = true indexmap.workspace = true jsonschema.workspace = true md5.workspace = true diff --git a/crates/shirabe/src/util/loop.rs b/crates/shirabe/src/util/loop.rs index 4efd7707..c2fa3b08 100644 --- a/crates/shirabe/src/util/loop.rs +++ b/crates/shirabe/src/util/loop.rs @@ -2,6 +2,8 @@ use crate::util::HttpDownloader; use crate::util::ProcessExecutor; +use futures::StreamExt; +use futures::stream::FuturesUnordered; use shirabe_external_packages::symfony::console::helper::ProgressBar; #[derive(Debug)] @@ -44,25 +46,24 @@ impl Loop { >, _progress: Option<&mut ProgressBar>, ) -> anyhow::Result<()> { + let mut pending: FuturesUnordered<_> = promises.into_iter().collect(); let mut uncaught: Option<anyhow::Error> = None; - // TODO(phase-c-promise): the asynchronous worker classes (HttpDownloader / ProcessExecutor) - // run single-threaded for now, so the promises are consumed serially. Once the workers run - // on a multi-thread runtime these futures should be driven concurrently instead of in order. + // TODO(phase-c-promise): promises are now polled concurrently via FuturesUnordered, but + // each individual future (HttpDownloader::add/add_copy etc.) still resolves through a + // blocking bridge (curl_runtime()/sync_executor::block_on), so real I/O overlap does not + // happen yet — the bridged future fully blocks the thread until it settles before the next + // one gets polled. That only changes once a single top-level Runtime replaces those bridges. // The PHP progress bar is tied to the worker active-job count and is also deferred until then. - for promise in promises { - if let Err(e) = promise.await + while let Some(result) = pending.next().await { + if let Err(e) = result && uncaught.is_none() { uncaught = Some(e); } } - if let Some(e) = uncaught { - return Err(e); - } - - Ok(()) + uncaught.map_or(Ok(()), Err) } pub fn abort_jobs(&self) { |