mirror of
https://github.com/lexmount/moli.git
synced 2026-10-05 08:00:56 +00:00
fix(curl): resume eligible HTTP jobs after completions
Include queued HTTP work in the owner's runnable decision after completions release active slots. Share the global and per-origin eligibility checks with the startup path so blocked queues still allow the owner to wait. Add native owner regressions that consume submission wakeups before releasing a held request, then require the queued request to finish within its short deadline. Cover both scheduler limits and priority queues whose head is blocked by an origin cap.
This commit is contained in:
ldm0
@@ -59,6 +59,12 @@ impl<H: Handler, C> HttpRegistry<H, C> {
|
||||
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<Instant> {
|
||||
self.pending
|
||||
.iter()
|
||||
@@ -174,20 +180,21 @@ impl<H: Handler, C> HttpRegistry<H, C> {
|
||||
}
|
||||
}
|
||||
|
||||
fn next_eligible_job_index(&self) -> Option<usize> {
|
||||
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();
|
||||
|
||||
@@ -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<H: Handler, C> {
|
||||
}
|
||||
|
||||
impl<H: Handler, C> CurlRuntimeOwner<H, C> {
|
||||
/// 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<CurlRuntimeCommand<H, C>>,
|
||||
@@ -95,7 +98,9 @@ impl<H: Handler, C> CurlRuntimeOwner<H, C> {
|
||||
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 {
|
||||
|
||||
@@ -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<u8>);
|
||||
|
||||
impl Handler for Capture {
|
||||
fn write(&mut self, data: &[u8]) -> Result<usize, WriteError> {
|
||||
self.0.extend_from_slice(data);
|
||||
Ok(data.len())
|
||||
}
|
||||
}
|
||||
|
||||
fn request(
|
||||
address: SocketAddr,
|
||||
path: &'static str,
|
||||
deadline: Option<Instant>,
|
||||
) -> CurlMultiJob<Capture, &'static str> {
|
||||
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::<Result<Vec<_>, _>>();
|
||||
// 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);
|
||||
}
|
||||
Reference in New Issue
Block a user