use std::cell::{self, Cell, RefCell, RefMut};
use std::cmp::max;
use std::collections::hash_map;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicIsize, Ordering};
use std::thread;
use crossbeam_channel::{Receiver, SendError, Sender, unbounded};
use dom_struct::dom_struct;
use js::context::JSContext;
use js::jsapi::{GCReason, JSGCParamKey, JSTracer};
use js::realm::CurrentRealm;
use js::rust::wrappers2::{JS_GC, JS_GetGCParameter};
use malloc_size_of::malloc_size_of_is_0;
use net_traits::policy_container::PolicyContainer;
use net_traits::request::{Destination, Origin, PreloadedResources, RequestClient};
use rustc_hash::FxHashMap;
use script_bindings::reflector::{Reflector, reflect_dom_object_with_cx};
use servo_base::id::PipelineId;
use servo_url::{ImmutableOrigin, ServoUrl};
use style::thread_state::{self, ThreadState};
use swapper::{Swapper, swapper};
use uuid::Uuid;
use crate::conversions::Convert;
use crate::dom::bindings::codegen::Bindings::RequestBinding::RequestCredentials;
use crate::dom::bindings::codegen::Bindings::WindowBinding::Window_Binding::WindowMethods;
use crate::dom::bindings::codegen::Bindings::WorkletBinding::{WorkletMethods, WorkletOptions};
use crate::dom::bindings::error::Error;
use crate::dom::bindings::inheritance::Castable;
use crate::dom::bindings::refcounted::TrustedPromise;
use crate::dom::bindings::root::{Dom, DomRoot};
use crate::dom::bindings::str::USVString;
use crate::dom::bindings::trace::{JSTraceable, RootedTraceableBox};
use crate::dom::globalscope::GlobalScope;
use crate::dom::promise::Promise;
#[cfg(feature = "testbinding")]
use crate::dom::testworkletglobalscope::TestWorkletTask;
use crate::dom::window::Window;
use crate::dom::workletglobalscope::{
WorkletGlobalScope, WorkletGlobalScopeInit, WorkletGlobalScopeType, WorkletTask,
};
use crate::messaging::{CommonScriptMsg, MainThreadScriptMsg, ScriptEventLoopSender};
use crate::microtask::MicrotaskQueue;
use crate::modules::script_module::fetch_a_module_script_graph;
use crate::realms::enter_auto_realm;
use crate::script_runtime::{IntroductionType, Runtime, ScriptThreadEventCategory};
use crate::tasks::task_source::TaskSourceName;
use crate::url::ensure_blob_referenced_by_url_is_kept_alive;
const WORKLET_THREAD_POOL_SIZE: u32 = 3;
const MIN_GC_THRESHOLD: u32 = 1_000_000;
type LazyCellWithBoxedInitializer<T> = cell::LazyCell<T, Box<dyn FnOnce() -> T>>;
#[derive(JSTraceable, MallocSizeOf)]
struct DroppableField {
worklet_id: WorkletId,
#[ignore_malloc_size_of = "Difficult to measure memory usage of Rc<...> types"]
thread_pool: LazyCellWithBoxedInitializer<Rc<WorkletThreadPool>>,
is_thread_pool_initialized: Cell<bool>,
}
impl Drop for DroppableField {
fn drop(&mut self) {
let worklet_id = self.worklet_id;
if self.is_thread_pool_initialized.get() {
self.thread_pool.exit_worklet(worklet_id);
}
}
}
#[dom_struct]
pub(crate) struct Worklet {
reflector: Reflector,
window: Dom<Window>,
global_type: WorkletGlobalScopeType,
droppable_field: DroppableField,
}
impl Worklet {
fn new_inherited(
window: &Window,
global_type: WorkletGlobalScopeType,
thread_pool_constructor: Box<dyn FnOnce() -> Rc<WorkletThreadPool>>,
) -> Worklet {
Worklet {
reflector: Reflector::new(),
window: Dom::from_ref(window),
global_type,
droppable_field: DroppableField {
worklet_id: WorkletId::new(),
thread_pool: LazyCellWithBoxedInitializer::new(thread_pool_constructor),
is_thread_pool_initialized: Cell::new(false),
},
}
}
pub(crate) fn new(
cx: &mut JSContext,
window: &Window,
global_type: WorkletGlobalScopeType,
thread_pool_constructor: Box<dyn FnOnce() -> Rc<WorkletThreadPool>>,
) -> DomRoot<Worklet> {
debug!("Creating worklet {:?}.", global_type);
reflect_dom_object_with_cx(
Box::new(Worklet::new_inherited(
window,
global_type,
thread_pool_constructor,
)),
window,
cx,
)
}
pub(crate) fn worklet_thread_pool(&self) -> &WorkletThreadPool {
self.droppable_field.is_thread_pool_initialized.set(true);
&self.droppable_field.thread_pool
}
#[cfg(feature = "testbinding")]
pub(crate) fn worklet_id(&self) -> WorkletId {
self.droppable_field.worklet_id
}
#[expect(dead_code)]
pub(crate) fn worklet_global_scope_type(&self) -> WorkletGlobalScopeType {
self.global_type
}
}
impl WorkletMethods<crate::DomTypeHolder> for Worklet {
fn AddModule(
&self,
realm: &mut CurrentRealm,
module_url: USVString,
options: &WorkletOptions,
) -> Rc<Promise> {
let promise = Promise::new_in_realm(realm);
let module_url_record = match self.window.Document().base_url().join(&module_url.0) {
Ok(url) => url,
Err(err) => {
debug!("URL {:?} parse error {:?}.", module_url.0, err);
promise.reject_error(realm, Error::Syntax(None));
return promise;
},
};
debug!("Adding Worklet module {}.", module_url_record);
let global_scope = self.window.as_global_scope();
let pending_tasks_struct = PendingTasksStruct::new();
self.worklet_thread_pool()
.fetch_and_invoke_a_worklet_script(
self.window.pipeline_id(),
self.droppable_field.worklet_id,
self.global_type,
self.window.origin().immutable().clone(),
global_scope.api_base_url(),
module_url_record,
global_scope.policy_container(),
options.credentials,
pending_tasks_struct,
&promise,
global_scope.inherited_secure_context(),
);
debug!("Returning promise.");
promise
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, JSTraceable, PartialEq)]
pub(crate) struct WorkletId(#[no_trace] Uuid);
malloc_size_of_is_0!(WorkletId);
impl WorkletId {
fn new() -> WorkletId {
WorkletId(Uuid::new_v4())
}
}
#[derive(Clone, Debug)]
pub(crate) struct PendingTasksStruct(Arc<AtomicIsize>);
impl PendingTasksStruct {
fn new() -> PendingTasksStruct {
PendingTasksStruct(Arc::new(AtomicIsize::new(
WORKLET_THREAD_POOL_SIZE as isize,
)))
}
fn set_counter_to(&self, value: isize) -> isize {
self.0.swap(value, Ordering::AcqRel)
}
fn decrement_counter_by(&self, offset: isize) -> isize {
self.0.fetch_sub(offset, Ordering::AcqRel)
}
}
#[derive(Clone, JSTraceable)]
pub(crate) struct WorkletThreadPool {
#[no_trace]
primary_sender: Sender<WorkletData>,
#[no_trace]
hot_backup_sender: Sender<WorkletData>,
#[no_trace]
cold_backup_sender: Sender<WorkletData>,
#[no_trace]
control_sender_0: Sender<WorkletControl>,
#[no_trace]
control_sender_1: Sender<WorkletControl>,
#[no_trace]
control_sender_2: Sender<WorkletControl>,
}
impl Drop for WorkletThreadPool {
fn drop(&mut self) {
let _ = self.cold_backup_sender.send(WorkletData::Quit);
let _ = self.hot_backup_sender.send(WorkletData::Quit);
let _ = self.primary_sender.send(WorkletData::Quit);
}
}
impl WorkletThreadPool {
pub(crate) fn spawn(global_init: WorkletGlobalScopeInit) -> WorkletThreadPool {
let primary_role = WorkletThreadRole::new(false, false);
let hot_backup_role = WorkletThreadRole::new(true, false);
let cold_backup_role = WorkletThreadRole::new(false, true);
let primary_sender = primary_role.sender.clone();
let hot_backup_sender = hot_backup_role.sender.clone();
let cold_backup_sender = cold_backup_role.sender.clone();
let init = WorkletThreadInit {
primary_sender: primary_sender.clone(),
hot_backup_sender: hot_backup_sender.clone(),
cold_backup_sender: cold_backup_sender.clone(),
global_init,
};
WorkletThreadPool {
primary_sender,
hot_backup_sender,
cold_backup_sender,
control_sender_0: WorkletThread::spawn(primary_role, init.clone(), 0),
control_sender_1: WorkletThread::spawn(hot_backup_role, init.clone(), 1),
control_sender_2: WorkletThread::spawn(cold_backup_role, init, 2),
}
}
#[allow(clippy::too_many_arguments)]
fn fetch_and_invoke_a_worklet_script(
&self,
pipeline_id: PipelineId,
worklet_id: WorkletId,
global_type: WorkletGlobalScopeType,
origin: ImmutableOrigin,
base_url: ServoUrl,
script_url: ServoUrl,
policy_container: PolicyContainer,
credentials: RequestCredentials,
pending_tasks_struct: PendingTasksStruct,
promise: &Rc<Promise>,
inherited_secure_context: Option<bool>,
) {
for sender in &[
&self.control_sender_0,
&self.control_sender_1,
&self.control_sender_2,
] {
let _ = sender.send(WorkletControl::FetchAndInvokeAWorkletScript {
pipeline_id,
worklet_id,
global_type,
origin: origin.clone(),
base_url: base_url.clone(),
script_url: script_url.clone(),
policy_container: policy_container.clone(),
credentials,
pending_tasks_struct: pending_tasks_struct.clone(),
promise: TrustedPromise::new(promise.clone()),
inherited_secure_context,
});
}
self.wake_threads();
}
pub(crate) fn exit_worklet(&self, worklet_id: WorkletId) {
for sender in &[
&self.control_sender_0,
&self.control_sender_1,
&self.control_sender_2,
] {
let _ = sender.send(WorkletControl::ExitWorklet(worklet_id));
}
self.wake_threads();
}
#[cfg(feature = "testbinding")]
pub(crate) fn test_worklet_lookup(&self, id: WorkletId, key: String) -> Option<String> {
let (sender, receiver) = unbounded();
let msg = WorkletData::Task(id, WorkletTask::Test(TestWorkletTask::Lookup(key, sender)));
let _ = self.primary_sender.send(msg);
receiver.recv().expect("Test worklet has died?")
}
fn wake_threads(&self) {
let _ = self.cold_backup_sender.send(WorkletData::WakeUp);
let _ = self.hot_backup_sender.send(WorkletData::WakeUp);
let _ = self.primary_sender.send(WorkletData::WakeUp);
}
}
enum WorkletData {
Task(WorkletId, WorkletTask),
StartSwapRoles(Sender<WorkletData>),
FinishSwapRoles(Swapper<WorkletThreadRole>),
WakeUp,
Quit,
}
pub(crate) enum WorkletControl {
ExitWorklet(WorkletId),
FetchAndInvokeAWorkletScript {
pipeline_id: PipelineId,
worklet_id: WorkletId,
global_type: WorkletGlobalScopeType,
origin: ImmutableOrigin,
base_url: ServoUrl,
script_url: ServoUrl,
policy_container: PolicyContainer,
credentials: RequestCredentials,
pending_tasks_struct: PendingTasksStruct,
promise: TrustedPromise,
inherited_secure_context: Option<bool>,
},
Common(CommonScriptMsg),
}
struct WorkletThreadRole {
receiver: Receiver<WorkletData>,
sender: Sender<WorkletData>,
is_hot_backup: bool,
is_cold_backup: bool,
}
impl WorkletThreadRole {
fn new(is_hot_backup: bool, is_cold_backup: bool) -> WorkletThreadRole {
let (sender, receiver) = unbounded();
WorkletThreadRole {
sender,
receiver,
is_hot_backup,
is_cold_backup,
}
}
}
#[derive(Clone)]
struct WorkletThreadInit {
primary_sender: Sender<WorkletData>,
hot_backup_sender: Sender<WorkletData>,
cold_backup_sender: Sender<WorkletData>,
global_init: WorkletGlobalScopeInit,
}
#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
struct WorkletThread {
role: WorkletThreadRole,
control_receiver: Receiver<WorkletControl>,
control_sender: Sender<WorkletControl>,
primary_sender: Sender<WorkletData>,
hot_backup_sender: Sender<WorkletData>,
cold_backup_sender: Sender<WorkletData>,
global_init: WorkletGlobalScopeInit,
global_scopes: FxHashMap<WorkletId, Dom<WorkletGlobalScope>>,
control_buffer: Option<WorkletControl>,
closing: Arc<AtomicBool>,
runtime: Runtime,
should_gc: bool,
gc_threshold: u32,
}
#[expect(unsafe_code)]
unsafe impl JSTraceable for WorkletThread {
unsafe fn trace(&self, trc: *mut JSTracer) {
debug!("Tracing worklet thread.");
unsafe { self.global_scopes.trace(trc) };
}
}
impl WorkletThread {
#[allow(unsafe_code)]
fn spawn(
role: WorkletThreadRole,
init: WorkletThreadInit,
thread_index: u8,
) -> Sender<WorkletControl> {
let (control_sender, control_receiver) = unbounded();
let control_sender_clone = control_sender.clone();
let _ = thread::Builder::new()
.name(format!("Worklet#{thread_index}"))
.spawn(move || {
debug!("Initializing worklet thread.");
thread_state::initialize(ThreadState::SCRIPT | ThreadState::IN_WORKER);
let runtime = Runtime::new(None);
let mut cx = unsafe { runtime.cx() };
let mut thread = RootedTraceableBox::new(WorkletThread {
role,
control_receiver,
control_sender: control_sender_clone,
primary_sender: init.primary_sender,
hot_backup_sender: init.hot_backup_sender,
cold_backup_sender: init.cold_backup_sender,
global_init: init.global_init,
global_scopes: FxHashMap::default(),
control_buffer: None,
runtime,
should_gc: false,
closing: Arc::new(AtomicBool::new(false)),
gc_threshold: MIN_GC_THRESHOLD,
});
thread.run(&mut cx);
})
.expect("Couldn't start worklet thread");
control_sender
}
fn run(&mut self, cx: &mut JSContext) {
loop {
let message = self.role.receiver.recv().unwrap();
match message {
WorkletData::Task(id, task) => {
self.perform_a_worklet_task(cx, id, task);
},
WorkletData::StartSwapRoles(sender) => {
let (our_swapper, their_swapper) = swapper();
match sender.send(WorkletData::FinishSwapRoles(their_swapper)) {
Ok(_) => {},
Err(_) => {
return;
},
};
let _ = our_swapper.swap(&mut self.role);
},
WorkletData::FinishSwapRoles(swapper) => {
let _ = swapper.swap(&mut self.role);
},
WorkletData::WakeUp => {},
WorkletData::Quit => {
return;
},
}
if self.role.is_cold_backup {
if let Some(control) = self.control_buffer.take() {
self.process_control(control, cx);
}
while let Ok(control) = self.control_receiver.try_recv() {
self.process_control(control, cx);
}
for worklet_global_scope in self.global_scopes.values() {
worklet_global_scope.perform_a_microtask_checkpoint(cx);
}
self.gc(cx);
} else if self.control_buffer.is_none() &&
let Ok(control) = self.control_receiver.try_recv()
{
self.control_buffer = Some(control);
let msg = WorkletData::StartSwapRoles(self.role.sender.clone());
let _ = self.cold_backup_sender.send(msg);
}
if self.current_memory_usage() > self.gc_threshold {
if self.role.is_hot_backup || self.role.is_cold_backup {
self.should_gc = false;
self.gc(cx);
} else if !self.should_gc {
self.should_gc = true;
let msg = WorkletData::StartSwapRoles(self.role.sender.clone());
let _ = self.hot_backup_sender.send(msg);
}
}
}
}
#[expect(unsafe_code)]
fn current_memory_usage(&self) -> u32 {
unsafe { JS_GetGCParameter(self.runtime.cx_no_gc(), JSGCParamKey::JSGC_BYTES) }
}
#[expect(unsafe_code)]
fn gc(&mut self, cx: &mut JSContext) {
debug!(
"BEGIN GC (usage = {}, threshold = {}).",
self.current_memory_usage(),
self.gc_threshold
);
unsafe { JS_GC(cx, GCReason::API) };
self.gc_threshold = max(MIN_GC_THRESHOLD, self.current_memory_usage() * 2);
debug!(
"END GC (usage = {}, threshold = {}).",
self.current_memory_usage(),
self.gc_threshold
);
}
#[expect(clippy::too_many_arguments)]
fn get_worklet_global_scope(
&mut self,
cx: &mut JSContext,
pipeline_id: PipelineId,
worklet_id: WorkletId,
inherited_secure_context: Option<bool>,
global_type: WorkletGlobalScopeType,
base_url: ServoUrl,
microtask_queue: Rc<MicrotaskQueue>,
) -> DomRoot<WorkletGlobalScope> {
match self.global_scopes.entry(worklet_id) {
hash_map::Entry::Occupied(entry) => DomRoot::from_ref(entry.get()),
hash_map::Entry::Vacant(entry) => {
debug!("Creating new worklet global scope.");
let executor = WorkletExecutor {
worklet_id,
primary_sender: self.primary_sender.clone(),
hot_backup_sender: self.hot_backup_sender.clone(),
cold_backup_sender: self.cold_backup_sender.clone(),
control_sender: self.control_sender.clone(),
};
let result = WorkletGlobalScope::new(
global_type,
pipeline_id,
base_url,
inherited_secure_context,
executor,
&self.global_init,
cx,
self.closing.clone(),
microtask_queue,
);
entry.insert(Dom::from_ref(&*result));
result
},
}
}
#[allow(clippy::too_many_arguments)]
fn fetch_and_invoke_a_worklet_script(
&self,
global_scope: &WorkletGlobalScope,
pipeline_id: PipelineId,
origin: ImmutableOrigin,
script_url: ServoUrl,
policy_container: PolicyContainer,
credentials: RequestCredentials,
pending_tasks_struct: PendingTasksStruct,
promise: TrustedPromise,
cx: &mut JSContext,
) {
debug!("Fetching from {}.", script_url);
let global = global_scope.upcast::<GlobalScope>();
let request_client = RequestClient {
preloaded_resources: PreloadedResources::default(),
policy_container,
origin: Origin::Origin(origin),
is_nested_browsing_context: global.is_nested_browsing_context(),
insecure_requests_policy: global.insecure_requests_policy(),
has_trustworthy_ancestor_origin: global.has_trustworthy_ancestor_origin(),
};
let promise_task = Rc::new(RefCell::new(Some(promise)));
let script_thread_sender = self.global_init.to_script_thread_sender.clone();
let rooted_global = DomRoot::from_ref(global);
let script_url = ensure_blob_referenced_by_url_is_kept_alive(global, script_url);
fetch_a_module_script_graph(
cx,
global,
script_url,
request_client,
Destination::PaintWorklet,
global.get_referrer(),
credentials.convert(),
Some(IntroductionType::WORKLET),
move |cx, module_tree| {
match module_tree {
None => {
debug!("Failed to load script.");
reject_promise(
&pending_tasks_struct,
promise_task.borrow_mut(),
script_thread_sender.clone(),
);
},
Some(script) => {
let mut realm = enter_auto_realm(cx, &*rooted_global);
let cx = &mut realm.current_realm();
if script.get_rethrow_error().take().is_some() {
reject_promise(
&pending_tasks_struct,
promise_task.borrow_mut(),
script_thread_sender.clone(),
);
return;
}
rooted_global.run_a_module_script(cx, script, false);
let old_counter = pending_tasks_struct.decrement_counter_by(1);
if old_counter == 1 {
debug!("Resolving promise.");
let msg = MainThreadScriptMsg::WorkletLoaded(pipeline_id);
script_thread_sender
.send(msg)
.expect("Worklet thread outlived script thread.");
let task = promise_task
.borrow_mut()
.take()
.expect("promise_task must be consumed exactly once")
.resolve_task(());
let msg = CommonScriptMsg::Task(
ScriptThreadEventCategory::WorkletEvent,
Box::new(task),
None,
TaskSourceName::Networking,
);
let msg = MainThreadScriptMsg::Common(msg);
script_thread_sender
.send(msg)
.expect("Worklet thread outlived script thread.");
}
},
}
},
);
}
fn perform_a_worklet_task(&self, cx: &mut JSContext, worklet_id: WorkletId, task: WorkletTask) {
match self.global_scopes.get(&worklet_id) {
Some(global) => global.perform_a_worklet_task(cx, task),
None => warn!("No such worklet as {:?}.", worklet_id),
}
}
fn process_control(&mut self, control: WorkletControl, cx: &mut js::context::JSContext) {
match control {
WorkletControl::ExitWorklet(worklet_id) => {
self.global_scopes.remove(&worklet_id);
},
WorkletControl::FetchAndInvokeAWorkletScript {
pipeline_id,
worklet_id,
global_type,
origin,
base_url,
script_url,
policy_container,
credentials,
pending_tasks_struct,
promise,
inherited_secure_context,
} => {
let global = self.get_worklet_global_scope(
cx,
pipeline_id,
worklet_id,
inherited_secure_context,
global_type,
base_url,
self.runtime.microtask_queue.clone(),
);
self.fetch_and_invoke_a_worklet_script(
&global,
pipeline_id,
origin,
script_url,
policy_container,
credentials,
pending_tasks_struct,
promise,
cx,
)
},
WorkletControl::Common(script_msg) => {
if let CommonScriptMsg::Task(_, task, _, _) = script_msg {
task.run_box(cx);
}
},
}
}
}
pub(crate) fn reject_promise(
pending_tasks_struct: &PendingTasksStruct,
mut promise_task: RefMut<'_, Option<TrustedPromise>>,
script_thread_sender: Sender<MainThreadScriptMsg>,
) {
let old_counter = pending_tasks_struct.set_counter_to(-1);
if old_counter > 0 {
let task = promise_task
.take()
.expect("promise_task must be consumed exactly once")
.reject_task(Error::Abort(None));
let msg = CommonScriptMsg::Task(
ScriptThreadEventCategory::WorkletEvent,
Box::new(task),
None,
TaskSourceName::Networking,
);
let msg = MainThreadScriptMsg::Common(msg);
script_thread_sender
.send(msg)
.expect("Worklet thread outlived script thread.");
}
}
#[derive(Clone, JSTraceable, MallocSizeOf)]
pub(crate) struct WorkletExecutor {
worklet_id: WorkletId,
#[no_trace]
primary_sender: Sender<WorkletData>,
#[no_trace]
hot_backup_sender: Sender<WorkletData>,
#[no_trace]
cold_backup_sender: Sender<WorkletData>,
#[no_trace]
control_sender: Sender<WorkletControl>,
}
impl WorkletExecutor {
pub(crate) fn wake_threads(&self) -> Result<(), SendError<()>> {
self.cold_backup_sender
.send(WorkletData::WakeUp)
.map_err(|_| SendError(()))?;
self.hot_backup_sender
.send(WorkletData::WakeUp)
.map_err(|_| SendError(()))?;
self.primary_sender
.send(WorkletData::WakeUp)
.map_err(|_| SendError(()))
}
pub(crate) fn schedule_a_worklet_task(&self, task: WorkletTask) {
let _ = self
.primary_sender
.send(WorkletData::Task(self.worklet_id, task));
}
pub(crate) fn send_control_message(
&self,
control_message: WorkletControl,
) -> Result<(), SendError<()>> {
self.control_sender
.send(control_message)
.map_err(|_| SendError(()))?;
self.wake_threads()
}
pub(crate) fn event_loop_sender(&self) -> ScriptEventLoopSender {
ScriptEventLoopSender::Worklet(self.clone())
}
}