Mirrors/servo
1
0
Fork 0
mirror of https://github.com/servo/servo.git synced 2025-07-23 15:23:42 +01:00
Code Issues Projects Releases Packages Wiki Activity

Auto merge of #21388 - gterzian:introduce_task_queues, r=jdm

Browse source 
Introduce task queues, and throttling performance timeline tasksource

<!-- Please describe your changes on the following line: -->

---
<!-- Thank you for contributing to Servo! Please replace each `[ ]` by `[X]` when the step is complete, and replace `__` with appropriate data: -->
- [ ] `./mach build -d` does not report any errors
- [ ] `./mach test-tidy` does not report any errors
- [ ] These changes fix #19997 (github issue number if applicable).

<!-- Either: -->
- [ ] There are tests for these changes OR
- [ ] These changes do not require tests because _____

<!-- Also, please make sure that "Allow edits from maintainers" checkbox is checked, so that we can help you if you get stuck somewhere along the way.-->

<!-- Pull requests that do not address these steps are welcome, but they will require additional verification as part of the review process. -->

<!-- Reviewable:start -->
---
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/servo/servo/21388)
<!-- Reviewable:end -->
This commit is contained in:
bors-servo 2018-08-30 14:56:26 -04:00 • committed by GitHub
commit 1ee3deea27
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 629 additions and 138 deletions
Download patch file Download diff file Expand all files Collapse all files

131
components/script/dom/abstractworkerglobalscope.rs
View file

