Auto merge of #21325 - gterzian:crossbeam_integration, r=SimonSapin,jdm

Replace mpsc with crossbeam-channel Follow up on https://github.com/servo/servo/pull/19515 --- Selecting over multiple channels in `std::sync::mpsc` is not stable and likely never will be: https://github.com/rust-lang/rust/issues/27800#issuecomment-260136777 > It seems the only thing keeping `mpsc_select` around is Servo. crossbeam-channel is designed specifically to replace `std::sync::mpsc` and fix many of its shortcomings: https://github.com/stjepang/rfcs-crossbeam/blob/channel/text/2017-11-09-channel.md This is to be landed together with https://github.com/servo/ipc-channel/pull/183.  --- This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/servo/servo/21325)
2025-08-05 21:50:18 +01:00 · 2018-09-12 13:33:45 -04:00 · 2018-09-12 13:33:45 -04:00 · 910cc23a6e
commit 910cc23a6e
parent 9a83ab6297 2a996fbc8f
93 changed files with 590 additions and 385 deletions
--- a/components/script/dom/abstractworkerglobalscope.rs
+++ b/components/script/dom/abstractworkerglobalscope.rs
@ -11,7 +11,7 @@ use dom::globalscope::GlobalScope;
 use dom::worker::TrustedWorkerAddress;
 use dom::workerglobalscope::WorkerGlobalScope;
 use script_runtime::{ScriptChan, CommonScriptMsg, ScriptPort};
-use std::sync::mpsc::{Receiver, Select, Sender};
+use servo_channel::{Receiver, Sender};
 use task_queue::{QueuedTaskConversion, TaskQueue};

 /// A ScriptChan that can be cloned freely and will silently send a TrustedWorkerAddress with
@ -65,9 +65,9 @@ impl ScriptChan for WorkerThreadWorkerChan {
 impl ScriptPort for Receiver<DedicatedWorkerScriptMsg> {
    fn recv(&self) -> Result<CommonScriptMsg, ()> {
        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!")
+            Some(DedicatedWorkerScriptMsg::CommonWorker(_worker, common_msg)) => common_msg,
+            None => return Err(()),
+            Some(DedicatedWorkerScriptMsg::WakeUp) => panic!("unexpected worker event message!")
        };
        match common_msg {
            WorkerScriptMsg::Common(script_msg) => Ok(script_msg),
@ -89,7 +89,6 @@ pub trait WorkerEventLoopMethods {
    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>)
@ -101,31 +100,18 @@ where
    + DomObject {
    let scope = worker_scope.upcast::<WorkerGlobalScope>();
    let timer_event_port = worker_scope.timer_event_port();
-    let devtools_port = scope.from_devtools_receiver();
+    let devtools_port = match scope.from_devtools_sender() {
+        Some(_) => Some(scope.from_devtools_receiver().select()),
+        None => None,
+    };
    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();
+    let event = select! {
+        recv(task_queue.select(), msg) => {
+            task_queue.take_tasks(msg.unwrap());
            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!")
-        }
+        },
+        recv(timer_event_port.select(), msg) => worker_scope.from_timer_msg(msg.unwrap()),
+        recv(devtools_port, msg) => worker_scope.from_devtools_msg(msg.unwrap()),
    };
    let mut sequential = vec![];
    sequential.push(event);
@ -138,14 +124,14 @@ where
        // 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)),
+            None => match timer_event_port.try_recv() {
+                None => match devtools_port.and_then(|port| port.try_recv()) {
+                    None => break,
+                    Some(ev) => sequential.push(worker_scope.from_devtools_msg(ev)),
                },
-                Ok(ev) => sequential.push(worker_scope.from_timer_msg(ev)),
+                Some(ev) => sequential.push(worker_scope.from_timer_msg(ev)),
            },
-            Ok(ev) => sequential.push(worker_scope.from_worker_msg(ev)),
+            Some(ev) => sequential.push(worker_scope.from_worker_msg(ev)),
        }
    }
    // Step 3
--- a/components/script/dom/bindings/trace.rs
+++ b/components/script/dom/bindings/trace.rs
@ -88,6 +88,7 @@ use selectors::matching::ElementSelectorFlags;
 use serde::{Deserialize, Serialize};
 use servo_arc::Arc as ServoArc;
 use servo_atoms::Atom;
+use servo_channel::{Receiver, Sender};
 use servo_media::Backend;
 use servo_media::audio::buffer_source_node::AudioBuffer;
 use servo_media::audio::context::AudioContext;
@ -104,7 +105,6 @@ use std::path::PathBuf;
 use std::rc::Rc;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicUsize};
