script/tasks/
task_queue.rs

1/* This Source Code Form is subject to the terms of the Mozilla Public
2 * License, v. 2.0. If a copy of the MPL was not distributed with this
3 * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
4
5//! Machinery for [task-queue](https://html.spec.whatwg.org/multipage/#task-queue).
6
7use std::collections::VecDeque;
8use std::default::Default;
9
10use crossbeam_channel::{self, Receiver};
11use rustc_hash::{FxHashMap, FxHashSet};
12use script_bindings::cell::DomRefCell;
13use servo_base::id::PipelineId;
14
15use crate::dom::worker::TrustedWorkerAddress;
16use crate::runtime::script_runtime::ScriptThreadEventCategory;
17use crate::tasks::task::TaskBox;
18use crate::tasks::task_source::TaskSourceName;
19
20#[derive(MallocSizeOf)]
21pub(crate) struct QueuedTask {
22    pub(crate) worker: Option<TrustedWorkerAddress>,
23    pub(crate) event_category: ScriptThreadEventCategory,
24    #[ignore_malloc_size_of = "TaskBox is difficult"]
25    pub(crate) task: Box<dyn TaskBox>,
26    pub(crate) pipeline_id: Option<PipelineId>,
27    pub(crate) task_source: TaskSourceName,
28}
29
30/// Defining the operations used to convert from a msg T to a QueuedTask.
31pub(crate) trait QueuedTaskConversion {
32    fn pipeline_id(&self) -> Option<PipelineId>;
33    fn into_queued_task(self) -> Option<QueuedTask>;
34    fn from_queued_task(queued_task: QueuedTask) -> Self;
35    fn inactive_msg() -> Self;
36    fn wake_up_msg() -> Self;
37    fn is_wake_up(&self) -> bool;
38}
39
40#[derive(MallocSizeOf)]
41pub(crate) struct TaskQueue<T> {
42    /// The original port on which the task-sources send tasks as messages.
43    port: Receiver<T>,
44    /// A queue from which the event-loop can drain tasks.
45    msg_queue: DomRefCell<VecDeque<T>>,
46    /// Tasks for not fully-active documents.
47    inactive: DomRefCell<FxHashMap<PipelineId, VecDeque<QueuedTask>>>,
48    /// List of pipelines that have been owned and closed by the parent script thread.
49    closed_pipelines: DomRefCell<FxHashSet<PipelineId>>,
50}
51
52impl<T: QueuedTaskConversion> TaskQueue<T> {
53    pub(crate) fn new(port: Receiver<T>) -> TaskQueue<T> {
54        TaskQueue {
55            port,
56            msg_queue: DomRefCell::new(VecDeque::new()),
57            inactive: Default::default(),
58            closed_pipelines: Default::default(),
59        }
60    }
61
62    /// Remove any tasks for an exiting pipeline. This only clears tasks that
63    /// were stored from a previously inactive pipeline, as others will be drained
64    /// the next time the queue is processed.
65    pub(crate) fn remove_tasks_for_exiting_pipeline(&self, pipeline_id: &PipelineId) {
66        self.inactive.borrow_mut().remove(pipeline_id);
67        self.closed_pipelines.borrow_mut().insert(*pipeline_id);
68    }
69
70    pub(crate) fn is_pipeline_closed(&self, pipeline_id: &PipelineId) -> bool {
71        self.closed_pipelines.borrow().contains(pipeline_id)
72    }
73
74    /// Release previously held-back tasks for documents that are now fully-active.
75    /// <https://html.spec.whatwg.org/multipage/#event-loop-processing-model:fully-active>
76    fn release_tasks_for_fully_active_documents(
77        &self,
78        fully_active: &FxHashSet<PipelineId>,
79    ) -> Vec<T> {
80        self.inactive
81            .borrow_mut()
82            .iter_mut()
83            .filter(|(pipeline_id, _)| fully_active.contains(pipeline_id))
84            .flat_map(|(_, inactive_queue)| {
85                inactive_queue
86                    .drain(0..)
87                    .map(|queued_task| T::from_queued_task(queued_task))
88            })
89            .collect()
90    }
91
92    /// Hold back tasks for currently not fully-active documents.
93    /// <https://html.spec.whatwg.org/multipage/#event-loop-processing-model:fully-active>
94    fn store_task_for_inactive_pipeline(&self, msg: T, pipeline_id: &PipelineId) {
95        if !self.is_pipeline_closed(pipeline_id) {
96            let mut inactive = self.inactive.borrow_mut();
97            let inactive_queue = inactive.entry(*pipeline_id).or_default();
98            inactive_queue.push_back(
99                msg.into_queued_task()
100                    .expect("Incoming messages should always be convertible into queued tasks"),
101            );
102        }
103
104        let mut msg_queue = self.msg_queue.borrow_mut();
105        if msg_queue.is_empty() {
106            // Ensure there is at least one message.
107            // Otherwise if the just stored inactive message (or dropped message)
108            // was the first and last of this iteration,
109            // it will result in a spurious wake-up of the event-loop.
110            msg_queue.push_back(T::inactive_msg());
111        }
112    }
113
114    /// Drain the queue for the current iteration of the event-loop.
115    pub(crate) fn take_tasks(&self, first_message: T, fully_active: &FxHashSet<PipelineId>) {
116        // 1. Make any previously stored task from now fully-active document available.
117        let mut incoming = self.release_tasks_for_fully_active_documents(fully_active);
118
119        // 2. Process the first message (artifact of the fact that select always returns a message).
120        if !first_message.is_wake_up() {
121            incoming.push(first_message);
122        }
123
124        // 3. Process any other incoming message.
125        while let Ok(message) = self.port.try_recv() {
126            if !message.is_wake_up() {
127                incoming.push(message);
128            }
129        }
130
131        for msg in incoming {
132            if let Some(pipeline_id) = msg.pipeline_id() &&
133                !fully_active.contains(&pipeline_id)
134            {
135                self.store_task_for_inactive_pipeline(msg, &pipeline_id);
136                continue;
137            }
138            self.msg_queue.borrow_mut().push_back(msg);
139        }
140    }
141
142    /// Reset the queue for a new iteration of the event-loop,
143    /// returning the port about whose readiness we want to be notified.
144    pub(crate) fn select(&self) -> &crossbeam_channel::Receiver<T> {
145        // We want to be notified when the script-port is ready to receive.
146        // Hence that's the one we need to include in the select.
147        &self.port
148    }
149
150    /// Take a message from the front of the queue, without waiting if empty.
151    pub(crate) fn recv(&self) -> Result<T, ()> {
152        self.msg_queue.borrow_mut().pop_front().ok_or(())
153    }
154
155    /// Take all tasks again and then run `recv()`.
156    pub(crate) fn take_tasks_and_recv(
157        &self,
158        fully_active: &FxHashSet<PipelineId>,
159    ) -> Result<T, ()> {
160        self.take_tasks(T::wake_up_msg(), fully_active);
161        self.recv()
162    }
163}
script/tasks/task_queue.rs

script/tasks/
task_queue.rs
