diff options
| -rw-r--r-- | Cargo.lock | 12 | ||||
| -rw-r--r-- | Cargo.toml | 2 | ||||
| -rw-r--r-- | crates/shirabe/src/downloader/file_downloader.rs | 27 | ||||
| -rw-r--r-- | crates/shirabe/src/installer/installation_manager.rs | 29 | ||||
| -rw-r--r-- | crates/shirabe/src/package/version/version_guesser.rs | 3 | ||||
| -rw-r--r-- | crates/shirabe/src/repository/composer_repository.rs | 11 | ||||
| -rw-r--r-- | crates/shirabe/src/util.rs | 1 | ||||
| -rw-r--r-- | crates/shirabe/src/util/http/curl_downloader.rs | 149 | ||||
| -rw-r--r-- | crates/shirabe/src/util/sync_executor.rs | 28 | ||||
| -rw-r--r-- | crates/shirabe/src/util/sync_helper.rs | 4 |
10 files changed, 147 insertions, 119 deletions
@@ -569,6 +569,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -589,6 +590,12 @@ dependencies = [ ] [[package]] +name = "futures-io" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" + +[[package]] name = "futures-sink" version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -607,7 +614,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ "futures-core", + "futures-io", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] @@ -1653,7 +1663,9 @@ dependencies = [ "base64", "bytes", "encoding_rs", + "futures-channel", "futures-core", + "futures-util", "h2", "http", "http-body", @@ -24,7 +24,7 @@ jsonschema = { version = "0.46.6", default-features = false } md5 = "0.7.0" mockall = "0.14.0" regex = "1.12.3" -reqwest = "0.13.4" +reqwest = { version = "0.13.4", features = ["blocking"] } serde = { version = "1.0.228", features = ["derive"] } serde_json = { version = "1.0.149", features = ["preserve_order"] } serde_urlencoded = "0.7.1" diff --git a/crates/shirabe/src/downloader/file_downloader.rs b/crates/shirabe/src/downloader/file_downloader.rs index d3e1f6c..a6334fb 100644 --- a/crates/shirabe/src/downloader/file_downloader.rs +++ b/crates/shirabe/src/downloader/file_downloader.rs @@ -23,6 +23,7 @@ use crate::util::Platform; use crate::util::ProcessExecutor; use crate::util::Silencer; use crate::util::Url as UrlUtil; +use crate::util::sync_executor; use anyhow::Result; use indexmap::IndexMap; use shirabe_php_shim::{ @@ -629,21 +630,17 @@ impl ChangeReportInterface for FileDownloader { .remove_directory(format!("{}_compare", target_dir))?; } - tokio::runtime::Runtime::new() - .unwrap() - .block_on(self.download( - package.clone(), - &format!("{}_compare", target_dir), - None, - false, - ))?; - tokio::runtime::Runtime::new() - .unwrap() - .block_on(self.install( - package.clone(), - &format!("{}_compare", target_dir), - false, - ))?; + sync_executor::block_on(self.download( + package.clone(), + &format!("{}_compare", target_dir), + None, + false, + ))?; + sync_executor::block_on(self.install( + package.clone(), + &format!("{}_compare", target_dir), + false, + ))?; let mut comparer = Comparer::new(); comparer.set_source(format!("{}_compare", target_dir)); diff --git a/crates/shirabe/src/installer/installation_manager.rs b/crates/shirabe/src/installer/installation_manager.rs index 3ce8076..a2cc228 100644 --- a/crates/shirabe/src/installer/installation_manager.rs +++ b/crates/shirabe/src/installer/installation_manager.rs @@ -17,6 +17,7 @@ use crate::package::PackageInterfaceHandle; use crate::repository::InstalledRepositoryInterface; use crate::util::Platform; use crate::util::r#loop::Loop; +use crate::util::sync_executor; use anyhow::Result; use indexmap::IndexMap; use shirabe_external_packages::seld::signal::SignalHandler; @@ -231,17 +232,15 @@ impl InstallationManager { } for batch_to_execute in batches { - tokio::runtime::Runtime::new().unwrap().block_on( - self.download_and_execute_batch( - repo, - batch_to_execute, - &mut cleanup_promises, - dev_mode, - run_scripts, - download_only, - all_operations.clone(), - ), - )?; + sync_executor::block_on(self.download_and_execute_batch( + repo, + batch_to_execute, + &mut cleanup_promises, + dev_mode, + run_scripts, + download_only, + all_operations.clone(), + ))?; } Ok(()) @@ -253,9 +252,7 @@ impl InstallationManager { match result { Ok(()) => {} Err(e) => { - tokio::runtime::Runtime::new() - .unwrap() - .block_on(self.run_cleanup(&cleanup_promises)); + sync_executor::block_on(self.run_cleanup(&cleanup_promises)); return Err(e); } } @@ -672,7 +669,7 @@ impl InstallationManager { PhpMixed::Array(http.into_iter().collect()), ); - tokio::runtime::Runtime::new().unwrap().block_on( + sync_executor::block_on( self.loop_ .borrow() .get_http_downloader() @@ -734,7 +731,7 @@ impl InstallationManager { PhpMixed::Array(http.into_iter().collect()), ); - tokio::runtime::Runtime::new().unwrap().block_on( + sync_executor::block_on( self.loop_ .borrow() .get_http_downloader() diff --git a/crates/shirabe/src/package/version/version_guesser.rs b/crates/shirabe/src/package/version/version_guesser.rs index f18f363..8e8cc32 100644 --- a/crates/shirabe/src/package/version/version_guesser.rs +++ b/crates/shirabe/src/package/version/version_guesser.rs @@ -10,6 +10,7 @@ use crate::util::HttpDownloader; use crate::util::Platform; use crate::util::ProcessExecutor; use crate::util::Svn as SvnUtil; +use crate::util::sync_executor; use anyhow::Result; use indexmap::IndexMap; use shirabe_external_packages::composer::pcre::{CaptureKey, Preg}; @@ -543,7 +544,7 @@ impl VersionGuesser { }, &scm_cmdline, ); - let mut process = tokio::runtime::Runtime::new().unwrap().block_on( + let mut process = sync_executor::block_on( self.process .borrow_mut() .execute_async(&cmd_line, Some(path)), diff --git a/crates/shirabe/src/repository/composer_repository.rs b/crates/shirabe/src/repository/composer_repository.rs index e3a5711..baf5768 100644 --- a/crates/shirabe/src/repository/composer_repository.rs +++ b/crates/shirabe/src/repository/composer_repository.rs @@ -32,6 +32,7 @@ use crate::util::HttpDownloader; use crate::util::Url; use crate::util::http::Response; use crate::util::r#loop::Loop; +use crate::util::sync_executor; use indexmap::IndexMap; use shirabe_external_packages::composer::pcre::{CaptureKey, Preg}; use shirabe_metadata_minifier::MetadataMinifier; @@ -964,9 +965,8 @@ impl ComposerRepository { continue; } - let spec = tokio::runtime::Runtime::new() - .unwrap() - .block_on(self.start_cached_async_download(&name, Some(&name)))?; + let spec = + sync_executor::block_on(self.start_cached_async_download(&name, Some(&name)))?; // [$response] = $spec; let response = spec @@ -1726,9 +1726,8 @@ impl ComposerRepository { } let version_parser = self.version_parser.clone(); - let spec = tokio::runtime::Runtime::new() - .unwrap() - .block_on(self.start_cached_async_download(&name, Some(&real_name)))?; + let spec = + sync_executor::block_on(self.start_cached_async_download(&name, Some(&real_name)))?; // [$response, $packagesSource] = $spec; let spec_list = spec.as_list().cloned().unwrap_or_default(); diff --git a/crates/shirabe/src/util.rs b/crates/shirabe/src/util.rs index bd51b66..35402d7 100644 --- a/crates/shirabe/src/util.rs +++ b/crates/shirabe/src/util.rs @@ -25,6 +25,7 @@ pub mod remote_filesystem; pub mod silencer; pub mod stream_context_factory; pub mod svn; +pub mod sync_executor; pub mod sync_helper; pub mod tar; pub mod tls_helper; diff --git a/crates/shirabe/src/util/http/curl_downloader.rs b/crates/shirabe/src/util/http/curl_downloader.rs index 019a5ab..0d8248b 100644 --- a/crates/shirabe/src/util/http/curl_downloader.rs +++ b/crates/shirabe/src/util/http/curl_downloader.rs @@ -1,9 +1,10 @@ //! ref: composer/src/Composer/Util/Http/CurlDownloader.php //! //! reqwest-based re-implementation. The libcurl multi-handle event loop of the PHP original is -//! replaced by reqwest + a tokio runtime: `download()` queues a job, and `tick()` drives one job -//! to completion by `block_on`-ing the HTTP request (mirroring the existing sync bridge used -//! elsewhere in the codebase, e.g. file_downloader.rs / installation_manager.rs). +//! replaced by a blocking reqwest client: `download()` queues a job, and `tick()` drives one job to +//! completion with a synchronous HTTP request. A blocking client is used (rather than the async +//! client + a per-downloader tokio runtime) so the request never starts a tokio runtime from within +//! another one when invoked through the codebase's sync bridges (see util/sync_executor.rs). //! //! The PHP control flow (insecure-URL check, redirect following, transport/status retries, //! authenticated-retry detection, max_file_size enforcement, atomic rename of the `~` temp file) @@ -66,9 +67,7 @@ pub struct CurlDownloader { /// Connection pool / cookie / TLS-session sharing — reqwest::Client handles this internally, /// replacing the PHP multiHandle + shareHandle. Redirects are disabled because we follow them /// manually (to control auth-header re-attachment), matching CURLOPT_FOLLOWLOCATION = false. - client: reqwest::Client, - /// tokio runtime used to `block_on` the async reqwest calls from the sync `tick()`. - runtime: tokio::runtime::Runtime, + client: reqwest::blocking::Client, jobs: IndexMap<i64, CurlJob>, next_id: i64, io: std::rc::Rc<std::cell::RefCell<dyn IOInterface>>, @@ -95,11 +94,11 @@ impl CurlDownloader { // - cookie_store(true) ~ CURL_LOCK_DATA_COOKIE // - redirect(none) ~ CURLOPT_FOLLOWLOCATION = false (we follow manually) // The libcurl version-specific multiplexing / accept-encoding workarounds are not needed. - // TODO: a brand-new tokio runtime + reqwest client is created per CurlDownloader; that is - // acceptable here (one HttpDownloader owns one CurlDownloader) but not pooled across them. + // TODO: a brand-new reqwest client is created per CurlDownloader; that is acceptable here + // (one HttpDownloader owns one CurlDownloader) but not pooled across them. // TODO: cookie sharing (CURL_LOCK_DATA_COOKIE) would need reqwest's `cookies` feature // (.cookie_store(true)); omitted as it is not required for package downloads. - let client = reqwest::Client::builder() + let client = reqwest::blocking::Client::builder() .pool_max_idle_per_host(8) .redirect(reqwest::redirect::Policy::none()) .build() @@ -107,14 +106,10 @@ impl CurlDownloader { // cannot proceed, mirroring PHP aborting when curl is missing. .expect("failed to build reqwest client for CurlDownloader"); - let runtime = tokio::runtime::Runtime::new() - .expect("failed to build tokio runtime for CurlDownloader"); - let auth_helper = AuthHelper::new(io.clone(), config.clone()); Self { client, - runtime, jobs: IndexMap::new(), next_id: 1, io, @@ -282,7 +277,7 @@ impl CurlDownloader { } // Drive every queued job to completion. The PHP multi-handle progressed all easy handles - // a little per tick(); here each tick() fully resolves one job (block_on per request), + // a little per tick(); here each tick() fully resolves one job (one blocking request), // which is observationally equivalent for the sync wait_id() loop that calls tick(). let ids: Vec<i64> = self.jobs.keys().copied().collect(); for id in ids { @@ -531,7 +526,7 @@ impl CurlDownloader { } } - /// Performs one HTTP request via reqwest (`block_on`), enforcing max_file_size and streaming the + /// Performs one blocking HTTP request via reqwest, enforcing max_file_size and streaming the /// body to the `~` temp file when in file mode. Replaces PHP's curl_setopt block + curl I/O. fn send_once( &self, @@ -570,76 +565,74 @@ impl CurlDownloader { let reqwest_method = reqwest::Method::from_bytes(method.as_bytes()).unwrap_or(reqwest::Method::GET); - self.runtime.block_on(async { - let mut builder = self.client.request(reqwest_method, url); - for header in &headers { - if let Some((name, value)) = header.split_once(':') { - // Skip the Connection header reqwest manages itself. - let name = name.trim(); - if name.eq_ignore_ascii_case("connection") { - continue; - } - builder = builder.header(name, value.trim()); + let mut builder = self.client.request(reqwest_method, url); + for header in &headers { + if let Some((name, value)) = header.split_once(':') { + // Skip the Connection header reqwest manages itself. + let name = name.trim(); + if name.eq_ignore_ascii_case("connection") { + continue; } + builder = builder.header(name, value.trim()); } - builder = builder.timeout(std::time::Duration::from_secs(timeout_secs as u64)); - if let Some(body) = body { - builder = builder.body(body); - } + } + builder = builder.timeout(std::time::Duration::from_secs(timeout_secs as u64)); + if let Some(body) = body { + builder = builder.body(body); + } - let resp = builder.send().await.map_err(|e| TransportError { - message: e.to_string(), - retryable: e.is_timeout() || e.is_connect() || e.is_request(), - is_connect: e.is_connect(), - was_timeout: e.is_timeout(), - is_max_file_size: false, - })?; + let resp = builder.send().map_err(|e| TransportError { + message: e.to_string(), + retryable: e.is_timeout() || e.is_connect() || e.is_request(), + is_connect: e.is_connect(), + was_timeout: e.is_timeout(), + is_max_file_size: false, + })?; - let status = resp.status().as_u16() as i64; - let status_line = format!( - "HTTP/{} {}", - match resp.version() { - reqwest::Version::HTTP_09 => "0.9", - reqwest::Version::HTTP_10 => "1.0", - reqwest::Version::HTTP_11 => "1.1", - reqwest::Version::HTTP_2 => "2", - reqwest::Version::HTTP_3 => "3", - _ => "1.1", - }, - resp.status() - ); - // Header lines start with the status line so Response::get_status_message can find it. - let mut headers_out: Vec<String> = vec![status_line]; - for (k, v) in resp.headers().iter() { - headers_out.push(format!("{}: {}", k, v.to_str().unwrap_or(""))); - } + let status = resp.status().as_u16() as i64; + let status_line = format!( + "HTTP/{} {}", + match resp.version() { + reqwest::Version::HTTP_09 => "0.9", + reqwest::Version::HTTP_10 => "1.0", + reqwest::Version::HTTP_11 => "1.1", + reqwest::Version::HTTP_2 => "2", + reqwest::Version::HTTP_3 => "3", + _ => "1.1", + }, + resp.status() + ); + // Header lines start with the status line so Response::get_status_message can find it. + let mut headers_out: Vec<String> = vec![status_line]; + for (k, v) in resp.headers().iter() { + headers_out.push(format!("{}: {}", k, v.to_str().unwrap_or(""))); + } - let body = Self::read_body_with_limit(resp, max_file_size, filename) - .await - .map_err(|(message, is_max_file_size)| TransportError { - message, - retryable: false, - is_connect: false, - was_timeout: false, - is_max_file_size, - })?; + let body = Self::read_body_with_limit(resp, max_file_size, filename).map_err( + |(message, is_max_file_size)| TransportError { + message, + retryable: false, + is_connect: false, + was_timeout: false, + is_max_file_size, + }, + )?; - Ok(RawResponse { - status, - headers: headers_out, - body, - }) + Ok(RawResponse { + status, + headers: headers_out, + body, }) } /// Reads the body, enforcing max_file_size, writing to the `~` temp file when in file mode. /// The `bool` in the error is `true` when the failure is a max_file_size violation. - async fn read_body_with_limit( - resp: reqwest::Response, + fn read_body_with_limit( + resp: reqwest::blocking::Response, max_file_size: Option<u64>, filename: Option<&str>, ) -> Result<Body, (String, bool)> { - use std::io::Write; + use std::io::{Read, Write}; let mut stream = resp; let mut written: u64 = 0; @@ -654,13 +647,15 @@ impl CurlDownloader { None => Sink::Memory(Vec::new()), }; + let mut buffer = [0u8; 16 * 1024]; loop { - let chunk = match stream.chunk().await { - Ok(Some(c)) => c, - Ok(None) => break, + let n = match stream.read(&mut buffer) { + Ok(0) => break, + Ok(n) => n, Err(e) => return Err((e.to_string(), false)), }; - written += chunk.len() as u64; + let chunk = &buffer[..n]; + written += n as u64; if let Some(max) = max_file_size && written > max { @@ -673,8 +668,8 @@ impl CurlDownloader { )); } match &mut sink { - Sink::File(f) => f.write_all(&chunk).map_err(|e| (e.to_string(), false))?, - Sink::Memory(buf) => buf.extend_from_slice(&chunk), + Sink::File(f) => f.write_all(chunk).map_err(|e| (e.to_string(), false))?, + Sink::Memory(buf) => buf.extend_from_slice(chunk), } } diff --git a/crates/shirabe/src/util/sync_executor.rs b/crates/shirabe/src/util/sync_executor.rs new file mode 100644 index 0000000..b7512c2 --- /dev/null +++ b/crates/shirabe/src/util/sync_executor.rs @@ -0,0 +1,28 @@ +//! Minimal synchronous future executor used as a drop-in for the `tokio::runtime::Runtime::new()` +//! + `block_on` sync bridges scattered across the codebase (repository / installer / downloader). +//! +//! Those bridges drive `async fn`s whose `.await` points all resolve synchronously: the only async +//! I/O (reqwest in `CurlDownloader`) is performed through a blocking client, so no reactor is +//! required. Nesting `tokio` runtimes is forbidden ("Cannot start a runtime from within a runtime"), +//! which is why this no-reactor executor exists; it can be nested freely. +//! +//! TODO(phase-e): remove this module once the async bridges are either made genuinely synchronous or +//! consolidated onto a single shared runtime driven from `main`. + +use std::future::Future; +use std::task::{Context, Poll}; + +/// Polls `fut` to completion on the current thread without any tokio runtime. +/// +/// Relies on the invariant that no awaited future parks on a reactor (it would otherwise spin). +pub fn block_on<F: Future>(fut: F) -> F::Output { + let mut fut = std::pin::pin!(fut); + let waker = std::task::Waker::noop(); + let mut cx = Context::from_waker(waker); + loop { + if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { + return value; + } + std::hint::spin_loop(); + } +} diff --git a/crates/shirabe/src/util/sync_helper.rs b/crates/shirabe/src/util/sync_helper.rs index 8988795..1014b86 100644 --- a/crates/shirabe/src/util/sync_helper.rs +++ b/crates/shirabe/src/util/sync_helper.rs @@ -179,9 +179,7 @@ impl SyncHelper { promise: Option<std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + '_>>>, ) -> Result<()> { if let Some(promise) = promise { - tokio::runtime::Runtime::new() - .unwrap() - .block_on(r#loop.borrow_mut().wait(vec![promise], None))?; + crate::util::sync_executor::block_on(r#loop.borrow_mut().wait(vec![promise], None))?; } Ok(()) } |