diff --git a/moli-curl/src/http/registry.rs b/moli-curl/src/http/registry.rs index 086f4a5ab4..ac7510fdb9 100644 --- a/moli-curl/src/http/registry.rs +++ b/moli-curl/src/http/registry.rs @@ -59,6 +59,12 @@ impl HttpRegistry { self.has_active() || !self.pending.is_empty() } + /// Pending work can run after completions release scheduler capacity. + /// Use the same global and per-origin limits as the actual startup path. + pub(crate) fn has_eligible_jobs(&self) -> bool { + self.next_eligible_job_index().is_some() + } + pub(crate) fn next_deadline(&self) -> Option { self.pending .iter() @@ -174,20 +180,21 @@ impl HttpRegistry { } } + fn next_eligible_job_index(&self) -> Option { + if self.closed || self.active.len() >= self.config.max_active.get() { + return None; + } + self.pending.iter().position(|pending| { + job_is_eligible( + pending.job.origin.as_ref(), + &self.active, + self.config.max_host_active, + ) + }) + } + fn start_eligible_jobs(&mut self, multi: &mut Multi) { - loop { - if self.closed || self.active.len() >= self.config.max_active.get() { - return; - } - let Some(index) = self.pending.iter().position(|pending| { - job_is_eligible( - pending.job.origin.as_ref(), - &self.active, - self.config.max_host_active, - ) - }) else { - return; - }; + while let Some(index) = self.next_eligible_job_index() { let pending = self .pending .remove(index) @@ -534,6 +541,59 @@ mod tests { CurlTransferId::from_token(sequence).expect("test transfer ID is non-zero") } + #[test] + fn eligible_jobs_respect_global_and_origin_capacity_after_completions() { + let config = CurlMultiRuntimeConfig { + max_active: NonZeroUsize::new(2).unwrap(), + max_host_active: NonZeroUsize::new(1), + ..CurlMultiRuntimeConfig::default() + }; + let (completion_tx, _completed) = crossbeam_channel::unbounded(); + let mut multi = Multi::new(); + let mut registry = HttpRegistry::new(config, completion_tx); + let origin = |host: &str| CurlOriginKey { + scheme: "https".to_owned(), + host: host.to_owned(), + port: Some(443), + }; + let first = test_transfer_id(1); + let blocked = test_transfer_id(2); + let other_origin = test_transfer_id(3); + let no_origin = test_transfer_id(4); + + registry.admit(first, test_job("first", 1, Some(origin("a.test")))); + registry.advance(&mut multi); + assert!(registry.contains(first)); + assert!(!registry.has_eligible_jobs()); + + // A free global slot does not bypass the origin cap. + registry.admit(blocked, test_job("blocked", 2, Some(origin("a.test")))); + assert!(!registry.has_eligible_jobs()); + // A blocked high-priority head must not hide eligible work elsewhere. + registry.admit(other_origin, test_job("other", 0, Some(origin("b.test")))); + assert!(registry.has_eligible_jobs()); + registry.advance(&mut multi); + assert!(registry.contains(other_origin)); + assert!(!registry.contains(blocked)); + + // The global cap also applies to jobs without an origin key. + registry.admit(no_origin, test_job("no-origin", 0, None)); + assert!(!registry.has_eligible_jobs()); + registry.complete(&mut multi, vec![(other_origin, Ok(()))]); + assert!(registry.has_eligible_jobs()); + registry.advance(&mut multi); + assert!(registry.contains(no_origin)); + assert!(!registry.contains(blocked)); + + registry.complete(&mut multi, vec![(first, Ok(()))]); + assert!(registry.has_eligible_jobs()); + registry.advance(&mut multi); + assert!(registry.contains(blocked)); + assert!(!registry.has_eligible_jobs()); + registry.shutdown(&mut multi); + assert!(!registry.has_eligible_jobs()); + } + #[test] fn active_transfer_wait_is_capped_by_the_earliest_queued_deadline() { let now = Instant::now(); diff --git a/moli-curl/src/runtime/owner.rs b/moli-curl/src/runtime/owner.rs index 61846e649e..a2e981a2ea 100644 --- a/moli-curl/src/runtime/owner.rs +++ b/moli-curl/src/runtime/owner.rs @@ -1,6 +1,9 @@ //! One native owner drives both protocols. Registries own their easy handles; //! only this loop performs curl work, drains CURLMSG_DONE and waits for readiness. +#[cfg(test)] +mod tests; + use std::{ sync::{ Arc, @@ -48,7 +51,7 @@ pub(super) struct CurlRuntimeOwner { } impl CurlRuntimeOwner { - /// Construct all native state on its owner thread, including the WS cache. + /// Construct all native state on its owner thread, including the shared pool. pub(super) fn run( config: CurlMultiRuntimeConfig, command_rx: Receiver>, @@ -95,7 +98,9 @@ impl CurlRuntimeOwner { return; } - let runnable = progressed || !self.command_rx.is_empty(); + // HTTP completions can release slots after advance() has run. + let runnable = + progressed || !self.command_rx.is_empty() || self.http.has_eligible_jobs(); if !runnable && !self.http.has_curl_work() && self.websockets.is_empty() { self.wait_for_next_owner_event(); } else { diff --git a/moli-curl/src/runtime/owner/tests.rs b/moli-curl/src/runtime/owner/tests.rs new file mode 100644 index 0000000000..658c77ef30 --- /dev/null +++ b/moli-curl/src/runtime/owner/tests.rs @@ -0,0 +1,188 @@ +use std::{ + io::{Read, Write}, + net::{SocketAddr, TcpListener, TcpStream}, + num::NonZeroUsize, + thread, +}; + +use curl::easy::{Easy2, WriteError}; + +use super::*; +use crate::websocket::CurlWebSocketConnector; +use crate::{CurlDnsResolution, CurlHttpSender, CurlMultiJob, CurlOriginKey}; + +const TEST_TIMEOUT: Duration = Duration::from_secs(5); + +#[derive(Default)] +struct Capture(Vec); + +impl Handler for Capture { + fn write(&mut self, data: &[u8]) -> Result { + self.0.extend_from_slice(data); + Ok(data.len()) + } +} + +fn request( + address: SocketAddr, + path: &'static str, + deadline: Option, +) -> CurlMultiJob { + let mut easy = Easy2::new(Capture::default()); + easy.url(&format!("http://{address}{path}")).unwrap(); + easy.proxy("").unwrap(); + CurlMultiJob { + easy, + context: path, + origin: Some(CurlOriginKey { + scheme: "http".to_owned(), + host: address.ip().to_string(), + port: Some(address.port()), + }), + deadline, + dns_resolution: CurlDnsResolution::curl_managed(), + priority: 1, + label: path.to_owned(), + } +} + +fn read_path(stream: &mut TcpStream) -> String { + stream.set_read_timeout(Some(TEST_TIMEOUT)).unwrap(); + let mut request = Vec::new(); + while !request.ends_with(b"\r\n\r\n") { + let mut byte = [0]; + stream.read_exact(&mut byte).unwrap(); + request.push(byte[0]); + assert!(request.len() < 4096); + } + String::from_utf8(request) + .unwrap() + .split_whitespace() + .nth(1) + .unwrap() + .to_owned() +} + +fn completion_starts_queued_job(per_origin: bool) { + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let address = listener.local_addr().unwrap(); + let (accepted, ready) = crossbeam_channel::bounded(1); + let (release, released) = crossbeam_channel::bounded(1); + let server = thread::spawn(move || { + let mut paths = Vec::new(); + loop { + let (mut stream, _) = listener.accept().unwrap(); + let path = read_path(&mut stream); + if path == "/stop" { + return paths; + } + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\n") + .unwrap(); + if path == "/held" { + accepted.send(()).unwrap(); + released.recv_timeout(TEST_TIMEOUT).unwrap(); + } + stream.write_all(b"ok").unwrap(); + paths.push(path); + } + }); + + let config = CurlMultiRuntimeConfig { + max_active: NonZeroUsize::new(if per_origin { 2 } else { 1 }).unwrap(), + max_host_active: per_origin.then(|| NonZeroUsize::new(1).unwrap()), + ..CurlMultiRuntimeConfig::default() + }; + let multi = make_runtime_multi(&config); + let (command_tx, command_rx) = crossbeam_channel::unbounded(); + let (completion_tx, completed) = crossbeam_channel::unbounded(); + let (_websocket_tx, websocket_rx) = CurlWebSocketConnector::channel(); + let shutdown_requested = Arc::new(AtomicBool::new(false)); + let sender = CurlHttpSender { + command_tx, + owner_waker: multi.waker(), + shutdown_requested: shutdown_requested.clone(), + }; + let mut owner = CurlRuntimeOwner { + command_rx, + shutdown_requested, + closed: false, + poll_interval: config.poll_interval, + diagnostics: Diagnostics::from_env(), + http: HttpRegistry::new(config, completion_tx), + websockets: WebSocketRegistry::new(websocket_rx), + multi, + }; + + // Prepare a real active transfer on this thread so the test can explicitly + // consume all submission wakeups while the peer still holds its response. + let first = sender.submit(request(address, "/held", None)).unwrap(); + owner.drain_commands(); + owner.http.advance(&mut owner.multi); + let setup_deadline = Instant::now() + TEST_TIMEOUT; + loop { + owner.process_completed_transfers(); + if ready.try_recv().is_ok() { + break; + } + assert!( + Instant::now() < setup_deadline, + "first request did not start" + ); + owner.wait_for_curl_progress(false); + } + let second = sender + .submit(request( + address, + "/queued", + Some(Instant::now() + Duration::from_millis(500)), + )) + .unwrap(); + owner.drain_commands(); + owner.http.advance(&mut owner.multi); + assert!(owner.http.contains(first)); + assert!(!owner.http.contains(second)); + // In particular, B's submit() wakeup must not rescue the owner after A + // completes. The native poll consumes it before we release A's body. + owner.wait_for_curl_progress(false); + + let finished = thread::spawn(move || { + let completions = (0..2) + .map(|_| completed.recv_timeout(TEST_TIMEOUT)) + .collect::, _>>(); + // No new command is sent after A completes. Only B's terminal (or the + // test watchdog) permits shutdown to wake the native owner. + sender + .command_tx + .send(CurlRuntimeCommand::Shutdown) + .unwrap(); + sender.owner_waker.wakeup().unwrap(); + completions + }); + release.send(()).unwrap(); + owner.drive(); + let completions = finished.join().unwrap(); + let mut stop = TcpStream::connect(address).unwrap(); + stop.write_all(b"GET /stop HTTP/1.1\r\nHost: localhost\r\n\r\n") + .unwrap(); + let paths = server.join().unwrap(); + + for (completion, id) in completions.unwrap().into_iter().zip([first, second]) { + assert_eq!(completion.transfer_id, id); + completion + .result + .expect("a queued request must start when its active slot is released"); + assert_eq!(completion.easy.unwrap().get_ref().0, b"ok"); + } + assert_eq!(paths, ["/held", "/queued"]); +} + +#[test] +fn http_completion_starts_queued_job_before_global_deadline() { + completion_starts_queued_job(false); +} + +#[test] +fn http_completion_starts_queued_job_before_per_origin_deadline() { + completion_starts_queued_job(true); +}