Mirrors/servo
1
0
Fork 0
mirror of https://github.com/servo/servo.git synced 2025-08-03 04:30:10 +01:00
Code Issues Projects Releases Packages Wiki Activity

add a control chan to workers, use to signal shutdown

Browse source
This commit is contained in:
Gregory Terzian 2020-05-24 16:36:10 +08:00
parent ed688fe2c1
commit 947fa8bbb7
6 changed files with 165 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -82,12 +82,15 @@ impl ScriptPort for Receiver<DedicatedWorkerScriptMsg> {
pub trait WorkerEventLoopMethods {
type WorkerMsg: QueuedTaskConversion + Send;
type ControlMsg;
type Event;
fn task_queue(&self) -> &TaskQueue<Self::WorkerMsg>;
fn handle_event(&self, event: Self::Event);
fn handle_event(&self, event: Self::Event) -> bool;
fn handle_worker_post_event(&self, worker: &TrustedWorkerAddress) -> Option<AutoWorkerReset>;
fn from_control_msg(&self, msg: Self::ControlMsg) -> Self::Event;
fn from_worker_msg(&self, msg: Self::WorkerMsg) -> Self::Event;
fn from_devtools_msg(&self, msg: DevtoolScriptControlMsg) -> Self::Event;
fn control_receiver(&self) -> &Receiver<Self::ControlMsg>;
}
// https://html.spec.whatwg.org/multipage/#worker-event-loop
@ -108,6 +111,7 @@ pub fn run_worker_event_loop<T, WorkerMsg, Event>(
};
let task_queue = worker_scope.task_queue();
let event = select! {
recv(worker_scope.control_receiver()) -> msg => worker_scope.from_control_msg(msg.unwrap()),
recv(task_queue.select()) -> msg => {
task_queue.take_tasks(msg.unwrap());
worker_scope.from_worker_msg(task_queue.recv().unwrap())
@ -136,7 +140,10 @@ pub fn run_worker_event_loop<T, WorkerMsg, Event>(
}
// Step 3
for event in sequential {
worker_scope.handle_event(event);
if !worker_scope.handle_event(event) {
// Shutdown
return;
}
// Step 6
let _ar = match worker {
Some(worker) => worker_scope.handle_worker_post_event(worker),

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

@ -86,6 +86,12 @@ impl<'a> Drop for AutoWorkerReset<'a> {
}
}
/// Messages sent from the owning global.
pub enum DedicatedWorkerControlMsg {
/// Shutdown the worker.
Exit,
}
pub enum DedicatedWorkerScriptMsg {
/// Standard message from a worker.
CommonWorker(TrustedWorkerAddress, WorkerScriptMsg),
@ -96,6 +102,7 @@ pub enum DedicatedWorkerScriptMsg {
pub enum MixedMessage {
FromWorker(DedicatedWorkerScriptMsg),
FromDevtools(DevtoolScriptControlMsg),
FromControl(DedicatedWorkerControlMsg),
}
impl QueuedTaskConversion for DedicatedWorkerScriptMsg {
@ -183,18 +190,23 @@ pub struct DedicatedWorkerGlobalScope {
#[ignore_malloc_size_of = "Arc"]
image_cache: Arc<dyn ImageCache>,
browsing_context: Option<BrowsingContextId>,
/// A receiver of control messages,
/// currently only used to signal shutdown.
#[ignore_malloc_size_of = "Channels are hard"]
control_receiver: Receiver<DedicatedWorkerControlMsg>,
}
impl WorkerEventLoopMethods for DedicatedWorkerGlobalScope {
type WorkerMsg = DedicatedWorkerScriptMsg;
type ControlMsg = DedicatedWorkerControlMsg;
type Event = MixedMessage;
fn task_queue(&self) -> &TaskQueue<DedicatedWorkerScriptMsg> {
&self.task_queue
}
fn handle_event(&self, event: MixedMessage) {
self.handle_mixed_message(event);
fn handle_event(&self, event: MixedMessage) -> bool {
self.handle_mixed_message(event)
}
fn handle_worker_post_event(&self, worker: &TrustedWorkerAddress) -> Option<AutoWorkerReset> {
@ -202,6 +214,10 @@ impl WorkerEventLoopMethods for DedicatedWorkerGlobalScope {
Some(ar)
}
fn from_control_msg(&self, msg: DedicatedWorkerControlMsg) -> MixedMessage {
MixedMessage::FromControl(msg)
}
fn from_worker_msg(&self, msg: DedicatedWorkerScriptMsg) -> MixedMessage {
MixedMessage::FromWorker(msg)
}
@ -209,6 +225,10 @@ impl WorkerEventLoopMethods for DedicatedWorkerGlobalScope {
fn from_devtools_msg(&self, msg: DevtoolScriptControlMsg) -> MixedMessage {
MixedMessage::FromDevtools(msg)
}
fn control_receiver(&self) -> &Receiver<DedicatedWorkerControlMsg> {
&self.control_receiver
}
}
impl DedicatedWorkerGlobalScope {
@ -226,6 +246,7 @@ impl DedicatedWorkerGlobalScope {
image_cache: Arc<dyn ImageCache>,
browsing_context: Option<BrowsingContextId>,
gpu_id_hub: Arc<Mutex<Identities>>,
control_receiver: Receiver<DedicatedWorkerControlMsg>,
) -> DedicatedWorkerGlobalScope {
DedicatedWorkerGlobalScope {
workerglobalscope: WorkerGlobalScope::new_inherited(
@ -244,6 +265,7 @@ impl DedicatedWorkerGlobalScope {
worker: DomRefCell::new(None),
image_cache: image_cache,
browsing_context,
control_receiver,
}
}
@ -262,6 +284,7 @@ impl DedicatedWorkerGlobalScope {
image_cache: Arc<dyn ImageCache>,
browsing_context: Option<BrowsingContextId>,
gpu_id_hub: Arc<Mutex<Identities>>,
control_receiver: Receiver<DedicatedWorkerControlMsg>,
) -> DomRoot<DedicatedWorkerGlobalScope> {
let cx = runtime.cx();
let scope = Box::new(DedicatedWorkerGlobalScope::new_inherited(
@ -278,6 +301,7 @@ impl DedicatedWorkerGlobalScope {
image_cache,
browsing_context,
gpu_id_hub,
control_receiver,
));
unsafe { DedicatedWorkerGlobalScopeBinding::Wrap(SafeJSContext::from_ptr(cx), scope) }
}
@ -299,6 +323,7 @@ impl DedicatedWorkerGlobalScope {
image_cache: Arc<dyn ImageCache>,
browsing_context: Option<BrowsingContextId>,
gpu_id_hub: Arc<Mutex<Identities>>,
control_receiver: Receiver<DedicatedWorkerControlMsg>,
) -> JoinHandle<()> {
let serialized_worker_url = worker_url.to_string();
let name = format!("WebWorker for {}", serialized_worker_url);
@ -370,6 +395,7 @@ impl DedicatedWorkerGlobalScope {
image_cache,
browsing_context,
gpu_id_hub,
control_receiver,
);
// FIXME(njn): workers currently don't have a unique ID suitable for using in reporter
// registration (#6631), so we instead use a random number and cross our fingers.
@ -485,7 +511,7 @@ impl DedicatedWorkerGlobalScope {
}
}
fn handle_mixed_message(&self, msg: MixedMessage) {
fn handle_mixed_message(&self, msg: MixedMessage) -> bool {
// FIXME(#26324): `self.worker` is None in devtools messages.
match msg {
MixedMessage::FromDevtools(msg) => match msg {
@ -505,7 +531,11 @@ impl DedicatedWorkerGlobalScope {
self.handle_script_event(msg);
},
MixedMessage::FromWorker(DedicatedWorkerScriptMsg::WakeUp) => {},
MixedMessage::FromControl(DedicatedWorkerControlMsg::Exit) => {
return false;
},
}
true
}
// https://html.spec.whatwg.org/multipage/#runtime-script-errors-2

34
components/script/dom/globalscope.rs
View file

@ -26,7 +26,9 @@ use crate::dom::bindings::weakref::{DOMTracker, WeakRef};
use crate::dom::blob::Blob;
use crate::dom::broadcastchannel::BroadcastChannel;
use crate::dom::crypto::Crypto;
use crate::dom::dedicatedworkerglobalscope::DedicatedWorkerGlobalScope;
use crate::dom::dedicatedworkerglobalscope::{
DedicatedWorkerControlMsg, DedicatedWorkerGlobalScope,
};
use crate::dom::errorevent::ErrorEvent;
use crate::dom::event::{Event, EventBubbles, EventCancelable, EventStatus};
use crate::dom::eventsource::EventSource;
@ -65,6 +67,7 @@ use crate::task_source::TaskSourceName;
use crate::timers::{IsInterval, OneshotTimerCallback, OneshotTimerHandle};
use crate::timers::{OneshotTimers, TimerCallback};
use content_security_policy::CspList;
use crossbeam_channel::Sender;
use devtools_traits::{PageError, ScriptToDevtoolsControlMsg};
use dom_struct::dom_struct;
use embedder_traits::EmbedderMsg;
@ -118,8 +121,13 @@ use uuid::Uuid;
#[derive(JSTraceable)]
pub struct AutoCloseWorker {
/// https://html.spec.whatwg.org/multipage/#dom-workerglobalscope-closing
closing: Arc<AtomicBool>,
/// A handle to join on the worker thread.
join_handle: Option<JoinHandle<()>>,
/// A sender of control messages,
/// currently only used to signal shutdown.
control_sender: Sender<DedicatedWorkerControlMsg>,
}
impl Drop for AutoCloseWorker {
@ -128,13 +136,25 @@ impl Drop for AutoCloseWorker {
// Step 1.
self.closing.store(true, Ordering::SeqCst);
if self
.control_sender
.send(DedicatedWorkerControlMsg::Exit)
.is_err()
{
warn!("Couldn't send an exit message to a dedicated worker.");
}
// TODO: step 2 and 3.
// Step 4 is unnecessary since we don't use actual ports for dedicated workers.
self.join_handle
if self
.join_handle
.take()
.expect("No handle to join on worker.")
.join()
.expect("Couldn't join on worker thread.");
.is_err()
{
warn!("Failed to join on dedicated worker thread.");
}
}
}
@ -1808,12 +1828,18 @@ impl GlobalScope {
&self.permission_state_invocation_results
}
pub fn track_worker(&self, closing: Arc<AtomicBool>, join_handle: JoinHandle<()>) {
pub fn track_worker(
&self,
closing: Arc<AtomicBool>,
join_handle: JoinHandle<()>,
control_sender: Sender<DedicatedWorkerControlMsg>,
) {
self.list_auto_close_worker
.borrow_mut()
.push(AutoCloseWorker {
closing,
join_handle: Some(join_handle),
control_sender: control_sender,
});
}

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

@ -116,9 +116,16 @@ impl QueuedTaskConversion for ServiceWorkerScriptMsg {
}
}
/// Messages sent from the owning registration.
pub enum ServiceWorkerControlMsg {
/// Shutdown.
Exit,
}
pub enum MixedMessage {
FromServiceWorker(ServiceWorkerScriptMsg),
FromDevtools(DevtoolScriptControlMsg),
FromControl(ServiceWorkerControlMsg),
}
#[derive(Clone, JSTraceable)]
@ -165,24 +172,34 @@ pub struct ServiceWorkerGlobalScope {
swmanager_sender: IpcSender<ServiceWorkerMsg>,
scope_url: ServoUrl,
/// A receiver of control messages,
/// currently only used to signal shutdown.
#[ignore_malloc_size_of = "Channels are hard"]
control_receiver: Receiver<ServiceWorkerControlMsg>,
}
impl WorkerEventLoopMethods for ServiceWorkerGlobalScope {
type WorkerMsg = ServiceWorkerScriptMsg;
type ControlMsg = ServiceWorkerControlMsg;
type Event = MixedMessage;
fn task_queue(&self) -> &TaskQueue<ServiceWorkerScriptMsg> {
&self.task_queue
}
fn handle_event(&self, event: MixedMessage) {
self.handle_mixed_message(event);
fn handle_event(&self, event: MixedMessage) -> bool {
self.handle_mixed_message(event)
}
fn handle_worker_post_event(&self, _worker: &TrustedWorkerAddress) -> Option<AutoWorkerReset> {
None
}
fn from_control_msg(&self, msg: ServiceWorkerControlMsg) -> MixedMessage {
MixedMessage::FromControl(msg)
}
fn from_worker_msg(&self, msg: ServiceWorkerScriptMsg) -> MixedMessage {
MixedMessage::FromServiceWorker(msg)
}
@ -190,6 +207,10 @@ impl WorkerEventLoopMethods for ServiceWorkerGlobalScope {
fn from_devtools_msg(&self, msg: DevtoolScriptControlMsg) -> MixedMessage {
MixedMessage::FromDevtools(msg)
}
fn control_receiver(&self) -> &Receiver<ServiceWorkerControlMsg> {
&self.control_receiver
}
}
impl ServiceWorkerGlobalScope {
@ -203,6 +224,7 @@ impl ServiceWorkerGlobalScope {
time_out_port: Receiver<Instant>,
swmanager_sender: IpcSender<ServiceWorkerMsg>,
scope_url: ServoUrl,
control_receiver: Receiver<ServiceWorkerControlMsg>,
) -> ServiceWorkerGlobalScope {
ServiceWorkerGlobalScope {
workerglobalscope: WorkerGlobalScope::new_inherited(
@ -220,6 +242,7 @@ impl ServiceWorkerGlobalScope {
time_out_port,
swmanager_sender: swmanager_sender,
scope_url: scope_url,
control_receiver,
}
}
@ -234,6 +257,7 @@ impl ServiceWorkerGlobalScope {
time_out_port: Receiver<Instant>,
swmanager_sender: IpcSender<ServiceWorkerMsg>,
scope_url: ServoUrl,
control_receiver: Receiver<ServiceWorkerControlMsg>,
) -> DomRoot<ServiceWorkerGlobalScope> {
let cx = runtime.cx();
let scope = Box::new(ServiceWorkerGlobalScope::new_inherited(
@ -246,6 +270,7 @@ impl ServiceWorkerGlobalScope {
time_out_port,
swmanager_sender,
scope_url,
control_receiver,
));
unsafe { ServiceWorkerGlobalScopeBinding::Wrap(SafeJSContext::from_ptr(cx), scope) }
}
@ -259,6 +284,7 @@ impl ServiceWorkerGlobalScope {
devtools_receiver: IpcReceiver<DevtoolScriptControlMsg>,
swmanager_sender: IpcSender<ServiceWorkerMsg>,
scope_url: ServoUrl,
control_receiver: Receiver<ServiceWorkerControlMsg>,
) -> JoinHandle<()> {
let ScopeThings {
script_url,
@ -315,6 +341,7 @@ impl ServiceWorkerGlobalScope {
time_out_port,
swmanager_sender,
scope_url,
control_receiver,
);
let (_url, source) =
@ -366,23 +393,23 @@ impl ServiceWorkerGlobalScope {
fn handle_mixed_message(&self, msg: MixedMessage) -> bool {
match msg {
MixedMessage::FromDevtools(msg) => {
match msg {
DevtoolScriptControlMsg::EvaluateJS(_pipe_id, string, sender) => {
devtools::handle_evaluate_js(self.upcast(), string, sender)
},
DevtoolScriptControlMsg::WantsLiveNotifications(_pipe_id, bool_val) => {
devtools::handle_wants_live_notifications(self.upcast(), bool_val)
},
_ => debug!("got an unusable devtools control message inside the worker!"),
}
true
MixedMessage::FromDevtools(msg) => match msg {
DevtoolScriptControlMsg::EvaluateJS(_pipe_id, string, sender) => {
devtools::handle_evaluate_js(self.upcast(), string, sender)
},
DevtoolScriptControlMsg::WantsLiveNotifications(_pipe_id, bool_val) => {
devtools::handle_wants_live_notifications(self.upcast(), bool_val)
},
_ => debug!("got an unusable devtools control message inside the worker!"),
},
MixedMessage::FromServiceWorker(msg) => {
self.handle_script_event(msg);
true
},
MixedMessage::FromControl(ServiceWorkerControlMsg::Exit) => {
return false;
},
}
true
}
fn has_timed_out(&self) -> bool {

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

@ -124,6 +124,8 @@ impl Worker {
let init = prepare_workerscope_init(global, Some(devtools_sender), Some(worker_id));
let (control_sender, control_receiver) = unbounded();
let join_handle = DedicatedWorkerGlobalScope::run_worker_scope(
init,
worker_url,
@ -139,9 +141,10 @@ impl Worker {
global.image_cache(),
browsing_context,
global.wgpu_id_hub(),
control_receiver,
);
global.track_worker(closing, join_handle);
global.track_worker(closing, join_handle, control_sender);
Ok(worker)
}

55
components/script/serviceworker_manager.rs
View file

@ -8,7 +8,9 @@
//! active_workers map
use crate::dom::abstractworker::WorkerScriptMsg;
use crate::dom::serviceworkerglobalscope::{ServiceWorkerGlobalScope, ServiceWorkerScriptMsg};
use crate::dom::serviceworkerglobalscope::{
ServiceWorkerControlMsg, ServiceWorkerGlobalScope, ServiceWorkerScriptMsg,
};
use crate::dom::serviceworkerregistration::longest_prefix_match;
use crossbeam_channel::{unbounded, Receiver, RecvError, Sender};
use ipc_channel::ipc::{self, IpcSender};
@ -80,12 +82,27 @@ enum RegistrationUpdateTarget {
impl Drop for ServiceWorkerRegistration {
/// <https://html.spec.whatwg.org/multipage/#terminate-a-worker>
fn drop(&mut self) {
// Drop the channel to signal shutdown.
if self
.control_sender
.take()
.expect("No control sender to worker thread.")
.send(ServiceWorkerControlMsg::Exit)
.is_err()
{
warn!("Failed to send exit message to service worker scope.");
}
// TODO: Step 1, 2 and 3.
self.join_handle
if self
.join_handle
.take()
.expect("No handle to join on worker.")
.join()
.expect("Couldn't join on worker thread.");
.is_err()
{
warn!("Failed to join on service worker thread.");
}
}
}
@ -99,6 +116,9 @@ struct ServiceWorkerRegistration {
waiting_worker: Option<ServiceWorker>,
/// https://w3c.github.io/ServiceWorker/#dfn-installing-worker
installing_worker: Option<ServiceWorker>,
/// A channel to send control message to the worker,
/// currently only used to signal shutdown.
control_sender: Option<Sender<ServiceWorkerControlMsg>>,
/// A handle to join on the worker thread.
join_handle: Option<JoinHandle<()>>,
}
@ -111,12 +131,20 @@ impl ServiceWorkerRegistration {
waiting_worker: None,
installing_worker: None,
join_handle: None,
control_sender: None,
}
}
fn note_worker_thread(&mut self, join_handle: JoinHandle<()>) {
fn note_worker_thread(
&mut self,
join_handle: JoinHandle<()>,
control_sender: Sender<ServiceWorkerControlMsg>,
) {
assert!(self.join_handle.is_none());
self.join_handle = Some(join_handle);
assert!(self.control_sender.is_none());
self.control_sender = Some(control_sender);
}
/// <https://w3c.github.io/ServiceWorker/#get-newest-worker>
@ -203,6 +231,10 @@ impl ServiceWorkerManager {
Message::FromResource(msg) => self.handle_message_from_resource(msg),
};
if !should_continue {
for registration in self.registrations.drain() {
// Signal shut-down, and join on the thread.
drop(registration);
}
break;
}
}
@ -346,10 +378,11 @@ impl ServiceWorkerManager {
// Very roughly steps 5 to 18.
// TODO: implement all steps precisely.
let (new_worker, join_handle) =
let (new_worker, join_handle, control_sender) =
update_serviceworker(self.own_sender.clone(), job.scope_url.clone(), scope_things);
registration.note_worker_thread(join_handle);
// Since we've just started the worker thread, ensure we can shut it down later.
registration.note_worker_thread(join_handle, control_sender);
// Step 19, run Install.
@ -385,11 +418,17 @@ fn update_serviceworker(
own_sender: IpcSender<ServiceWorkerMsg>,
scope_url: ServoUrl,
scope_things: ScopeThings,
) -> (ServiceWorker, JoinHandle<()>) {
) -> (
ServiceWorker,
JoinHandle<()>,
Sender<ServiceWorkerControlMsg>,
) {
let (sender, receiver) = unbounded();
let (_devtools_sender, devtools_receiver) = ipc::channel().unwrap();
let worker_id = ServiceWorkerId::new();
let (control_sender, control_receiver) = unbounded();
let join_handle = ServiceWorkerGlobalScope::run_serviceworker_scope(
scope_things.clone(),
sender.clone(),
@ -397,11 +436,13 @@ fn update_serviceworker(
devtools_receiver,
own_sender,
scope_url.clone(),
control_receiver,
);
(
ServiceWorker::new(scope_things.script_url, sender, worker_id),
join_handle,
control_sender,
)
}