Revert "GenericChannel: Migrate compositor channels to GenericChannel (#38782)" (#38940)

This reverts commit fb1c0a4c48.

Previously in `create_compositor_channel`, the [routing callback][1] was
setup so that a message received on the Compositor's IPC receiver will
be
forwarded to the local receiver using the `CompositorProxy` which also
takes care of waking up the event loop. In #38782, this was changed so
that the routing callbacks simply forwards the message directly without
going via the `CompositorProxy`. This breaks behaviours that rely on the
event loop being woken up on message sending, e.g. updating image frames
for animated gifs.

Since the GenericChannel API doesn't allow custom routing callbacks,
revert this change until we figure out a better solution.

[1]:
d2ccce6052/components/servo/lib.rs (L1114)

Signed-off-by: Mukilan Thiyagarajan <mukilan@igalia.com>

Signed-off-by: Mukilan Thiyagarajan <mukilan@igalia.com>

components/compositing/compositor.rs

@@ -11,9 +11,9 @@ use std::rc::Rc;
 use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
 
+use base::Epoch;
 use base::cross_process_instant::CrossProcessInstant;
 use base::id::{PipelineId, WebViewId};
-use base::{Epoch, generic_channel};
 use bitflags::bitflags;
 use compositing_traits::display_list::{CompositorDisplayListInfo, ScrollTree, ScrollType};
 use compositing_traits::rendering_context::RenderingContext;
@@ -22,7 +22,7 @@ use compositing_traits::{
     WebViewTrait,
 };
 use constellation_traits::{EmbedderToConstellationMessage, PaintMetricEvent};
-use crossbeam_channel::Sender;
+use crossbeam_channel::{Receiver, Sender};
 use dpi::PhysicalSize;
 use embedder_traits::{CompositorHitTestResult, InputEvent, ShutdownState, ViewportDetails};
 use euclid::{Point2D, Rect, Scale, Size2D, Transform3D};
@@ -92,7 +92,7 @@ pub struct ServoRenderer {
     shutdown_state: Rc<Cell<ShutdownState>>,
 
     /// The port on which we receive messages.
-    compositor_receiver: generic_channel::RoutedReceiver<CompositorMsg>,
+    compositor_receiver: Receiver<CompositorMsg>,
 
     /// The channel on which messages can be sent to the constellation.
     pub(crate) constellation_sender: Sender<EmbedderToConstellationMessage>,
@@ -1388,7 +1388,7 @@ impl IOCompositor {
     }
 
     /// Get the message receiver for this [`IOCompositor`].
-    pub fn receiver(&self) -> Ref<'_, generic_channel::RoutedReceiver<CompositorMsg>> {
+    pub fn receiver(&self) -> Ref<'_, Receiver<CompositorMsg>> {
         Ref::map(self.global.borrow(), |global| &global.compositor_receiver)
     }
 

components/compositing/lib.rs

@@ -7,11 +7,10 @@
 use std::cell::Cell;
 use std::rc::Rc;
 
-use base::generic_channel;
 use compositing_traits::rendering_context::RenderingContext;
 use compositing_traits::{CompositorMsg, CompositorProxy};
 use constellation_traits::EmbedderToConstellationMessage;
-use crossbeam_channel::Sender;
+use crossbeam_channel::{Receiver, Sender};
 use embedder_traits::{EventLoopWaker, ShutdownState};
 use profile_traits::{mem, time};
 use webrender::RenderApi;
@@ -33,7 +32,7 @@ pub struct InitialCompositorState {
     /// A channel to the compositor.
     pub sender: CompositorProxy,
     /// A port on which messages inbound to the compositor can be received.
-    pub receiver: generic_channel::RoutedReceiver<CompositorMsg>,
+    pub receiver: Receiver<CompositorMsg>,
     /// A channel to the constellation.
     pub constellation_chan: Sender<EmbedderToConstellationMessage>,
     /// A channel to the time profiler thread.

components/constellation/constellation.rs

@@ -290,11 +290,11 @@ pub struct Constellation<STF, SWF> {
 
     /// An IPC channel for script threads to send messages to the constellation.
     /// This is the script threads' view of `script_receiver`.
-    script_sender: GenericSender<(PipelineId, ScriptToConstellationMessage)>,
+    script_sender: IpcSender<(PipelineId, ScriptToConstellationMessage)>,
 
     /// A channel for the constellation to receive messages from script threads.
     /// This is the constellation's view of `script_sender`.
-    script_receiver: RoutedReceiver<(PipelineId, ScriptToConstellationMessage)>,
+    script_receiver: Receiver<Result<(PipelineId, ScriptToConstellationMessage), IpcError>>,
 
     /// A handle to register components for hang monitoring.
     /// None when in multiprocess mode.
@@ -607,8 +607,11 @@ where
             .name("Constellation".to_owned())
             .spawn(move || {
                 let (script_ipc_sender, script_ipc_receiver) =
-                    generic_channel::channel().expect("ipc channel failure");
-                let script_receiver = script_ipc_receiver.route_preserving_errors();
+                    ipc::channel().expect("ipc channel failure");
+                let script_receiver =
+                    route_ipc_receiver_to_new_crossbeam_receiver_preserving_errors(
+                        script_ipc_receiver,
+                    );
 
                 let (namespace_ipc_sender, namespace_ipc_receiver) =
                     generic_channel::channel().expect("ipc channel failure");
@@ -1243,7 +1246,7 @@ where
 
         let request = match request {
             Ok(request) => request,
-            Err(err) => return error!("Deserialization failed ({err:?})."),
+            Err(err) => return error!("Deserialization failed ({}).", err),
         };
 
         match request {

components/script/dom/workletglobalscope.rs

@@ -4,7 +4,6 @@
 
 use std::sync::Arc;
 
-use base::generic_channel::GenericSender;
 use base::id::PipelineId;
 use constellation_traits::{ScriptToConstellationChan, ScriptToConstellationMessage};
 use crossbeam_channel::Sender;
@@ -196,7 +195,7 @@ pub(crate) struct WorkletGlobalScopeInit {
     /// Channel to devtools
     pub(crate) devtools_chan: Option<IpcSender<ScriptToDevtoolsControlMsg>>,
     /// Messages to send to constellation
-    pub(crate) to_constellation_sender: GenericSender<(PipelineId, ScriptToConstellationMessage)>,
+    pub(crate) to_constellation_sender: IpcSender<(PipelineId, ScriptToConstellationMessage)>,
     /// The image cache
     pub(crate) image_cache: Arc<dyn ImageCache>,
     /// Identity manager for WebGPU resources

components/script/messaging.rs

@@ -340,7 +340,7 @@ pub(crate) struct ScriptThreadSenders {
     /// particular pipelines.
     #[no_trace]
     pub(crate) pipeline_to_constellation_sender:
-        GenericSender<(PipelineId, ScriptToConstellationMessage)>,
+        IpcSender<(PipelineId, ScriptToConstellationMessage)>,
 
     /// The shared [`IpcSender`] which is sent to the `ImageCache` when requesting an image. The
     /// messages on this channel are routed to crossbeam [`Sender`] on the router thread, which

components/servo/lib.rs

@@ -33,7 +33,6 @@ use std::rc::{Rc, Weak};
 use std::sync::{Arc, Mutex};
 use std::thread;
 
-use base::generic_channel;
 pub use base::id::WebViewId;
 use base::id::{PipelineNamespace, PipelineNamespaceId};
 #[cfg(feature = "bluetooth")]
@@ -83,6 +82,7 @@ use gaol::sandbox::{ChildSandbox, ChildSandboxMethods};
 pub use gleam::gl;
 use gleam::gl::RENDERER;
 use ipc_channel::ipc::{self, IpcSender};
+use ipc_channel::router::ROUTER;
 use javascript_evaluator::JavaScriptEvaluator;
 pub use keyboard_types::{
     Code, CompositionEvent, CompositionState, Key, KeyState, Location, Modifiers, NamedKey,
@@ -544,7 +544,7 @@ impl Servo {
             let mut compositor = self.compositor.borrow_mut();
             let mut messages = Vec::new();
             while let Ok(message) = compositor.receiver().try_recv() {
-                messages.push(message.expect("IPC serialization error"));
+                messages.push(message);
             }
             compositor.handle_messages(messages);
         }
@@ -1095,23 +1095,28 @@ fn create_embedder_channel(
 
 fn create_compositor_channel(
     event_loop_waker: Box<dyn EventLoopWaker>,
-) -> (
-    CompositorProxy,
-    generic_channel::RoutedReceiver<CompositorMsg>,
-) {
-    let routed_channel =
-        generic_channel::routed_channel_with_local_sender().expect("Create channel failure");
+) -> (CompositorProxy, Receiver<CompositorMsg>) {
+    let (sender, receiver) = unbounded();
 
-    let cross_process_compositor_api =
-        CrossProcessCompositorApi(routed_channel.generic_sender.clone());
+    let (compositor_ipc_sender, compositor_ipc_receiver) =
+        ipc::channel().expect("ipc channel failure");
 
-    let compositor_proxy = CompositorProxy::new(
-        routed_channel.local_sender,
+    let cross_process_compositor_api = CrossProcessCompositorApi(compositor_ipc_sender);
+    let compositor_proxy = CompositorProxy {
+        sender,
         cross_process_compositor_api,
         event_loop_waker,
+    };
+
+    let compositor_proxy_clone = compositor_proxy.clone();
+    ROUTER.add_typed_route(
+        compositor_ipc_receiver,
+        Box::new(move |message| {
+            compositor_proxy_clone.send(message.expect("Could not convert Compositor message"));
+        }),
     );
 
-    (compositor_proxy, routed_channel.local_receiver)
+    (compositor_proxy, receiver)
 }
 
 #[allow(clippy::too_many_arguments)]

components/shared/base/generic_channel.rs

@@ -395,18 +395,6 @@ where
     }
 }
 
-/// A GenericChannel, which was routed via the Router.
-pub struct GenericRoutedChannel<T: Serialize + for<'de> Deserialize<'de>> {
-    /// A GenericSender of the channel, i.e. it may be sent to other processes in multiprocess mode.
-    /// Connected to `local_receiver` via the [ROUTER].
-    pub generic_sender: GenericSender<T>,
-    /// A sender that directly sends to the local_receiver. Can only be used in the same process
-    /// as the `local_receiver`.
-    pub local_sender: crossbeam_channel::Sender<Result<T, ipc_channel::Error>>,
-    /// The receiving end of the channel. Only usable in the current process.
-    pub local_receiver: RoutedReceiver<T>,
-}
-
 /// Private helper function to create a crossbeam based channel.
 ///
 /// Do NOT make this function public!
@@ -447,34 +435,6 @@ where
     }
 }
 
-/// Returns a [GenericRoutedChannel], where the receiver is usable in the current process,
-/// and sending is possible both in remote and local process, via the generic_sender and the
-/// local_sender.
-pub fn routed_channel_with_local_sender<T>() -> Option<GenericRoutedChannel<T>>
-where
-    T: for<'de> Deserialize<'de> + Serialize + Send + 'static,
-{
-    let (crossbeam_sender, crossbeam_receiver) = crossbeam_channel::unbounded();
-    let generic_sender = if opts::get().multiprocess || opts::get().force_ipc {
-        let (ipc_sender, ipc_receiver) = ipc_channel::ipc::channel().ok()?;
-        let crossbeam_sender_clone = crossbeam_sender.clone();
-        ROUTER.add_typed_route(
-            ipc_receiver,
-            Box::new(move |message| {
-                let _ = crossbeam_sender_clone.send(message);
-            }),
-        );
-        GenericSender(GenericSenderVariants::Ipc(ipc_sender))
-    } else {
-        GenericSender(GenericSenderVariants::Crossbeam(crossbeam_sender.clone()))
-    };
-    Some(GenericRoutedChannel {
-        generic_sender,
-        local_sender: crossbeam_sender,
-        local_receiver: crossbeam_receiver,
-    })
-}
-
 #[cfg(test)]
 mod single_process_channel_tests {
     //! These unit-tests test that ipc_channel and crossbeam_channel Senders and Receivers

components/shared/compositing/lib.rs

@@ -23,7 +23,6 @@ pub mod viewport_description;
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
 
-use base::generic_channel::GenericSender;
 use bitflags::bitflags;
 use display_list::CompositorDisplayListInfo;
 use embedder_traits::ScreenGeometry;
@@ -45,12 +44,7 @@ use crate::viewport_description::ViewportDescription;
 /// Sends messages to the compositor.
 #[derive(Clone)]
 pub struct CompositorProxy {
-    /// A sender optimised for sending in the local process.
-    /// The type is a Result to match the API of ipc_channel after routing,
-    /// which contains a Result to propagate deserialization errors to the
-    /// recipient.
-    /// The field is private to hide the inner `Result` type.
-    sender: Sender<Result<CompositorMsg, ipc_channel::Error>>,
+    pub sender: Sender<CompositorMsg>,
     /// Access to [`Self::sender`] that is possible to send across an IPC
     /// channel. These messages are routed via the router thread to
     /// [`Self::sender`].
@@ -58,20 +52,6 @@ pub struct CompositorProxy {
     pub event_loop_waker: Box<dyn EventLoopWaker>,
 }
 
-impl CompositorProxy {
-    pub fn new(
-        local_process_sender: Sender<Result<CompositorMsg, ipc_channel::Error>>,
-        cross_process_compositor_api: CrossProcessCompositorApi,
-        event_loop_waker: Box<dyn EventLoopWaker>,
-    ) -> Self {
-        Self {
-            sender: local_process_sender,
-            cross_process_compositor_api,
-            event_loop_waker,
-        }
-    }
-}
-
 impl OpaqueSender<CompositorMsg> for CompositorProxy {
     fn send(&self, message: CompositorMsg) {
         CompositorProxy::send(self, message)
@@ -80,7 +60,7 @@ impl OpaqueSender<CompositorMsg> for CompositorProxy {
 
 impl CompositorProxy {
     pub fn send(&self, msg: CompositorMsg) {
-        if let Err(err) = self.sender.send(Ok(msg)) {
+        if let Err(err) = self.sender.send(msg) {
             warn!("Failed to send response ({:?}).", err);
         }
         self.event_loop_waker.wake();
@@ -190,18 +170,18 @@ pub struct CompositionPipeline {
 
 /// A mechanism to send messages from ScriptThread to the parent process' WebRender instance.
 #[derive(Clone, Deserialize, MallocSizeOf, Serialize)]
-pub struct CrossProcessCompositorApi(pub GenericSender<CompositorMsg>);
+pub struct CrossProcessCompositorApi(pub IpcSender<CompositorMsg>);
 
 impl CrossProcessCompositorApi {
     /// Create a new [`CrossProcessCompositorApi`] struct that does not have a listener on the other
     /// end to use for unit testing.
     pub fn dummy() -> Self {
-        let (sender, _) = base::generic_channel::channel().unwrap();
+        let (sender, _) = ipc::channel().unwrap();
         Self(sender)
     }
 
     /// Get the sender for this proxy.
-    pub fn sender(&self) -> &GenericSender<CompositorMsg> {
+    pub fn sender(&self) -> &IpcSender<CompositorMsg> {
         &self.0
     }
 

components/shared/constellation/from_script_message.rs

@@ -8,7 +8,6 @@ use std::collections::HashMap;
 use std::fmt;
 
 use base::Epoch;
-use base::generic_channel::{GenericSender, SendResult};
 use base::id::{
     BroadcastChannelRouterId, BrowsingContextId, HistoryStateId, MessagePortId,
     MessagePortRouterId, PipelineId, ServiceWorkerId, ServiceWorkerRegistrationId, WebViewId,
@@ -22,6 +21,7 @@ use embedder_traits::{
 };
 use euclid::default::Size2D as UntypedSize2D;
 use http::{HeaderMap, Method};
+use ipc_channel::Error as IpcError;
 use ipc_channel::ipc::{IpcReceiver, IpcSender};
 use malloc_size_of_derive::MallocSizeOf;
 use net_traits::policy_container::PolicyContainer;
@@ -46,14 +46,14 @@ use crate::{
 #[derive(Clone, Debug, Deserialize, MallocSizeOf, Serialize)]
 pub struct ScriptToConstellationChan {
     /// Sender for communicating with constellation thread.
-    pub sender: GenericSender<(PipelineId, ScriptToConstellationMessage)>,
+    pub sender: IpcSender<(PipelineId, ScriptToConstellationMessage)>,
     /// Used to identify the origin of the message.
     pub pipeline_id: PipelineId,
 }
 
 impl ScriptToConstellationChan {
     /// Send ScriptMsg and attach the pipeline_id to the message.
-    pub fn send(&self, msg: ScriptToConstellationMessage) -> SendResult {
+    pub fn send(&self, msg: ScriptToConstellationMessage) -> Result<(), IpcError> {
         self.sender.send((self.pipeline_id, msg))
     }
 }