Auto merge of #21645 - AgustinCB:add-websocket-task-queue, r=jdm

Add Websocket task source

According to the doc: https://html.spec.whatwg.org/multipage/web-sockets.html#network

The task source for all tasks queued in the websocket section are the
websocket task source, so this commit also updates those references to
use the appropriate one.

Also, while working on this, I made a typo here: 5dd6e21c2e/components/script/dom/window.rs (L191)

Setting the name incorrectly. The error, however, was this:

```bash
error[E0412]: cannot find type `WebsocketEventTaskSource` in this scope
     --> components/script/dom/window.rs:171:1
       |
171 | #[dom_struct]
       | ^^^^^^^^^^^^^ did you mean `WebsocketTaskSource`?
```

Which isn't useful at all. Not sure if it's a rustc problem or something related with htis code base, but I thought it was worth mentioning.

---
- [x] `./mach build -d` does not report any errors
- [x] `./mach test-tidy` does not report any errors
- [x] These changes fix #21590
- [ ] There are tests for these changes OR
- [x] These changes do not require tests because they don't include new behavior and existing tests should cover this code.

<!-- 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/21645)
<!-- Reviewable:end -->

components/script/dom/globalscope.rs

@@ -59,6 +59,7 @@ use task_source::file_reading::FileReadingTaskSource;
 use task_source::networking::NetworkingTaskSource;
 use task_source::performance_timeline::PerformanceTimelineTaskSource;
 use task_source::remote_event::RemoteEventTaskSource;
+use task_source::websocket::WebsocketTaskSource;
 use time::{Timespec, get_time};
 use timers::{IsInterval, OneshotTimerCallback, OneshotTimerHandle};
 use timers::{OneshotTimers, TimerCallback};
@@ -430,6 +431,18 @@ impl GlobalScope {
         unreachable!();
     }
 