-use std::sync::mpsc::{Receiver, Sender};
 use std::time::{SystemTime, Instant};
 use style::attr::{AttrIdentifier, AttrValue, LengthOrPercentageOrAuto};
 use style::context::QuirksMode;
--- a/components/script/dom/dedicatedworkerglobalscope.rs
+++ b/components/script/dom/dedicatedworkerglobalscope.rs
@ -36,12 +36,12 @@ use net_traits::request::{CredentialsMode, Destination, RequestInit};
 use script_runtime::{CommonScriptMsg, ScriptChan, ScriptPort, new_rt_and_cx, Runtime};
 use script_runtime::ScriptThreadEventCategory::WorkerEvent;
 use script_traits::{TimerEvent, TimerSource, WorkerGlobalScopeInit, WorkerScriptLoadOrigin};
+use servo_channel::{channel, route_ipc_receiver_to_new_servo_sender, Sender, Receiver};
 use servo_rand::random;
 use servo_url::ServoUrl;
 use std::mem::replace;
 use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
-use std::sync::mpsc::{Receiver, Sender, channel};
 use std::thread;
 use style::thread_state::{self, ThreadState};
 use task_queue::{QueuedTask, QueuedTaskConversion, TaskQueue};
@ -313,7 +313,7 @@ impl DedicatedWorkerGlobalScope {
            let runtime = unsafe { new_rt_and_cx() };

            let (devtools_mpsc_chan, devtools_mpsc_port) = channel();
-            ROUTER.route_ipc_receiver_to_mpsc_sender(from_devtools_receiver, devtools_mpsc_chan);
+            route_ipc_receiver_to_new_servo_sender(from_devtools_receiver, devtools_mpsc_chan);

            let (timer_tx, timer_rx) = channel();
            let (timer_ipc_chan, timer_ipc_port) = ipc::channel().unwrap();
--- a/components/script/dom/paintworkletglobalscope.rs
+++ b/components/script/dom/paintworkletglobalscope.rs
@ -49,6 +49,8 @@ use profile_traits::ipc;
 use script_traits::{DrawAPaintImageResult, PaintWorkletError};
 use script_traits::Painter;
 use servo_atoms::Atom;
+use servo_channel::{channel, Sender};
+use servo_channel::base_channel;
 use servo_config::prefs::PREFS;
 use servo_url::ServoUrl;
 use std::cell::Cell;
@ -58,8 +60,6 @@ use std::ptr::null_mut;
 use std::rc::Rc;
 use std::sync::Arc;
 use std::sync::Mutex;
-use std::sync::mpsc;
-use std::sync::mpsc::Sender;
 use std::thread;
 use std::time::Duration;
 use style_traits::CSSPixel;
@ -350,7 +350,7 @@ impl PaintWorkletGlobalScope {
                                  arguments: Vec<String>)
                                  -> Result<DrawAPaintImageResult, PaintWorkletError> {
                let name = self.name.clone();
-                let (sender, receiver) = mpsc::channel();
+                let (sender, receiver) = channel();
                let task = PaintWorkletTask::DrawAPaintImage(name,
                                                             size,
                                                             device_pixel_ratio,
@ -364,9 +364,12 @@ impl PaintWorkletGlobalScope {
                                   .as_u64()
                                   .unwrap_or(10u64);

-                let timeout_duration = Duration::from_millis(timeout);
-                receiver.recv_timeout(timeout_duration)
-                        .map_err(|e| PaintWorkletError::from(e))
+                select! {
+                    recv(base_channel::after(Duration::from_millis(timeout))) => {
+                        Err(PaintWorkletError::Timeout)
+                    }
+                    recv(receiver.select(), msg) => msg.ok_or(PaintWorkletError::Timeout)
+                }
            }
        }
        Box::new(WorkletPainter {
--- a/components/script/dom/serviceworkerglobalscope.rs
+++ b/components/script/dom/serviceworkerglobalscope.rs
@ -22,17 +22,16 @@ use dom::worker::TrustedWorkerAddress;
 use dom::workerglobalscope::WorkerGlobalScope;
 use dom_struct::dom_struct;
 use ipc_channel::ipc::{self, IpcSender, IpcReceiver};
-use ipc_channel::router::ROUTER;
 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_traits::{TimerEvent, WorkerGlobalScopeInit, ScopeThings, ServiceWorkerMsg, WorkerScriptLoadOrigin};
+use servo_channel::{channel, route_ipc_receiver_to_new_servo_sender, Receiver, Sender};
 use servo_config::prefs::PREFS;
 use servo_rand::random;
 use servo_url::ServoUrl;
-use std::sync::mpsc::{Receiver, Sender, channel};
 use std::thread;
 use std::time::Duration;
 use style::thread_state::{self, ThreadState};
@ -272,7 +271,7 @@ impl ServiceWorkerGlobalScope {
            let runtime = unsafe { new_rt_and_cx() };

            let (devtools_mpsc_chan, devtools_mpsc_port) = channel();
-            ROUTER.route_ipc_receiver_to_mpsc_sender(devtools_receiver, devtools_mpsc_chan);
+            route_ipc_receiver_to_new_servo_sender(devtools_receiver, devtools_mpsc_chan);
            // TODO XXXcreativcoder use this timer_ipc_port, when we have a service worker instance here
            let (timer_ipc_chan, _timer_ipc_port) = ipc::channel().unwrap();
            let (timer_chan, timer_port) = channel();
--- a/components/script/dom/servoparser/async_html.rs
+++ b/components/script/dom/servoparser/async_html.rs
@ -27,12 +27,12 @@ use html5ever::tendril::fmt::UTF8;
 use html5ever::tokenizer::{Tokenizer as HtmlTokenizer, TokenizerOpts, TokenizerResult};
 use html5ever::tree_builder::{ElementFlags, NodeOrText as HtmlNodeOrText, NextParserState, QuirksMode, TreeSink};
 use html5ever::tree_builder::{TreeBuilder, TreeBuilderOpts};
+use servo_channel::{channel, Receiver, Sender};
 use servo_url::ServoUrl;
 use std::borrow::Cow;
 use std::cell::Cell;
 use std::collections::HashMap;
 use std::collections::vec_deque::VecDeque;
-use std::sync::mpsc::{channel, Receiver, Sender};
 use std::thread;
 use style::context::QuirksMode as ServoQuirksMode;

--- a/components/script/dom/testworkletglobalscope.rs
+++ b/components/script/dom/testworkletglobalscope.rs
@ -13,9 +13,9 @@ use dom::workletglobalscope::WorkletGlobalScopeInit;
 use dom_struct::dom_struct;
 use js::rust::Runtime;
 use msg::constellation_msg::PipelineId;
+use servo_channel::Sender;
 use servo_url::ServoUrl;
 use std::collections::HashMap;
-use std::sync::mpsc::Sender;

 // check-tidy: no specs after this line

--- a/components/script/dom/vrdisplay.rs
+++ b/components/script/dom/vrdisplay.rs
@ -36,11 +36,11 @@ use profile_traits::ipc;
 use script_runtime::CommonScriptMsg;
 use script_runtime::ScriptThreadEventCategory::WebVREvent;
 use serde_bytes::ByteBuf;
+use servo_channel::{channel, Sender};
 use std::cell::Cell;
 use std::mem;
 use std::ops::Deref;
 use std::rc::Rc;
-use std::sync::mpsc;
 use std::thread;
 use task_source::TaskSourceName;
 use webvr_traits::{WebVRDisplayData, WebVRDisplayEvent, WebVRFrameData, WebVRLayer, WebVRMsg};
@ -505,7 +505,7 @@ impl VRDisplay {
        // while the render thread is syncing the VRFrameData to be used for the current frame.
        // This thread runs until the user calls ExitPresent, the tab is closed or some unexpected error happened.
        thread::Builder::new().name("WebVR_RAF".into()).spawn(move || {
-            let (raf_sender, raf_receiver) = mpsc::channel();
+            let (raf_sender, raf_receiver) = channel();
            let mut near = near_init;
            let mut far = far_init;

@ -581,7 +581,7 @@ impl VRDisplay {
        self.frame_data_status.set(status);
    }

-    fn handle_raf(&self, end_sender: &mpsc::Sender<Result<(f64, f64), ()>>) {
+    fn handle_raf(&self, end_sender: &Sender<Result<(f64, f64), ()>>) {
        self.frame_data_status.set(VRFrameDataStatus::Waiting);
        self.running_display_raf.set(true);

--- a/components/script/dom/window.rs
+++ b/components/script/dom/window.rs
@ -94,6 +94,7 @@ use script_traits::{TimerSchedulerMsg, UntrustedNodeAddress, WindowSizeData, Win
 use script_traits::webdriver_msg::{WebDriverJSError, WebDriverJSResult};
 use selectors::attr::CaseSensitivity;
 use servo_arc;
+use servo_channel::{channel, Sender};
 use servo_config::opts;
 use servo_geometry::{f32_rect_to_au_rect, MaxRect};
 use servo_url::{Host, MutableOrigin, ImmutableOrigin, ServoUrl};
@ -109,8 +110,6 @@ use std::mem;
 use std::rc::Rc;
 use std::sync::{Arc, Mutex};
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::mpsc::{Sender, channel};
-use std::sync::mpsc::TryRecvError::{Disconnected, Empty};
 use style::error_reporting::ParseErrorReporter;
 use style::media_queries;
 use style::parser::ParserContext as CssParserContext;
@ -1376,15 +1375,16 @@ impl Window {

        debug!("script: layout forked");

-        let complete = match join_port.try_recv() {
-            Err(Empty) => {
+        let complete = select! {
+            recv(join_port.select(), msg) => if let Some(reflow_complete) = msg {
+                reflow_complete
+            } else {
+                panic!("Layout thread failed while script was waiting for a result.");
+            },
+            default => {
                info!("script: waiting on layout");
                join_port.recv().unwrap()
            }
-            Ok(reflow_complete) => reflow_complete,
-            Err(Disconnected) => {
-                panic!("Layout thread failed while script was waiting for a result.");
-            }
        };

        debug!("script: layout joined");
--- a/components/script/dom/worker.rs
+++ b/components/script/dom/worker.rs
@ -25,10 +25,10 @@ use js::jsapi::{JSAutoCompartment, JSContext, JS_RequestInterruptCallback};
 use js::jsval::UndefinedValue;
 use js::rust::HandleValue;
 use script_traits::WorkerScriptLoadOrigin;
+use servo_channel::{channel, Sender};
 use std::cell::Cell;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::mpsc::{Sender, channel};
 use task::TaskOnce;

 pub type TrustedWorkerAddress = Trusted<Worker>;
--- a/components/script/dom/workerglobalscope.rs
+++ b/components/script/dom/workerglobalscope.rs
@ -36,12 +36,12 @@ use net_traits::request::{CredentialsMode, Destination, RequestInit as NetReques
 use script_runtime::{CommonScriptMsg, ScriptChan, ScriptPort, get_reports, Runtime};
 use script_traits::{TimerEvent, TimerEventId};
 use script_traits::WorkerGlobalScopeInit;
+use servo_channel::Receiver;
 use servo_url::{MutableOrigin, ServoUrl};
 use std::default::Default;
 use std::rc::Rc;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::mpsc::Receiver;
 use task::TaskCanceller;
 use task_source::file_reading::FileReadingTaskSource;
 use task_source::networking::NetworkingTaskSource;
--- a/components/script/dom/worklet.rs
+++ b/components/script/dom/worklet.rs
@ -48,6 +48,7 @@ use script_runtime::Runtime;
 use script_runtime::ScriptThreadEventCategory;
 use script_runtime::new_rt_and_cx;
 use script_thread::{MainThreadScriptMsg, ScriptThread};
+use servo_channel::{channel, Sender, Receiver};
 use servo_rand;
 use servo_url::ImmutableOrigin;
 use servo_url::ServoUrl;
@ -58,9 +59,6 @@ use std::rc::Rc;
 use std::sync::Arc;
 use std::sync::atomic::AtomicIsize;
 use std::sync::atomic::Ordering;
-use std::sync::mpsc;
-use std::sync::mpsc::Receiver;
-use std::sync::mpsc::Sender;
 use std::thread;
 use style::thread_state::{self, ThreadState};
 use swapper::Swapper;
@ -309,7 +307,7 @@ impl WorkletThreadPool {

    /// For testing.
    pub fn test_worklet_lookup(&self, id: WorkletId, key: String) -> Option<String> {
-        let (sender, receiver) = mpsc::channel();
+        let (sender, receiver) = channel();
        let msg = WorkletData::Task(id, WorkletTask::Test(TestWorkletTask::Lookup(key, sender)));
        let _ = self.primary_sender.send(msg);
        receiver.recv().expect("Test worklet has died?")
@ -355,7 +353,7 @@ struct WorkletThreadRole {

 impl WorkletThreadRole {
    fn new(is_hot_backup: bool, is_cold_backup: bool) -> WorkletThreadRole {
-        let (sender, receiver) = mpsc::channel();
+        let (sender, receiver) = channel();
        WorkletThreadRole {
            sender: sender,
            receiver: receiver,
@ -419,7 +417,7 @@ impl WorkletThread {
    #[allow(unsafe_code)]
    #[allow(unrooted_must_root)]
    fn spawn(role: WorkletThreadRole, init: WorkletThreadInit) -> Sender<WorkletControl> {
-        let (control_sender, control_receiver) = mpsc::channel();
+        let (control_sender, control_receiver) = channel();
        // TODO: name this thread
        thread::spawn(move || {
            // TODO: add a new IN_WORKLET thread state?
@ -488,12 +486,12 @@ impl WorkletThread {
                if let Some(control) = self.control_buffer.take() {
                    self.process_control(control);
                }
-                while let Ok(control) = self.control_receiver.try_recv() {
+                while let Some(control) = self.control_receiver.try_recv() {
                    self.process_control(control);
                }
                self.gc();
            } else if self.control_buffer.is_none() {
-                if let Ok(control) = self.control_receiver.try_recv() {
+                if let Some(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);
--- a/components/script/dom/workletglobalscope.rs
+++ b/components/script/dom/workletglobalscope.rs
@ -26,11 +26,11 @@ use script_thread::MainThreadScriptMsg;
 use script_traits::{Painter, ScriptMsg};
 use script_traits::{ScriptToConstellationChan, TimerSchedulerMsg};
 use servo_atoms::Atom;
+use servo_channel::Sender;
 use servo_url::ImmutableOrigin;
 use servo_url::MutableOrigin;
 use servo_url::ServoUrl;
 use std::sync::Arc;
-use std::sync::mpsc::Sender;

 #[dom_struct]
 /// <https://drafts.css-houdini.org/worklets/#workletglobalscope>