From 70edd60ed226161f3caa23e8c11bcda3b4bb10d2 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Tue, 8 Sep 2026 05:32:17 +0800 Subject: [PATCH] fix(renderer): release retired renderer I/O drivers Require Tokio 1.53.1 so dropping worker wake receivers releases their registered wakers even while browser-context senders remain live. This frees stopped renderer I/O drivers instead of retaining an eventpoll/eventfd pair per closed target. Cover deferred services, live services, and deferred-to-live initialization while preserving a live peer receiver. --- Cargo.lock | 4 +- moli-renderer-v8/Cargo.toml | 5 +- .../src/runtime/browser_context_runtime.rs | 3 + .../owner_wake_tests.rs | 104 ++++++++++++++++++ 4 files changed, 113 insertions(+), 3 deletions(-) create mode 100644 moli-renderer-v8/src/runtime/browser_context_runtime/owner_wake_tests.rs diff --git a/Cargo.lock b/Cargo.lock index 82903b196f..c4f557230f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5080,9 +5080,9 @@ dependencies = [ [[package]] name = "tokio" -version = "1.52.1" +version = "1.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b67dee974fe86fd92cc45b7a95fdd2f99a36a6d7b0d431a231178d3d670bbcc6" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" dependencies = [ "bytes", "libc", diff --git a/moli-renderer-v8/Cargo.toml b/moli-renderer-v8/Cargo.toml index a65a234ede..e8f0eac0d9 100644 --- a/moli-renderer-v8/Cargo.toml +++ b/moli-renderer-v8/Cargo.toml @@ -96,7 +96,10 @@ serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.145" serde_v8 = "=0.320.0" strum = { version = "0.28.0", features = ["derive"] } -tokio = { version = "1.51.0", features = ["io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] } +# Tokio >=1.53 releases a dropped mpsc receiver's waker even while context-wide +# senders remain (tokio-rs/tokio#8095). Older versions retain stopped renderer +# I/O drivers and leak an eventpoll/eventfd pair per closed target on Linux. +tokio = { version = "1.53.1", features = ["io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] } tracing = "0.1.41" url = "2.5.7" uuid = { version = "1.23.1", default-features = false } diff --git a/moli-renderer-v8/src/runtime/browser_context_runtime.rs b/moli-renderer-v8/src/runtime/browser_context_runtime.rs index 59bfa4b0f5..bc7c267c73 100644 --- a/moli-renderer-v8/src/runtime/browser_context_runtime.rs +++ b/moli-renderer-v8/src/runtime/browser_context_runtime.rs @@ -941,6 +941,9 @@ impl RendererWorkerContextRuntime { } } +#[cfg(test)] +mod owner_wake_tests; + #[cfg(test)] mod tests { use std::time::Duration; diff --git a/moli-renderer-v8/src/runtime/browser_context_runtime/owner_wake_tests.rs b/moli-renderer-v8/src/runtime/browser_context_runtime/owner_wake_tests.rs new file mode 100644 index 0000000000..989fa7f972 --- /dev/null +++ b/moli-renderer-v8/src/runtime/browser_context_runtime/owner_wake_tests.rs @@ -0,0 +1,104 @@ +use std::{ + sync::Arc, + task::{Context, Wake, Waker}, +}; + +use super::RendererBrowserContextRuntime; +use crate::{ + service_worker_runtime::service_worker_owner_wake_channel, + shared_worker_runtime::shared_worker_owner_wake_channel, +}; + +enum WorkerServices { + Deferred, + AlreadyLive, + InitializedAfterRegistration, +} + +struct ReceiverWake; + +// A static Waker::noop() cannot witness whether the channel retains its owner. +#[allow(clippy::manual_noop_waker)] +impl Wake for ReceiverWake { + fn wake(self: Arc) {} +} + +fn assert_retired_receiver_releases_waker(services: WorkerServices) { + let context = RendererBrowserContextRuntime::new(); + let initialize_workers = || { + context.inner.shared_worker_runtime.get_or_init(); + context.inner.service_worker_runtime.get_or_init(); + }; + if matches!(services, WorkerServices::AlreadyLive) { + initialize_workers(); + } + + let (shared_tx, mut shared_rx) = shared_worker_owner_wake_channel(); + let (service_tx, mut service_rx) = service_worker_owner_wake_channel(); + context.add_shared_worker_owner_wake_sender(shared_tx); + context.add_service_worker_owner_wake_sender(service_tx); + let (peer_shared_tx, mut peer_shared_rx) = shared_worker_owner_wake_channel(); + let (peer_service_tx, mut peer_service_rx) = service_worker_owner_wake_channel(); + context.add_shared_worker_owner_wake_sender(peer_shared_tx); + context.add_service_worker_owner_wake_sender(peer_service_tx); + + if matches!(services, WorkerServices::InitializedAfterRegistration) { + initialize_workers(); + } + + let shared_wake = Arc::new(ReceiverWake); + let shared_weak = Arc::downgrade(&shared_wake); + let service_wake = Arc::new(ReceiverWake); + let service_weak = Arc::downgrade(&service_wake); + { + let shared_waker = Waker::from(shared_wake); + let service_waker = Waker::from(service_wake); + assert!( + shared_rx + .poll_recv(&mut Context::from_waker(&shared_waker)) + .is_pending() + ); + assert!( + service_rx + .poll_recv(&mut Context::from_waker(&service_waker)) + .is_pending() + ); + } + drop(shared_rx); + drop(service_rx); + + // The context (including its senders) stays live. Before Tokio 1.53, + // receiver drop retained the last block_on waker and its I/O driver. + // Assert before any new registration/wake can happen to prune old routes. + assert!( + shared_weak.upgrade().is_none(), + "retired SharedWorker receiver retained its waker" + ); + assert!( + service_weak.upgrade().is_none(), + "retired ServiceWorker receiver retained its waker" + ); + assert!(matches!( + peer_shared_rx.try_recv(), + Err(tokio::sync::mpsc::error::TryRecvError::Empty) + )); + assert!(matches!( + peer_service_rx.try_recv(), + Err(tokio::sync::mpsc::error::TryRecvError::Empty) + )); +} + +#[test] +fn retired_deferred_worker_wake_receivers_release_wakers_while_context_lives() { + assert_retired_receiver_releases_waker(WorkerServices::Deferred); +} + +#[test] +fn retired_live_worker_wake_receivers_release_wakers_without_affecting_peer() { + assert_retired_receiver_releases_waker(WorkerServices::AlreadyLive); +} + +#[test] +fn retired_worker_wake_receivers_release_wakers_after_service_initialization() { + assert_retired_receiver_releases_waker(WorkerServices::InitializedAfterRegistration); +}