diff --git a/README.md b/README.md index 30ea45c..a720b31 100644 --- a/README.md +++ b/README.md @@ -17,9 +17,14 @@ for proxy in resolve_proxy(&url)? { ``` Results mirror PAC semantics: `"PROXY a:8080; DIRECT"` → an *ordered* fallback -list `[Http("a:8080"), Direct]`. The API is synchronous; PAC evaluation runs -on a dedicated worker thread. Calls may block on network I/O up to configured -timeouts, so use `spawn_blocking` (or similar) from async runtimes. +list `[Http("a:8080"), Direct]`. The base API is synchronous and may block on +network I/O up to configured timeouts. With the `tokio` feature, use +`resolve_proxy_async` instead: blocking resolution is scheduled onto a lazy background thread, identical concurrent calls share a result, and distinct targets +queue without consuming async-runtime blocking threads. + +```rust +let proxies = os_proxy_resolver::resolve_proxy_async(&url).await?; +``` ## Architecture diff --git a/src/lib.rs b/src/lib.rs index cddf96a..acbce0a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -147,3 +147,13 @@ pub fn pac_wasm_artifact_size() -> usize { pub fn resolve_proxy(url: &url::Url) -> Result> { ProxyResolver::global().resolve_proxy(url) } + +/// Resolve the ordered proxy list asynchronously using the process-wide +/// [`ProxyResolver`]. Blocking resolution is scheduled onto a dedicated background thread +/// (PAC evaluation uses the existing PAC worker thread), and identical concurrent calls share one result. +/// +/// See [`ProxyResolver::resolve_proxy_async`] for details. +#[cfg(feature = "tokio")] +pub async fn resolve_proxy_async(url: &url::Url) -> Result> { + ProxyResolver::global().resolve_proxy_async(url).await +} diff --git a/src/resolver.rs b/src/resolver.rs index 46510c3..2298a6b 100644 --- a/src/resolver.rs +++ b/src/resolver.rs @@ -15,6 +15,12 @@ use std::sync::{Arc, Mutex, MutexGuard, OnceLock}; use std::time::{Duration, Instant}; use url::Url; +#[cfg(feature = "tokio")] +use std::sync::Weak; + +#[cfg(feature = "tokio")] +const ASYNC_RESOLUTION_QUEUE_CAPACITY: usize = 256; + /// Tunables for a [`ProxyResolver`]. Construct with `Default::default()` and /// override fields as needed. #[derive(Debug, Clone)] @@ -108,6 +114,12 @@ struct Inner { my_ip: Mutex)>>, #[cfg(windows)] winhttp: OnceLock>, + #[cfg(feature = "tokio")] + async_worker: OnceLock>, + #[cfg(all(test, feature = "tokio"))] + async_resolution_starts: std::sync::atomic::AtomicUsize, + #[cfg(all(test, feature = "tokio"))] + async_resolution_delay: Mutex>, } struct ConfigCache { @@ -133,6 +145,172 @@ struct WpadCache { script: Option>, } +#[cfg(feature = "tokio")] +#[derive(Clone, PartialEq, Eq, Hash)] +struct AsyncResolutionKey { + generation: u64, + url: String, +} + +#[cfg(feature = "tokio")] +type AsyncResolutionSender = tokio::sync::watch::Sender>>>; + +#[cfg(feature = "tokio")] +struct AsyncResolutionRequest { + key: AsyncResolutionKey, + url: Url, +} + +#[cfg(feature = "tokio")] +struct AsyncSubmissionGuard { + key: AsyncResolutionKey, + inflight: Arc>>, + armed: bool, +} + +#[cfg(feature = "tokio")] +impl AsyncSubmissionGuard { + fn new( + key: AsyncResolutionKey, + inflight: Arc>>, + ) -> Self { + Self { + key, + inflight, + armed: true, + } + } + + fn disarm(&mut self) { + self.armed = false; + } +} + +#[cfg(feature = "tokio")] +impl Drop for AsyncSubmissionGuard { + fn drop(&mut self) { + if self.armed { + lock(&self.inflight).remove(&self.key); + } + } +} + +/// One blocking resolver thread per [`ProxyResolver`], created lazily by the +/// async API. The in-flight map lets identical callers await the same result +/// without occupying one blocking thread each; distinct targets queue on the +/// single worker so blocking OS/PAC/WPAD work has fixed thread consumption. +#[cfg(feature = "tokio")] +struct AsyncResolverWorker { + tx: tokio::sync::mpsc::Sender, + inflight: Arc>>, +} + +#[cfg(feature = "tokio")] +impl AsyncResolverWorker { + fn new(inner: Weak) -> std::result::Result { + let (tx, mut rx) = + tokio::sync::mpsc::channel::(ASYNC_RESOLUTION_QUEUE_CAPACITY); + let inflight: Arc>> = + Arc::new(Mutex::new(HashMap::new())); + let worker_inflight = inflight.clone(); + std::thread::Builder::new() + .name("os-proxy-resolver".to_string()) + .spawn(move || { + while let Some(request) = rx.blocking_recv() { + let result = match inner.upgrade() { + Some(inner) => { + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + #[cfg(all(test, feature = "tokio"))] + { + use std::sync::atomic::Ordering; + + inner.async_resolution_starts.fetch_add(1, Ordering::SeqCst); + if let Some(delay) = *lock(&inner.async_resolution_delay) { + std::thread::sleep(delay); + } + } + ProxyResolver { inner }.resolve_proxy(&request.url) + })) + .unwrap_or_else(|_| { + Err(Error::Platform( + "proxy resolver worker panicked during resolution".to_string(), + )) + }) + } + None => Err(Error::Platform( + "proxy resolver was dropped during async resolution".to_string(), + )), + }; + if let Some(sender) = lock(&worker_inflight).get(&request.key).cloned() { + let _ = sender.send(Some(result)); + lock(&worker_inflight).remove(&request.key); + } + } + }) + .map_err(|error| format!("failed to spawn proxy resolver worker: {error}"))?; + Ok(Self { tx, inflight }) + } + + async fn resolve(&self, generation: u64, url: &Url) -> Result> { + let mut key_url = url.clone(); + key_url.set_fragment(None); + let _ = key_url.set_username(""); + let _ = key_url.set_password(None); + let key = AsyncResolutionKey { + generation, + url: key_url.as_str().to_string(), + }; + let (mut receiver, leader) = { + let mut inflight = lock(&self.inflight); + match inflight.get(&key) { + Some(sender) => (sender.subscribe(), false), + None => { + let (sender, receiver) = tokio::sync::watch::channel(None); + inflight.insert(key.clone(), sender); + (receiver, true) + } + } + }; + + if leader { + // If this future is canceled while the bounded queue is full, + // remove its unresolved in-flight entry. Dropping the sender wakes + // followers with a closed-channel error so a later call can retry. + let mut submission = AsyncSubmissionGuard::new(key.clone(), self.inflight.clone()); + if self + .tx + .send(AsyncResolutionRequest { + key: key.clone(), + url: url.clone(), + }) + .await + .is_err() + { + let error = Error::Platform("proxy resolver worker is unavailable".to_string()); + if let Some(sender) = lock(&self.inflight).remove(&key) { + let _ = sender.send(Some(Err(error.clone()))); + } + submission.disarm(); + return Err(error); + } + submission.disarm(); + } + + let result = match receiver.wait_for(Option::is_some).await { + Ok(result) => match result.clone() { + Some(result) => result, + None => Err(Error::Platform( + "proxy resolver worker returned no result".to_string(), + )), + }, + Err(_) => Err(Error::Platform( + "proxy resolver worker stopped before returning a result".to_string(), + )), + }; + result + } +} + fn lock(m: &Mutex) -> MutexGuard<'_, T> { m.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) } @@ -183,6 +361,12 @@ impl ProxyResolver { my_ip: Mutex::new(None), #[cfg(windows)] winhttp: OnceLock::new(), + #[cfg(feature = "tokio")] + async_worker: OnceLock::new(), + #[cfg(all(test, feature = "tokio"))] + async_resolution_starts: std::sync::atomic::AtomicUsize::new(0), + #[cfg(all(test, feature = "tokio"))] + async_resolution_delay: Mutex::new(None), }), } } @@ -215,6 +399,28 @@ impl ProxyResolver { Ok(self.demote_bad(list)) } + /// Resolve the ordered proxy list without blocking an async-runtime worker. + /// + /// One lazily-created background thread owns all blocking resolution for + /// this resolver. Concurrent calls for the same URL and configuration + /// generation await one shared result; distinct URLs queue on that thread. + /// This bounds blocking thread use even when many HTTP requests arrive at + /// once. Enable the `tokio` feature to use this API. + #[cfg(feature = "tokio")] + pub async fn resolve_proxy_async(&self, url: &Url) -> Result> { + if url.host_str().is_none() { + return Err(Error::InvalidUrl(url.to_string())); + } + let worker = self + .inner + .async_worker + .get_or_init(|| AsyncResolverWorker::new(Arc::downgrade(&self.inner))); + match worker { + Ok(worker) => worker.resolve(self.config_generation(), url).await, + Err(error) => Err(Error::Platform(error.clone())), + } + } + /// Current configuration generation. Bumped by the platform watcher on /// every (possible) OS proxy config change; poll it to detect staleness /// of anything you derived from a resolution. @@ -654,6 +860,41 @@ mod tests { )); } + #[cfg(feature = "tokio")] + #[tokio::test(flavor = "current_thread")] + async fn async_resolution_coalesces_identical_concurrent_calls() { + use std::sync::atomic::Ordering; + + let resolver = ProxyResolver::with_env( + ResolverOptions::default(), + env(&[("https_proxy", "http://proxy.example:3128")]), + ); + *lock(&resolver.inner.async_resolution_delay) = Some(Duration::from_millis(100)); + + let mut tasks = Vec::new(); + for _ in 0..32 { + let resolver = resolver.clone(); + let target = url("https://example.com/"); + tasks.push(tokio::spawn(async move { + resolver.resolve_proxy_async(&target).await.unwrap() + })); + } + + for task in tasks { + assert_eq!( + task.await.unwrap(), + vec![ProxyKind::Http("proxy.example:3128".into())] + ); + } + assert_eq!( + resolver + .inner + .async_resolution_starts + .load(Ordering::SeqCst), + 1 + ); + } + #[cfg(not(windows))] #[test] fn evaluate_pac_public_api() { diff --git a/src/types.rs b/src/types.rs index 2d80567..a25db6d 100644 --- a/src/types.rs +++ b/src/types.rs @@ -94,7 +94,7 @@ pub type Result = std::result::Result; /// Errors from proxy resolution. PAC/WPAD failures are generally *not* /// surfaced here — resolution falls back to the next layer and ultimately to /// `DIRECT`. Errors mean the input was unusable or a platform call failed. -#[derive(Debug)] +#[derive(Debug, Clone)] #[non_exhaustive] pub enum Error { /// The URL has no host or is otherwise not resolvable against a proxy config.