@ -2,25 +2,31 @@
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
use devtools_traits::DevtoolScriptControlMsg;
use dom::abstractworker::WorkerScriptMsg;
use dom::bindings::refcounted::Trusted;
use dom::bindings::conversions::DerivedFrom;
use dom::bindings::reflector::DomObject;
use dom::bindings::trace::JSTraceable;
use dom::dedicatedworkerglobalscope::{AutoWorkerReset, DedicatedWorkerScriptMsg};
use dom::globalscope::GlobalScope;
use dom::worker::TrustedWorkerAddress;
use dom::workerglobalscope::WorkerGlobalScope;
use script_runtime::{ScriptChan, CommonScriptMsg, ScriptPort};
use std::sync::mpsc::{Receiver, Sender};
use std::sync::mpsc::{Receiver, Select, Sender};
use task_queue::{QueuedTaskConversion, TaskQueue};
/// A ScriptChan that can be cloned freely and will silently send a TrustedWorkerAddress with
/// common event loop messages. While this SendableWorkerScriptChan is alive, the associated
/// Worker object will remain alive.
#[derive(Clone, JSTraceable)]
pub struct SendableWorkerScriptChan<T: DomObject> {
pub sender: Sender<(Trusted<T>, CommonScriptMsg)>,
pub worker: Trusted<T>,
pub struct SendableWorkerScriptChan {
pub sender: Sender<DedicatedWorkerScriptMsg>,
pub worker: TrustedWorkerAddress,
}
impl<T: JSTraceable + DomObject + 'static> ScriptChan for SendableWorkerScriptChan<T> {
impl ScriptChan for SendableWorkerScriptChan {
fn send(&self, msg: CommonScriptMsg) -> Result<(), ()> {
self.sender.send((self.worker.clone(), msg)).map_err(|_| ())
let msg = DedicatedWorkerScriptMsg::CommonWorker(self.worker.clone(), WorkerScriptMsg::Common(msg));
self.sender.send(msg).map_err(|_| ())
}
fn clone(&self) -> Box<ScriptChan + Send> {
@ -35,15 +41,16 @@ impl<T: JSTraceable + DomObject + 'static> ScriptChan for SendableWorkerScriptCh
/// worker event loop messages. While this SendableWorkerScriptChan is alive, the associated
/// Worker object will remain alive.
#[derive(Clone, JSTraceable)]
pub struct WorkerThreadWorkerChan<T: DomObject> {
pub sender: Sender<(Trusted<T>, WorkerScriptMsg)>,
pub worker: Trusted<T>,
pub struct WorkerThreadWorkerChan {
pub sender: Sender<DedicatedWorkerScriptMsg>,
pub worker: TrustedWorkerAddress,
}
impl<T: JSTraceable + DomObject + 'static> ScriptChan for WorkerThreadWorkerChan<T> {
impl ScriptChan for WorkerThreadWorkerChan {
fn send(&self, msg: CommonScriptMsg) -> Result<(), ()> {
let msg = DedicatedWorkerScriptMsg::CommonWorker(self.worker.clone(), WorkerScriptMsg::Common(msg));
self.sender
.send((self.worker.clone(), WorkerScriptMsg::Common(msg)))
.send(msg)
.map_err(|_| ())
}
@ -55,12 +62,100 @@ impl<T: JSTraceable + DomObject + 'static> ScriptChan for WorkerThreadWorkerChan
}
}
impl<T: DomObject> ScriptPort for Receiver<(Trusted<T>, WorkerScriptMsg)> {
impl ScriptPort for Receiver<DedicatedWorkerScriptMsg> {
fn recv(&self) -> Result<CommonScriptMsg, ()> {
match self.recv().map(|(_, msg)| msg) {
Ok(WorkerScriptMsg::Common(script_msg)) => Ok(script_msg),
Ok(WorkerScriptMsg::DOMMessage(_)) => panic!("unexpected worker event message!"),
Err(_) => Err(()),
let common_msg = match self.recv() {
Ok(DedicatedWorkerScriptMsg::CommonWorker(_worker, common_msg)) => common_msg,
Err(_) => return Err(()),
Ok(DedicatedWorkerScriptMsg::WakeUp) => panic!("unexpected worker event message!")
};
match common_msg {
WorkerScriptMsg::Common(script_msg) => Ok(script_msg),
WorkerScriptMsg::DOMMessage(_) => panic!("unexpected worker event message!"),
}
}
}
pub trait WorkerEventLoopMethods {
type TimerMsg: Send;
type WorkerMsg: QueuedTaskConversion + Send;
type Event;
fn timer_event_port(&self) -> &Receiver<Self::TimerMsg>;
fn task_queue(&self) -> &TaskQueue<Self::WorkerMsg>;
fn handle_event(&self, event: Self::Event);
fn handle_worker_post_event(&self, worker: &TrustedWorkerAddress) -> Option<AutoWorkerReset>;
fn from_worker_msg(&self, msg: Self::WorkerMsg) -> Self::Event;
fn from_timer_msg(&self, msg: Self::TimerMsg) -> Self::Event;
fn from_devtools_msg(&self, msg: DevtoolScriptControlMsg) -> Self::Event;
}
#[allow(unsafe_code)]
// https://html.spec.whatwg.org/multipage/#worker-event-loop
pub fn run_worker_event_loop<T, TimerMsg, WorkerMsg, Event>(worker_scope: &T,
worker: Option<&TrustedWorkerAddress>)
where
TimerMsg: Send,
WorkerMsg: QueuedTaskConversion + Send,
T: WorkerEventLoopMethods<TimerMsg = TimerMsg, WorkerMsg = WorkerMsg, Event = Event>
+ DerivedFrom<WorkerGlobalScope> + DerivedFrom<GlobalScope>
+ DomObject {
let scope = worker_scope.upcast::<WorkerGlobalScope>();
let timer_event_port = worker_scope.timer_event_port();
let devtools_port = scope.from_devtools_receiver();
let task_queue = worker_scope.task_queue();
let sel = Select::new();
let mut worker_handle = sel.handle(task_queue.select());
let mut timer_event_handle = sel.handle(timer_event_port);
let mut devtools_handle = sel.handle(devtools_port);
unsafe {
worker_handle.add();
timer_event_handle.add();
if scope.from_devtools_sender().is_some() {
devtools_handle.add();
}
}
let ret = sel.wait();
let event = {
if ret == worker_handle.id() {
task_queue.take_tasks();
worker_scope.from_worker_msg(task_queue.recv().unwrap())
} else if ret == timer_event_handle.id() {
worker_scope.from_timer_msg(timer_event_port.recv().unwrap())
} else if ret == devtools_handle.id() {
worker_scope.from_devtools_msg(devtools_port.recv().unwrap())
} else {
panic!("unexpected select result!")
}
};
let mut sequential = vec![];
sequential.push(event);
// https://html.spec.whatwg.org/multipage/#worker-event-loop
// Once the WorkerGlobalScope's closing flag is set to true,
// the event loop's task queues must discard any further tasks
// that would be added to them
// (tasks already on the queue are unaffected except where otherwise specified).
while !scope.is_closing() {
// Batch all events that are ready.
// The task queue will throttle non-priority tasks if necessary.
match task_queue.try_recv() {
Err(_) => match timer_event_port.try_recv() {
Err(_) => match devtools_port.try_recv() {
Err(_) => break,
Ok(ev) => sequential.push(worker_scope.from_devtools_msg(ev)),
},
Ok(ev) => sequential.push(worker_scope.from_timer_msg(ev)),
},
Ok(ev) => sequential.push(worker_scope.from_worker_msg(ev)),
}
}
// Step 3
for event in sequential {
worker_scope.handle_event(event);
// Step 6
let _ar = match worker {
Some(worker) => worker_scope.handle_worker_post_event(worker),
None => None
};
worker_scope.upcast::<GlobalScope>().perform_a_microtask_checkpoint();
}
}

180
components/script/dom/dedicatedworkerglobalscope.rs
View file

@ -6,6 +6,7 @@ use devtools;
use devtools_traits::DevtoolScriptControlMsg;
use dom::abstractworker::{SimpleWorkerErrorHandler, WorkerScriptMsg};
use dom::abstractworkerglobalscope::{SendableWorkerScriptChan, WorkerThreadWorkerChan};
use dom::abstractworkerglobalscope::{WorkerEventLoopMethods, run_worker_event_loop};
use dom::bindings::cell::DomRefCell;
use dom::bindings::codegen::Bindings::DedicatedWorkerGlobalScopeBinding;
use dom::bindings::codegen::Bindings::DedicatedWorkerGlobalScopeBinding::DedicatedWorkerGlobalScopeMethods;
@ -32,7 +33,7 @@ use js::rust::HandleValue;
use msg::constellation_msg::TopLevelBrowsingContextId;
use net_traits::{IpcSend, load_whole_resource};
use net_traits::request::{CredentialsMode, Destination, RequestInit};
use script_runtime::{CommonScriptMsg, ScriptChan, ScriptPort, new_rt_and_cx, Runtime};
use script_runtime::{CommonScriptMsg, ScriptChan, ScriptPort, ScriptThreadEventCategory, new_rt_and_cx, Runtime};
use script_runtime::ScriptThreadEventCategory::WorkerEvent;
use script_traits::{TimerEvent, TimerSource, WorkerGlobalScopeInit, WorkerScriptLoadOrigin};
use servo_rand::random;
@ -40,15 +41,16 @@ use servo_url::ServoUrl;
use std::mem::replace;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::mpsc::{Receiver, RecvError, Select, Sender, channel};
use std::sync::mpsc::{Receiver, Sender, channel};
use std::thread;
use style::thread_state::{self, ThreadState};
use task_queue::{QueuedTask, QueuedTaskConversion, TaskQueue};
/// Set the `worker` field of a related DedicatedWorkerGlobalScope object to a particular
/// value for the duration of this object's lifetime. This ensures that the related Worker
/// object only lives as long as necessary (ie. while events are being executed), while
/// providing a reference that can be cloned freely.
struct AutoWorkerReset<'a> {
pub struct AutoWorkerReset<'a> {
workerscope: &'a DedicatedWorkerGlobalScope,
old_worker: Option<TrustedWorkerAddress>,
}
@ -70,20 +72,81 @@ impl<'a> Drop for AutoWorkerReset<'a> {
}
}
enum MixedMessage {
FromWorker((TrustedWorkerAddress, WorkerScriptMsg)),
pub enum DedicatedWorkerScriptMsg {
/// Standard message from a worker.
CommonWorker(TrustedWorkerAddress, WorkerScriptMsg),
/// Wake-up call from the task queue.
WakeUp,
}
pub enum MixedMessage {
FromWorker(DedicatedWorkerScriptMsg),
FromScheduler((TrustedWorkerAddress, TimerEvent)),
FromDevtools(DevtoolScriptControlMsg)
}
impl QueuedTaskConversion for DedicatedWorkerScriptMsg {
fn task_category(&self) -> Option<&ScriptThreadEventCategory> {
let common_worker_msg = match self {
DedicatedWorkerScriptMsg::CommonWorker(_, common_worker_msg) => common_worker_msg,
_ => return None,
};
let script_msg = match common_worker_msg {
WorkerScriptMsg::Common(ref script_msg) => script_msg,
_ => return None,
};
let category = match script_msg {
CommonScriptMsg::Task(category, _boxed, _pipeline_id) => category,
_ => return None,
};
Some(category)
}
fn into_queued_task(self) -> Option<QueuedTask> {
let (worker, common_worker_msg) = match self {
DedicatedWorkerScriptMsg::CommonWorker(worker, common_worker_msg) => (worker, common_worker_msg),
_ => return None,
};
let script_msg = match common_worker_msg {
WorkerScriptMsg::Common(script_msg) => script_msg,
_ => return None,
};
let (category, boxed, pipeline_id) = match script_msg {
CommonScriptMsg::Task(category, boxed, pipeline_id) =>
(category, boxed, pipeline_id),
_ => return None,
};
Some((Some(worker), category, boxed, pipeline_id))
}
fn from_queued_task(queued_task: QueuedTask) -> Self {
let (worker, category, boxed, pipeline_id) = queued_task;
let script_msg = CommonScriptMsg::Task(category, boxed, pipeline_id);
DedicatedWorkerScriptMsg::CommonWorker(worker.unwrap(), WorkerScriptMsg::Common(script_msg))
}
fn wake_up_msg() -> Self {
DedicatedWorkerScriptMsg::WakeUp
}
fn is_wake_up(&self) -> bool {
match self {
DedicatedWorkerScriptMsg::WakeUp => true,
_ => false,
}
}
}
unsafe_no_jsmanaged_fields!(TaskQueue<DedicatedWorkerScriptMsg>);
// https://html.spec.whatwg.org/multipage/#dedicatedworkerglobalscope
#[dom_struct]
pub struct DedicatedWorkerGlobalScope {
workerglobalscope: WorkerGlobalScope,
#[ignore_malloc_size_of = "Defined in std"]
receiver: Receiver<(TrustedWorkerAddress, WorkerScriptMsg)>,
task_queue: TaskQueue<DedicatedWorkerScriptMsg>,
#[ignore_malloc_size_of = "Defined in std"]
own_sender: Sender<(TrustedWorkerAddress, WorkerScriptMsg)>,
own_sender: Sender<DedicatedWorkerScriptMsg>,
#[ignore_malloc_size_of = "Defined in std"]
timer_event_port: Receiver<(TrustedWorkerAddress, TimerEvent)>,
#[ignore_malloc_size_of = "Trusted<T> has unclear ownership like Dom<T>"]
@ -93,14 +156,49 @@ pub struct DedicatedWorkerGlobalScope {
parent_sender: Box<ScriptChan + Send>,
}
impl WorkerEventLoopMethods for DedicatedWorkerGlobalScope {
type TimerMsg = (TrustedWorkerAddress, TimerEvent);
type WorkerMsg = DedicatedWorkerScriptMsg;
type Event = MixedMessage;
fn timer_event_port(&self) -> &Receiver<(TrustedWorkerAddress, TimerEvent)> {
&self.timer_event_port
}
fn task_queue(&self) -> &TaskQueue<DedicatedWorkerScriptMsg> {
&self.task_queue
}
fn handle_event(&self, event: MixedMessage) {
self.handle_mixed_message(event);
}
fn handle_worker_post_event(&self, worker: &TrustedWorkerAddress) -> Option<AutoWorkerReset> {
let ar = AutoWorkerReset::new(&self, worker.clone());
Some(ar)
}
fn from_worker_msg(&self, msg: DedicatedWorkerScriptMsg) -> MixedMessage {
MixedMessage::FromWorker(msg)
}
fn from_timer_msg(&self, msg: (TrustedWorkerAddress, TimerEvent)) -> MixedMessage {
MixedMessage::FromScheduler(msg)
}
fn from_devtools_msg(&self, msg: DevtoolScriptControlMsg) -> MixedMessage {
MixedMessage::FromDevtools(msg)
}
}
impl DedicatedWorkerGlobalScope {
fn new_inherited(init: WorkerGlobalScopeInit,
worker_url: ServoUrl,
from_devtools_receiver: Receiver<DevtoolScriptControlMsg>,
runtime: Runtime,
parent_sender: Box<ScriptChan + Send>,
own_sender: Sender<(TrustedWorkerAddress, WorkerScriptMsg)>,
receiver: Receiver<(TrustedWorkerAddress, WorkerScriptMsg)>,
own_sender: Sender<DedicatedWorkerScriptMsg>,
receiver: Receiver<DedicatedWorkerScriptMsg>,
timer_event_chan: IpcSender<TimerEvent>,
timer_event_port: Receiver<(TrustedWorkerAddress, TimerEvent)>,
closing: Arc<AtomicBool>)
@ -112,7 +210,7 @@ impl DedicatedWorkerGlobalScope {
from_devtools_receiver,
timer_event_chan,
Some(closing)),
receiver: receiver,
task_queue: TaskQueue::new(receiver, own_sender.clone()),
own_sender: own_sender,
timer_event_port: timer_event_port,
parent_sender: parent_sender,
@ -126,8 +224,8 @@ impl DedicatedWorkerGlobalScope {
from_devtools_receiver: Receiver<DevtoolScriptControlMsg>,
runtime: Runtime,
parent_sender: Box<ScriptChan + Send>,
own_sender: Sender<(TrustedWorkerAddress, WorkerScriptMsg)>,
receiver: Receiver<(TrustedWorkerAddress, WorkerScriptMsg)>,
own_sender: Sender<DedicatedWorkerScriptMsg>,
receiver: Receiver<DedicatedWorkerScriptMsg>,
timer_event_chan: IpcSender<TimerEvent>,
timer_event_port: Receiver<(TrustedWorkerAddress, TimerEvent)>,
closing: Arc<AtomicBool>)
@ -151,13 +249,14 @@ impl DedicatedWorkerGlobalScope {
}
#[allow(unsafe_code)]
// https://html.spec.whatwg.org/multipage/#run-a-worker
pub fn run_worker_scope(init: WorkerGlobalScopeInit,
worker_url: ServoUrl,
from_devtools_receiver: IpcReceiver<DevtoolScriptControlMsg>,
worker: TrustedWorkerAddress,
parent_sender: Box<ScriptChan + Send>,
own_sender: Sender<(TrustedWorkerAddress, WorkerScriptMsg)>,
receiver: Receiver<(TrustedWorkerAddress, WorkerScriptMsg)>,
own_sender: Sender<DedicatedWorkerScriptMsg>,
receiver: Receiver<DedicatedWorkerScriptMsg>,
worker_load_origin: WorkerScriptLoadOrigin,
closing: Arc<AtomicBool>) {
let serialized_worker_url = worker_url.to_string();
@ -242,17 +341,11 @@ impl DedicatedWorkerGlobalScope {
let reporter_name = format!("dedicated-worker-reporter-{}", random::<u64>());
scope.upcast::<GlobalScope>().mem_profiler_chan().run_with_memory_reporting(|| {
// https://html.spec.whatwg.org/multipage/#event-loop-processing-model
// Step 1
while let Ok(event) = global.receive_event() {
if scope.is_closing() {
break;
}
// Step 3
global.handle_event(event);
// Step 6
let _ar = AutoWorkerReset::new(&global, worker.clone());
global.upcast::<GlobalScope>().perform_a_microtask_checkpoint();
// Step 29, Run the responsible event loop specified by inside settings until it is destroyed.
// The worker processing model remains on this step until the event loop is destroyed,
// which happens after the closing flag is set to true.
while !scope.is_closing() {
run_worker_event_loop(&*global, Some(&worker));
}
}, reporter_name, parent_sender, CommonScriptMsg::CollectReports);
}).expect("Thread spawning failed");
@ -274,36 +367,6 @@ impl DedicatedWorkerGlobalScope {
(chan, Box::new(rx))
}
#[allow(unsafe_code)]
fn receive_event(&self) -> Result<MixedMessage, RecvError> {
let scope = self.upcast::<WorkerGlobalScope>();
let worker_port = &self.receiver;
let timer_event_port = &self.timer_event_port;
let devtools_port = scope.from_devtools_receiver();
let sel = Select::new();
let mut worker_handle = sel.handle(worker_port);
let mut timer_event_handle = sel.handle(timer_event_port);
let mut devtools_handle = sel.handle(devtools_port);
unsafe {
worker_handle.add();
timer_event_handle.add();
if scope.from_devtools_sender().is_some() {
devtools_handle.add();
}
}
let ret = sel.wait();
if ret == worker_handle.id() {
Ok(MixedMessage::FromWorker(worker_port.recv()?))
} else if ret == timer_event_handle.id() {
Ok(MixedMessage::FromScheduler(timer_event_port.recv()?))
} else if ret == devtools_handle.id() {
Ok(MixedMessage::FromDevtools(devtools_port.recv()?))
} else {
panic!("unexpected select result!")
}
}
fn handle_script_event(&self, msg: WorkerScriptMsg) {
match msg {
WorkerScriptMsg::DOMMessage(data) => {
@ -321,8 +384,8 @@ impl DedicatedWorkerGlobalScope {
}
}
fn handle_event(&self, event: MixedMessage) {
match event {
fn handle_mixed_message(&self, msg: MixedMessage) {
match msg {
MixedMessage::FromDevtools(msg) => {
match msg {
DevtoolScriptControlMsg::EvaluateJS(_pipe_id, string, sender) =>
@ -346,10 +409,11 @@ impl DedicatedWorkerGlobalScope {
}
}
}
MixedMessage::FromWorker((linked_worker, msg)) => {
MixedMessage::FromWorker(DedicatedWorkerScriptMsg::CommonWorker(linked_worker, msg)) => {
let _ar = AutoWorkerReset::new(self, linked_worker);
self.handle_script_event(msg);
}
MixedMessage::FromWorker(DedicatedWorkerScriptMsg::WakeUp) => {},
}
}

151
components/script/dom/serviceworkerglobalscope.rs
View file

@ -5,17 +5,20 @@
use devtools;
use devtools_traits::DevtoolScriptControlMsg;
use dom::abstractworker::WorkerScriptMsg;
use dom::abstractworkerglobalscope::{WorkerEventLoopMethods, run_worker_event_loop};
use dom::bindings::codegen::Bindings::ServiceWorkerGlobalScopeBinding;
use dom::bindings::codegen::Bindings::ServiceWorkerGlobalScopeBinding::ServiceWorkerGlobalScopeMethods;
use dom::bindings::inheritance::Castable;
use dom::bindings::reflector::DomObject;
use dom::bindings::root::{DomRoot, RootCollection, ThreadLocalStackRoots};
use dom::bindings::str::DOMString;
use dom::dedicatedworkerglobalscope::AutoWorkerReset;
use dom::event::Event;
use dom::eventtarget::EventTarget;
use dom::extendableevent::ExtendableEvent;
use dom::extendablemessageevent::ExtendableMessageEvent;
use dom::globalscope::GlobalScope;
use dom::worker::TrustedWorkerAddress;
use dom::workerglobalscope::WorkerGlobalScope;
use dom_struct::dom_struct;
use ipc_channel::ipc::{self, IpcSender, IpcReceiver};
@ -24,22 +27,69 @@ use js::jsapi::{JSAutoCompartment, JSContext, JS_AddInterruptCallback};
use js::jsval::UndefinedValue;
use net_traits::{load_whole_resource, IpcSend, CustomResponseMediator};
use net_traits::request::{CredentialsMode, Destination, RequestInit};
use script_runtime::{CommonScriptMsg, ScriptChan, new_rt_and_cx, Runtime};
use script_runtime::{CommonScriptMsg, ScriptChan, ScriptThreadEventCategory, new_rt_and_cx, Runtime};
use script_traits::{TimerEvent, WorkerGlobalScopeInit, ScopeThings, ServiceWorkerMsg, WorkerScriptLoadOrigin};
use servo_config::prefs::PREFS;
use servo_rand::random;
use servo_url::ServoUrl;
use std::sync::mpsc::{Receiver, RecvError, Select, Sender, channel};
use std::sync::mpsc::{Receiver, Sender, channel};
use std::thread;
use std::time::Duration;
use style::thread_state::{self, ThreadState};
use task_queue::{QueuedTask, QueuedTaskConversion, TaskQueue};
/// Messages used to control service worker event loop
pub enum ServiceWorkerScriptMsg {
/// Message common to all workers
CommonWorker(WorkerScriptMsg),
// Message to request a custom response by the service worker
Response(CustomResponseMediator)
/// Message to request a custom response by the service worker
Response(CustomResponseMediator),
/// Wake-up call from the task queue.
WakeUp,
}
impl QueuedTaskConversion for ServiceWorkerScriptMsg {
fn task_category(&self) -> Option<&ScriptThreadEventCategory> {
let script_msg = match self {
ServiceWorkerScriptMsg::CommonWorker(WorkerScriptMsg::Common(script_msg)) => script_msg,
_ => return None,
};
let category = match script_msg {
CommonScriptMsg::Task(category, _boxed, _pipeline_id) => category,
_ => return None,
};
Some(&category)
}
fn into_queued_task(self) -> Option<QueuedTask> {
let script_msg = match self {
ServiceWorkerScriptMsg::CommonWorker(WorkerScriptMsg::Common(script_msg)) => script_msg,
_ => return None,
};
let (category, boxed, pipeline_id) = match script_msg {
CommonScriptMsg::Task(category, boxed, pipeline_id) =>
(category, boxed, pipeline_id),
_ => return None,
};
Some((None, category, boxed, pipeline_id))
}
fn from_queued_task(queued_task: QueuedTask) -> Self {
let (_worker, category, boxed, pipeline_id) = queued_task;
let script_msg = CommonScriptMsg::Task(category, boxed, pipeline_id);
ServiceWorkerScriptMsg::CommonWorker(WorkerScriptMsg::Common(script_msg))
}
fn wake_up_msg() -> Self {
ServiceWorkerScriptMsg::WakeUp
}
fn is_wake_up(&self) -> bool {
match self {
ServiceWorkerScriptMsg::WakeUp => true,
_ => false,
}
}
}
pub enum MixedMessage {
@ -67,11 +117,13 @@ impl ScriptChan for ServiceWorkerChan {
}
}
unsafe_no_jsmanaged_fields!(TaskQueue<ServiceWorkerScriptMsg>);
#[dom_struct]
pub struct ServiceWorkerGlobalScope {
workerglobalscope: WorkerGlobalScope,
#[ignore_malloc_size_of = "Defined in std"]
receiver: Receiver<ServiceWorkerScriptMsg>,
task_queue: TaskQueue<ServiceWorkerScriptMsg>,
#[ignore_malloc_size_of = "Defined in std"]
own_sender: Sender<ServiceWorkerScriptMsg>,
#[ignore_malloc_size_of = "Defined in std"]
@ -81,6 +133,40 @@ pub struct ServiceWorkerGlobalScope {
scope_url: ServoUrl,
}
impl WorkerEventLoopMethods for ServiceWorkerGlobalScope {
type TimerMsg = ();
type WorkerMsg = ServiceWorkerScriptMsg;
type Event = MixedMessage;
fn timer_event_port(&self) -> &Receiver<()> {
&self.timer_event_port
}
fn task_queue(&self) -> &TaskQueue<ServiceWorkerScriptMsg> {
&self.task_queue
}
fn handle_event(&self, event: MixedMessage) {
self.handle_mixed_message(event);
}
fn handle_worker_post_event(&self, _worker: &TrustedWorkerAddress) -> Option<AutoWorkerReset> {
None
}
fn from_worker_msg(&self, msg: ServiceWorkerScriptMsg) -> MixedMessage {
MixedMessage::FromServiceWorker(msg)
}
fn from_timer_msg(&self, msg: ()) -> MixedMessage {
MixedMessage::FromTimeoutThread(msg)
}
fn from_devtools_msg(&self, msg: DevtoolScriptControlMsg) -> MixedMessage {
MixedMessage::FromDevtools(msg)
}
}
impl ServiceWorkerGlobalScope {
fn new_inherited(init: WorkerGlobalScopeInit,
worker_url: ServoUrl,
@ -100,7 +186,7 @@ impl ServiceWorkerGlobalScope {
from_devtools_receiver,
timer_event_chan,
None),
receiver: receiver,
task_queue: TaskQueue::new(receiver, own_sender.clone()),
timer_event_port: timer_event_port,
own_sender: own_sender,
swmanager_sender: swmanager_sender,
@ -139,6 +225,7 @@ impl ServiceWorkerGlobalScope {
}
#[allow(unsafe_code)]
// https://html.spec.whatwg.org/multipage/#run-a-worker
pub fn run_serviceworker_scope(scope_things: ScopeThings,
own_sender: Sender<ServiceWorkerScriptMsg>,
receiver: Receiver<ServiceWorkerScriptMsg>,
@ -211,22 +298,18 @@ impl ServiceWorkerGlobalScope {
global.dispatch_activate();
let reporter_name = format!("service-worker-reporter-{}", random::<u64>());
scope.upcast::<GlobalScope>().mem_profiler_chan().run_with_memory_reporting(|| {
// https://html.spec.whatwg.org/multipage/#event-loop-processing-model
// Step 1
while let Ok(event) = global.receive_event() {
// Step 3
if !global.handle_event(event) {
break;
}
// Step 6
global.upcast::<GlobalScope>().perform_a_microtask_checkpoint();
// Step 29, Run the responsible event loop specified by inside settings until it is destroyed.
// The worker processing model remains on this step until the event loop is destroyed,
// which happens after the closing flag is set to true.
while !scope.is_closing() {
run_worker_event_loop(&*global, None);
}
}, reporter_name, scope.script_chan(), CommonScriptMsg::CollectReports);
}).expect("Thread spawning failed");
}
fn handle_event(&self, event: MixedMessage) -> bool {
match event {
fn handle_mixed_message(&self, msg: MixedMessage) -> bool {
match msg {
MixedMessage::FromDevtools(msg) => {
match msg {
DevtoolScriptControlMsg::EvaluateJS(_pipe_id, string, sender) =>
@ -271,38 +354,8 @@ impl ServiceWorkerGlobalScope {
// https://slightlyoff.github.io/ServiceWorker/spec/service_worker_1/index.html#fetch-event-section
self.upcast::<EventTarget>().fire_event(atom!("fetch"));
let _ = mediator.response_chan.send(None);
}
}
}
#[allow(unsafe_code)]
fn receive_event(&self) -> Result<MixedMessage, RecvError> {
let scope = self.upcast::<WorkerGlobalScope>();
let worker_port = &self.receiver;
let devtools_port = scope.from_devtools_receiver();
let timer_event_port = &self.timer_event_port;
let sel = Select::new();
let mut worker_handle = sel.handle(worker_port);
let mut devtools_handle = sel.handle(devtools_port);
let mut timer_port_handle = sel.handle(timer_event_port);
unsafe {
worker_handle.add();
if scope.from_devtools_sender().is_some() {
devtools_handle.add();
}
timer_port_handle.add();
}
let ret = sel.wait();
if ret == worker_handle.id() {
Ok(MixedMessage::FromServiceWorker(worker_port.recv()?))
}else if ret == devtools_handle.id() {
Ok(MixedMessage::FromDevtools(devtools_port.recv()?))
} else if ret == timer_port_handle.id() {
Ok(MixedMessage::FromTimeoutThread(timer_event_port.recv()?))
} else {
panic!("unexpected select result!")
},
WakeUp => {},
}
}

10
components/script/dom/worker.rs
View file

@ -14,7 +14,7 @@ use dom::bindings::reflector::{DomObject, reflect_dom_object};
use dom::bindings::root::DomRoot;
use dom::bindings::str::DOMString;
use dom::bindings::structuredclone::StructuredCloneData;
use dom::dedicatedworkerglobalscope::DedicatedWorkerGlobalScope;
use dom::dedicatedworkerglobalscope::{DedicatedWorkerGlobalScope, DedicatedWorkerScriptMsg};
use dom::eventtarget::EventTarget;
use dom::globalscope::GlobalScope;
use dom::messageevent::MessageEvent;
@ -40,14 +40,14 @@ pub struct Worker {
#[ignore_malloc_size_of = "Defined in std"]
/// Sender to the Receiver associated with the DedicatedWorkerGlobalScope
/// this Worker created.
sender: Sender<(TrustedWorkerAddress, WorkerScriptMsg)>,
sender: Sender<DedicatedWorkerScriptMsg>,
#[ignore_malloc_size_of = "Arc"]
closing: Arc<AtomicBool>,
terminated: Cell<bool>,
}
impl Worker {
fn new_inherited(sender: Sender<(TrustedWorkerAddress, WorkerScriptMsg)>,
fn new_inherited(sender: Sender<DedicatedWorkerScriptMsg>,
closing: Arc<AtomicBool>) -> Worker {
Worker {
eventtarget: EventTarget::new_inherited(),
@ -58,7 +58,7 @@ impl Worker {
}
pub fn new(global: &GlobalScope,
sender: Sender<(TrustedWorkerAddress, WorkerScriptMsg)>,
sender: Sender<DedicatedWorkerScriptMsg>,
closing: Arc<AtomicBool>) -> DomRoot<Worker> {
reflect_dom_object(Box::new(Worker::new_inherited(sender, closing)),
global,
@ -148,7 +148,7 @@ impl WorkerMethods for Worker {
// NOTE: step 9 of https://html.spec.whatwg.org/multipage/#dom-messageport-postmessage
// indicates that a nonexistent communication channel should result in a silent error.
let _ = self.sender.send((address, WorkerScriptMsg::DOMMessage(data)));
let _ = self.sender.send(DedicatedWorkerScriptMsg::CommonWorker(address, WorkerScriptMsg::DOMMessage(data)));
Ok(())
}

2
components/script/dom/workerglobalscope.rs
View file

@ -407,8 +407,6 @@ impl WorkerGlobalScope {
reports_chan.send(reports);
},
}
// FIXME(jdm): Should we do a microtask checkpoint here?
}
pub fn handle_fire_timer(&self, timer_id: TimerEventId) {

2
components/script/lib.rs
View file

@ -5,6 +5,7 @@
#![cfg_attr(feature = "unstable", feature(core_intrinsics))]
#![cfg_attr(feature = "unstable", feature(on_unimplemented))]
#![feature(const_fn)]
#![feature(drain_filter)]
#![feature(mpsc_select)]
#![feature(plugin)]
#![feature(try_from)]
@ -123,6 +124,7 @@ pub mod script_thread;
mod serviceworker_manager;
mod serviceworkerjob;
mod stylesheet_loader;
mod task_queue;
mod task_source;
pub mod test;
pub mod textinput;

67
components/script/script_thread.rs
View file

@ -116,6 +116,7 @@ use std::sync::Arc;
use std::sync::mpsc::{Receiver, Select, Sender, channel};
use std::thread;
use style::thread_state::{self, ThreadState};
use task_queue::{QueuedTask, QueuedTaskConversion, TaskQueue};
use task_source::dom_manipulation::DOMManipulationTaskSource;
use task_source::file_reading::FileReadingTaskSource;
use task_source::history_traversal::HistoryTraversalTaskSource;
@ -242,6 +243,52 @@ pub enum MainThreadScriptMsg {
},
/// Dispatches a job queue.
DispatchJobQueue { scope_url: ServoUrl },
/// Wake-up call from the task queue.
WakeUp,
}
impl QueuedTaskConversion for MainThreadScriptMsg {
fn task_category(&self) -> Option<&ScriptThreadEventCategory> {
let script_msg = match self {
MainThreadScriptMsg::Common(script_msg) => script_msg,
_ => return None,
};
let category = match script_msg {
CommonScriptMsg::Task(category, _boxed, _pipeline_id) => category,
_ => return None,
};
Some(&category)
}
fn into_queued_task(self) -> Option<QueuedTask> {
let script_msg = match self {
MainThreadScriptMsg::Common(script_msg) => script_msg,
_ => return None,
};
let (category, boxed, pipeline_id) = match script_msg {
CommonScriptMsg::Task(category, boxed, pipeline_id) =>
(category, boxed, pipeline_id),
_ => return None,
};
Some((None, category, boxed, pipeline_id))
}
fn from_queued_task(queued_task: QueuedTask) -> Self {
let (_worker, category, boxed, pipeline_id) = queued_task;
let script_msg = CommonScriptMsg::Task(category, boxed, pipeline_id);
MainThreadScriptMsg::Common(script_msg)
}
fn wake_up_msg() -> Self {
MainThreadScriptMsg::WakeUp
}
fn is_wake_up(&self) -> bool {
match self {
MainThreadScriptMsg::WakeUp => true,
_ => false,
}
}
}
impl OpaqueSender<CommonScriptMsg> for Box<ScriptChan + Send> {
@ -398,6 +445,8 @@ impl<'a> Iterator for DocumentsIter<'a> {
type IncompleteParserContexts = Vec<(PipelineId, ParserContext)>;
unsafe_no_jsmanaged_fields!(RefCell<IncompleteParserContexts>);
unsafe_no_jsmanaged_fields!(TaskQueue<MainThreadScriptMsg>);
#[derive(JSTraceable)]
// ScriptThread instances are rooted on creation, so this is okay
#[allow(unrooted_must_root)]
@ -423,8 +472,9 @@ pub struct ScriptThread {
/// A handle to the bluetooth thread.
bluetooth_thread: IpcSender<BluetoothRequest>,
/// The port on which the script thread receives messages (load URL, exit, etc.)
port: Receiver<MainThreadScriptMsg>,
/// A queue of tasks to be executed in this script-thread.
task_queue: TaskQueue<MainThreadScriptMsg>,
/// A channel to hand out to script thread-based entities that need to be able to enqueue
/// events in the event queue.
chan: MainThreadScriptChan,
@ -856,6 +906,8 @@ impl ScriptThread {
let (image_cache_channel, image_cache_port) = channel();
let task_queue = TaskQueue::new(port, chan.clone());
ScriptThread {
documents: DomRefCell::new(Documents::new()),
window_proxies: DomRefCell::new(HashMap::new()),
@ -871,7 +923,7 @@ impl ScriptThread {
resource_threads: state.resource_threads,
bluetooth_thread: state.bluetooth_thread,
port: port,
task_queue,
chan: MainThreadScriptChan(chan.clone()),
dom_manipulation_task_sender: chan.clone(),
@ -968,7 +1020,7 @@ impl ScriptThread {
debug!("Waiting for event.");
let mut event = {
let sel = Select::new();
let mut script_port = sel.handle(&self.port);
let mut script_port = sel.handle(self.task_queue.select());
let mut control_port = sel.handle(&self.control_port);
let mut timer_event_port = sel.handle(&self.timer_event_port);
let mut devtools_port = sel.handle(&self.devtools_port);
@ -984,7 +1036,8 @@ impl ScriptThread {
}
let ret = sel.wait();
if ret == script_port.id() {
FromScript(self.port.recv().unwrap())
self.task_queue.take_tasks();
FromScript(self.task_queue.recv().unwrap())
} else if ret == control_port.id() {
FromConstellation(self.control_port.recv().unwrap())
} else if ret == timer_event_port.id() {
@ -1077,7 +1130,7 @@ impl ScriptThread {
// and check for more resize events. If there are no events pending, we'll move
// on and execute the sequential non-resize events we've seen.
match self.control_port.try_recv() {
Err(_) => match self.port.try_recv() {
Err(_) => match self.task_queue.try_recv() {
Err(_) => match self.timer_event_port.try_recv() {
Err(_) => match self.devtools_port.try_recv() {
Err(_) => match self.image_cache_port.try_recv() {
@ -1233,6 +1286,7 @@ impl ScriptThread {
MainThreadScriptMsg::WorkletLoaded(pipeline_id) => Some(pipeline_id),
MainThreadScriptMsg::RegisterPaintWorklet { pipeline_id, .. } => Some(pipeline_id),
MainThreadScriptMsg::DispatchJobQueue { .. } => None,
MainThreadScriptMsg::WakeUp => None,
}
},
MixedMessage::FromImageCache((pipeline_id, _)) => Some(pipeline_id),
@ -1404,6 +1458,7 @@ impl ScriptThread {
MainThreadScriptMsg::DispatchJobQueue { scope_url } => {
self.job_queue_map.run_job(scope_url, self)
}
MainThreadScriptMsg::WakeUp => {},
}
}

165
components/script/task_queue.rs Normal file
View file

@ -0,0 +1,165 @@
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
//! Machinery for [task-queue](https://html.spec.whatwg.org/multipage/#task-queue).
use dom::bindings::cell::DomRefCell;
use dom::worker::TrustedWorkerAddress;
use msg::constellation_msg::PipelineId;
use script_runtime::ScriptThreadEventCategory;
use std::cell::Cell;
use std::collections::{HashMap, VecDeque};
use std::default::Default;
use std::sync::mpsc::{Receiver, Sender};
use task::TaskBox;
use task_source::TaskSourceName;
pub type QueuedTask = (Option<TrustedWorkerAddress>, ScriptThreadEventCategory, Box<TaskBox>, Option<PipelineId>);
/// Defining the operations used to convert from a msg T to a QueuedTask.
pub trait QueuedTaskConversion {
fn task_category(&self) -> Option<&ScriptThreadEventCategory>;
fn into_queued_task(self) -> Option<QueuedTask>;
fn from_queued_task(queued_task: QueuedTask) -> Self;
fn wake_up_msg() -> Self;
fn is_wake_up(&self) -> bool;
}
pub struct TaskQueue<T> {
/// The original port on which the task-sources send tasks as messages.
port: Receiver<T>,
/// A sender to ensure the port doesn't block on select while there are throttled tasks.
wake_up_sender: Sender<T>,
/// A queue from which the event-loop can drain tasks.
msg_queue: DomRefCell<VecDeque<T>>,
/// A "business" counter, reset for each iteration of the event-loop
taken_task_counter: Cell<u64>,
/// Tasks that will be throttled for as long as we are "busy".
throttled: DomRefCell<HashMap<TaskSourceName, VecDeque<QueuedTask>>>
}
impl<T: QueuedTaskConversion> TaskQueue<T> {
pub fn new(port: Receiver<T>, wake_up_sender: Sender<T>) -> TaskQueue<T> {
TaskQueue {
port,
wake_up_sender,
msg_queue: DomRefCell::new(VecDeque::new()),
taken_task_counter: Default::default(),
throttled: Default::default(),
}
}
/// Process incoming tasks, immediately sending priority ones downstream,
/// and categorizing potential throttles.
fn process_incoming_tasks(&self) {
let mut non_throttled: Vec<T> = self.port
.try_iter()
.filter(|msg| !msg.is_wake_up())
.collect();
let to_be_throttled: Vec<T> = non_throttled.drain_filter(|msg|{
let category = match msg.task_category() {
Some(category) => category,
None => return false,
};
match category {
ScriptThreadEventCategory::PerformanceTimelineTask => return true,
_ => {
// A task that will not be throttled, start counting "business"
self.taken_task_counter.set(self.taken_task_counter.get() + 1);
return false
},
}
}).collect();
for msg in non_throttled {
// Immediately send non-throttled tasks for processing.
let _ = self.msg_queue.borrow_mut().push_back(msg);
}
for msg in to_be_throttled {
// Categorize tasks per task queue.
let (worker, category, boxed, pipeline_id) = match msg.into_queued_task() {
Some((worker, category, boxed, pipeline_id)) => (worker, category, boxed, pipeline_id),
None => unreachable!("A message to be throttled should always be convertible into a queued task"),
};
// FIXME: Add the task-source name directly to CommonScriptMsg::Task.
let task_source = match category {
ScriptThreadEventCategory::PerformanceTimelineTask => TaskSourceName::PerformanceTimeline,
_ => unreachable!(),
};
let mut throttled_tasks = self.throttled.borrow_mut();
throttled_tasks
.entry(task_source)
.or_insert(VecDeque::new())
.push_back((worker, category, boxed, pipeline_id));
}
}
/// Reset the queue for a new iteration of the event-loop,
/// returning the port about whose readiness we want to be notified.
pub fn select(&self) -> &Receiver<T> {
// This is a new iteration of the event-loop, so we reset the "business" counter.
self.taken_task_counter.set(0);
// We want to be notified when the script-port is ready to receive.
// Hence that's the one we need to include in the select.
&self.port
}
/// Take a message from the front of the queue, without waiting if empty.
pub fn recv(&self) -> Result<T, ()> {
self.msg_queue.borrow_mut().pop_front().ok_or(())
}
/// Same as recv.
pub fn try_recv(&self) -> Result<T, ()> {
self.recv()
}
/// Drain the queue for the current iteration of the event-loop.
/// Holding-back throttles above a given high-water mark.
pub fn take_tasks(&self) {
// High-watermark: once reached, throttled tasks will be held-back.
const PER_ITERATION_MAX: u64 = 5;
// Always first check for new tasks, but don't reset 'taken_task_counter'.
self.process_incoming_tasks();
let mut throttled = self.throttled.borrow_mut();
let mut throttled_length: usize = throttled.values().map(|queue| queue.len()).sum();
let task_source_names = TaskSourceName::all();
let mut task_source_cycler = task_source_names.iter().cycle();
// "being busy", is defined as having more than x tasks for this loop's iteration.
// As long as we're not busy, and there are throttled tasks left:
loop {
let max_reached = self.taken_task_counter.get() > PER_ITERATION_MAX;
let none_left = throttled_length == 0;
match (max_reached, none_left) {
(_, true) => break,
(true, false) => {
// We have reached the high-watermark for this iteration of the event-loop,
// yet also have throttled messages left in the queue.
// Ensure the select wakes up in the next iteration of the event-loop
let _ = self.wake_up_sender.send(T::wake_up_msg());
break;
},
(false, false) => {
// Cycle through non-priority task sources, taking one throttled task from each.
let task_source = task_source_cycler.next().unwrap();
let throttled_queue = match throttled.get_mut(&task_source) {
Some(queue) => queue,
None => continue,
};
let queued_task = match throttled_queue.pop_front() {
Some(queued_task) => queued_task,
None => continue,
};
let msg = T::from_queued_task(queued_task);
let _ = self.msg_queue.borrow_mut().push_back(msg);
self.taken_task_counter.set(self.taken_task_counter.get() + 1);
throttled_length = throttled_length - 1;
},
}
}
}
}

14
tests/wpt/mozilla/meta/MANIFEST.json
View file

@ -20674,6 +20674,16 @@
{}
]
],
"mozilla/task_queue_throttling.any.js": [
[
"/_mozilla/mozilla/task_queue_throttling.any.html",
{}
],
[
"/_mozilla/mozilla/task_queue_throttling.any.worker.html",
{}
]
],
"mozilla/textcontent.html": [
[
"/_mozilla/mozilla/textcontent.html",
@ -32667,6 +32677,10 @@
"321132b7e7e21d542ce86e4322db04aba3e9d11e",
"support"
],
"mozilla/task_queue_throttling.any.js": [
"bbf464f384d55a44db6948d6cc7ab5f54509b800",
"testharness"
],
"mozilla/test.jpg": [
"110b3d6b666d35a117ca8c8ada32bf1002943044",
"support"

45
tests/wpt/mozilla/tests/mozilla/task_queue_throttling.any.js Normal file
View file

@ -0,0 +1,45 @@
// META: title=Throttling the performance timeline task queue.
async_test(function(t) {
var counter = 0;
function perf_observer(list, observer) {
// The timeline event should be throttled,
// while the event-loop is busy,
// and only handled after at least 6 other events,
// across several iterations of the event-loop.
assert_true(counter > 6)
}
var observer2 = new PerformanceObserver(t.step_func_done(perf_observer));
observer2.observe({entryTypes: ["mark"]});
for (var i = 0; i < 4; i++) {
var reader = new FileReader();
reader.onload = function() {
counter++;
};
var blob = new Blob();
reader.readAsText(blob);
}
var reader = new FileReader();
reader.onload = function() {
counter++;
// In a subsequent iteration of the event-loop,
// start reading another 5 blobs
for (var i = 0; i < 5; i++) {
var reader = new FileReader();
reader.onload = function() {
counter++;
};
var blob = new Blob();
reader.readAsText(blob);
}
};
var blob = new Blob();
reader.readAsText(blob);
// We've started reading 5 blobs in this iteration of the event-loop.
// Do this in the current iteration of the event-loop.
performance.mark("start");
});