chromiumoxide/
runtime_release.rs

1//! Background batcher for `Runtime.releaseObject` calls.
2//!
3//! # Lock-freedom
4//!
5//! Every hot-path operation reduces to one or two relaxed-ordering atomic
6//! ops, with no mutex, no `Once`-style sync after the first init:
7//!
8//! | operation       | primitive                                                |
9//! | --------------- | -------------------------------------------------------- |
10//! | `try_release`   | `OnceLock::get()` (atomic load) + `UnboundedSender::send` (tokio's lock-free MPSC list) |
11//! | `init_worker`   | `OnceLock::get()` fast-path check; `get_or_init` only on the *first* call ever |
12//! | `Drop` guards   | `Arc::clone` (atomic fetch-add) + `try_release`          |
13//!
14//! The **first** `init_worker` call runs `OnceLock::get_or_init`, which
15//! briefly uses the `Once` sync primitive to serialise racing initialisers.
16//! This is a one-time per-process cost; subsequent calls skip it entirely
17//! via an explicit `get()` fast-path check.
18//!
19//! # Deadlock-freedom
20//!
21//! - `Drop` handlers never await, never acquire a lock another async task
22//!   holds, and never allocate beyond an atomic pointer push into the MPSC.
23//! - The worker holds no locks across its awaits; `rx.recv().await` parks
24//!   on a lock-free `Notify`, and `page.execute` goes through the existing
25//!   CDP command pipeline (no new locks introduced).
26//! - If the worker task panics or the runtime shuts down, `try_release`
27//!   continues to succeed (channel send does not require a live receiver);
28//!   releases are silently dropped and V8 reclaims on context teardown.
29//!
30//! # Batching
31//!
32//! The worker drains up to [`MAX_BATCH`] pending releases per round and
33//! fires them **concurrently** through `futures::join_all`, multiplexing on
34//! the shared CDP connection rather than stalling per call.
35
36use crate::page::Page;
37use chromiumoxide_cdp::cdp::js_protocol::runtime::{ReleaseObjectParams, RemoteObjectId};
38use std::sync::OnceLock;
39use tokio::sync::mpsc;
40
41/// Max items drained per worker round.  Cap keeps a burst from holding a
42/// giant `Vec` while we issue concurrent CDP commands.
43const MAX_BATCH: usize = 64;
44
45static RELEASE_TX: OnceLock<mpsc::UnboundedSender<(Page, RemoteObjectId)>> = OnceLock::new();
46
47/// Spawn the worker and return its sender.  Only ever invoked once, from
48/// inside the `OnceLock::get_or_init` closure on the very first init.
49fn spawn_worker() -> mpsc::UnboundedSender<(Page, RemoteObjectId)> {
50    let (tx, mut rx) = mpsc::unbounded_channel::<(Page, RemoteObjectId)>();
51
52    tokio::spawn(async move {
53        let mut batch: Vec<(Page, RemoteObjectId)> = Vec::with_capacity(MAX_BATCH);
54        loop {
55            // `recv_many` awaits at least one item and then drains up to
56            // `limit` without additional awaits — single atomic drain
57            // rather than the `recv + N×try_recv` CAS loop.
58            let n = rx.recv_many(&mut batch, MAX_BATCH).await;
59            if n == 0 {
60                break; // channel closed — no more producers
61            }
62
63            // Fire all release commands concurrently.  Each `page.execute`
64            // multiplexes onto the shared CDP connection; `join_all` lets
65            // Chrome process them in parallel rather than strict serial.
66            let futs = batch.drain(..).map(|(page, id)| async move {
67                let _ = page.execute(ReleaseObjectParams::new(id)).await;
68            });
69            futures_util::future::join_all(futs).await;
70        }
71    });
72
73    tx
74}
75
76/// Ensure the background release worker is running.
77///
78/// Hot path is a single atomic `OnceLock::get()` load; only the very first
79/// call ever touches `get_or_init` (and thus the `Once` sync primitive).
80/// Must be invoked from a tokio runtime context on the first call.
81#[inline]
82pub fn init_worker() {
83    // Explicit fast path: one atomic load, zero sync primitives after the
84    // first init.
85    if RELEASE_TX.get().is_some() {
86        return;
87    }
88    let _ = RELEASE_TX.get_or_init(spawn_worker);
89}
90
91/// Returns `true` if the worker has been initialised in this process.
92#[inline]
93pub fn worker_inited() -> bool {
94    RELEASE_TX.get().is_some()
95}
96
97/// Enqueue a remote-object release for background processing.
98///
99/// Lock-free (one atomic load + one wait-free MPSC push), safe to invoke
100/// from a `Drop` implementation on any thread.  If the worker has not yet
101/// been initialised the release is silently dropped; V8 reclaims the
102/// object on the next execution-context teardown.
103#[inline]
104pub fn try_release(page: Page, object_id: RemoteObjectId) {
105    if let Some(tx) = RELEASE_TX.get() {
106        let _ = tx.send((page, object_id));
107    }
108}
chromiumoxide/runtime_release.rs

chromiumoxide/
runtime_release.rs