+    /// `ScriptChan` to send messages to the websocket task source of
+    /// this global scope.
+    pub fn websocket_task_source(&self) -> WebsocketTaskSource {
+        if let Some(window) = self.downcast::<Window>() {
+            return window.websocket_task_source();
+        }
+        if let Some(worker) = self.downcast::<WorkerGlobalScope>() {
+            return worker.websocket_task_source();
+        }
+        unreachable!();
+    }
+
     /// Evaluate JS code on this global scope.
     pub fn evaluate_js_on_global_with_result(
             &self, code: &str, rval: MutableHandleValue) -> bool {

components/script/dom/websocket.rs

@@ -39,8 +39,8 @@ use std::cell::Cell;
 use std::ptr;
 use std::thread;
 use task::{TaskOnce, TaskCanceller};
-use task_source::{TaskSource, TaskSourceName};
-use task_source::networking::NetworkingTaskSource;
+use task_source::TaskSource;
+use task_source::websocket::WebsocketTaskSource;
 
 #[derive(Clone, Copy, Debug, JSTraceable, MallocSizeOf, PartialEq)]
 enum WebSocketRequestState {
@@ -70,7 +70,7 @@ mod close_code {
 
 pub fn close_the_websocket_connection(
     address: Trusted<WebSocket>,
-    task_source: &NetworkingTaskSource,
+    task_source: &WebsocketTaskSource,
     canceller: &TaskCanceller,
     code: Option<u16>,
     reason: String,
@@ -86,7 +86,7 @@ pub fn close_the_websocket_connection(
 
 pub fn fail_the_websocket_connection(
     address: Trusted<WebSocket>,
-    task_source: &NetworkingTaskSource,
+    task_source: &WebsocketTaskSource,
     canceller: &TaskCanceller,
 ) {
     let close_task = CloseTask {
@@ -199,11 +199,8 @@ impl WebSocket {
         };
         let _ = global.core_resource_thread().send(CoreResourceMsg::Fetch(request, channels));
 
-        // TODO: use a dedicated task source,
-        // https://html.spec.whatwg.org/multipage/#websocket-task-source
-        // When making the switch, also update the task_canceller call.
-        let task_source = global.networking_task_source();
-        let canceller = global.task_canceller(TaskSourceName::Networking);
+        let task_source = global.websocket_task_source();
+        let canceller = global.task_canceller(WebsocketTaskSource::NAME);
         thread::spawn(move || {
             while let Ok(event) = dom_event_receiver.recv() {
                 match event {
@@ -273,7 +270,7 @@ impl WebSocket {
                     WebSocketEvent,
                     task,
                     Some(pipeline_id),
-                    TaskSourceName::Networking,
+                    WebsocketTaskSource::NAME,
                 ))
                 .unwrap();
         }
@@ -407,10 +404,10 @@ impl WebSocketMethods for WebSocket {
                 // TODO: use a dedicated task source,
                 // https://html.spec.whatwg.org/multipage/#websocket-task-source
                 // When making the switch, also update the task_canceller call.
-                let task_source = self.global().networking_task_source();
+                let task_source = self.global().websocket_task_source();
                 fail_the_websocket_connection(address,
                                               &task_source,
-                                              &self.global().task_canceller(TaskSourceName::Networking));
+                                              &self.global().task_canceller(WebsocketTaskSource::NAME));
             }
             WebSocketRequestState::Open => {
                 self.ready_state.set(WebSocketRequestState::Closing);

components/script/dom/window.rs

@@ -128,6 +128,7 @@ use task_source::networking::NetworkingTaskSource;
 use task_source::performance_timeline::PerformanceTimelineTaskSource;
 use task_source::remote_event::RemoteEventTaskSource;
 use task_source::user_interaction::UserInteractionTaskSource;
+use task_source::websocket::WebsocketTaskSource;
 use time;
 use timers::{IsInterval, TimerCallback};
 use url::Position;
@@ -186,6 +187,8 @@ pub struct Window {
     performance_timeline_task_source: PerformanceTimelineTaskSource,
     #[ignore_malloc_size_of = "task sources are hard"]
     remote_event_task_source: RemoteEventTaskSource,
+    #[ignore_malloc_size_of = "task sources are hard"]
+    websocket_task_source: WebsocketTaskSource,
     navigator: MutNullableDom<Navigator>,
     #[ignore_malloc_size_of = "Arc"]
     image_cache: Arc<ImageCache>,
@@ -376,6 +379,10 @@ impl Window {
         self.remote_event_task_source.clone()
     }
 
+    pub fn websocket_task_source(&self) -> WebsocketTaskSource {
+        self.websocket_task_source.clone()
+    }
+
     pub fn main_thread_script_chan(&self) -> &Sender<MainThreadScriptMsg> {
         &self.script_chan.0
     }
@@ -1904,6 +1911,7 @@ impl Window {
         file_reading_task_source: FileReadingTaskSource,
         performance_timeline_task_source: PerformanceTimelineTaskSource,
         remote_event_task_source: RemoteEventTaskSource,
+        websocket_task_source: WebsocketTaskSource,
         image_cache_chan: Sender<ImageCacheMsg>,
         image_cache: Arc<ImageCache>,
         resource_threads: ResourceThreads,
@@ -1957,6 +1965,7 @@ impl Window {
             file_reading_task_source,
             performance_timeline_task_source,
             remote_event_task_source,
+            websocket_task_source,
             image_cache_chan,
             image_cache,
             navigator: Default::default(),

components/script/dom/workerglobalscope.rs

@@ -47,6 +47,7 @@ use task_source::file_reading::FileReadingTaskSource;
 use task_source::networking::NetworkingTaskSource;
 use task_source::performance_timeline::PerformanceTimelineTaskSource;
 use task_source::remote_event::RemoteEventTaskSource;
+use task_source::websocket::WebsocketTaskSource;
 use time::precise_time_ns;
 use timers::{IsInterval, TimerCallback};
 
@@ -386,6 +387,10 @@ impl WorkerGlobalScope {
         RemoteEventTaskSource(self.script_chan(), self.pipeline_id())
     }
 
+    pub fn websocket_task_source(&self) -> WebsocketTaskSource {
+        WebsocketTaskSource(self.script_chan(), self.pipeline_id())
+    }
+
     pub fn new_script_pair(&self) -> (Box<ScriptChan + Send>, Box<ScriptPort + Send>) {
         let dedicated = self.downcast::<DedicatedWorkerGlobalScope>();
         if let Some(dedicated) = dedicated {

components/script/script_thread.rs

@@ -125,6 +125,7 @@ use task_source::networking::NetworkingTaskSource;
 use task_source::performance_timeline::PerformanceTimelineTaskSource;
 use task_source::remote_event::RemoteEventTaskSource;
 use task_source::user_interaction::UserInteractionTaskSource;
+use task_source::websocket::WebsocketTaskSource;
 use time::{get_time, precise_time_ns, Tm};
 use url::Position;
 use url::percent_encoding::percent_decode;
@@ -1896,6 +1897,10 @@ impl ScriptThread {
         RemoteEventTaskSource(self.remote_event_task_sender.clone(), pipeline_id)
     }
 
+    pub fn websocket_task_source(&self, pipeline_id: PipelineId) -> WebsocketTaskSource {
+        WebsocketTaskSource(self.remote_event_task_sender.clone(), pipeline_id)
+    }
+
     /// Handles a request for the window title.
     fn handle_get_title_msg(&self, pipeline_id: PipelineId) {
         let document = match { self.documents.borrow().find_document(pipeline_id) } {
@@ -2220,6 +2225,7 @@ impl ScriptThread {
             self.file_reading_task_source(incomplete.pipeline_id),
             self.performance_timeline_task_source(incomplete.pipeline_id).clone(),
             self.remote_event_task_source(incomplete.pipeline_id),
+            self.websocket_task_source(incomplete.pipeline_id),
             self.image_cache_channel.clone(),
             self.image_cache.clone(),
             self.resource_threads.clone(),

components/script/task_source/mod.rs

@@ -10,6 +10,7 @@ pub mod networking;
 pub mod performance_timeline;
 pub mod remote_event;
 pub mod user_interaction;
+pub mod websocket;
 
 use dom::globalscope::GlobalScope;
 use enum_iterator::IntoEnumIterator;
@@ -28,7 +29,8 @@ pub enum TaskSourceName {
     Networking,
     PerformanceTimeline,
     UserInteraction,
-    RemoteEvent
+    RemoteEvent,
+    Websocket,
 }
 
 impl TaskSourceName {

components/script/task_source/websocket.rs

@@ -0,0 +1,37 @@
+/* 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/. */
+
+use msg::constellation_msg::PipelineId;
+use script_runtime::{CommonScriptMsg, ScriptChan, ScriptThreadEventCategory};
+use task::{TaskCanceller, TaskOnce};
+use task_source::{TaskSource, TaskSourceName};
+
+#[derive(JSTraceable)]
+pub struct WebsocketTaskSource(pub Box<ScriptChan + Send + 'static>, pub PipelineId);
+
+impl Clone for WebsocketTaskSource {
+    fn clone(&self) -> WebsocketTaskSource {
+        WebsocketTaskSource(self.0.clone(), self.1.clone())
+    }
+}
+
+impl TaskSource for WebsocketTaskSource {
+    const NAME: TaskSourceName = TaskSourceName::Websocket;
+
+    fn queue_with_canceller<T>(
+        &self,
+        task: T,
+        canceller: &TaskCanceller,
+    ) -> Result<(), ()>
+    where
+        T: TaskOnce + 'static,
+    {
+        self.0.send(CommonScriptMsg::Task(
+            ScriptThreadEventCategory::NetworkEvent,
+            Box::new(canceller.wrap_task(task)),
+            Some(self.1),
+            WebsocketTaskSource::NAME,
+        ))
+    }
+}