From a7015fef4867db6a2d70599474850fd0dc19a7fc Mon Sep 17 00:00:00 2001 From: Perry Bot Date: Fri, 25 Sep 2026 00:49:17 +0000 Subject: [PATCH 1/2] http: createConnection and the Agent facade expiry on turnloop; perry-ext-http drops tokio The last two tokio users in perry-ext-http move onto the agent's loop: - the agent.createConnection / createSocket (and request-level createConnection) exchange runs in client_turnloop::raw_socket. Instead of a tokio task polling the raw-net vtable every 1 ms, perry-ext-net calls a new perry_ffi::raw_net_notify when a raw-mode socket gains bytes or goes terminal, and the client drains it on a 0 ms loop timer; - the keep-alive Agent's 40 ms socket-facade idle expiry, and req.setTimeout's early 'timeout', are loop deadlines (push_after). perry-ext-http no longer depends on tokio: cargo tree -p perry-ext-http -i tokio -e normal,dev finds no tokio. tokio inventory: 5 -> 4 edges. --- Cargo.lock | 1 - crates/perry-ext-http/Cargo.toml | 1 - crates/perry-ext-http/src/agent.rs | 20 +- .../src/client_connect_override.rs | 130 +------- .../src/client_turnloop/conn.rs | 11 +- .../perry-ext-http/src/client_turnloop/mod.rs | 127 ++++++- .../src/client_turnloop/raw_socket.rs | 309 ++++++++++++++++++ crates/perry-ext-http/src/lib.rs | 1 - .../tests/turnloop_client_exchange.rs | 89 +++++ crates/perry-ext-net/src/raw_bridge.rs | 6 +- crates/perry-ffi/src/lib.rs | 4 +- crates/perry-ffi/src/raw_net.rs | 39 +++ scripts/tokio_inventory.json | 15 +- 13 files changed, 586 insertions(+), 167 deletions(-) create mode 100644 crates/perry-ext-http/src/client_turnloop/raw_socket.rs diff --git a/Cargo.lock b/Cargo.lock index f238b2acd4..6f862ec23c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5450,7 +5450,6 @@ dependencies = [ "rustls-pemfile", "rustls-webpki", "serde_json", - "tokio", "turnloop-http", "url", "webpki-roots 1.0.9", diff --git a/crates/perry-ext-http/Cargo.toml b/crates/perry-ext-http/Cargo.toml index dc52a7c0e5..021e851d56 100644 --- a/crates/perry-ext-http/Cargo.toml +++ b/crates/perry-ext-http/Cargo.toml @@ -24,7 +24,6 @@ http = "1" rustls = { workspace = true, features = ["std", "ring", "tls12"] } rustls_webpki = { package = "rustls-webpki", version = "0.103" } rustls-pemfile.workspace = true -tokio = { workspace = true } # The client transport (`client_turnloop`): the rustls session that runs # `https:` above a turnloop socket, and the URL type requests are parsed into. perry-tls-session.workspace = true diff --git a/crates/perry-ext-http/src/agent.rs b/crates/perry-ext-http/src/agent.rs index 87d248d7ca..c4d9199479 100644 --- a/crates/perry-ext-http/src/agent.rs +++ b/crates/perry-ext-http/src/agent.rs @@ -761,20 +761,20 @@ fn release_request_inner( ); } let key = key.to_string(); - perry_ffi::spawn_async(async move { - // The transport owns the physical pooled connection, so the public - // net.Socket facade cannot receive its idle read/EOF edge. - // Conservatively retire an unclaimed facade after the I/O - // guard window; immediate/next-tick reuse cancels this via the - // generation check below. - tokio::time::sleep(std::time::Duration::from_millis(40)).await; - crate::push_event(crate::PendingHttpEvent::AgentIdleExpire { + // The transport owns the physical pooled connection, so the public + // net.Socket facade cannot receive its idle read/EOF edge. + // Conservatively retire an unclaimed facade after the I/O guard + // window; immediate/next-tick reuse cancels this via the generation + // check below. A deadline on the loop, not a tokio sleep. + crate::client_turnloop::push_after( + 40, + crate::PendingHttpEvent::AgentIdleExpire { agent_handle: handle, key, socket, generation, - }); - }); + }, + ); } None } diff --git a/crates/perry-ext-http/src/client_connect_override.rs b/crates/perry-ext-http/src/client_connect_override.rs index 55cf067bb4..518892c481 100644 --- a/crates/perry-ext-http/src/client_connect_override.rs +++ b/crates/perry-ext-http/src/client_connect_override.rs @@ -8,10 +8,10 @@ use std::collections::HashMap; -use perry_ffi::{spawn_blocking_with_reactor as spawn_blocking, Handle}; +use perry_ffi::Handle; use super::agent; -use crate::{parse_http_response, push_event, ClientInflightGuard, PendingHttpEvent}; +use crate::{push_event, PendingHttpEvent}; /// Look up `request_handle`'s own `createConnection` (if any) and, when /// set, dispatch over it. `None` means "not set / not usable" — the @@ -96,12 +96,13 @@ fn serialize_http_request( /// `Connection: close` and read to EOF. A `101` response to an upgrade request /// detaches the still-live socket from the raw reader and pushes `Upgrade` with /// any bytes following the header block. Other responses are parsed with -/// [`parse_http_response`] and produce the same `Response` / `Error` events as +/// `plain_client::parse_http_response` and produce the same `Response` / `Error` events as /// the default transport. /// /// The socket I/O goes through perry-ffi's raw-net vtable (published by -/// perry-ext-net), so this crate needs no link edge to perry-ext-net. If no -/// net backend is linked the request errors out (the override couldn't have +/// perry-ext-net) and runs on the agent's event loop, woken by +/// `perry_ffi::raw_net_notify` (`client_turnloop::raw_socket`). If no net +/// backend is linked the request errors out (the override couldn't have /// produced a socket without `net`, so this is a defensive guard). pub(crate) fn dispatch_request_over_socket( request_handle: Handle, @@ -137,118 +138,13 @@ pub(crate) fn dispatch_request_over_socket( } let req_bytes = serialize_http_request(&method, &path, &host_header, &headers, &body); let wants_upgrade = crate::client_upgrade::wants_upgrade(&headers); - let deadline = std::time::Duration::from_millis(timeout_ms.unwrap_or(30_000)); - - spawn_blocking(move || { - let try_h = tokio::runtime::Handle::try_current(); - std::hint::black_box(&try_h); - if try_h.is_err() { - push_event(PendingHttpEvent::Error { - request_handle, - error_message: "http client runtime unavailable".to_string(), - }); - return; - } - let handle = tokio::runtime::Handle::current(); - // #5779 follow-up: keep this fetch counted in-flight for its whole - // lifetime so the idle-kick recovers a lost worker-unpark. - let inflight_guard = ClientInflightGuard::new(request_handle); - let jh = handle.spawn(async move { - let _inflight = inflight_guard; - let vtable = match perry_ffi::raw_net() { - Some(v) => v, - None => { - push_event(PendingHttpEvent::Error { - request_handle, - error_message: "agent.createConnection requires node:net (not linked)" - .to_string(), - }); - return; - } - }; - // Attach is idempotent — the request path also attaches on the - // main thread before this task runs, to close any data race. - (vtable.attach)(socket_id); - if (vtable.write)(socket_id, req_bytes.as_ptr(), req_bytes.len()) == 0 { - push_event(PendingHttpEvent::Error { - request_handle, - error_message: "failed to write request to agent socket".to_string(), - }); - return; - } - - let mut raw = Vec::new(); - let mut chunk = [0u8; 16 * 1024]; - let start = tokio::time::Instant::now(); - loop { - let n = (vtable.poll_read)(socket_id, chunk.as_mut_ptr(), chunk.len()); - if n > 0 { - raw.extend_from_slice(&chunk[..n as usize]); - if wants_upgrade { - if let Some(header_end) = raw.windows(4).position(|w| w == b"\r\n\r\n") { - let header_end = header_end + 4; - let status = std::str::from_utf8(&raw[..header_end]) - .ok() - .and_then(|head| head.lines().next()) - .and_then(|line| line.split_whitespace().nth(1)) - .and_then(|code| code.parse::().ok()); - if status == Some(101) { - match parse_http_response(&raw[..header_end]) { - Ok(parsed) => { - (vtable.detach)(socket_id); - push_event(PendingHttpEvent::Upgrade { - request_handle, - status: parsed.status, - status_message: parsed.status_message, - headers: parsed.headers, - socket_handle: socket_id, - head: raw[header_end..].to_vec(), - }); - } - Err(error_message) => { - (vtable.close)(socket_id); - push_event(PendingHttpEvent::Error { - request_handle, - error_message, - }); - } - } - return; - } - } - } - } else if n == 0 { - break; // clean EOF — peer closed after the response - } else { - if start.elapsed() >= deadline { - (vtable.close)(socket_id); - push_event(PendingHttpEvent::Timeout { request_handle }); - return; - } - tokio::time::sleep(std::time::Duration::from_millis(1)).await; - } - } - (vtable.close)(socket_id); - - match parse_http_response(&raw) { - Ok(parsed) => push_event(PendingHttpEvent::Response { - request_handle, - status: parsed.status, - status_message: parsed.status_message, - headers: parsed.headers, - trailers: parsed.trailers, - body: parsed.body, - http_version: parsed.http_version, - }), - Err(error_message) => push_event(PendingHttpEvent::Error { - request_handle, - error_message, - }), - } - }); - std::hint::black_box(&jh); - std::mem::forget(jh); - }); + crate::client_turnloop::raw_socket::start_raw_exchange( + request_handle, + req_bytes, + wants_upgrade, + timeout_ms, + socket_id, + ); } #[cfg(test)] diff --git a/crates/perry-ext-http/src/client_turnloop/conn.rs b/crates/perry-ext-http/src/client_turnloop/conn.rs index 5e3a22942c..31f1a4955f 100644 --- a/crates/perry-ext-http/src/client_turnloop/conn.rs +++ b/crates/perry-ext-http/src/client_turnloop/conn.rs @@ -1098,11 +1098,14 @@ pub(super) fn on_timer(st: &mut State, timer: i64) -> Vec { close(st, conn, &mut fx); } } - Timer::Standalone { request } => { - fx.push(Effect::Push(PendingHttpEvent::Timeout { - request_handle: request, - })); + Timer::Deferred(event) => { + super::DEFERRED_FIRED.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + fx.push(Effect::Push(event)); } + Timer::RawDrain { socket } => { + fx.extend(super::raw_socket::on_raw_drain_timer(st, socket, timer)) + } + Timer::RawDeadline { socket } => fx.extend(super::raw_socket::on_raw_deadline(st, socket)), } fx } diff --git a/crates/perry-ext-http/src/client_turnloop/mod.rs b/crates/perry-ext-http/src/client_turnloop/mod.rs index a5bdcd732c..403a4e6e21 100644 --- a/crates/perry-ext-http/src/client_turnloop/mod.rs +++ b/crates/perry-ext-http/src/client_turnloop/mod.rs @@ -54,6 +54,7 @@ mod conn; mod pool; mod proxy; +pub(crate) mod raw_socket; pub(crate) mod tls; mod wire; @@ -96,6 +97,8 @@ static COMPLETED: AtomicU64 = AtomicU64::new(0); static REUSED: AtomicU64 = AtomicU64::new(0); static HANDSHAKES: AtomicU64 = AtomicU64::new(0); static TIMED_OUT: AtomicU64 = AtomicU64::new(0); +static RAW_COMPLETED: AtomicU64 = AtomicU64::new(0); +static DEFERRED_FIRED: AtomicU64 = AtomicU64::new(0); /// Exchanges handed to this module (directly or posted to the loop owner). pub fn accepted_total() -> u64 { @@ -122,6 +125,17 @@ pub fn timed_out_total() -> u64 { TIMED_OUT.load(Ordering::Relaxed) } +/// `createConnection` / `createSocket` exchanges ([`raw_socket`]) that +/// delivered a response or an upgrade. +pub fn raw_completed_total() -> u64 { + RAW_COMPLETED.load(Ordering::Relaxed) +} + +/// Deferred events ([`push_after`]) whose loop deadline fired. +pub fn deferred_fired_total() -> u64 { + DEFERRED_FIRED.load(Ordering::Relaxed) +} + // ── The request ───────────────────────────────────────────────────────────── /// Which of the four exchange shapes a request is. @@ -162,14 +176,18 @@ pub(crate) struct Outbound { // ── Shared state ──────────────────────────────────────────────────────────── /// A deadline this module armed, and what it is for. -#[derive(Clone, Copy, Debug)] enum Timer { /// `options.timeout` for an in-flight exchange. Deadline { conn: i64, request: Handle }, /// An idle pooled connection's expiry. Idle { conn: i64 }, - /// `req.setTimeout(ms, cb)` armed before (or independently of) dispatch. - Standalone { request: Handle }, + /// An event queued for later: `req.setTimeout(ms, cb)` armed before (or + /// independently of) dispatch, or an Agent socket facade's idle expiry. + Deferred(PendingHttpEvent), + /// Read a `createConnection` socket that `perry-ext-net` said is ready. + RawDrain { socket: i64 }, + /// A `createConnection` exchange's deadline. + RawDeadline { socket: i64 }, } #[derive(Default)] @@ -180,6 +198,8 @@ struct State { by_request: HashMap, timers: HashMap, idle: HashMap>, + /// `createConnection` exchanges, keyed by the user's socket id. + raw: HashMap, } fn state() -> &'static Mutex { @@ -209,6 +229,10 @@ enum Effect { Push(PendingHttpEvent), /// Hand a connection to `net` after a `101`, then publish the event. Handoff(i64, PendingHttpEvent), + /// Read a `createConnection` socket ([`raw_socket::drain_raw_socket`]). + RawDrain(i64), + /// Close a `createConnection` socket through the raw-net vtable. + RawClose(i64), /// Run a request again on a fresh connection (a reused one died before /// the response began), keeping its in-flight guard. Redispatch(Box, crate::ClientInflightGuard), @@ -273,6 +297,14 @@ fn run(effects: Vec) { push_event(event); Vec::new() } + Effect::RawDrain(socket) => { + raw_socket::drain_raw_socket(socket); + Vec::new() + } + Effect::RawClose(socket) => { + raw_socket::close_raw_socket(socket); + Vec::new() + } Effect::Handoff(id, event) => { handoff(id, event); Vec::new() @@ -353,10 +385,20 @@ impl perry_ffi::agent_post::AgentJob for LoopJob { /// How often a transiently refused post is retried before the request fails. const POST_ATTEMPTS: usize = 64; +thread_local! { + /// This thread has already been found to own the agent's loop by + /// [`on_loop`]. Asking [`available`] *claims* the route for the first + /// thread that asks, so [`push_after`] consults this instead of asking: + /// scheduling a deferred event must never be what decides which thread + /// owns the loop. + static OWNS_LOOP_HERE: std::cell::Cell = const { std::cell::Cell::new(false) }; +} + /// Run `op` on the loop: here when this thread owns it, else on the owner. /// `false` means no loop exists for this agent at all; `op` was not run. fn on_loop(op: impl FnOnce() + Send + 'static) -> bool { if available() { + OWNS_LOOP_HERE.with(|owns| owns.set(true)); op(); return true; } @@ -611,29 +653,51 @@ pub fn purge_agent(agent_handle: Handle) { /// `req.setTimeout(ms[, cb])` / `options.timeout` armed at request creation: /// a one-shot `'timeout'` for the request, independent of any exchange. pub(crate) fn arm_request_timeout(request_handle: Handle, ms: u64) { + push_after(ms, PendingHttpEvent::Timeout { request_handle }); +} + +/// Queue `event` for the drain `ms` milliseconds from now, on a loop deadline. +/// +/// The deadline is unreferenced (`tl::timer_arm`): like the tokio sleeps it +/// replaces for `'timeout'` and the Agent facade's idle expiry, it never keeps +/// the process alive on its own. +pub(crate) fn push_after(ms: u64, event: PendingHttpEvent) { + if !OWNS_LOOP_HERE.with(std::cell::Cell::get) { + // Not (yet) known to be the loop owner — a `'timeout'` armed before + // this thread's first request, or bookkeeping on a thread that never + // carried one. A plain thread keeps the promise that the event fires + // without claiming the loop route as a side effect. + std::thread::spawn(move || { + std::thread::sleep(std::time::Duration::from_millis(ms)); + push_event(event); + }); + return; + } + // `on_loop` drops its closure unrun when it refuses, so the event is held + // in a shared slot the "no loop at all" fallback below can take back. + let slot = std::sync::Arc::new(Mutex::new(Some(event))); + let posted = slot.clone(); let carried = on_loop(move || { + let Some(event) = posted.lock().unwrap_or_else(|e| e.into_inner()).take() else { + return; + }; let id = next_id(); if id == perry_ffi::INVALID_HANDLE { - push_event(PendingHttpEvent::Timeout { request_handle }); + push_event(event); return; } - with_state(|st| { - st.timers.insert( - id, - Timer::Standalone { - request: request_handle, - }, - ) - }); + with_state(|st| st.timers.insert(id, Timer::Deferred(event))); run(vec![Effect::ArmTimer(id, ms)]); }); if !carried { // No loop anywhere for this agent: a plain thread keeps the promise - // that `'timeout'` fires, without a second event loop. - std::thread::spawn(move || { - std::thread::sleep(std::time::Duration::from_millis(ms)); - push_event(PendingHttpEvent::Timeout { request_handle }); - }); + // that the event fires, without a second event loop. + if let Some(event) = slot.lock().unwrap_or_else(|e| e.into_inner()).take() { + std::thread::spawn(move || { + std::thread::sleep(std::time::Duration::from_millis(ms)); + push_event(event); + }); + } } } @@ -729,6 +793,35 @@ pub fn try_dispatch_pooled( on_loop(move || start(outbound)) } +/// Run an HTTP exchange over an already-open `perry-ext-net` socket, as +/// `agent.createConnection` does. For `tests/turnloop_client_exchange.rs`. +#[allow(clippy::too_many_arguments)] +pub fn try_dispatch_over_socket( + request_handle: Handle, + method: &str, + url: &str, + headers: &HashMap, + body: &[u8], + timeout_ms: Option, + socket_id: i64, +) { + crate::client_connect_override::dispatch_request_over_socket( + request_handle, + method.to_string(), + url.to_string(), + headers.clone(), + body.to_vec(), + timeout_ms, + socket_id, + ); +} + +/// Queue a `'timeout'` for `request_handle` in `ms` ([`push_after`]). For +/// `tests/turnloop_client_exchange.rs`. +pub fn schedule_timeout_for_test(request_handle: Handle, ms: u64) { + arm_request_timeout(request_handle, ms); +} + // ── The completion sink ───────────────────────────────────────────────────── extern "C" fn sink(completion: *const tl::NetCompletion) { diff --git a/crates/perry-ext-http/src/client_turnloop/raw_socket.rs b/crates/perry-ext-http/src/client_turnloop/raw_socket.rs new file mode 100644 index 0000000000..2e4441d9e6 --- /dev/null +++ b/crates/perry-ext-http/src/client_turnloop/raw_socket.rs @@ -0,0 +1,309 @@ +//! An HTTP exchange over a socket JS produced: `agent.createConnection` / +//! `createSocket` (#2154) and the request-level `createConnection` (#10469). +//! +//! The socket belongs to `perry-ext-net`, which already runs it on the agent's +//! turnloop loop (#11105) and, for `tls.connect`, decrypts above it. This +//! module never touches the handle directly: it speaks to the socket through +//! perry-ffi's raw-net vtable (`attach` / `write` / `poll_read` / `detach` / +//! `close`), so the user's socket keeps its own connect/TLS state and JS +//! identity, exactly as before. +//! +//! What changed is how the bytes are *waited for*. This used to be a tokio task +//! that called `poll_read` in a loop with a 1 ms `tokio::time::sleep` between +//! empty reads. Now `perry-ext-net` calls `perry_ffi::raw_net_notify` when a +//! raw-mode socket gains bytes or goes terminal, and [`raw_socket_ready`] schedules a +//! drain on a 0 ms loop timer, so the socket is read when there is something +//! to read and not otherwise. The drain runs from this module's own completion +//! (never inside `perry-ext-net`'s dispatch, per `RawNetNotify`'s contract). +//! +//! The exchange itself is unchanged: the request goes out in one write with +//! `Connection: close` (`client_connect_override::serialize_http_request`), the +//! response is read to EOF and parsed with `plain_client::parse_http_response`, +//! and a `101` to an upgrade request detaches the socket and hands it to the +//! request's `'upgrade'` listener with the bytes after the head. The deadline +//! keeps its old default of 30 s; it is a loop timer now. + +use std::sync::atomic::Ordering; + +use perry_ffi::Handle; + +use super::{next_id, on_loop, run, with_state, Effect, State, Timer, RAW_COMPLETED}; +use crate::plain_client::parse_http_response; +use crate::{push_event, ClientInflightGuard, PendingHttpEvent}; + +/// The deadline the tokio task applied when the request set none. +const DEFAULT_DEADLINE_MS: u64 = 30_000; + +pub(super) struct RawExchange { + request_handle: Handle, + wants_upgrade: bool, + /// Response bytes read so far; parsed once the peer closes. + raw: Vec, + /// Armed drain timer, `0` when none is pending. + drain_timer: i64, + deadline_timer: i64, + _inflight: ClientInflightGuard, +} + +/// Start the exchange. Called on the JS thread with the socket already in +/// raw mode (`attach`ed by the caller so no byte can reach a JS `'data'` +/// listener first). Delivers exactly one terminal event for the request. +pub(crate) fn start_raw_exchange( + request_handle: Handle, + request: Vec, + wants_upgrade: bool, + timeout_ms: Option, + socket_id: i64, +) { + let Some(vtable) = perry_ffi::raw_net() else { + push_event(PendingHttpEvent::Error { + request_handle, + error_message: "agent.createConnection requires node:net (not linked)".to_string(), + }); + return; + }; + perry_ffi::register_raw_net_notify(raw_socket_ready); + let inflight = ClientInflightGuard::new(request_handle); + let carried = on_loop(move || { + (vtable.attach)(socket_id); + if (vtable.write)(socket_id, request.as_ptr(), request.len()) == 0 { + push_event(PendingHttpEvent::Error { + request_handle, + error_message: "failed to write request to agent socket".to_string(), + }); + drop(inflight); + return; + } + let effects = with_state(|st| { + let mut fx = Vec::new(); + let deadline = next_id(); + let deadline_timer = if deadline == perry_ffi::INVALID_HANDLE { + 0 + } else { + st.timers + .insert(deadline, Timer::RawDeadline { socket: socket_id }); + fx.push(Effect::ArmTimer( + deadline, + timeout_ms.unwrap_or(DEFAULT_DEADLINE_MS), + )); + deadline + }; + st.raw.insert( + socket_id, + RawExchange { + request_handle, + wants_upgrade, + raw: Vec::new(), + drain_timer: 0, + deadline_timer, + _inflight: inflight, + }, + ); + // Bytes may already be buffered (a server that answers before the + // request finished writing): drain once without waiting for a + // notification. + schedule_raw_drain(st, socket_id, &mut fx); + fx + }); + run(effects); + }); + if !carried { + push_event(PendingHttpEvent::TransportError { + request_handle, + message: format!("connect {}", super::NO_LOOP_CODE), + code: super::NO_LOOP_CODE.to_string(), + syscall: "connect".to_string(), + errno: perry_ffi::turnloop_net::errno_for_code(super::NO_LOOP_CODE) as i64, + }); + } +} + +/// `perry_ffi::raw_net_notify`'s target. Runs inside `perry-ext-net`'s +/// completion handling, so it only schedules; see [`drain_raw_socket`]. +extern "C" fn raw_socket_ready(socket_id: i64) { + let effects = with_state(|st| { + let mut fx = Vec::new(); + schedule_raw_drain(st, socket_id, &mut fx); + fx + }); + run(effects); +} + +fn schedule_raw_drain(st: &mut State, socket_id: i64, fx: &mut Vec) { + let Some(exchange) = st.raw.get_mut(&socket_id) else { + return; + }; + if exchange.drain_timer != 0 { + return; + } + let timer = next_id(); + if timer == perry_ffi::INVALID_HANDLE { + // No id for a timer: drain from the effect queue instead, which runs + // after the current dispatch has returned. + fx.push(Effect::RawDrain(socket_id)); + return; + } + exchange.drain_timer = timer; + st.timers + .insert(timer, Timer::RawDrain { socket: socket_id }); + fx.push(Effect::ArmTimer(timer, 0)); +} + +/// A drain timer fired: forget it, then read (outside the lock). +pub(super) fn on_raw_drain_timer(st: &mut State, socket_id: i64, timer: i64) -> Vec { + if let Some(exchange) = st.raw.get_mut(&socket_id) { + if exchange.drain_timer == timer { + exchange.drain_timer = 0; + } + } + vec![Effect::RawDrain(socket_id)] +} + +/// The deadline fired with the exchange still open. +pub(super) fn on_raw_deadline(st: &mut State, socket_id: i64) -> Vec { + let Some(exchange) = st.raw.remove(&socket_id) else { + return Vec::new(); + }; + let mut fx = Vec::new(); + if exchange.drain_timer != 0 && st.timers.remove(&exchange.drain_timer).is_some() { + fx.push(Effect::CancelTimer(exchange.drain_timer)); + } + fx.push(Effect::RawClose(socket_id)); + fx.push(Effect::Push(PendingHttpEvent::Timeout { + request_handle: exchange.request_handle, + })); + fx +} + +/// Settle an exchange: take it out of the table and cancel its timers. +fn settle_raw_exchange( + st: &mut State, + socket_id: i64, + fx: &mut Vec, +) -> Option { + let exchange = st.raw.remove(&socket_id)?; + for timer in [exchange.drain_timer, exchange.deadline_timer] { + if timer != 0 && st.timers.remove(&timer).is_some() { + fx.push(Effect::CancelTimer(timer)); + } + } + Some(exchange) +} + +/// Read everything `perry-ext-net` has buffered for the socket, and finish the +/// exchange if the response is complete. Runs from the effect queue with no +/// lock held: the vtable calls re-enter `perry-ext-net`, whose teardown can +/// call [`raw_socket_ready`] synchronously. +pub(super) fn drain_raw_socket(socket_id: i64) { + let Some(vtable) = perry_ffi::raw_net() else { + return; + }; + let mut chunk = [0u8; 16 * 1024]; + loop { + if !with_state(|st| st.raw.contains_key(&socket_id)) { + return; + } + let n = (vtable.poll_read)(socket_id, chunk.as_mut_ptr(), chunk.len()); + if n < 0 { + // Would block: the next notification brings more. + return; + } + if n == 0 { + // Clean EOF: the response is whatever arrived. + let (effects, exchange) = with_state(|st| { + let mut fx = Vec::new(); + let exchange = settle_raw_exchange(st, socket_id, &mut fx); + (fx, exchange) + }); + run(effects); + (vtable.close)(socket_id); + if let Some(exchange) = exchange { + deliver_raw_response(exchange); + } + return; + } + let bytes = &chunk[..n as usize]; + let upgraded = with_state(|st| { + let exchange = st.raw.get_mut(&socket_id)?; + exchange.raw.extend_from_slice(bytes); + if !exchange.wants_upgrade { + return None; + } + let end = exchange.raw.windows(4).position(|w| w == b"\r\n\r\n")? + 4; + let status = std::str::from_utf8(&exchange.raw[..end]) + .ok() + .and_then(|head| head.lines().next()) + .and_then(|line| line.split_whitespace().nth(1)) + .and_then(|code| code.parse::().ok()); + (status == Some(101)).then_some(end) + }); + if let Some(end) = upgraded { + let (effects, exchange) = with_state(|st| { + let mut fx = Vec::new(); + let exchange = settle_raw_exchange(st, socket_id, &mut fx); + (fx, exchange) + }); + run(effects); + let Some(exchange) = exchange else { + return; + }; + hand_off_raw_upgrade(exchange, socket_id, end); + return; + } + } +} + +fn hand_off_raw_upgrade(exchange: RawExchange, socket_id: i64, head_end: usize) { + let Some(vtable) = perry_ffi::raw_net() else { + return; + }; + RAW_COMPLETED.fetch_add(1, Ordering::Relaxed); + match parse_http_response(&exchange.raw[..head_end]) { + Ok(parsed) => { + (vtable.detach)(socket_id); + push_event(PendingHttpEvent::Upgrade { + request_handle: exchange.request_handle, + status: parsed.status, + status_message: parsed.status_message, + headers: parsed.headers, + socket_handle: socket_id, + head: exchange.raw[head_end..].to_vec(), + }); + } + Err(error_message) => { + (vtable.close)(socket_id); + push_event(PendingHttpEvent::Error { + request_handle: exchange.request_handle, + error_message, + }); + } + } +} + +fn deliver_raw_response(exchange: RawExchange) { + match parse_http_response(&exchange.raw) { + Ok(parsed) => { + RAW_COMPLETED.fetch_add(1, Ordering::Relaxed); + push_event(PendingHttpEvent::Response { + request_handle: exchange.request_handle, + status: parsed.status, + status_message: parsed.status_message, + headers: parsed.headers, + trailers: parsed.trailers, + body: parsed.body, + http_version: parsed.http_version, + }); + } + Err(error_message) => push_event(PendingHttpEvent::Error { + request_handle: exchange.request_handle, + error_message, + }), + } +} + +/// Close the socket for a deadline (from the effect queue, no lock held). +pub(super) fn close_raw_socket(socket_id: i64) { + if let Some(vtable) = perry_ffi::raw_net() { + (vtable.close)(socket_id); + } +} diff --git a/crates/perry-ext-http/src/lib.rs b/crates/perry-ext-http/src/lib.rs index 106460c611..731f14fbca 100644 --- a/crates/perry-ext-http/src/lib.rs +++ b/crates/perry-ext-http/src/lib.rs @@ -53,7 +53,6 @@ mod tls_client; mod client_connect_override; mod client_upgrade; mod plain_client; -use plain_client::parse_http_response; // `Expect: 100-continue` (#5080): arms the head-first exchange and hands the // withheld body to it at `end()`. diff --git a/crates/perry-ext-http/tests/turnloop_client_exchange.rs b/crates/perry-ext-http/tests/turnloop_client_exchange.rs index bcf6899196..4eb5c607f1 100644 --- a/crates/perry-ext-http/tests/turnloop_client_exchange.rs +++ b/crates/perry-ext-http/tests/turnloop_client_exchange.rs @@ -83,6 +83,8 @@ const POOLED_B: i64 = GET_307 + 3; const DEADLINE: i64 = GET_307 + 4; const TRAILERS: i64 = GET_307 + 5; const SECURE: i64 = GET_307 + 6; +const OVER_SOCKET: i64 = GET_307 + 7; +const DEFERRED: i64 = GET_307 + 8; const CERT_PEM: &[u8] = include_bytes!("../../../test-parity/node-suite/tls/fixtures/localhost-cert.pem"); @@ -196,6 +198,8 @@ fn every_client_shape_is_carried_end_to_end_on_turnloop() { a_deadline_tears_down_an_exchange_the_server_never_answers(); a_te_trailers_response_is_decoded_to_its_end(); an_https_request_handshakes_with_the_callers_ca(); + a_create_connection_socket_is_read_when_net_says_it_is_ready(); + a_deferred_event_fires_from_a_loop_deadline(); } fn a_cleartext_get_is_carried_and_a_307_is_not_followed() { @@ -533,3 +537,88 @@ fn an_https_request_handshakes_with_the_callers_ca() { assert!(head.starts_with("GET /secure HTTP/1.1\r\n"), "{head}"); assert_eq!(alpn, None, "Node's https client offers no ALPN"); } + +/// `agent.createConnection`: the exchange runs over a socket `perry-ext-net` +/// owns, through the raw-net vtable. The socket here is a real ext-net socket +/// (a connected stream adopted onto the loop exactly as an HTTP upgrade hands +/// one over), in raw mode, so this is the production read path end to end. +/// +/// What it proves beyond "a response arrived": the response is written by the +/// server only AFTER the request has been read, so the one drain `start` +/// schedules up front finds nothing, and every byte after that is read only +/// because `perry-ext-net` called `perry_ffi::raw_net_notify`. Sabotage-checked: +/// with the notify registration removed, this shape never completes. +fn a_create_connection_socket_is_read_when_net_says_it_is_ready() { + let listener = TcpListener::bind("127.0.0.1:0").expect("an ephemeral port"); + let port = listener.local_addr().expect("a bound address").port(); + let (heads, received) = mpsc::channel::(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("one connection"); + let (head, _) = read_request(&mut stream).expect("a request head"); + // Two writes with a gap: the second arrives on a later read, so the + // exchange needs a second notification to finish. + let _ = stream.write_all(b"HTTP/1.1 200 Over Your Socket\r\ncontent-length: 5\r\n\r\nhe"); + let _ = stream.flush(); + std::thread::sleep(Duration::from_millis(50)); + let _ = stream.write_all(b"llo"); + let _ = heads.send(head); + // Close: this path reads to EOF (it sends `Connection: close`). + }); + + let stream = TcpStream::connect(("127.0.0.1", port)).expect("the client side connects"); + let socket = perry_ext_net::adopt_upgraded_tcp_stream(stream); + assert_ne!( + socket, + perry_ffi::INVALID_HANDLE, + "ext-net adopted the socket" + ); + // On the loop owner: finish the adoption and publish the raw-net vtable. + perry_ext_net::ensure_adopted_socket_dispatch(); + assert!( + perry_ffi::raw_net().is_some(), + "perry-ext-net's raw-net vtable must be published, or this shape tests nothing" + ); + + let completed_before = client_turnloop::raw_completed_total(); + let url = format!("http://127.0.0.1:{port}/over-socket"); + client_turnloop::try_dispatch_over_socket( + OVER_SOCKET, + "GET", + &url, + &no_headers(), + &[], + None, + socket, + ); + drive("createConnection"); + assert_eq!( + client_turnloop::raw_completed_total(), + completed_before + 1, + "the response must have been read to EOF and delivered" + ); + let head = received + .recv_timeout(Duration::from_secs(10)) + .expect("the server must have received the request over the adopted socket"); + join(server, "createConnection server"); + settle(); + assert!(head.starts_with("GET /over-socket HTTP/1.1\r\n"), "{head}"); + assert!(head.contains("Connection: close\r\n"), "{head}"); +} + +/// The Agent facade's idle expiry and `req.setTimeout`'s early `'timeout'` are +/// deadlines on the loop now (they were tokio sleeps). +fn a_deferred_event_fires_from_a_loop_deadline() { + let fired_before = client_turnloop::deferred_fired_total(); + let started = Instant::now(); + client_turnloop::schedule_timeout_for_test(DEFERRED, 40); + let deadline = Instant::now() + Duration::from_secs(10); + while client_turnloop::deferred_fired_total() == fired_before { + turn(); + assert!(Instant::now() < deadline, "the loop deadline never fired"); + } + assert!( + started.elapsed() >= Duration::from_millis(40), + "it fired early: {:?}", + started.elapsed() + ); +} diff --git a/crates/perry-ext-net/src/raw_bridge.rs b/crates/perry-ext-net/src/raw_bridge.rs index 024a22c35a..e78973630b 100644 --- a/crates/perry-ext-net/src/raw_bridge.rs +++ b/crates/perry-ext-net/src/raw_bridge.rs @@ -57,6 +57,9 @@ pub(crate) fn route_data(id: i64, bytes: &[u8]) -> bool { if let Ok(mut st) = raw.lock() { st.buf.extend(bytes.iter().copied()); } + // After the buffer lock is released: the consumer drains on a + // later turn (see `perry_ffi::RawNetNotify`'s contract). + perry_ffi::raw_net_notify(id); true } None => false, @@ -71,6 +74,7 @@ pub(crate) fn mark_terminal(id: i64, error: Option) -> bool { match raw_state_for(id) { Some(raw) => { raw_mark_closed(&raw, error); + perry_ffi::raw_net_notify(id); true } None => false, @@ -114,7 +118,7 @@ extern "C" fn perry_net_raw_write(socket_id: i64, ptr: *const u8, len: usize) -> /// Drain up to `max` buffered inbound bytes from socket `socket_id` into `out`. /// Returns the byte count (`> 0`), `0` for clean EOF once drained and the peer /// closed, or `-1` when nothing is available but the socket is still open -/// ("would block" — the caller should yield and retry). +/// ("would block" — wait for the next `perry_ffi::raw_net_notify`). extern "C" fn perry_net_raw_poll_read(socket_id: i64, out: *mut u8, max: usize) -> isize { if out.is_null() || max == 0 { return -1; diff --git a/crates/perry-ffi/src/lib.rs b/crates/perry-ffi/src/lib.rs index e4bfba4514..cf18842e20 100644 --- a/crates/perry-ffi/src/lib.rs +++ b/crates/perry-ffi/src/lib.rs @@ -130,7 +130,9 @@ mod event_pump; pub use event_pump::{notify_main_thread, register_aux_event_pump}; mod raw_net; -pub use raw_net::{raw_net, register_raw_net, RawNetVtable}; +pub use raw_net::{ + raw_net, raw_net_notify, register_raw_net, register_raw_net_notify, RawNetNotify, RawNetVtable, +}; // `runtime-link` gates this `extern crate` so external npm-packaged // wrappers (which lack perry-runtime in their Cargo graph) compile diff --git a/crates/perry-ffi/src/raw_net.rs b/crates/perry-ffi/src/raw_net.rs index 3d914b6396..d2f89a4649 100644 --- a/crates/perry-ffi/src/raw_net.rs +++ b/crates/perry-ffi/src/raw_net.rs @@ -73,3 +73,42 @@ pub fn register_raw_net(vtable: RawNetVtable) { pub fn raw_net() -> Option<&'static RawNetVtable> { RAW_NET.get() } + +/// Readiness callback for raw-mode sockets. +/// +/// [`RawNetVtable::poll_read`] only answers "what is buffered now". Without a +/// push, a consumer can learn that bytes arrived only by polling, which is +/// what the `createConnection` exchange used to do from a tokio task on a 1 ms +/// sleep. The `net` backend calls [`raw_net_notify`] whenever a raw-mode +/// socket gains buffered bytes or reaches a terminal state (EOF, error, +/// destroy), so the consumer can drain it then. +/// +/// Scope, same as [`raw_net`]'s slot: this is a `static` in perry-ffi, so it +/// is shared only by code linked against one perry-ffi instance. The default +/// (auto-optimize) build links one; a `PERRY_NO_AUTO_OPTIMIZE` link of +/// separately built extension archives can carry several, and then neither +/// slot reaches across them (tracked in the perry-ffi cross-archive state +/// issue, together with the duplicated perry-ext-net crate that makes moving +/// only these two slots insufficient). +/// +/// Contract for the consumer's callback: it runs on the thread that owns the +/// agent's event loop, from inside the `net` backend's completion handling, +/// with no `net` lock held. It must not call back into the vtable +/// synchronously (the backend is mid-dispatch); it should schedule the drain +/// for a later turn instead. +pub type RawNetNotify = extern "C" fn(socket_id: i64); + +static RAW_NET_NOTIFY: OnceLock = OnceLock::new(); + +/// Install the raw-mode readiness callback. The first registration wins. +pub fn register_raw_net_notify(notify: RawNetNotify) { + let _ = RAW_NET_NOTIFY.set(notify); +} + +/// Called by the `net` backend when raw-mode socket `socket_id` has new +/// buffered bytes or has gone terminal. A no-op when no consumer registered. +pub fn raw_net_notify(socket_id: i64) { + if let Some(notify) = RAW_NET_NOTIFY.get() { + notify(socket_id); + } +} diff --git a/scripts/tokio_inventory.json b/scripts/tokio_inventory.json index dbbd817809..e113890d7e 100644 --- a/scripts/tokio_inventory.json +++ b/scripts/tokio_inventory.json @@ -25,18 +25,6 @@ "that says how much code sits behind an edge." ], "edges": [ - { - "crate": "perry-ext-http", - "dep": "tokio", - "kind": "normal", - "optional": false, - "target": null, - "surface": "two small node:http CLIENT paths and nothing else: an `agent.createConnection` / `createSocket` (or request-level `createConnection`) exchange, which `client_connect_override.rs` drives from a tokio task polling perry-ext-net's raw vtable with a 1 ms `tokio::time::sleep`; and the keep-alive Agent's socket-facade idle expiry (`agent.rs`, `perry_ffi::spawn_async` + a 40 ms `tokio::time::sleep`). NO LONGER the client transport: every exchange reqwest carried and the three raw-TcpStream bypasses (`TE: trailers`, `Expect: 100-continue`, `Connection: Upgrade`) run on turnloop in `client_turnloop`. NO LONGER any server (plan A closed) and NO LONGER `http2.connect`.", - "reached_when": "only when a request's Agent (or the request itself) supplies `createConnection` / `createSocket`, or when a keep-alive Agent returns a socket facade to its free pool", - "blocker": "plan A, C and D's `h2` half are done; `reqwest` and `tokio-rustls` left this crate with the client transport. Two separable uses hold the edge open: (1) the `createConnection` exchange loop, which polls because perry-ffi's `raw_net` vtable has no completion push -- driving it from perry-ext-net's socket events (the socket is already on the loop) removes it; (2) the facade idle-expiry sleep, which becomes a `tl::timer_arm` deadline exactly as `client_outgoing::arm_client_timeout` did. After both, the manifest line deletes.", - "issue": "unfiled \u2014 P8", - "plan": "D" - }, { "crate": "perry-ext-mongodb", "dep": "mongodb", @@ -69,7 +57,7 @@ "target": null, "surface": "the tokio HALF of the async bridge only (turnloop P8 lane L split it out): `common::tokio_bridge` \u2014 the current-thread `RUNTIME`, its wait-driver tick, `spawn` / `spawn_for_promise*` \u2014 plus the `perry_ffi_spawn_async` / `perry_ffi_spawn_blocking_with_reactor` C ABI and `perry_ffi_spawn_blocking`'s tokio-pool arm. The promise bridge itself (`common::async_bridge`'s settle queue and pump, and the promise / pool / blocking `perry_ffi_*` shims) is tokio-free under the `async-bridge` feature, which is what the auto-optimize driver now force-enables.", "reached_when": "the `async-runtime` feature, which is selected only by (1) a Cargo feature whose code hands tokio a future \u2014 `web-fetch` (reqwest), `bundled-net` / `tls-runtime` / `external-tls-server` / `external-net-tls` / `bundled-ws` (tokio sockets), `external-net-pump` / `external-ws-pump` / `external-http-server-pump` / `external-http-client-pump` (perry-ext-net / -ws / -http hand futures to `perry_ffi_spawn_async`); or (2) the auto-optimize driver, for every shared-tokio wrapper (`binding_needs_shared_tokio`: net, ws, http, https, http2, undici, fastify, mongodb, nodemailer) and for pg / mysql2. `full` still implies it, so every PERRY_NO_AUTO_OPTIMIZE / prebuilt-archive build links tokio. It is NO LONGER forced onto every auto-optimized program: one that uses only crypto, bcrypt, argon2, zlib (bundled or perry-ext-zlib), readline, worker_threads, timers or a UI backend links no tokio.", - "blocker": "each selector in `reached_when` has to go; then `tokio_bridge.rs` and the three `cfg(feature = \"async-runtime\")` shims in `perry_ffi_async.rs` delete whole and `async-runtime` collapses into `async-bridge` \u2014 nothing in `async_bridge.rs` has to move. In order: G (#11101) leaves `web-fetch` tokio-free, after which it needs only `async-bridge`; H (#11102) plus P1 put the bundled net / tls / ws sockets on turnloop handles; A moved perry-ext-net off tokio and tokio-rustls (#11105, landed in merge train 266); perry-ext-http's `perry_ffi_spawn_async` / `_with_reactor` use is what remains of that step; B removes the db wrappers' decline paths, which call `Handle::current()` inside `perry_ffi_spawn_blocking` (with perry-ext-net's `upgradeTLS` reply wait, the only reason that shim still needs tokio's pool in a tokio build). K took `container` off it: the compose engine runs on turnloop through `perry_container_compose::rt`, so `container` implies only `async-bridge`. PerryTS/turnloop#42 is NOT a blocker any more: `Occupancy::Long` shipped in turnloop 0.1.0-alpha.5 (Perry pins alpha.6), and the tokio-free `perry_ffi_spawn_blocking` already runs on it through `turnloop_pool::submit_long`.", + "blocker": "each selector in `reached_when` has to go; then `tokio_bridge.rs` and the three `cfg(feature = \"async-runtime\")` shims in `perry_ffi_async.rs` delete whole and `async-runtime` collapses into `async-bridge` \u2014 nothing in `async_bridge.rs` has to move. In order: G (#11101) leaves `web-fetch` tokio-free, after which it needs only `async-bridge`; H (#11102) plus P1 put the bundled net / tls / ws sockets on turnloop handles; A moved perry-ext-net off tokio and tokio-rustls (#11105, landed in merge train 266); perry-ext-http no longer hands tokio anything (C: #11205 moved the client transport; D: the `createConnection` exchange and the Agent facade expiry run on turnloop and the crate's `tokio` edge is gone), so what remains of that step is driver-side: `external-http-server-pump` / `external-http-client-pump` still imply `async-runtime`, and `binding_needs_shared_tokio` still lists `http` / `https` / `http2`, which keeps linking tokio into every http program though nothing in perry-ext-http uses it; B removes the db wrappers' decline paths, which call `Handle::current()` inside `perry_ffi_spawn_blocking` (with perry-ext-net's `upgradeTLS` reply wait, the only reason that shim still needs tokio's pool in a tokio build). K took `container` off it: the compose engine runs on turnloop through `perry_container_compose::rt`, so `container` implies only `async-bridge`. PerryTS/turnloop#42 is NOT a blocker any more: `Occupancy::Long` shipped in turnloop 0.1.0-alpha.5 (Perry pins alpha.6), and the tokio-free `perry_ffi_spawn_blocking` already runs on it through `turnloop_pool::submit_long`.", "issue": "unfiled \u2014 P8; PerryTS/turnloop#42 closed (Occupancy::Long, turnloop 0.1.0-alpha.5+); lane L split the bridge from the runtime", "plan": "L" }, @@ -110,7 +98,6 @@ "perry": 3, "perry-container-compose": 27, "perry-ext-ads": 5, - "perry-ext-http": 7, "perry-ext-mongodb": 29, "perry-ffi": 2, "perry-stdlib": 69, From 9d9a0af296d11fbd04784af3c0851af68d8cbbe3 Mon Sep 17 00:00:00 2001 From: Perry Bot Date: Fri, 25 Sep 2026 00:50:07 +0000 Subject: [PATCH 2/2] changelog: perry-ext-http drops tokio (#11265) --- changelog.d/11265-ext-http-drop-tokio.md | 36 ++++++++++++++++++++++++ 1 file changed, 36 insertions(+) create mode 100644 changelog.d/11265-ext-http-drop-tokio.md diff --git a/changelog.d/11265-ext-http-drop-tokio.md b/changelog.d/11265-ext-http-drop-tokio.md new file mode 100644 index 0000000000..e1bdb2c735 --- /dev/null +++ b/changelog.d/11265-ext-http-drop-tokio.md @@ -0,0 +1,36 @@ +`perry-ext-http` no longer depends on tokio (tokio lane D). Together with +#11205 (reqwest, tokio-rustls) and #11144 (hyper), the crate now has no +tokio-family dependency at all: `cargo tree -p perry-ext-http -i tokio -e +normal,dev` finds no tokio, and the tokio inventory drops from 5 to 4 manifest +edges. + +The last two tokio users in the crate moved onto the agent's turnloop loop: + +- **`agent.createConnection` / `createSocket`, and a request-level + `createConnection`.** The HTTP exchange over the socket JS produced used to + run in a tokio task that called the raw-net vtable's `poll_read` in a loop, + sleeping 1 ms after each empty read. It now runs in + `client_turnloop::raw_socket`. perry-ext-net calls a new perry-ffi hook, + `raw_net_notify`, whenever a raw-mode socket gains bytes or goes terminal; + the client then schedules a drain on a 0 ms loop timer. The socket stays + perry-ext-net's, including its connect and TLS state, and the exchange is + unchanged: + - `Connection: close`, read to EOF, and the same parser. + - A `101` detaches the socket for `'upgrade'`. + - The 30 s default deadline is kept, now as a loop timer. + - In `PERRY_NO_AUTO_OPTIMIZE=1` builds, `createConnection` also needs #11263 + (one perry-ffi and one perry-ext-net per link); auto-optimize builds work + now. +- **The keep-alive Agent's socket-facade idle expiry.** The 40 ms + `tokio::time::sleep` is now a loop deadline through + `client_turnloop::push_after`. `req.setTimeout`'s early `'timeout'` uses the + same mechanism. + +Neither loop deadline keeps the process alive by itself. + +Not part of this change: compiled http programs still link tokio. The compiler +driver's `binding_needs_shared_tokio` still lists `http` / `https` / `http2`, +and perry-stdlib's `external-http-{client,server}-pump` features still imply +`async-runtime`, although nothing in perry-ext-http uses either any more. That +driver change is the next step, recorded in the tokio inventory's +`perry-stdlib -> tokio` row.