diff --git a/moli-curl/README.md b/moli-curl/README.md index 9a1cb17284..701b6dc56e 100644 --- a/moli-curl/README.md +++ b/moli-curl/README.md @@ -1,4 +1,27 @@ -# Native WebSocket owner diagnostics +# Native HTTP and WebSocket runtime + +## Module boundaries + +| Module | Responsibility | +| --- | --- | +| `runtime.rs`, `runtime/owner.rs` | Runtime ownership and the single native thread: submit handling, shutdown, `perform`, completion dispatch and `poll` | +| `http.rs`, `http/registry.rs`, `http/scheduling.rs` | HTTP request capability, priority/origin admission, DNS/deadlines and transfer completion | +| `websocket/connector.rs`, `websocket/connection.rs` | WebSocket submission capability and the caller/owner I/O endpoints, including independent send completion | +| `websocket/registry.rs`, `websocket/session.rs` | Resident native connections, handshake and frame I/O | +| `websocket/readiness.rs` | I/O admission, extra socket interests and applying the shared poll's readiness results | +| `websocket/standalone.rs` | Convenience owner using the same runtime for standalone callers | +| `dns_adapter.rs`, `tls.rs` | Shared DNS and TLS configuration | + +The owner dispatches to concrete HTTP and WebSocket registries. HTTP completion +removes a transfer; WebSocket handshake completion keeps its handle resident. +Both registries preserve the easy handle's error details when interpreting +`CURLMSG_DONE`. Only `runtime/owner.rs` calls `perform`, `messages` and `poll`. +Read that file first when investigating scheduling or thread shutdown. + +Public exports remain at `moli_curl::*` for HTTP/runtime and +`moli_curl::websocket::*` for WebSocket. Neither protocol's request capability +owns the native thread. Shared owner counters live in `runtime/diagnostics.rs`; +the diagnostics switch and log fields below remain compatible. `moli-curl` owns native connections and frame I/O. Browser message assembly, UTF-8 checks, command admission and the close handshake live in `moli-websocket`. diff --git a/moli-curl/src/http.rs b/moli-curl/src/http.rs new file mode 100644 index 0000000000..10d14a7be2 --- /dev/null +++ b/moli-curl/src/http.rs @@ -0,0 +1,154 @@ +//! HTTP submission types and owner-local request scheduling. +//! The shared runtime drives libcurl; this module owns HTTP jobs until completion. + +pub(crate) mod registry; +mod scheduling; + +use crate::{ + CurlDnsResolution, CurlTransferId, + runtime::{CurlRuntimeCommand, identity::next_transfer_id}, +}; +use anyhow::{Result, anyhow}; +use crossbeam_channel::Sender; +use curl::{ + easy::{Easy2, Handler}, + multi::MultiWaker, +}; +use std::{ + fmt, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + time::Instant, +}; + +/// Origin key used by the curl scheduler for per-origin active transfer caps. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct CurlOriginKey { + pub scheme: String, + pub host: String, + pub port: Option, +} + +/// A configured curl transfer plus scheduler metadata. +pub struct CurlMultiJob { + pub easy: Easy2, + pub context: C, + pub origin: Option, + /// Absolute deadline for the whole scheduler-owned transfer attempt. + /// + /// libcurl cannot account for time spent in Moli's priority queue or in + /// the shared DNS residence because both happen before the easy handle is + /// added to the multi handle. The owner enforces this deadline in those + /// residences and gives libcurl only the remaining duration. + pub deadline: Option, + /// DNS ownership chosen by the caller before this transfer enters curl. + /// + /// A curl-managed policy preserves libcurl's resolver behavior. A shared + /// origin policy parks the transfer outside the curl multi handle set until + /// the bounded system resolver publishes an answer. + pub dns_resolution: CurlDnsResolution, + /// Higher values start before lower values when jobs are queued. + pub priority: u8, + pub label: String, +} + +impl fmt::Debug for CurlMultiJob { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("CurlMultiJob") + .field("origin", &self.origin) + .field("deadline", &self.deadline) + .field("dns_resolution", &self.dns_resolution) + .field("priority", &self.priority) + .field("label", &self.label) + .finish_non_exhaustive() + } +} + +/// Completion emitted by `CurlMultiRuntime`. +pub struct CurlMultiCompletion { + pub transfer_id: CurlTransferId, + pub easy: Option>, + pub context: C, + pub result: Result<()>, +} + +impl fmt::Debug for CurlMultiCompletion { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("CurlMultiCompletion") + .field("transfer_id", &self.transfer_id) + .field("has_easy", &self.easy.is_some()) + .field("result", &self.result.as_ref().map(|_| ())) + .finish_non_exhaustive() + } +} + +/// Error returned when a job cannot be submitted and is returned to the caller. +pub struct CurlSubmitError { + pub job: CurlMultiJob, + pub error: anyhow::Error, +} + +impl fmt::Debug for CurlSubmitError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("CurlSubmitError") + .field("job", &self.job) + .field("error", &self.error) + .finish() + } +} + +/// Submits HTTP work to one owner. Clones carry no shutdown or join authority. +#[derive(Debug)] +pub struct CurlHttpSender { + pub(crate) command_tx: Sender>, + pub(crate) owner_waker: MultiWaker, + pub(crate) shutdown_requested: Arc, +} + +impl Clone for CurlHttpSender { + fn clone(&self) -> Self { + Self { + command_tx: self.command_tx.clone(), + owner_waker: self.owner_waker.clone(), + shutdown_requested: self.shutdown_requested.clone(), + } + } +} + +impl CurlHttpSender { + pub fn submit( + &self, + job: CurlMultiJob, + ) -> std::result::Result> { + if self.shutdown_requested.load(Ordering::SeqCst) { + return Err(CurlSubmitError { + job, + error: anyhow!("curl multi runtime is shutting down"), + }); + } + let transfer_id = match next_transfer_id() { + Ok(transfer_id) => transfer_id, + Err(error) => return Err(CurlSubmitError { job, error }), + }; + match self + .command_tx + .send(CurlRuntimeCommand::Request { transfer_id, job }) + { + Ok(()) => { + let _ = self.owner_waker.wakeup(); + Ok(transfer_id) + } + Err(error) => { + let CurlRuntimeCommand::Request { job, .. } = error.into_inner() else { + unreachable!("submit only sends request commands"); + }; + Err(CurlSubmitError { + job, + error: anyhow!("curl multi runtime is shutting down"), + }) + } + } + } +} diff --git a/moli-curl/src/http/registry.rs b/moli-curl/src/http/registry.rs new file mode 100644 index 0000000000..086f4a5ab4 --- /dev/null +++ b/moli-curl/src/http/registry.rs @@ -0,0 +1,568 @@ +//! HTTP jobs move through the priority queue, DNS and active easy handles. +//! The runtime alone calls perform/messages/poll and dispatches completions here. + +use super::{ + CurlMultiCompletion, CurlMultiJob, + scheduling::{ + CurlActiveTransfer, CurlPendingJob, active_origin_count, enqueue_existing_pending_job, + enqueue_pending_job, job_is_eligible, pending_origin_count, take_expired_pending_jobs, + take_transfers_in_notification_order, + }, +}; +use crate::{ + CurlMultiRuntimeConfig, CurlTransferId, + dns_adapter::{CurlDnsOwnerCompletion, CurlDnsOwnerResidence, CurlDnsReady}, +}; +use anyhow::{Context, Result, anyhow}; +use crossbeam_channel::{Receiver, Sender}; +use curl::{ + easy::Handler, + multi::{Message, Multi}, +}; +use std::{ + collections::{HashMap, VecDeque}, + num::NonZeroUsize, + sync::OnceLock, + time::{Duration, Instant}, +}; + +pub(crate) struct HttpRegistry { + config: CurlMultiRuntimeConfig, + completion_tx: Sender>, + closed: bool, + pending: VecDeque>, + dns: CurlDnsOwnerResidence>, + active: HashMap>, +} + +impl HttpRegistry { + pub(crate) fn new( + config: CurlMultiRuntimeConfig, + completion_tx: Sender>, + ) -> Self { + Self { + config, + completion_tx, + closed: false, + pending: VecDeque::new(), + dns: CurlDnsOwnerResidence::default(), + active: HashMap::new(), + } + } + + pub(crate) fn has_active(&self) -> bool { + !self.active.is_empty() + } + + /// DNS-only work can wait on its completion channel without polling curl. + pub(crate) fn has_curl_work(&self) -> bool { + self.has_active() || !self.pending.is_empty() + } + + pub(crate) fn next_deadline(&self) -> Option { + self.pending + .iter() + .filter_map(|pending| pending.job.deadline) + .chain(self.dns.next_deadline(|pending| pending.job.deadline)) + .min() + } + + pub(crate) fn dns_completions(&self) -> &Receiver> { + self.dns.completion_receiver() + } + + pub(crate) fn advance(&mut self, multi: &mut Multi) { + self.drain_dns_completions(); + self.expire_waiting_jobs(); + self.start_eligible_jobs(multi); + } + + pub(crate) fn contains(&self, id: CurlTransferId) -> bool { + self.active.contains_key(&id) + } + + pub(crate) fn completion_result( + &self, + id: CurlTransferId, + message: &Message<'_>, + ) -> Option> { + message.result_for2(&self.active.get(&id)?.handle) + } + + pub(crate) fn complete( + &mut self, + multi: &mut Multi, + completed: Vec<(CurlTransferId, Result<(), curl::Error>)>, + ) { + for (transfer_id, active, result) in + take_transfers_in_notification_order(&mut self.active, completed) + { + self.finish_active_transfer(multi, transfer_id, active, result.map_err(Into::into)); + } + } + + pub(crate) fn admit(&mut self, transfer_id: CurlTransferId, job: CurlMultiJob) { + if self.closed { + self.send_completion(CurlMultiCompletion { + transfer_id, + easy: Some(job.easy), + context: job.context, + result: Err(anyhow!("curl multi runtime is shutting down")), + }); + return; + } + if curl_runtime_trace_enabled() { + let origin = job.origin.as_ref(); + tracing::info!( + target: "moli_cdp_nav_timing", + transfer_id = %transfer_id, + label = %job.label, + origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""), + origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""), + origin_port = ?origin.and_then(|origin| origin.port), + origin = ?job.origin, + priority = job.priority, + pending_before = self.pending.len(), + pending_same_origin_before = origin + .map(|origin| pending_origin_count(&self.pending, origin)) + .unwrap_or(0), + stage = "curl_runtime_job_queued", + ); + } + enqueue_pending_job(&mut self.pending, transfer_id, job); + } + + pub(crate) fn shutdown(&mut self, multi: &mut Multi) { + if self.closed { + return; + } + self.closed = true; + while let Some(pending) = self.pending.pop_front() { + let CurlPendingJob { + transfer_id, job, .. + } = pending; + self.send_completion(CurlMultiCompletion { + transfer_id, + easy: Some(job.easy), + context: job.context, + result: Err(anyhow!("curl multi runtime is shutting down")), + }); + } + for pending in self.dns.drain() { + let CurlPendingJob { + transfer_id, job, .. + } = pending; + let _ = self.completion_tx.send(CurlMultiCompletion { + transfer_id, + easy: Some(job.easy), + context: job.context, + result: Err(anyhow!( + "curl multi runtime DNS request cancelled during shutdown" + )), + }); + } + for (transfer_id, active) in self.active.drain() { + let easy = multi.remove2(active.handle).ok(); + let _ = self.completion_tx.send(CurlMultiCompletion { + transfer_id, + easy, + context: active.context, + result: Err(anyhow!( + "curl multi runtime request cancelled during shutdown" + )), + }); + } + } + + 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; + }; + let pending = self + .pending + .remove(index) + .expect("pending curl job index should exist"); + let dns_target = pending.job.dns_resolution.target().cloned(); + match dns_target { + Some(target) => self + .dns + .start(pending.transfer_id, pending, target, multi.waker()), + None => self.start_job(multi, pending), + } + } + } + + fn drain_dns_completions(&mut self) { + while let Some(ready) = self.dns.try_claim_next() { + self.handle_dns_completion(ready); + } + } + + pub(crate) fn claim_dns_completion( + &mut self, + completion: CurlDnsOwnerCompletion, + ) { + let Some(ready) = self.dns.claim(completion) else { + return; + }; + self.handle_dns_completion(ready); + } + + fn handle_dns_completion(&mut self, ready: CurlDnsReady>) { + let mut pending = ready.pending; + if self.closed { + let CurlPendingJob { + transfer_id, job, .. + } = pending; + self.send_completion(CurlMultiCompletion { + transfer_id, + easy: Some(job.easy), + context: job.context, + result: Err(anyhow!("curl multi runtime is shutting down")), + }); + return; + } + if pending.deadline_reached(Instant::now()) { + self.complete_timed_out_job(pending, "while waiting for DNS"); + return; + } + match ready.result { + Ok(addresses) => { + if let Err(error) = pending + .job + .dns_resolution + .install(&mut pending.job.easy, addresses.as_ref()) + { + let CurlPendingJob { + transfer_id, job, .. + } = pending; + self.send_completion(CurlMultiCompletion { + transfer_id, + easy: Some(job.easy), + context: job.context, + result: Err(error), + }); + return; + } + enqueue_existing_pending_job(&mut self.pending, pending); + } + Err(error) => { + let CurlPendingJob { + transfer_id, job, .. + } = pending; + self.send_completion(CurlMultiCompletion { + transfer_id, + easy: Some(job.easy), + context: job.context, + result: Err(anyhow!(error.to_string())), + }); + } + } + } + + fn start_job(&mut self, multi: &mut Multi, pending: CurlPendingJob) { + if pending.deadline_reached(Instant::now()) { + self.complete_timed_out_job(pending, "while waiting to start"); + return; + } + let transfer_id = pending.transfer_id; + let queued_for = pending.enqueued_at.elapsed(); + let mut job = pending.job; + let label = job.label.clone(); + if let Some(deadline) = job.deadline { + let Some(remaining) = curl_timeout_for_deadline(deadline, Instant::now()) else { + self.send_completion(CurlMultiCompletion { + transfer_id, + easy: Some(job.easy), + context: job.context, + result: Err(curl_runtime_timeout_error("while waiting to start")), + }); + return; + }; + if let Err(error) = job.easy.timeout(remaining) { + self.send_completion(CurlMultiCompletion { + transfer_id, + easy: Some(job.easy), + context: job.context, + result: Err(error).context("failed to apply remaining curl request deadline"), + }); + return; + } + } + match multi + .add2(job.easy) + .with_context(|| anyhow!("failed to add curl easy handle for {label}")) + { + Ok(mut handle) => { + handle + .set_token(transfer_id.token()) + .expect("active curl handle must accept its transfer token"); + if curl_runtime_trace_enabled() { + let origin = job.origin.as_ref(); + tracing::info!( + target: "moli_cdp_nav_timing", + transfer_id = %transfer_id, + label = %label, + origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""), + origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""), + origin_port = ?origin.and_then(|origin| origin.port), + origin = ?job.origin, + priority = job.priority, + queued_ms = queued_for.as_millis(), + active_before = self.active.len(), + active_same_origin_before = origin + .map(|origin| active_origin_count(&self.active, origin)) + .unwrap_or(0), + pending_after = self.pending.len(), + pending_same_origin_after = origin + .map(|origin| pending_origin_count(&self.pending, origin)) + .unwrap_or(0), + max_active = self.config.max_active.get(), + max_host_active = ?self.config.max_host_active.map(NonZeroUsize::get), + max_host_connections = ?self.config.max_host_connections.map(NonZeroUsize::get), + max_total_connections = ?self.config.max_total_connections.map(NonZeroUsize::get), + max_concurrent_streams = ?self.config.max_concurrent_streams.map(NonZeroUsize::get), + multiplex = self.config.multiplex, + stage = "curl_runtime_job_start", + ); + } + let previous = self.active.insert( + transfer_id, + CurlActiveTransfer { + handle, + context: job.context, + origin: job.origin, + priority: job.priority, + label, + started_at: Instant::now(), + queued_for, + }, + ); + assert!(previous.is_none(), "curl transfer identity is unique"); + } + Err(error) => self.send_completion(CurlMultiCompletion { + transfer_id, + easy: None, + context: job.context, + result: Err(error), + }), + } + } + + fn expire_waiting_jobs(&mut self) { + let now = Instant::now(); + for pending in take_expired_pending_jobs(&mut self.pending, now) { + self.complete_timed_out_job(pending, "while waiting in the scheduler"); + } + for pending in self.dns.take_expired(now, |pending| pending.job.deadline) { + self.complete_timed_out_job(pending, "while waiting for DNS"); + } + } + + fn complete_timed_out_job(&self, pending: CurlPendingJob, stage: &'static str) { + let CurlPendingJob { + transfer_id, job, .. + } = pending; + self.send_completion(CurlMultiCompletion { + transfer_id, + easy: Some(job.easy), + context: job.context, + result: Err(curl_runtime_timeout_error(stage)), + }); + } + + fn finish_active_transfer( + &self, + multi: &mut Multi, + transfer_id: CurlTransferId, + active: CurlActiveTransfer, + result: Result<()>, + ) { + let easy = match multi.remove2(active.handle) { + Ok(easy) => Some(easy), + Err(error) => { + self.send_completion(CurlMultiCompletion { + transfer_id, + easy: None, + context: active.context, + result: Err(anyhow!( + "failed to remove curl easy handle for {}: {error}", + active.label + )), + }); + return; + } + }; + if curl_runtime_trace_enabled() { + let origin = active.origin.as_ref(); + match &result { + Ok(()) => { + tracing::info!( + target: "moli_cdp_nav_timing", + transfer_id = %transfer_id, + label = %active.label, + origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""), + origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""), + origin_port = ?origin.and_then(|origin| origin.port), + origin = ?active.origin, + priority = active.priority, + ok = true, + active_ms = active.started_at.elapsed().as_millis(), + queued_ms = active.queued_for.as_millis(), + active_remaining = self.active.len(), + active_same_origin_remaining = origin + .map(|origin| active_origin_count(&self.active, origin)) + .unwrap_or(0), + pending_after = self.pending.len(), + pending_same_origin_after = origin + .map(|origin| pending_origin_count(&self.pending, origin)) + .unwrap_or(0), + stage = "curl_runtime_job_done", + ); + } + Err(error) => { + tracing::info!( + target: "moli_cdp_nav_timing", + transfer_id = %transfer_id, + label = %active.label, + origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""), + origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""), + origin_port = ?origin.and_then(|origin| origin.port), + origin = ?active.origin, + priority = active.priority, + ok = false, + error = %error, + active_ms = active.started_at.elapsed().as_millis(), + queued_ms = active.queued_for.as_millis(), + active_remaining = self.active.len(), + active_same_origin_remaining = origin + .map(|origin| active_origin_count(&self.active, origin)) + .unwrap_or(0), + pending_after = self.pending.len(), + pending_same_origin_after = origin + .map(|origin| pending_origin_count(&self.pending, origin)) + .unwrap_or(0), + stage = "curl_runtime_job_done", + ); + } + } + } + let result = result.with_context(|| { + anyhow!( + "curl request failed for {} after active={}ms queued={}ms", + active.label, + active.started_at.elapsed().as_millis(), + active.queued_for.as_millis() + ) + }); + self.send_completion(CurlMultiCompletion { + transfer_id, + easy, + context: active.context, + result, + }); + } + + fn send_completion(&self, completion: CurlMultiCompletion) { + let _ = self.completion_tx.send(completion); + } +} + +fn curl_runtime_timeout_error(stage: &str) -> anyhow::Error { + anyhow!("curl multi runtime request timed out {stage}") +} + +fn curl_timeout_for_deadline(deadline: Instant, now: Instant) -> Option { + let remaining = deadline.saturating_duration_since(now); + // curl-rust converts CURLOPT_TIMEOUT_MS with `Duration::as_millis()`. A + // positive sub-millisecond value would therefore become zero, which + // libcurl interprets as disabling the timeout entirely. + (remaining >= Duration::from_millis(1)).then_some(remaining) +} + +fn curl_runtime_trace_enabled() -> bool { + static ENABLED: OnceLock = OnceLock::new(); + *ENABLED.get_or_init(|| { + env_flag_enabled("MOLI_CDP_NAV_TIMING") || env_flag_enabled("MOLI_CURL_RUNTIME_TRACE") + }) +} + +fn env_flag_enabled(name: &str) -> bool { + std::env::var(name).is_ok_and(|value| { + let value = value.trim(); + !value.is_empty() && value != "0" && !value.eq_ignore_ascii_case("false") + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{CurlDnsResolution, CurlOriginKey}; + use curl::easy::Easy2; + #[derive(Debug)] + struct TestHandler; + + impl Handler for TestHandler {} + + fn test_job( + label: &str, + priority: u8, + origin: Option, + ) -> CurlMultiJob { + CurlMultiJob { + easy: Easy2::new(TestHandler), + context: label.to_owned(), + origin, + deadline: None, + dns_resolution: CurlDnsResolution::curl_managed(), + priority, + label: label.to_owned(), + } + } + + fn test_transfer_id(sequence: usize) -> CurlTransferId { + CurlTransferId::from_token(sequence).expect("test transfer ID is non-zero") + } + + #[test] + fn active_transfer_wait_is_capped_by_the_earliest_queued_deadline() { + let now = Instant::now(); + let later = now + Duration::from_secs(2); + let earlier = now + Duration::from_secs(1); + let mut later_job = test_job("later", 1, None); + later_job.deadline = Some(later); + let mut earlier_job = test_job("earlier", 1, None); + earlier_job.deadline = Some(earlier); + let mut pending = VecDeque::new(); + enqueue_pending_job(&mut pending, test_transfer_id(1), later_job); + enqueue_pending_job(&mut pending, test_transfer_id(2), earlier_job); + let (completion_tx, _) = crossbeam_channel::unbounded(); + let mut registry = HttpRegistry::new(CurlMultiRuntimeConfig::default(), completion_tx); + registry.pending = pending; + + assert_eq!(registry.next_deadline(), Some(earlier)); + } + + #[test] + fn sub_millisecond_deadline_never_disables_the_libcurl_timeout() { + let now = Instant::now(); + assert_eq!( + curl_timeout_for_deadline(now + Duration::from_micros(999), now), + None + ); + assert_eq!( + curl_timeout_for_deadline(now + Duration::from_millis(1), now), + Some(Duration::from_millis(1)) + ); + } +} diff --git a/moli-curl/src/runtime/residence.rs b/moli-curl/src/http/scheduling.rs similarity index 79% rename from moli-curl/src/runtime/residence.rs rename to moli-curl/src/http/scheduling.rs index 8504486493..c9e0e93a4b 100644 --- a/moli-curl/src/runtime/residence.rs +++ b/moli-curl/src/http/scheduling.rs @@ -1,3 +1,5 @@ +//! Priority ordering, per-origin admission and ordered HTTP completions. + use std::{ collections::{HashMap, VecDeque}, hash::Hash, @@ -7,38 +9,8 @@ use std::{ use curl::{easy::Handler, multi::Easy2Handle}; -use crate::dns_adapter::CurlDnsOwnerResidence; - use super::{CurlMultiJob, CurlOriginKey, CurlTransferId}; -pub(super) struct CurlOwnerState { - pub(super) closed: bool, - pub(super) pending: VecDeque>, - pub(super) dns: CurlDnsOwnerResidence>, - pub(super) active: HashMap>, -} - -impl Default for CurlOwnerState { - fn default() -> Self { - Self { - closed: false, - pending: VecDeque::new(), - dns: CurlDnsOwnerResidence::default(), - active: HashMap::new(), - } - } -} - -impl CurlOwnerState { - pub(super) fn next_waiting_deadline(&self) -> Option { - self.pending - .iter() - .filter_map(|pending| pending.job.deadline) - .chain(self.dns.next_deadline(|pending| pending.job.deadline)) - .min() - } -} - pub(super) struct CurlActiveTransfer { pub(super) handle: Easy2Handle, pub(super) context: C, @@ -109,11 +81,11 @@ pub(super) fn enqueue_existing_pending_job( pub(super) fn job_is_eligible( origin: Option<&CurlOriginKey>, - state: &CurlOwnerState, + active: &HashMap>, max_active_per_host: Option, ) -> bool { match (origin, max_active_per_host) { - (Some(origin), Some(limit)) => active_origin_count(&state.active, origin) < limit.get(), + (Some(origin), Some(limit)) => active_origin_count(active, origin) < limit.get(), _ => true, } } @@ -272,28 +244,6 @@ mod tests { ); } - #[test] - fn active_transfer_wait_is_capped_by_the_earliest_queued_deadline() { - let now = Instant::now(); - let later = now + Duration::from_secs(2); - let earlier = now + Duration::from_secs(1); - let mut later_job = test_job("later", 1, None); - later_job.deadline = Some(later); - let mut earlier_job = test_job("earlier", 1, None); - earlier_job.deadline = Some(earlier); - let mut pending = VecDeque::new(); - enqueue_pending_job(&mut pending, test_transfer_id(1), later_job); - enqueue_pending_job(&mut pending, test_transfer_id(2), earlier_job); - let state = CurlOwnerState { - closed: false, - pending, - dns: CurlDnsOwnerResidence::default(), - active: HashMap::new(), - }; - - assert_eq!(state.next_waiting_deadline(), Some(earlier)); - } - #[test] fn completed_jobs_preserve_libcurl_notification_order() { let mut active = HashMap::from([ @@ -354,15 +304,10 @@ mod tests { started_at: Instant::now(), queued_for: Duration::ZERO, }; - let state = CurlOwnerState { - closed: false, - pending: VecDeque::new(), - dns: CurlDnsOwnerResidence::default(), - active: HashMap::from([(test_transfer_id(1), active)]), - }; + let active = HashMap::from([(test_transfer_id(1), active)]); let cap = NonZeroUsize::new(1); - assert!(!job_is_eligible(Some(&capped_origin), &state, cap)); - assert!(job_is_eligible(Some(&other_origin), &state, cap)); + assert!(!job_is_eligible(Some(&capped_origin), &active, cap)); + assert!(job_is_eligible(Some(&other_origin), &active, cap)); } } diff --git a/moli-curl/src/lib.rs b/moli-curl/src/lib.rs index 4c56672434..02c10e1baa 100644 --- a/moli-curl/src/lib.rs +++ b/moli-curl/src/lib.rs @@ -1,13 +1,12 @@ //! Shared libcurl multi scheduler for Moli network requests. mod dns_adapter; +mod http; mod runtime; mod tls; pub mod websocket; pub use dns_adapter::CurlDnsResolution; -pub use runtime::{ - CurlHttpSender, CurlMultiCompletion, CurlMultiJob, CurlMultiRuntime, CurlMultiRuntimeConfig, - CurlOriginKey, CurlSubmitError, CurlTransferId, -}; +pub use http::{CurlHttpSender, CurlMultiCompletion, CurlMultiJob, CurlOriginKey, CurlSubmitError}; +pub use runtime::{CurlMultiRuntime, CurlMultiRuntimeConfig, CurlTransferId}; pub use tls::CurlTlsConfig; diff --git a/moli-curl/src/runtime.rs b/moli-curl/src/runtime.rs index 056331d7e6..3a9c064a89 100644 --- a/moli-curl/src/runtime.rs +++ b/moli-curl/src/runtime.rs @@ -1,109 +1,36 @@ +//! Owns the common native thread; HTTP and WebSocket handles only submit work. + mod config; +pub(crate) mod diagnostics; pub(crate) mod identity; mod owner; -mod residence; +#[cfg(test)] +mod tests; +use crate::{CurlHttpSender, CurlMultiCompletion, CurlMultiJob, websocket::CurlWebSocketConnector}; +use anyhow::{Context, Result}; +use crossbeam_channel::Receiver; +use curl::easy::Handler; +use parking_lot::Mutex; use std::{ - fmt, sync::{ Arc, atomic::{AtomicBool, Ordering}, }, thread, - time::Instant, }; -use anyhow::{Context, Result, anyhow}; -use crossbeam_channel::{Receiver, Sender}; -use curl::{ - easy::{Easy2, Handler}, - multi::MultiWaker, -}; -use parking_lot::Mutex; - -use crate::dns_adapter::CurlDnsResolution; -use crate::websocket::CurlWebSocketConnector; - pub use config::CurlMultiRuntimeConfig; pub use identity::CurlTransferId; +use owner::CurlRuntimeOwner; -use identity::next_transfer_id; -use owner::{CurlRuntimeCommand, CurlRuntimeOwner}; - -/// Origin key used by the curl scheduler for per-origin active transfer caps. -#[derive(Debug, Clone, PartialEq, Eq, Hash)] -pub struct CurlOriginKey { - pub scheme: String, - pub host: String, - pub port: Option, -} - -/// A configured curl transfer plus scheduler metadata. -pub struct CurlMultiJob { - pub easy: Easy2, - pub context: C, - pub origin: Option, - /// Absolute deadline for the whole scheduler-owned transfer attempt. - /// - /// libcurl cannot account for time spent in Moli's priority queue or in - /// the shared DNS residence because both happen before the easy handle is - /// added to the multi handle. The owner enforces this deadline in those - /// residences and gives libcurl only the remaining duration. - pub deadline: Option, - /// DNS ownership chosen by the caller before this transfer enters curl. - /// - /// A curl-managed policy preserves libcurl's resolver behavior. A shared - /// origin policy parks the transfer outside the curl multi handle set until - /// the bounded system resolver publishes an answer. - pub dns_resolution: CurlDnsResolution, - /// Higher values start before lower values when jobs are queued. - pub priority: u8, - pub label: String, -} - -impl fmt::Debug for CurlMultiJob { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("CurlMultiJob") - .field("origin", &self.origin) - .field("deadline", &self.deadline) - .field("dns_resolution", &self.dns_resolution) - .field("priority", &self.priority) - .field("label", &self.label) - .finish_non_exhaustive() - } -} - -/// Completion emitted by `CurlMultiRuntime`. -pub struct CurlMultiCompletion { - pub transfer_id: CurlTransferId, - pub easy: Option>, - pub context: C, - pub result: Result<()>, -} - -impl fmt::Debug for CurlMultiCompletion { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("CurlMultiCompletion") - .field("transfer_id", &self.transfer_id) - .field("has_easy", &self.easy.is_some()) - .field("result", &self.result.as_ref().map(|_| ())) - .finish_non_exhaustive() - } -} - -/// Error returned when a job cannot be submitted and is returned to the caller. -pub struct CurlSubmitError { - pub job: CurlMultiJob, - pub error: anyhow::Error, -} - -impl fmt::Debug for CurlSubmitError { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("CurlSubmitError") - .field("job", &self.job) - .field("error", &self.error) - .finish() - } +#[derive(Debug)] +pub(crate) enum CurlRuntimeCommand { + Request { + transfer_id: CurlTransferId, + job: CurlMultiJob, + }, + Shutdown, } /// Owns one native thread. Request handles cannot extend its lifetime or join it. @@ -116,24 +43,6 @@ pub struct CurlMultiRuntime { owner_handle: Mutex>>, } -/// Submits HTTP work to one owner. Clones carry no shutdown or join authority. -#[derive(Debug)] -pub struct CurlHttpSender { - command_tx: Sender>, - owner_waker: MultiWaker, - shutdown_requested: Arc, -} - -impl Clone for CurlHttpSender { - fn clone(&self) -> Self { - Self { - command_tx: self.command_tx.clone(), - owner_waker: self.owner_waker.clone(), - shutdown_requested: self.shutdown_requested.clone(), - } - } -} - impl CurlMultiRuntime { pub fn new( config: CurlMultiRuntimeConfig, @@ -146,20 +55,24 @@ impl CurlMultiRuntime { let shutdown_requested = Arc::new(AtomicBool::new(false)); #[cfg(test)] let owner_started = Arc::new(AtomicBool::new(false)); - let owner = CurlRuntimeOwner::new( - config, - command_rx, - completion_tx, - waker_tx, - Arc::clone(&shutdown_requested), - websocket_rx, - #[cfg(test)] - Arc::clone(&owner_started), - ); - let thread_name = owner.thread_name().to_owned(); + let thread_name = config.thread_name.clone(); + let owner_shutdown = shutdown_requested.clone(); + #[cfg(test)] + let started = owner_started.clone(); let owner_handle = thread::Builder::new() .name(thread_name) - .spawn(move || owner.run()) + .spawn(move || { + CurlRuntimeOwner::run( + config, + command_rx, + completion_tx, + waker_tx, + owner_shutdown, + websocket_rx, + #[cfg(test)] + started, + ) + }) .context("failed to spawn curl multi runtime owner thread")?; let owner_waker = waker_rx .recv() @@ -213,144 +126,3 @@ impl Drop for CurlMultiRuntime CurlHttpSender { - pub fn submit( - &self, - job: CurlMultiJob, - ) -> std::result::Result> { - if self.shutdown_requested.load(Ordering::SeqCst) { - return Err(CurlSubmitError { - job, - error: anyhow!("curl multi runtime is shutting down"), - }); - } - let transfer_id = match next_transfer_id() { - Ok(transfer_id) => transfer_id, - Err(error) => return Err(CurlSubmitError { job, error }), - }; - match self - .command_tx - .send(CurlRuntimeCommand::Request { transfer_id, job }) - { - Ok(()) => { - let _ = self.owner_waker.wakeup(); - Ok(transfer_id) - } - Err(error) => { - let CurlRuntimeCommand::Request { job, .. } = error.into_inner() else { - unreachable!("submit only sends request commands"); - }; - Err(CurlSubmitError { - job, - error: anyhow!("curl multi runtime is shutting down"), - }) - } - } - } -} - -#[cfg(test)] -mod tests { - use std::{ - io::{Read, Write}, - net::TcpListener, - time::Duration, - }; - - use super::*; - - #[derive(Debug)] - struct TestHandler; - - impl Handler for TestHandler {} - - #[test] - fn submitted_identity_reaches_the_matching_runtime_completion() { - let listener = TcpListener::bind(("127.0.0.1", 0)) - .expect("test HTTP listener should bind to a local port"); - let address = listener - .local_addr() - .expect("test HTTP listener should have an address"); - let server = thread::spawn(move || { - let (mut stream, _) = listener - .accept() - .expect("curl should connect to the test HTTP listener"); - stream - .set_read_timeout(Some(Duration::from_secs(5))) - .expect("test HTTP connection should accept a read timeout"); - let mut request = [0; 4096]; - let _ = stream - .read(&mut request) - .expect("test HTTP request should be readable"); - stream - .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok") - .expect("test HTTP response should be writable"); - }); - - let (runtime, completion_rx) = CurlMultiRuntime::new(CurlMultiRuntimeConfig { - poll_interval: Duration::from_millis(5), - ..CurlMultiRuntimeConfig::default() - }) - .expect("test curl runtime should start"); - let mut easy = Easy2::new(TestHandler); - easy.url(&format!("http://{address}/identity")) - .expect("test curl URL should be valid"); - let transfer_id = runtime - .http_sender() - .submit(CurlMultiJob { - easy, - context: "matching-context".to_owned(), - origin: None, - deadline: None, - dns_resolution: CurlDnsResolution::curl_managed(), - priority: 1, - label: "identity-test".to_owned(), - }) - .expect("test curl transfer should be accepted"); - - let completion = completion_rx - .recv_timeout(Duration::from_secs(5)) - .expect("test curl transfer should reach terminal completion"); - assert_eq!(completion.transfer_id, transfer_id); - assert_eq!(completion.context, "matching-context"); - assert!(completion.easy.is_some()); - completion - .result - .expect("test curl transfer should complete successfully"); - - runtime.shutdown(); - server.join().expect("test HTTP server should finish"); - } - - #[test] - fn http_sender_does_not_keep_owner_alive_and_returns_rejected_job() { - let (runtime, completed) = - CurlMultiRuntime::>::new(Default::default()).unwrap(); - let sender = runtime.http_sender(); - let retained = sender.clone(); - drop(runtime); - for sender in [sender, retained] { - let mut easy = Easy2::new(TestHandler); - easy.url("http://127.0.0.1:1/must-not-connect").unwrap(); - let error = sender - .submit(CurlMultiJob { - easy, - context: vec![7; 1024], - origin: None, - deadline: None, - dns_resolution: CurlDnsResolution::curl_managed(), - priority: 1, - label: "closed".into(), - }) - .unwrap_err(); - assert!(error.error.to_string().contains("shutting down")); - assert_eq!(error.job.context, vec![7; 1024]); - assert_eq!(error.job.label, "closed"); - } - assert!(matches!( - completed.try_recv(), - Err(crossbeam_channel::TryRecvError::Disconnected) - )); - } -} diff --git a/moli-curl/src/websocket/diagnostics.rs b/moli-curl/src/runtime/diagnostics.rs similarity index 95% rename from moli-curl/src/websocket/diagnostics.rs rename to moli-curl/src/runtime/diagnostics.rs index 23c2dcb39a..0f37e413cc 100644 --- a/moli-curl/src/websocket/diagnostics.rs +++ b/moli-curl/src/runtime/diagnostics.rs @@ -3,7 +3,7 @@ use std::time::{Duration, Instant}; #[derive(Default, Debug)] -pub(super) struct Counters { +pub(crate) struct Counters { pub turns: u64, pub progressed_turns: u64, pub read_bytes: u64, @@ -47,10 +47,10 @@ struct Window { counters: Counters, } -pub(super) struct Diagnostics(Option); +pub(crate) struct Diagnostics(Option); impl Diagnostics { - pub(super) fn from_env() -> Self { + pub(crate) fn from_env() -> Self { Self::new( std::env::var("MOLI_CURL_WEBSOCKET_DIAGNOSTICS").is_ok_and(|value| { let value = value.trim(); @@ -66,15 +66,15 @@ impl Diagnostics { })) } - pub(super) fn counters(&mut self) -> Option<&mut Counters> { + pub(crate) fn counters(&mut self) -> Option<&mut Counters> { self.0.as_mut().map(|window| &mut window.counters) } - pub(super) fn poll_start(&self) -> Option { + pub(crate) fn poll_start(&self) -> Option { self.0.as_ref().map(|_| Instant::now()) } - pub(super) fn polled(&mut self, start: Option, timeout: Duration, progressed: bool) { + pub(crate) fn polled(&mut self, start: Option, timeout: Duration, progressed: bool) { if let Some(start) = start { self.0 .as_mut() @@ -84,7 +84,7 @@ impl Diagnostics { } } - pub(super) fn report(&mut self, sessions: usize, finished: bool) { + pub(crate) fn report(&mut self, sessions: usize, finished: bool) { let Some(window) = &mut self.0 else { return; }; diff --git a/moli-curl/src/runtime/owner.rs b/moli-curl/src/runtime/owner.rs index f2c2491262..61846e649e 100644 --- a/moli-curl/src/runtime/owner.rs +++ b/moli-curl/src/runtime/owner.rs @@ -1,62 +1,55 @@ +//! One native owner drives both protocols. Registries own their easy handles; +//! only this loop performs curl work, drains CURLMSG_DONE and waits for readiness. + use std::{ - num::NonZeroUsize, sync::{ - Arc, OnceLock, + Arc, atomic::{AtomicBool, Ordering}, }, time::{Duration, Instant}, }; -use anyhow::{Context, Result, anyhow}; -use crossbeam_channel::{self, Receiver, Sender}; +use crossbeam_channel::{Receiver, Sender}; use curl::{ easy::Handler, multi::{Multi, MultiWaker}, }; use tracing::debug; -use crate::dns_adapter::{CurlDnsOwnerCompletion, CurlDnsReady}; -use crate::websocket::{Submission, registry::WebSocketRegistry}; - use super::{ - CurlMultiCompletion, CurlMultiJob, CurlMultiRuntimeConfig, CurlTransferId, + CurlMultiRuntimeConfig, CurlRuntimeCommand, config::{make_runtime_multi, runtime_wait_timeout}, - residence::{ - CurlActiveTransfer, CurlOwnerState, CurlPendingJob, active_origin_count, - enqueue_existing_pending_job, enqueue_pending_job, job_is_eligible, pending_origin_count, - take_expired_pending_jobs, take_transfers_in_notification_order, - }, + diagnostics::Diagnostics, +}; +use crate::{ + CurlMultiCompletion, CurlTransferId, + dns_adapter::CurlDnsOwnerCompletion, + http::registry::HttpRegistry, + websocket::{Submission, registry::WebSocketRegistry}, }; -#[derive(Debug)] -pub(super) enum CurlRuntimeCommand { - Request { - transfer_id: CurlTransferId, - job: CurlMultiJob, - }, - Shutdown, -} - enum CurlOwnerEvent { - Command(std::result::Result, crossbeam_channel::RecvError>), - Dns(std::result::Result, crossbeam_channel::RecvError>), - WebSocket(std::result::Result), + Command(Result, crossbeam_channel::RecvError>), + Dns(Result, crossbeam_channel::RecvError>), + WebSocket(Result), Deadline, } -pub(super) struct CurlRuntimeOwner { - config: CurlMultiRuntimeConfig, +pub(super) struct CurlRuntimeOwner { command_rx: Receiver>, - completion_tx: Sender>, - waker_tx: Sender, shutdown_requested: Arc, - websocket_rx: Receiver, - #[cfg(test)] - owner_started: Arc, + closed: bool, + poll_interval: Duration, + diagnostics: Diagnostics, + http: HttpRegistry, + websockets: WebSocketRegistry, + // Drop the registries' easy handles before their Multi, including on unwind. + multi: Multi, } -impl CurlRuntimeOwner { - pub(super) fn new( +impl CurlRuntimeOwner { + /// Construct all native state on its owner thread, including the WS cache. + pub(super) fn run( config: CurlMultiRuntimeConfig, command_rx: Receiver>, completion_tx: Sender>, @@ -64,640 +57,176 @@ impl CurlRuntimeOwner { shutdown_requested: Arc, websocket_rx: Receiver, #[cfg(test)] owner_started: Arc, - ) -> Self { - Self { - config, - command_rx, - completion_tx, - waker_tx, - shutdown_requested, - websocket_rx, - #[cfg(test)] - owner_started, - } - } - - pub(super) fn thread_name(&self) -> &str { - &self.config.thread_name - } - - pub(super) fn run(self) { + ) { #[cfg(test)] - self.owner_started.store(true, Ordering::SeqCst); - let mut multi = make_runtime_multi(&self.config); - let _ = self.waker_tx.send(multi.waker()); - let mut state = CurlOwnerState::default(); - let mut websockets = WebSocketRegistry::new(self.websocket_rx.clone()); + owner_started.store(true, Ordering::SeqCst); + let multi = make_runtime_multi(&config); + let _ = waker_tx.send(multi.waker()); + let mut owner = Self { + 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, + }; + owner.drive(); + } + fn drive(&mut self) { loop { - self.drain_commands(&mut state, &mut multi); - if state.closed { - websockets.shutdown(&mut multi); + self.drain_commands(); + self.http.advance(&mut self.multi); + self.process_completed_transfers(); + let progressed = self + .websockets + .advance(&mut self.multi, &mut self.diagnostics); + if let Some(counters) = self.diagnostics.counters() { + counters.turns += 1; + counters.progressed_turns += u64::from(progressed); } - self.drain_dns_completions(&mut state); - self.expire_waiting_jobs(&mut state); - self.start_eligible_jobs(&mut state, &mut multi); - self.process_completed_transfers(&mut state, &mut multi, &mut websockets); - let progressed = websockets.advance(&mut multi); - if state.closed - && state.pending.is_empty() - && state.dns.is_empty() - && state.active.is_empty() - && self.command_rx.is_empty() - { + // close() retires both registries. Drain racing HTTP submissions so + // every accepted job still receives its terminal completion. + if self.closed && self.command_rx.is_empty() { + self.diagnostics.report(0, true); return; } let runnable = progressed || !self.command_rx.is_empty(); - if !runnable - && state.active.is_empty() - && state.pending.is_empty() - && websockets.is_empty() - { - self.wait_for_next_owner_event(&mut state, &mut multi, &mut websockets); + if !runnable && !self.http.has_curl_work() && self.websockets.is_empty() { + self.wait_for_next_owner_event(); } else { - self.wait_for_curl_progress(&mut multi, &state, &mut websockets, runnable); + self.wait_for_curl_progress(runnable); } } } - fn drain_commands(&self, state: &mut CurlOwnerState, multi: &mut Multi) { + fn drain_commands(&mut self) { // Continuous HTTP submissions must not starve native I/O or WebSockets. for _ in 0..256 { match self.command_rx.try_recv() { - Ok(command) => self.handle_command(state, multi, command), + Ok(command) => self.handle_command(command), Err(crossbeam_channel::TryRecvError::Empty) => break, Err(crossbeam_channel::TryRecvError::Disconnected) => { - self.close(state, multi); + self.close(); break; } } } } - fn wait_for_next_owner_event( - &self, - state: &mut CurlOwnerState, - multi: &mut Multi, - websockets: &mut WebSocketRegistry, - ) { - if state.closed { + fn handle_command(&mut self, command: CurlRuntimeCommand) { + match command { + CurlRuntimeCommand::Request { transfer_id, job } => self.http.admit(transfer_id, job), + CurlRuntimeCommand::Shutdown => self.close(), + } + } + + fn close(&mut self) { + if self.closed { return; } - let event = if let Some(deadline) = state.dns.next_deadline(|pending| pending.job.deadline) - { + self.closed = true; + self.shutdown_requested.store(true, Ordering::SeqCst); + self.http.shutdown(&mut self.multi); + self.websockets.shutdown(&mut self.multi); + } + + fn wait_for_next_owner_event(&mut self) { + if self.closed { + return; + } + let event = if let Some(deadline) = self.http.next_deadline() { let deadline_rx = crossbeam_channel::after(deadline.saturating_duration_since(Instant::now())); crossbeam_channel::select! { recv(self.command_rx) -> command => CurlOwnerEvent::Command(command), - recv(websockets.submissions()) -> submission => CurlOwnerEvent::WebSocket(submission), - recv(state.dns.completion_receiver()) -> completion => CurlOwnerEvent::Dns(completion), + recv(self.websockets.submissions()) -> submission => CurlOwnerEvent::WebSocket(submission), + recv(self.http.dns_completions()) -> completion => CurlOwnerEvent::Dns(completion), recv(deadline_rx) -> _ => CurlOwnerEvent::Deadline, } } else { crossbeam_channel::select! { recv(self.command_rx) -> command => CurlOwnerEvent::Command(command), - recv(websockets.submissions()) -> submission => CurlOwnerEvent::WebSocket(submission), - recv(state.dns.completion_receiver()) -> completion => CurlOwnerEvent::Dns(completion), + recv(self.websockets.submissions()) -> submission => CurlOwnerEvent::WebSocket(submission), + recv(self.http.dns_completions()) -> completion => CurlOwnerEvent::Dns(completion), } }; match event { - CurlOwnerEvent::WebSocket(Ok(submission)) => websockets.admit(submission, multi), - CurlOwnerEvent::WebSocket(Err(_)) => self.close(state, multi), - CurlOwnerEvent::Command(command) => match command { - Ok(command) => self.handle_command(state, multi, command), - Err(_) => self.close(state, multi), - }, - CurlOwnerEvent::Dns(completion) => { - if let Ok(completion) = completion { - self.claim_dns_completion(state, completion); - } + CurlOwnerEvent::Command(Ok(command)) => self.handle_command(command), + CurlOwnerEvent::WebSocket(Ok(submission)) => { + self.websockets.admit(submission, &mut self.multi) } - CurlOwnerEvent::Deadline => {} + CurlOwnerEvent::Command(Err(_)) | CurlOwnerEvent::WebSocket(Err(_)) => self.close(), + CurlOwnerEvent::Dns(Ok(completion)) => self.http.claim_dns_completion(completion), + CurlOwnerEvent::Dns(Err(_)) | CurlOwnerEvent::Deadline => {} } } - fn handle_command( - &self, - state: &mut CurlOwnerState, - multi: &mut Multi, - command: CurlRuntimeCommand, - ) { - match command { - CurlRuntimeCommand::Request { transfer_id, job } if state.closed => { - self.send_completion(CurlMultiCompletion { - transfer_id, - easy: Some(job.easy), - context: job.context, - result: Err(anyhow!("curl multi runtime is shutting down")), - }); - } - CurlRuntimeCommand::Request { transfer_id, job } => { - self.admit_job(state, transfer_id, job) - } - CurlRuntimeCommand::Shutdown => self.close(state, multi), - } - } - - fn admit_job( - &self, - state: &mut CurlOwnerState, - transfer_id: CurlTransferId, - job: CurlMultiJob, - ) { - if curl_runtime_trace_enabled() { - let origin = job.origin.as_ref(); - tracing::info!( - target: "moli_cdp_nav_timing", - transfer_id = %transfer_id, - label = %job.label, - origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""), - origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""), - origin_port = ?origin.and_then(|origin| origin.port), - origin = ?job.origin, - priority = job.priority, - pending_before = state.pending.len(), - pending_same_origin_before = origin - .map(|origin| pending_origin_count(&state.pending, origin)) - .unwrap_or(0), - stage = "curl_runtime_job_queued", - ); - } - enqueue_pending_job(&mut state.pending, transfer_id, job); - } - - fn close(&self, state: &mut CurlOwnerState, multi: &mut Multi) { - if state.closed { - return; - } - state.closed = true; - self.shutdown_requested.store(true, Ordering::SeqCst); - while let Some(pending) = state.pending.pop_front() { - let CurlPendingJob { - transfer_id, job, .. - } = pending; - self.send_completion(CurlMultiCompletion { - transfer_id, - easy: Some(job.easy), - context: job.context, - result: Err(anyhow!("curl multi runtime is shutting down")), - }); - } - for pending in state.dns.drain() { - let CurlPendingJob { - transfer_id, job, .. - } = pending; - self.send_completion(CurlMultiCompletion { - transfer_id, - easy: Some(job.easy), - context: job.context, - result: Err(anyhow!( - "curl multi runtime DNS request cancelled during shutdown" - )), - }); - } - for (transfer_id, active) in state.active.drain() { - let easy = multi.remove2(active.handle).ok(); - self.send_completion(CurlMultiCompletion { - transfer_id, - easy, - context: active.context, - result: Err(anyhow!( - "curl multi runtime request cancelled during shutdown" - )), - }); - } - } - - fn start_eligible_jobs(&self, state: &mut CurlOwnerState, multi: &mut Multi) { - loop { - if state.closed || state.active.len() >= self.config.max_active.get() { - return; - } - let Some(index) = state.pending.iter().position(|pending| { - job_is_eligible( - pending.job.origin.as_ref(), - state, - self.config.max_host_active, - ) - }) else { - return; - }; - let pending = state - .pending - .remove(index) - .expect("pending curl job index should exist"); - let dns_target = pending.job.dns_resolution.target().cloned(); - match dns_target { - Some(target) => { - state - .dns - .start(pending.transfer_id, pending, target, multi.waker()) - } - None => self.start_job(state, multi, pending), - } - } - } - - fn drain_dns_completions(&self, state: &mut CurlOwnerState) { - while let Some(ready) = state.dns.try_claim_next() { - self.handle_dns_completion(state, ready); - } - } - - fn claim_dns_completion( - &self, - state: &mut CurlOwnerState, - completion: CurlDnsOwnerCompletion, - ) { - let Some(ready) = state.dns.claim(completion) else { - return; - }; - self.handle_dns_completion(state, ready); - } - - fn handle_dns_completion( - &self, - state: &mut CurlOwnerState, - ready: CurlDnsReady>, - ) { - let mut pending = ready.pending; - if state.closed { - let CurlPendingJob { - transfer_id, job, .. - } = pending; - self.send_completion(CurlMultiCompletion { - transfer_id, - easy: Some(job.easy), - context: job.context, - result: Err(anyhow!("curl multi runtime is shutting down")), - }); - return; - } - if pending.deadline_reached(Instant::now()) { - self.complete_timed_out_job(pending, "while waiting for DNS"); - return; - } - match ready.result { - Ok(addresses) => { - if let Err(error) = pending - .job - .dns_resolution - .install(&mut pending.job.easy, addresses.as_ref()) - { - let CurlPendingJob { - transfer_id, job, .. - } = pending; - self.send_completion(CurlMultiCompletion { - transfer_id, - easy: Some(job.easy), - context: job.context, - result: Err(error), - }); - return; - } - enqueue_existing_pending_job(&mut state.pending, pending); - } - Err(error) => { - let CurlPendingJob { - transfer_id, job, .. - } = pending; - self.send_completion(CurlMultiCompletion { - transfer_id, - easy: Some(job.easy), - context: job.context, - result: Err(anyhow!(error.to_string())), - }); - } - } - } - - fn start_job( - &self, - state: &mut CurlOwnerState, - multi: &mut Multi, - pending: CurlPendingJob, - ) { - if pending.deadline_reached(Instant::now()) { - self.complete_timed_out_job(pending, "while waiting to start"); - return; - } - let transfer_id = pending.transfer_id; - let queued_for = pending.enqueued_at.elapsed(); - let mut job = pending.job; - let label = job.label.clone(); - if let Some(deadline) = job.deadline { - let Some(remaining) = curl_timeout_for_deadline(deadline, Instant::now()) else { - self.send_completion(CurlMultiCompletion { - transfer_id, - easy: Some(job.easy), - context: job.context, - result: Err(curl_runtime_timeout_error("while waiting to start")), - }); - return; - }; - if let Err(error) = job.easy.timeout(remaining) { - self.send_completion(CurlMultiCompletion { - transfer_id, - easy: Some(job.easy), - context: job.context, - result: Err(error).context("failed to apply remaining curl request deadline"), - }); - return; - } - } - match multi - .add2(job.easy) - .with_context(|| anyhow!("failed to add curl easy handle for {label}")) - { - Ok(mut handle) => { - handle - .set_token(transfer_id.token()) - .expect("active curl handle must accept its transfer token"); - if curl_runtime_trace_enabled() { - let origin = job.origin.as_ref(); - tracing::info!( - target: "moli_cdp_nav_timing", - transfer_id = %transfer_id, - label = %label, - origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""), - origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""), - origin_port = ?origin.and_then(|origin| origin.port), - origin = ?job.origin, - priority = job.priority, - queued_ms = queued_for.as_millis(), - active_before = state.active.len(), - active_same_origin_before = origin - .map(|origin| active_origin_count(&state.active, origin)) - .unwrap_or(0), - pending_after = state.pending.len(), - pending_same_origin_after = origin - .map(|origin| pending_origin_count(&state.pending, origin)) - .unwrap_or(0), - max_active = self.config.max_active.get(), - max_host_active = ?self.config.max_host_active.map(NonZeroUsize::get), - max_host_connections = ?self.config.max_host_connections.map(NonZeroUsize::get), - max_total_connections = ?self.config.max_total_connections.map(NonZeroUsize::get), - max_concurrent_streams = ?self.config.max_concurrent_streams.map(NonZeroUsize::get), - multiplex = self.config.multiplex, - stage = "curl_runtime_job_start", - ); - } - let previous = state.active.insert( - transfer_id, - CurlActiveTransfer { - handle, - context: job.context, - origin: job.origin, - priority: job.priority, - label, - started_at: Instant::now(), - queued_for, - }, - ); - assert!(previous.is_none(), "curl transfer identity is unique"); - } - Err(error) => self.send_completion(CurlMultiCompletion { - transfer_id, - easy: None, - context: job.context, - result: Err(error), - }), - } - } - - fn process_completed_transfers( - &self, - state: &mut CurlOwnerState, - multi: &mut Multi, - websockets: &mut WebSocketRegistry, - ) { - if let Err(error) = multi.perform() { + fn process_completed_transfers(&mut self) { + if let Err(error) = self.multi.perform() { debug!("curl multi runtime perform failed: {error}"); - websockets.fail_sessions(multi, &error.to_string()); + self.websockets + .fail_sessions(&mut self.multi, &error.to_string()); } // Drain CURLMSG_DONE once. A WebSocket DONE starts its open residence; // HTTP DONE removes a finished transfer, preserving notification order. let mut completed = Vec::new(); - multi.messages(|message| { + self.multi.messages(|message| { let Some(id) = message.token().ok().and_then(CurlTransferId::from_token) else { return; }; - // Keep the easy handle's error buffer as well as the CURLcode. - let result = match state.active.get(&id) { - Some(transfer) => message.result_for2(&transfer.handle), - None => websockets.handshake_result(id, &message), - }; + // Both registries use result_for2 to retain the easy error buffer. + let result = self + .http + .completion_result(id, &message) + .or_else(|| self.websockets.handshake_result(id, &message)); if let Some(result) = result { completed.push((id, result)); } }); for (id, result) in &completed { - if !state.active.contains_key(id) { - websockets.complete_handshake(*id, result.clone(), multi); + if !self.http.contains(*id) { + self.websockets + .complete_handshake(*id, result.clone(), &mut self.multi); } } - for (transfer_id, active, result) in - take_transfers_in_notification_order(&mut state.active, completed) - { - self.finish_active_transfer( - state, - multi, - transfer_id, - active, - result.map_err(Into::into), - ); - } + self.http.complete(&mut self.multi, completed); } - fn expire_waiting_jobs(&self, state: &mut CurlOwnerState) { - let now = Instant::now(); - for pending in take_expired_pending_jobs(&mut state.pending, now) { - self.complete_timed_out_job(pending, "while waiting in the scheduler"); - } - for pending in state.dns.take_expired(now, |pending| pending.job.deadline) { - self.complete_timed_out_job(pending, "while waiting for DNS"); - } - } - - fn complete_timed_out_job(&self, pending: CurlPendingJob, stage: &'static str) { - let CurlPendingJob { - transfer_id, job, .. - } = pending; - self.send_completion(CurlMultiCompletion { - transfer_id, - easy: Some(job.easy), - context: job.context, - result: Err(curl_runtime_timeout_error(stage)), - }); - } - - fn finish_active_transfer( - &self, - state: &CurlOwnerState, - multi: &mut Multi, - transfer_id: CurlTransferId, - active: CurlActiveTransfer, - result: Result<()>, - ) { - let easy = match multi.remove2(active.handle) { - Ok(easy) => Some(easy), - Err(error) => { - self.send_completion(CurlMultiCompletion { - transfer_id, - easy: None, - context: active.context, - result: Err(anyhow!( - "failed to remove curl easy handle for {}: {error}", - active.label - )), - }); - return; - } - }; - if curl_runtime_trace_enabled() { - let origin = active.origin.as_ref(); - match &result { - Ok(()) => { - tracing::info!( - target: "moli_cdp_nav_timing", - transfer_id = %transfer_id, - label = %active.label, - origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""), - origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""), - origin_port = ?origin.and_then(|origin| origin.port), - origin = ?active.origin, - priority = active.priority, - ok = true, - active_ms = active.started_at.elapsed().as_millis(), - queued_ms = active.queued_for.as_millis(), - active_remaining = state.active.len(), - active_same_origin_remaining = origin - .map(|origin| active_origin_count(&state.active, origin)) - .unwrap_or(0), - pending_after = state.pending.len(), - pending_same_origin_after = origin - .map(|origin| pending_origin_count(&state.pending, origin)) - .unwrap_or(0), - stage = "curl_runtime_job_done", - ); - } - Err(error) => { - tracing::info!( - target: "moli_cdp_nav_timing", - transfer_id = %transfer_id, - label = %active.label, - origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""), - origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""), - origin_port = ?origin.and_then(|origin| origin.port), - origin = ?active.origin, - priority = active.priority, - ok = false, - error = %error, - active_ms = active.started_at.elapsed().as_millis(), - queued_ms = active.queued_for.as_millis(), - active_remaining = state.active.len(), - active_same_origin_remaining = origin - .map(|origin| active_origin_count(&state.active, origin)) - .unwrap_or(0), - pending_after = state.pending.len(), - pending_same_origin_after = origin - .map(|origin| pending_origin_count(&state.pending, origin)) - .unwrap_or(0), - stage = "curl_runtime_job_done", - ); - } - } - } - let result = result.with_context(|| { - anyhow!( - "curl request failed for {} after active={}ms queued={}ms", - active.label, - active.started_at.elapsed().as_millis(), - active.queued_for.as_millis() - ) - }); - self.send_completion(CurlMultiCompletion { - transfer_id, - easy, - context: active.context, - result, - }); - } - - fn wait_for_curl_progress( - &self, - multi: &mut Multi, - state: &CurlOwnerState, - websockets: &mut WebSocketRegistry, - progressed: bool, - ) { + fn wait_for_curl_progress(&mut self, runnable: bool) { // HTTP cancellation uses its configured progress interval. Idle WS // sessions use socket/waker readiness and need not inherit that cadence. - let interval = if state.active.is_empty() { - Duration::from_secs(1) + let interval = if self.http.has_active() { + self.poll_interval } else { - self.config.poll_interval + Duration::from_secs(1) }; - let mut wait_timeout = runtime_wait_timeout(multi, interval).unwrap_or(interval); - if let Some(deadline) = state.next_waiting_deadline() { - wait_timeout = wait_timeout.min(deadline.saturating_duration_since(Instant::now())); + let mut timeout = runtime_wait_timeout(&self.multi, interval).unwrap_or(interval); + for deadline in [self.http.next_deadline(), self.websockets.next_deadline()] + .into_iter() + .flatten() + { + timeout = timeout.min(deadline.saturating_duration_since(Instant::now())); } - if let Some(deadline) = websockets.next_deadline() { - wait_timeout = wait_timeout.min(deadline.saturating_duration_since(Instant::now())); + if runnable { + timeout = Duration::ZERO; } - if progressed { - wait_timeout = Duration::ZERO; + // libcurl adds HTTP/handshake sockets and its waker to these open WS fds. + let fds = self.websockets.poll_fds(); + let started = self.diagnostics.poll_start(); + let result = self.multi.poll(fds, timeout); + self.diagnostics.polled(started, timeout, runnable); + match result { + Ok(_) => self.websockets.apply_readiness(), + Err(error) => self + .websockets + .fail_sessions(&mut self.multi, &error.to_string()), } - websockets.wait(multi, wait_timeout, progressed); - } - - fn send_completion(&self, completion: CurlMultiCompletion) { - let _ = self.completion_tx.send(completion); - } -} - -fn curl_runtime_timeout_error(stage: &str) -> anyhow::Error { - anyhow!("curl multi runtime request timed out {stage}") -} - -fn curl_timeout_for_deadline(deadline: Instant, now: Instant) -> Option { - let remaining = deadline.saturating_duration_since(now); - // curl-rust converts CURLOPT_TIMEOUT_MS with `Duration::as_millis()`. A - // positive sub-millisecond value would therefore become zero, which - // libcurl interprets as disabling the timeout entirely. - (remaining >= Duration::from_millis(1)).then_some(remaining) -} - -fn curl_runtime_trace_enabled() -> bool { - static ENABLED: OnceLock = OnceLock::new(); - *ENABLED.get_or_init(|| { - env_flag_enabled("MOLI_CDP_NAV_TIMING") || env_flag_enabled("MOLI_CURL_RUNTIME_TRACE") - }) -} - -fn env_flag_enabled(name: &str) -> bool { - std::env::var(name).is_ok_and(|value| { - let value = value.trim(); - !value.is_empty() && value != "0" && !value.eq_ignore_ascii_case("false") - }) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn sub_millisecond_deadline_never_disables_the_libcurl_timeout() { - let now = Instant::now(); - assert_eq!( - curl_timeout_for_deadline(now + Duration::from_micros(999), now), - None - ); - assert_eq!( - curl_timeout_for_deadline(now + Duration::from_millis(1), now), - Some(Duration::from_millis(1)) - ); + self.diagnostics + .report(self.websockets.session_count(), false); } } diff --git a/moli-curl/src/runtime/tests.rs b/moli-curl/src/runtime/tests.rs new file mode 100644 index 0000000000..c096c2f4ed --- /dev/null +++ b/moli-curl/src/runtime/tests.rs @@ -0,0 +1,103 @@ +use std::{ + io::{Read, Write}, + net::TcpListener, + time::Duration, +}; + +use super::*; +use crate::{CurlDnsResolution, CurlMultiJob}; +use curl::easy::Easy2; + +#[derive(Debug)] +struct TestHandler; + +impl Handler for TestHandler {} + +#[test] +fn submitted_identity_reaches_the_matching_runtime_completion() { + let listener = TcpListener::bind(("127.0.0.1", 0)) + .expect("test HTTP listener should bind to a local port"); + let address = listener + .local_addr() + .expect("test HTTP listener should have an address"); + let server = thread::spawn(move || { + let (mut stream, _) = listener + .accept() + .expect("curl should connect to the test HTTP listener"); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .expect("test HTTP connection should accept a read timeout"); + let mut request = [0; 4096]; + let _ = stream + .read(&mut request) + .expect("test HTTP request should be readable"); + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok") + .expect("test HTTP response should be writable"); + }); + + let (runtime, completion_rx) = CurlMultiRuntime::new(CurlMultiRuntimeConfig { + poll_interval: Duration::from_millis(5), + ..CurlMultiRuntimeConfig::default() + }) + .expect("test curl runtime should start"); + let mut easy = Easy2::new(TestHandler); + easy.url(&format!("http://{address}/identity")) + .expect("test curl URL should be valid"); + let transfer_id = runtime + .http_sender() + .submit(CurlMultiJob { + easy, + context: "matching-context".to_owned(), + origin: None, + deadline: None, + dns_resolution: CurlDnsResolution::curl_managed(), + priority: 1, + label: "identity-test".to_owned(), + }) + .expect("test curl transfer should be accepted"); + + let completion = completion_rx + .recv_timeout(Duration::from_secs(5)) + .expect("test curl transfer should reach terminal completion"); + assert_eq!(completion.transfer_id, transfer_id); + assert_eq!(completion.context, "matching-context"); + assert!(completion.easy.is_some()); + completion + .result + .expect("test curl transfer should complete successfully"); + + runtime.shutdown(); + server.join().expect("test HTTP server should finish"); +} + +#[test] +fn http_sender_does_not_keep_owner_alive_and_returns_rejected_job() { + let (runtime, completed) = + CurlMultiRuntime::>::new(Default::default()).unwrap(); + let sender = runtime.http_sender(); + let retained = sender.clone(); + drop(runtime); + for sender in [sender, retained] { + let mut easy = Easy2::new(TestHandler); + easy.url("http://127.0.0.1:1/must-not-connect").unwrap(); + let error = sender + .submit(CurlMultiJob { + easy, + context: vec![7; 1024], + origin: None, + deadline: None, + dns_resolution: CurlDnsResolution::curl_managed(), + priority: 1, + label: "closed".into(), + }) + .unwrap_err(); + assert!(error.error.to_string().contains("shutting down")); + assert_eq!(error.job.context, vec![7; 1024]); + assert_eq!(error.job.label, "closed"); + } + assert!(matches!( + completed.try_recv(), + Err(crossbeam_channel::TryRecvError::Disconnected) + )); +} diff --git a/moli-curl/src/websocket.rs b/moli-curl/src/websocket.rs index c314db42db..36b54af619 100644 --- a/moli-curl/src/websocket.rs +++ b/moli-curl/src/websocket.rs @@ -1,42 +1,36 @@ -//! Persistent native WebSocket sessions, owned by a separate libcurl multi thread. +//! Native WebSocket frames on the shared libcurl runtime. //! //! This layer transports frames. Browser handshake policy, message assembly and //! close-handshake semantics belong to the caller. Dropping the receiver cancels //! its session, including DNS and handshake work, independently of queue capacity. //! -//! Internally, the shared runtime drives DNS and scheduling; session owns the native -//! handle and its Opening/Open/ReceivedClose lifecycle; scheduling holds I/O -//! admission state and maps polled sockets back to their sessions. Returning +//! Connector submits work; connection holds the caller/owner I/O endpoints; +//! registry owns DNS and attached sessions; session drives handshake and frame +//! I/O. Readiness maps the runtime's poll results back to sessions. Returning //! AGAIN parks that I/O until its socket is signalled. Application wakeups only //! resume paused work, such as a new frame or restored receive capacity. +mod connection; mod connection_pool; -mod diagnostics; +mod connector; +mod readiness; pub(crate) mod registry; mod request; -mod scheduling; mod session; +mod standalone; #[cfg(test)] mod tests; -use std::{ - sync::{ - Arc, - atomic::{AtomicBool, Ordering}, - }, - time::Duration, -}; +use crate::{CurlDnsResolution, CurlTlsConfig}; +use anyhow::{Result, bail}; +use std::time::Duration; -use anyhow::{Context, Result, bail}; -use curl::multi::MultiWaker; -use parking_lot::Mutex; -use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc, oneshot}; - -use crate::{ - CurlDnsResolution, CurlMultiRuntime, CurlMultiRuntimeConfig, CurlTlsConfig, CurlTransferId, - runtime::identity::next_transfer_id, -}; +use connection::SessionIo; +pub use connection::{CurlWebSocketConnection, CurlWebSocketSender}; +pub use connector::CurlWebSocketConnector; +pub(crate) use connector::Submission; pub use curl::easy::{WsFlags, WsFrame}; +pub use standalone::CurlWebSocketRuntime; /// Fragment large messages above this layer to bound native write residence. pub const MAX_SEND_FRAME_BYTES: usize = 64 * 1024; @@ -120,287 +114,3 @@ impl CurlWebSocketSend { Ok(()) } } - -#[derive(Debug)] -struct Control { - cancelled: AtomicBool, - closed: AtomicBool, - reading: AtomicBool, - send: Mutex>, - terminal: Mutex>>, - waker: MultiWaker, - #[cfg(test)] - read_blocked: tokio::sync::Notify, - #[cfg(test)] - write_blocked: tokio::sync::Notify, - #[cfg(test)] - read_attempts: std::sync::atomic::AtomicUsize, - #[cfg(test)] - receive_allocations: std::sync::atomic::AtomicUsize, - #[cfg(test)] - read_waiting: tokio::sync::Notify, - #[cfg(test)] - owner_thread: Mutex>, -} - -impl Control { - fn wake(&self) { - let _ = self.waker.wakeup(); - } - fn cancel(&self) { - self.cancelled.store(true, Ordering::Release); - self.wake(); - } -} - -#[derive(Debug)] -struct PendingSend { - frame: CurlWebSocketSend, - offset: usize, - completed: oneshot::Sender, -} - -/// Cloneable write/control capability; it does not keep a dropped receiver alive. -#[derive(Clone, Debug)] -pub struct CurlWebSocketSender { - control: Arc, -} - -impl CurlWebSocketSender { - /// Writes one frame and returns its payload size once libcurl consumes it. - /// - /// Only one frame may be pending per connection; overlapping sends return an - /// error. The caller owns data/control ordering and any message queue. - /// Completion is independent of reads and received event delivery. - /// - /// Dropping an unpolled future submits nothing. Once submitted, the frame - /// remains pending until written or the connection is cancelled/closed, even - /// if this future is dropped. Transport failure or cancellation returns an error. - pub async fn send_frame(&self, frame: CurlWebSocketSend) -> Result { - frame.validate()?; - let completion = { - let mut send = self.control.send.lock(); - if self.control.closed.load(Ordering::Acquire) - || self.control.cancelled.load(Ordering::Acquire) - { - bail!("WebSocket transport is closed"); - } - if send.is_some() { - bail!("WebSocket frame is already pending"); - } - let (completed, completion) = oneshot::channel(); - *send = Some(PendingSend { - frame, - offset: 0, - completed, - }); - completion - }; - self.control.wake(); - completion - .await - .context("WebSocket transport closed before frame completion") - } - - /// Handshake delivery starts paused, so application data cannot outrun Open. - /// Pausing does not retract queued chunks or a read already in progress. - /// Send completions progress independently of reads and received events. - pub fn set_reading(&self, enabled: bool) { - self.control.reading.store(enabled, Ordering::Release); - self.control.wake(); - } - - pub fn cancel(&self) { - self.control.cancel(); - } -} - -#[derive(Debug)] -pub struct CurlWebSocketConnection { - id: CurlTransferId, - sender: CurlWebSocketSender, - events: mpsc::Receiver, - terminal_delivered: bool, -} - -impl CurlWebSocketConnection { - pub fn id(&self) -> CurlTransferId { - self.id - } - pub fn sender(&self) -> CurlWebSocketSender { - self.sender.clone() - } - - pub async fn recv(&mut self) -> Option { - if let Some(event) = self.events.recv().await { - self.sender.control.wake(); - return Some(event); - } - if self.terminal_delivered { - return None; - } - self.terminal_delivered = true; - let result = self - .sender - .control - .terminal - .lock() - .take() - .unwrap_or_else(|| Err("WebSocket runtime stopped".to_owned())); - Some(CurlWebSocketEvent::Closed { result }) - } -} - -impl Drop for CurlWebSocketConnection { - fn drop(&mut self) { - self.sender.cancel(); - } -} - -struct SessionIo { - events: mpsc::Sender, - control: Arc, - _slot: OwnedSemaphorePermit, -} - -impl SessionIo { - fn cancelled(&self) -> bool { - self.control.cancelled.load(Ordering::Acquire) - } - fn finish(&self, result: std::result::Result<(), String>) { - *self.control.terminal.lock() = Some(result); - } -} - -impl Drop for SessionIo { - fn drop(&mut self) { - self.control.closed.store(true, Ordering::Release); - // Settle the pending write before closing the receive event channel. - self.control.send.lock().take(); - } -} - -pub(crate) struct Submission { - id: CurlTransferId, - request: CurlWebSocketRequest, - io: SessionIo, -} - -/// Admission capability for one native owner. Clones do not keep that owner -/// alive; shutting down its runtime closes connections and rejects new ones. -#[derive(Clone, Debug)] -pub struct CurlWebSocketConnector { - inner: Arc, -} - -#[derive(Debug)] -struct ConnectorInner { - submissions: crossbeam_channel::Sender, - slots: Arc, - waker: MultiWaker, - shutdown: Arc, -} - -impl CurlWebSocketConnector { - pub(crate) fn channel() -> ( - crossbeam_channel::Sender, - crossbeam_channel::Receiver, - ) { - crossbeam_channel::bounded(SESSION_CAPACITY) - } - - pub(crate) fn new( - submissions: crossbeam_channel::Sender, - waker: MultiWaker, - shutdown: Arc, - ) -> Self { - Self { - inner: Arc::new(ConnectorInner { - submissions, - slots: Arc::new(Semaphore::new(SESSION_CAPACITY)), - waker, - shutdown, - }), - } - } - - pub fn connect(&self, request: CurlWebSocketRequest) -> Result { - if self.inner.shutdown.load(Ordering::Acquire) { - bail!("curl WebSocket runtime is closed"); - } - let slot = self - .inner - .slots - .clone() - .try_acquire_owned() - .context("too many curl WebSocket sessions")?; - let id = next_transfer_id()?; - let control = Arc::new(Control { - cancelled: AtomicBool::new(false), - closed: AtomicBool::new(false), - reading: AtomicBool::new(false), - send: Mutex::new(None), - terminal: Mutex::new(None), - waker: self.inner.waker.clone(), - #[cfg(test)] - read_blocked: tokio::sync::Notify::new(), - #[cfg(test)] - write_blocked: tokio::sync::Notify::new(), - #[cfg(test)] - read_attempts: std::sync::atomic::AtomicUsize::new(0), - #[cfg(test)] - receive_allocations: std::sync::atomic::AtomicUsize::new(0), - #[cfg(test)] - read_waiting: tokio::sync::Notify::new(), - #[cfg(test)] - owner_thread: Mutex::new(None), - }); - let (event_tx, events) = mpsc::channel(MAX_PENDING_EVENTS); - let io = SessionIo { - events: event_tx, - control: control.clone(), - _slot: slot, - }; - self.inner - .submissions - .try_send(Submission { id, request, io }) - .map_err(|_| anyhow::anyhow!("curl WebSocket runtime cannot accept a session"))?; - control.wake(); - Ok(CurlWebSocketConnection { - id, - sender: CurlWebSocketSender { control }, - events, - terminal_delivered: false, - }) - } -} - -/// Standalone owner for callers without an existing HTTP runtime. Uses the -/// same Multi driver as HTTP, with no HTTP submissions. -#[derive(Debug)] -pub struct CurlWebSocketRuntime { - runtime: CurlMultiRuntime, -} - -#[derive(Debug)] -struct StandaloneHandler; -impl curl::easy::Handler for StandaloneHandler {} - -impl CurlWebSocketRuntime { - pub fn new() -> Result { - let (runtime, _) = CurlMultiRuntime::new(CurlMultiRuntimeConfig { - thread_name: "moli-curl-websocket".to_owned(), - poll_interval: Duration::from_secs(1), - ..CurlMultiRuntimeConfig::default() - })?; - Ok(Self { runtime }) - } - - pub fn connector(&self) -> CurlWebSocketConnector { - self.runtime.websocket_connector() - } - - pub fn connect(&self, request: CurlWebSocketRequest) -> Result { - self.connector().connect(request) - } -} diff --git a/moli-curl/src/websocket/connection.rs b/moli-curl/src/websocket/connection.rs new file mode 100644 index 0000000000..7490976f84 --- /dev/null +++ b/moli-curl/src/websocket/connection.rs @@ -0,0 +1,214 @@ +//! The caller's receive/write handles and the owner's I/O endpoint share one +//! lifetime. Pending send completion stays independent of received event capacity. + +use super::{CurlWebSocketEvent, CurlWebSocketSend, MAX_PENDING_EVENTS}; +use crate::CurlTransferId; +use anyhow::{Context, Result, bail}; +use curl::multi::MultiWaker; +use parking_lot::Mutex; +use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, +}; +use tokio::sync::{OwnedSemaphorePermit, mpsc, oneshot}; + +#[derive(Debug)] +pub(super) struct Control { + cancelled: AtomicBool, + closed: AtomicBool, + pub(super) reading: AtomicBool, + pub(super) send: Mutex>, + terminal: Mutex>>, + waker: MultiWaker, + #[cfg(test)] + pub(super) read_blocked: tokio::sync::Notify, + #[cfg(test)] + pub(super) write_blocked: tokio::sync::Notify, + #[cfg(test)] + pub(super) read_attempts: std::sync::atomic::AtomicUsize, + #[cfg(test)] + pub(super) receive_allocations: std::sync::atomic::AtomicUsize, + #[cfg(test)] + pub(super) read_waiting: tokio::sync::Notify, + #[cfg(test)] + pub(super) owner_thread: Mutex>, +} + +impl Control { + fn wake(&self) { + let _ = self.waker.wakeup(); + } + fn cancel(&self) { + self.cancelled.store(true, Ordering::Release); + self.wake(); + } +} + +#[derive(Debug)] +pub(super) struct PendingSend { + pub(super) frame: CurlWebSocketSend, + pub(super) offset: usize, + pub(super) completed: oneshot::Sender, +} + +/// Cloneable write/control capability; it does not keep a dropped receiver alive. +#[derive(Clone, Debug)] +pub struct CurlWebSocketSender { + pub(super) control: Arc, +} + +impl CurlWebSocketSender { + /// Writes one frame and returns its payload size once libcurl consumes it. + /// + /// Only one frame may be pending per connection; overlapping sends return an + /// error. The caller owns data/control ordering and any message queue. + /// Completion is independent of reads and received event delivery. + /// + /// Dropping an unpolled future submits nothing. Once submitted, the frame + /// remains pending until written or the connection is cancelled/closed, even + /// if this future is dropped. Transport failure or cancellation returns an error. + pub async fn send_frame(&self, frame: CurlWebSocketSend) -> Result { + frame.validate()?; + let completion = { + let mut send = self.control.send.lock(); + if self.control.closed.load(Ordering::Acquire) + || self.control.cancelled.load(Ordering::Acquire) + { + bail!("WebSocket transport is closed"); + } + if send.is_some() { + bail!("WebSocket frame is already pending"); + } + let (completed, completion) = oneshot::channel(); + *send = Some(PendingSend { + frame, + offset: 0, + completed, + }); + completion + }; + self.control.wake(); + completion + .await + .context("WebSocket transport closed before frame completion") + } + + /// Handshake delivery starts paused, so application data cannot outrun Open. + /// Pausing does not retract queued chunks or a read already in progress. + /// Send completions progress independently of reads and received events. + pub fn set_reading(&self, enabled: bool) { + self.control.reading.store(enabled, Ordering::Release); + self.control.wake(); + } + + pub fn cancel(&self) { + self.control.cancel(); + } +} + +#[derive(Debug)] +pub struct CurlWebSocketConnection { + id: CurlTransferId, + sender: CurlWebSocketSender, + pub(super) events: mpsc::Receiver, + terminal_delivered: bool, +} + +impl CurlWebSocketConnection { + pub(super) fn channel( + id: CurlTransferId, + waker: MultiWaker, + slot: OwnedSemaphorePermit, + ) -> (Self, SessionIo) { + let control = Arc::new(Control { + cancelled: AtomicBool::new(false), + closed: AtomicBool::new(false), + reading: AtomicBool::new(false), + send: Mutex::new(None), + terminal: Mutex::new(None), + waker, + #[cfg(test)] + read_blocked: tokio::sync::Notify::new(), + #[cfg(test)] + write_blocked: tokio::sync::Notify::new(), + #[cfg(test)] + read_attempts: std::sync::atomic::AtomicUsize::new(0), + #[cfg(test)] + receive_allocations: std::sync::atomic::AtomicUsize::new(0), + #[cfg(test)] + read_waiting: tokio::sync::Notify::new(), + #[cfg(test)] + owner_thread: Mutex::new(None), + }); + let (event_tx, events) = mpsc::channel(MAX_PENDING_EVENTS); + let io = SessionIo { + events: event_tx, + control: control.clone(), + _slot: slot, + }; + ( + Self { + id, + sender: CurlWebSocketSender { control }, + events, + terminal_delivered: false, + }, + io, + ) + } + + pub fn id(&self) -> CurlTransferId { + self.id + } + pub fn sender(&self) -> CurlWebSocketSender { + self.sender.clone() + } + + pub async fn recv(&mut self) -> Option { + if let Some(event) = self.events.recv().await { + self.sender.control.wake(); + return Some(event); + } + if self.terminal_delivered { + return None; + } + self.terminal_delivered = true; + let result = self + .sender + .control + .terminal + .lock() + .take() + .unwrap_or_else(|| Err("WebSocket runtime stopped".to_owned())); + Some(CurlWebSocketEvent::Closed { result }) + } +} + +impl Drop for CurlWebSocketConnection { + fn drop(&mut self) { + self.sender.cancel(); + } +} + +pub(super) struct SessionIo { + pub(super) events: mpsc::Sender, + pub(super) control: Arc, + _slot: OwnedSemaphorePermit, +} + +impl SessionIo { + pub(super) fn cancelled(&self) -> bool { + self.control.cancelled.load(Ordering::Acquire) + } + pub(super) fn finish(&self, result: std::result::Result<(), String>) { + *self.control.terminal.lock() = Some(result); + } +} + +impl Drop for SessionIo { + fn drop(&mut self) { + self.control.closed.store(true, Ordering::Release); + // Settle the pending write before closing the receive event channel. + self.control.send.lock().take(); + } +} diff --git a/moli-curl/src/websocket/connector.rs b/moli-curl/src/websocket/connector.rs new file mode 100644 index 0000000000..15ebc6b6d2 --- /dev/null +++ b/moli-curl/src/websocket/connector.rs @@ -0,0 +1,78 @@ +//! Submission capability for a specific owner, without thread ownership. + +use super::{ + CurlWebSocketConnection, CurlWebSocketRequest, SESSION_CAPACITY, connection::SessionIo, +}; +use crate::{CurlTransferId, runtime::identity::next_transfer_id}; +use anyhow::{Context, Result, bail}; +use curl::multi::MultiWaker; +use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, +}; +use tokio::sync::Semaphore; + +pub(crate) struct Submission { + pub(super) id: CurlTransferId, + pub(super) request: CurlWebSocketRequest, + pub(super) io: SessionIo, +} + +/// Admission capability for one native owner. Clones do not keep that owner +/// alive; shutting down its runtime closes connections and rejects new ones. +#[derive(Clone, Debug)] +pub struct CurlWebSocketConnector { + inner: Arc, +} + +#[derive(Debug)] +struct ConnectorInner { + submissions: crossbeam_channel::Sender, + slots: Arc, + waker: MultiWaker, + shutdown: Arc, +} + +impl CurlWebSocketConnector { + pub(crate) fn channel() -> ( + crossbeam_channel::Sender, + crossbeam_channel::Receiver, + ) { + crossbeam_channel::bounded(SESSION_CAPACITY) + } + + pub(crate) fn new( + submissions: crossbeam_channel::Sender, + waker: MultiWaker, + shutdown: Arc, + ) -> Self { + Self { + inner: Arc::new(ConnectorInner { + submissions, + slots: Arc::new(Semaphore::new(SESSION_CAPACITY)), + waker, + shutdown, + }), + } + } + + pub fn connect(&self, request: CurlWebSocketRequest) -> Result { + if self.inner.shutdown.load(Ordering::Acquire) { + bail!("curl WebSocket runtime is closed"); + } + let slot = self + .inner + .slots + .clone() + .try_acquire_owned() + .context("too many curl WebSocket sessions")?; + let id = next_transfer_id()?; + let (connection, io) = CurlWebSocketConnection::channel(id, self.inner.waker.clone(), slot); + self.inner + .submissions + .try_send(Submission { id, request, io }) + .map_err(|_| anyhow::anyhow!("curl WebSocket runtime cannot accept a session"))?; + let _ = self.inner.waker.wakeup(); + Ok(connection) + } +} diff --git a/moli-curl/src/websocket/scheduling.rs b/moli-curl/src/websocket/readiness.rs similarity index 73% rename from moli-curl/src/websocket/scheduling.rs rename to moli-curl/src/websocket/readiness.rs index 9ae4f008c1..404feca761 100644 --- a/moli-curl/src/websocket/scheduling.rs +++ b/moli-curl/src/websocket/readiness.rs @@ -2,11 +2,11 @@ //! work/capacity; WaitingForSocket I/O has already returned AGAIN. A successful //! operation stays Runnable so libcurl's buffered data can be drained as well. -use std::{collections::HashMap, time::Duration}; +use std::collections::HashMap; -use curl::multi::{Multi, WaitFd}; +use curl::multi::WaitFd; -use super::{diagnostics::Diagnostics, session::Session}; +use super::session::Session; use crate::CurlTransferId; #[derive(Default, PartialEq, Eq)] @@ -47,22 +47,15 @@ impl IoState { } /// Keeps each extra fd paired with its session across curl_multi_poll. -/// libcurl also polls opening handshakes and the cross-thread waker here. +/// The runtime polls these alongside HTTP/handshake sockets and curl's waker. #[derive(Default)] -pub(super) struct SocketPoll { +pub(super) struct SocketReadiness { ids: Vec, fds: Vec, } -impl SocketPoll { - pub(super) fn wait( - &mut self, - multi: &Multi, - sessions: &mut HashMap, - timeout: Duration, - progressed: bool, - diagnostics: &mut Diagnostics, - ) -> Result<(), curl::MultiError> { +impl SocketReadiness { + pub(super) fn prepare(&mut self, sessions: &HashMap) -> &mut [WaitFd] { self.ids.clear(); self.fds.clear(); for (id, session) in sessions.iter() { @@ -71,10 +64,10 @@ impl SocketPoll { self.fds.push(fd); } } - let started = diagnostics.poll_start(); - let result = multi.poll(&mut self.fds, timeout); - diagnostics.polled(started, timeout, progressed); - result?; + &mut self.fds + } + + pub(super) fn dispatch(&self, sessions: &mut HashMap) { for (id, fd) in self.ids.iter().zip(&self.fds) { if fd.received_read() || fd.received_write() { // A socket event permits a retry, not guaranteed progress. @@ -86,6 +79,5 @@ impl SocketPoll { .socket_ready(); } } - Ok(()) } } diff --git a/moli-curl/src/websocket/registry.rs b/moli-curl/src/websocket/registry.rs index cfd54cb61d..5fe6a68610 100644 --- a/moli-curl/src/websocket/registry.rs +++ b/moli-curl/src/websocket/registry.rs @@ -1,23 +1,24 @@ //! Persistent WebSocket residences on the common HTTP/WebSocket Multi. //! The runtime owns perform, completion dispatch and the single wait loop. -use std::{ - collections::HashMap, - rc::Rc, - time::{Duration, Instant}, -}; +use std::{collections::HashMap, rc::Rc, time::Instant}; -use curl::{easy::Easy2, multi::Multi}; +use curl::{ + easy::Easy2, + multi::{Multi, WaitFd}, +}; use super::{ SessionIo, Submission, connection_pool::ConnectionPool, - diagnostics::Diagnostics, + readiness::SocketReadiness, request::{self, Handshake}, - scheduling::SocketPoll, session::{Session, Step}, }; -use crate::{CurlDnsResolution, CurlTransferId, dns_adapter::CurlDnsOwnerResidence}; +use crate::{ + CurlDnsResolution, CurlTransferId, dns_adapter::CurlDnsOwnerResidence, + runtime::diagnostics::Diagnostics, +}; struct Pending { id: CurlTransferId, @@ -31,9 +32,8 @@ pub(crate) struct WebSocketRegistry { submissions: crossbeam_channel::Receiver, sessions: HashMap, dns: CurlDnsOwnerResidence, - poll: SocketPoll, + readiness: SocketReadiness, receive: Vec, - diagnostics: Diagnostics, pool: Option>, closed: bool, } @@ -44,9 +44,8 @@ impl WebSocketRegistry { submissions, sessions: HashMap::new(), dns: CurlDnsOwnerResidence::default(), - poll: SocketPoll::default(), + readiness: SocketReadiness::default(), receive: Vec::new(), - diagnostics: Diagnostics::from_env(), pool: None, closed: false, } @@ -76,7 +75,6 @@ impl WebSocketRegistry { .io .finish(Err("curl WebSocket runtime shut down".to_owned())); } - self.diagnostics.report(0, true); } pub(crate) fn admit(&mut self, submission: Submission, multi: &mut Multi) { @@ -188,7 +186,7 @@ impl WebSocketRegistry { self.sessions.get(&id)?.handshake_result(message) } - pub(crate) fn advance(&mut self, multi: &mut Multi) -> bool { + pub(crate) fn advance(&mut self, multi: &mut Multi, diagnostics: &mut Diagnostics) -> bool { for _ in 0..super::SESSION_CAPACITY { let Ok(submission) = self.submissions.try_recv() else { break; @@ -199,7 +197,7 @@ impl WebSocketRegistry { let mut retired = Vec::new(); let mut progressed = false; for (id, session) in &mut self.sessions { - match session.advance(&mut self.receive, &mut self.diagnostics) { + match session.advance(&mut self.receive, diagnostics) { Ok(Step::Progress) => progressed = true, Ok(Step::Idle) => {} terminal => retired.push((*id, terminal)), @@ -208,10 +206,6 @@ impl WebSocketRegistry { for (id, step) in retired { self.apply_step(id, step, multi); } - if let Some(counters) = self.diagnostics.counters() { - counters.turns += 1; - counters.progressed_turns += u64::from(progressed); - } progressed } @@ -234,17 +228,16 @@ impl WebSocketRegistry { .min() } - pub(crate) fn wait(&mut self, multi: &mut Multi, timeout: Duration, progressed: bool) { - if let Err(error) = self.poll.wait( - multi, - &mut self.sessions, - timeout, - progressed, - &mut self.diagnostics, - ) { - self.fail_sessions(multi, &error.to_string()); - } - self.diagnostics.report(self.sessions.len(), false); + pub(crate) fn session_count(&self) -> usize { + self.sessions.len() + } + + pub(crate) fn poll_fds(&mut self) -> &mut [WaitFd] { + self.readiness.prepare(&self.sessions) + } + + pub(crate) fn apply_readiness(&mut self) { + self.readiness.dispatch(&mut self.sessions); } pub(crate) fn fail_sessions(&mut self, multi: &mut Multi, error: &str) { diff --git a/moli-curl/src/websocket/session.rs b/moli-curl/src/websocket/session.rs index 184a488416..c8cc20c594 100644 --- a/moli-curl/src/websocket/session.rs +++ b/moli-curl/src/websocket/session.rs @@ -8,9 +8,9 @@ use curl::{ multi::{Easy2Handle, Multi, WaitFd}, }; -use super::diagnostics::Diagnostics; -use super::{CurlWebSocketEvent, SessionIo, WsFlags, request::Handshake, scheduling::IoState}; +use super::{CurlWebSocketEvent, SessionIo, WsFlags, readiness::IoState, request::Handshake}; use crate::CurlTransferId; +use crate::runtime::diagnostics::Diagnostics; const CHUNK_BYTES: usize = 16 * 1024; const IO_BUDGET: usize = 8; diff --git a/moli-curl/src/websocket/standalone.rs b/moli-curl/src/websocket/standalone.rs new file mode 100644 index 0000000000..a5234d529c --- /dev/null +++ b/moli-curl/src/websocket/standalone.rs @@ -0,0 +1,36 @@ +//! Convenience owner for callers without a shared network runtime. + +use super::{CurlWebSocketConnection, CurlWebSocketConnector, CurlWebSocketRequest}; +use crate::{CurlMultiRuntime, CurlMultiRuntimeConfig}; +use anyhow::Result; +use std::time::Duration; + +/// Standalone owner for callers without an existing HTTP runtime. Uses the +/// same Multi driver as HTTP, with no HTTP submissions. +#[derive(Debug)] +pub struct CurlWebSocketRuntime { + runtime: CurlMultiRuntime, +} + +#[derive(Debug)] +struct StandaloneHandler; +impl curl::easy::Handler for StandaloneHandler {} + +impl CurlWebSocketRuntime { + pub fn new() -> Result { + let (runtime, _) = CurlMultiRuntime::new(CurlMultiRuntimeConfig { + thread_name: "moli-curl-websocket".to_owned(), + poll_interval: Duration::from_secs(1), + ..CurlMultiRuntimeConfig::default() + })?; + Ok(Self { runtime }) + } + + pub fn connector(&self) -> CurlWebSocketConnector { + self.runtime.websocket_connector() + } + + pub fn connect(&self, request: CurlWebSocketRequest) -> Result { + self.connector().connect(request) + } +} diff --git a/moli-curl/src/websocket/tests.rs b/moli-curl/src/websocket/tests.rs index d7ea99783f..7f0e41a8ac 100644 --- a/moli-curl/src/websocket/tests.rs +++ b/moli-curl/src/websocket/tests.rs @@ -1,10 +1,11 @@ use std::{ io::{Read, Write}, net::{TcpListener, TcpStream}, + sync::{Arc, atomic::Ordering}, thread, }; -use tokio::time::timeout; +use tokio::{sync::oneshot, time::timeout}; use tokio_tungstenite::tungstenite::{self, Message, handshake::derive_accept_key}; use super::*; @@ -324,7 +325,7 @@ async fn native_full_delivery_queue_does_not_block_other_sessions_or_cancellatio let mut blocked = runtime.connect(CurlWebSocketRequest::new(url)).unwrap(); opened(&mut blocked).await; blocked.sender().set_reading(true); - timeout(DEADLINE, blocked.sender.control.read_blocked.notified()) + timeout(DEADLINE, blocked.sender().control.read_blocked.notified()) .await .unwrap(); assert_eq!(blocked.events.len(), MAX_PENDING_EVENTS); @@ -359,9 +360,12 @@ async fn native_queue_capacity_restoration_wakes_owner() { let mut connection = runtime.connect(CurlWebSocketRequest::new(url)).unwrap(); opened(&mut connection).await; connection.sender().set_reading(true); - timeout(DEADLINE, connection.sender.control.read_blocked.notified()) - .await - .unwrap(); + timeout( + DEADLINE, + connection.sender().control.read_blocked.notified(), + ) + .await + .unwrap(); for expected in 0..32 { assert!( matches!(event(&mut connection).await, CurlWebSocketEvent::Chunk { data, .. } if data == [expected]) diff --git a/moli-curl/src/websocket/tests/readiness.rs b/moli-curl/src/websocket/tests/readiness.rs index 0d255468ae..5972fae106 100644 --- a/moli-curl/src/websocket/tests/readiness.rs +++ b/moli-curl/src/websocket/tests/readiness.rs @@ -84,7 +84,7 @@ async fn native_socket_readiness_is_serviced_during_continuous_traffic() { .unwrap(); opened(&mut quiet).await; quiet.sender().set_reading(true); - timeout(DEADLINE, quiet.sender.control.read_waiting.notified()) + timeout(DEADLINE, quiet.sender().control.read_waiting.notified()) .await .unwrap(); let mut active = runtime.connect(CurlWebSocketRequest::new(url)).unwrap(); @@ -181,9 +181,12 @@ async fn native_peer_eof_wakes_waiting_reader() { let mut connection = runtime.connect(CurlWebSocketRequest::new(url)).unwrap(); opened(&mut connection).await; connection.sender().set_reading(true); - timeout(DEADLINE, connection.sender.control.read_waiting.notified()) - .await - .unwrap(); + timeout( + DEADLINE, + connection.sender().control.read_waiting.notified(), + ) + .await + .unwrap(); finish_tx.send(()).unwrap(); assert!(matches!( event(&mut connection).await, diff --git a/moli-curl/src/websocket/tests/shared.rs b/moli-curl/src/websocket/tests/shared.rs index ae6de810eb..8f55dfe3bf 100644 --- a/moli-curl/src/websocket/tests/shared.rs +++ b/moli-curl/src/websocket/tests/shared.rs @@ -3,7 +3,7 @@ use std::num::NonZeroUsize; use curl::easy::{Easy2, Handler, WriteError}; use super::*; -use crate::{CurlMultiJob, CurlOriginKey}; +use crate::{CurlMultiJob, CurlMultiRuntime, CurlMultiRuntimeConfig, CurlOriginKey}; #[derive(Default, Debug)] struct HttpCapture { @@ -149,7 +149,7 @@ async fn shared_owner_keeps_http_and_websocket_pools_independent() { assert_eq!(easy.get_ref().body, b"ok"); for mut connection in connections { assert_eq!( - *connection.sender.control.owner_thread.lock(), + *connection.sender().control.owner_thread.lock(), easy.get_ref().owner ); connection.sender().set_reading(true); diff --git a/moli-curl/src/websocket/tests/storage.rs b/moli-curl/src/websocket/tests/storage.rs index 3e8458c811..36f388736e 100644 --- a/moli-curl/src/websocket/tests/storage.rs +++ b/moli-curl/src/websocket/tests/storage.rs @@ -18,9 +18,12 @@ async fn native_receive_storage_is_shared_across_idle_probes_and_transferred_to_ let mut connection = runtime.connect(CurlWebSocketRequest::new(url)).unwrap(); opened(&mut connection).await; connection.sender().set_reading(true); - timeout(DEADLINE, connection.sender.control.read_waiting.notified()) - .await - .unwrap(); + timeout( + DEADLINE, + connection.sender().control.read_waiting.notified(), + ) + .await + .unwrap(); connections.push(connection); peers.push(peer); writers.push(write); @@ -30,7 +33,7 @@ async fn native_receive_storage_is_shared_across_idle_probes_and_transferred_to_ .iter() .map(|connection| { connection - .sender + .sender() .control .receive_allocations .load(Ordering::Acquire) @@ -53,7 +56,7 @@ async fn native_receive_storage_is_shared_across_idle_probes_and_transferred_to_ retained.push(data); timeout( DEADLINE, - connections[index].sender.control.read_waiting.notified(), + connections[index].sender().control.read_waiting.notified(), ) .await .unwrap();