Port StorageThreadMsg to GenericChannel (#38932)

This change includes the following additions to GenericChannel:
- Add a GenericSend trait which is meant to replace the `IpcSend` trait
over time, as channels are migrated. For the time being this means, that
we often need to use `GenericSend::send()` to disambiguate from the
`IpcSend::send` function, until all usages of `IpcSend` have been
replaced.
- Add an OpaqueSender impl for GenericSender
- Add a profiled version of GenericChannel. The profiling is 1:1 the
same as for the existing profiled IPC channel, namely that only the
blocked time during `recv` is measured.


Testing: No functional changes, covered by existing tests
Part of #38912

---------

Signed-off-by: Jonathan Schwender <schwenderjonathan@gmail.com>

components/constellation/constellation.rs

@@ -2631,10 +2631,10 @@ where
         }
 
         debug!("Exiting storage resource threads.");
-        if let Err(e) = self
-            .public_resource_threads
-            .send(StorageThreadMsg::Exit(storage_ipc_sender))
-        {
+        if let Err(e) = generic_channel::GenericSend::send(
+            &self.public_resource_threads,
+            StorageThreadMsg::Exit(storage_ipc_sender),
+        ) {
             warn!("Exit storage thread failed ({})", e);
         }
 

components/fonts/tests/font_context.rs

@@ -14,6 +14,7 @@ mod font_context {
     use std::thread;
 
     use app_units::Au;
+    use base::generic_channel;
     use compositing_traits::CrossProcessCompositorApi;
     use fonts::platform::font::PlatformFont;
     use fonts::{
@@ -48,7 +49,7 @@ mod font_context {
         fn new() -> TestContext {
             let (system_font_service, system_font_service_proxy) = MockSystemFontService::spawn();
             let (core_sender, _) = ipc::channel().unwrap();
-            let (storage_sender, _) = ipc::channel().unwrap();
+            let (storage_sender, _) = generic_channel::channel().unwrap();
             let (indexeddb_sender, _) = ipc::channel().unwrap();
             let mock_resource_threads =
                 ResourceThreads::new(core_sender, storage_sender, indexeddb_sender);

components/net/resource_thread.rs

@@ -14,6 +14,7 @@ use std::sync::{Arc, Mutex, RwLock, Weak};
 use std::thread;
 use std::time::Duration;
 
+use base::generic_channel::GenericSender;
 use base::id::CookieStoreId;
 use cookie::Cookie;
 use crossbeam_channel::Sender;
@@ -112,7 +113,7 @@ pub fn new_resource_threads(
         protocols,
     );
     let idb: IpcSender<IndexedDBThreadMsg> = IndexedDBThreadFactory::new(config_dir.clone());
-    let storage: IpcSender<StorageThreadMsg> =
+    let storage: GenericSender<StorageThreadMsg> =
         StorageThreadFactory::new(config_dir, mem_profiler_chan);
     (
         ResourceThreads::new(public_core, storage.clone(), idb.clone()),

components/net/storage_thread.rs

@@ -7,8 +7,9 @@ use std::collections::{BTreeMap, HashMap};
 use std::path::PathBuf;
 use std::thread;
 
+use base::generic_channel::{self, GenericReceiver, GenericSender};
 use base::id::WebViewId;
-use ipc_channel::ipc::{self, IpcReceiver, IpcSender};
+use ipc_channel::ipc::IpcSender;
 use malloc_size_of::MallocSizeOf;
 use net_traits::storage_thread::{StorageThreadMsg, StorageType};
 use profile_traits::mem::{
@@ -25,13 +26,13 @@ pub trait StorageThreadFactory {
     fn new(config_dir: Option<PathBuf>, mem_profiler_chan: MemProfilerChan) -> Self;
 }
 
-impl StorageThreadFactory for IpcSender<StorageThreadMsg> {
+impl StorageThreadFactory for GenericSender<StorageThreadMsg> {
     /// Create a storage thread
     fn new(
         config_dir: Option<PathBuf>,
         mem_profiler_chan: MemProfilerChan,
-    ) -> IpcSender<StorageThreadMsg> {
-        let (chan, port) = ipc::channel().unwrap();
+    ) -> GenericSender<StorageThreadMsg> {
+        let (chan, port) = generic_channel::channel().unwrap();
         let chan2 = chan.clone();
         thread::Builder::new()
             .name("StorageManager".to_owned())
@@ -51,14 +52,14 @@ impl StorageThreadFactory for IpcSender<StorageThreadMsg> {
 type OriginEntry = (usize, BTreeMap<String, String>);
 
 struct StorageManager {
-    port: IpcReceiver<StorageThreadMsg>,
+    port: GenericReceiver<StorageThreadMsg>,
     session_data: HashMap<WebViewId, HashMap<String, OriginEntry>>,
     local_data: HashMap<String, OriginEntry>,
     config_dir: Option<PathBuf>,
 }
 
 impl StorageManager {
-    fn new(port: IpcReceiver<StorageThreadMsg>, config_dir: Option<PathBuf>) -> StorageManager {
+    fn new(port: GenericReceiver<StorageThreadMsg>, config_dir: Option<PathBuf>) -> StorageManager {
         let mut local_data = HashMap::new();
         if let Some(ref config_dir) = config_dir {
             resource_thread::read_json_from_file(&mut local_data, config_dir, "local_data.json");
@@ -224,7 +225,7 @@ impl StorageManager {
 
     fn keys(
         &self,
-        sender: IpcSender<Vec<String>>,
+        sender: GenericSender<Vec<String>>,
         storage_type: StorageType,
         webview_id: WebViewId,
         url: ServoUrl,

components/script/dom/storage.rs

@@ -1,13 +1,12 @@
 /* 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 https://mozilla.org/MPL/2.0/. */
-
+use base::generic_channel::{GenericSend, SendResult};
 use base::id::WebViewId;
 use constellation_traits::ScriptToConstellationMessage;
 use dom_struct::dom_struct;
 use net_traits::storage_thread::{StorageThreadMsg, StorageType};
-use net_traits::{IpcSend, IpcSendResult};
-use profile_traits::ipc;
+use profile_traits::{generic_channel, ipc};
 use servo_url::ServoUrl;
 
 use crate::dom::bindings::codegen::Bindings::StorageBinding::StorageMethods;
@@ -57,8 +56,8 @@ impl Storage {
         self.global().get_url()
     }
 
-    fn send_storage_msg(&self, msg: StorageThreadMsg) -> IpcSendResult {
-        self.global().resource_threads().send(msg)
+    fn send_storage_msg(&self, msg: StorageThreadMsg) -> SendResult {
+        GenericSend::send(self.global().resource_threads(), msg)
     }
 }
 
@@ -173,7 +172,8 @@ impl StorageMethods<crate::DomTypeHolder> for Storage {
 
     // https://html.spec.whatwg.org/multipage/#the-storage-interface:supported-property-names
     fn SupportedPropertyNames(&self) -> Vec<DOMString> {
-        let (sender, receiver) = ipc::channel(self.global().time_profiler_chan().clone()).unwrap();
+        let time_profiler = self.global().time_profiler_chan().clone();
+        let (sender, receiver) = generic_channel::channel(time_profiler).unwrap();
 
         self.send_storage_msg(StorageThreadMsg::Keys(
             sender,

components/script/dom/windowproxy.rs

@@ -5,6 +5,7 @@
 use std::cell::Cell;
 use std::ptr;
 
+use base::generic_channel::GenericSend;
 use base::id::{BrowsingContextId, PipelineId, WebViewId};
 use constellation_traits::{
     AuxiliaryWebViewCreationRequest, LoadData, LoadOrigin, NavigationHistoryBehavior,
@@ -32,7 +33,6 @@ use js::jsval::{NullValue, PrivateValue, UndefinedValue};
 use js::rust::wrappers::{JS_TransplantObject, NewWindowProxy, SetWindowProxy};
 use js::rust::{Handle, MutableHandle, MutableHandleValue, get_object_class};
 use malloc_size_of::{MallocSizeOf, MallocSizeOfOps};
-use net_traits::IpcSend;
 use net_traits::request::Referrer;
 use net_traits::storage_thread::StorageThreadMsg;
 use script_traits::NewLayoutInfo;
@@ -356,7 +356,7 @@ impl WindowProxy {
                 dest: response.new_webview_id,
             };
 
-            document.global().resource_threads().send(msg).unwrap();
+            GenericSend::send(document.global().resource_threads(), msg).unwrap();
             receiver.recv().unwrap();
         }
         Some(new_window_proxy)

components/shared/base/generic_channel.rs

@@ -15,6 +15,18 @@ use serde::de::VariantAccess;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use servo_config::opts;
 
+/// Abstraction of the ability to send a particular type of message cross-process.
+/// This can be used to ease the use of GenericSender sub-fields.
+pub trait GenericSend<T>
+where
+    T: serde::Serialize + for<'de> serde::Deserialize<'de>,
+{
+    /// send message T
+    fn send(&self, _: T) -> SendResult;
+    /// get underlying sender
+    fn sender(&self) -> GenericSender<T>;
+}
+
 /// A GenericSender that sends messages to a [GenericReceiver].
 ///
 /// The sender supports sending messages cross-process, if servo is run in multiprocess mode.

components/shared/net/lib.rs

@@ -10,6 +10,7 @@ use std::sync::{LazyLock, OnceLock};
 use std::thread::{self, JoinHandle};
 
 use base::cross_process_instant::CrossProcessInstant;
+use base::generic_channel::{GenericSend, GenericSender, SendResult};
 use base::id::{CookieStoreId, HistoryStateId};
 use content_security_policy::{self as csp};
 use cookie::Cookie;
@@ -421,14 +422,14 @@ where
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct ResourceThreads {
     pub core_thread: CoreResourceThread,
-    storage_thread: IpcSender<StorageThreadMsg>,
+    storage_thread: GenericSender<StorageThreadMsg>,
     idb_thread: IpcSender<IndexedDBThreadMsg>,
 }
 
 impl ResourceThreads {
     pub fn new(
         c: CoreResourceThread,
-        s: IpcSender<StorageThreadMsg>,
+        s: GenericSender<StorageThreadMsg>,
         i: IpcSender<IndexedDBThreadMsg>,
     ) -> ResourceThreads {
         ResourceThreads {
@@ -463,12 +464,12 @@ impl IpcSend<IndexedDBThreadMsg> for ResourceThreads {
     }
 }
 
-impl IpcSend<StorageThreadMsg> for ResourceThreads {
-    fn send(&self, msg: StorageThreadMsg) -> IpcSendResult {
+impl GenericSend<StorageThreadMsg> for ResourceThreads {
+    fn send(&self, msg: StorageThreadMsg) -> SendResult {
         self.storage_thread.send(msg)
     }
 
-    fn sender(&self) -> IpcSender<StorageThreadMsg> {
+    fn sender(&self) -> GenericSender<StorageThreadMsg> {
         self.storage_thread.clone()
     }
 }

components/shared/net/storage_thread.rs

@@ -2,6 +2,7 @@
  * License, v. 2.0. If a copy of the MPL was not distributed with this
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
+use base::generic_channel::GenericSender;
 use base::id::WebViewId;
 use ipc_channel::ipc::IpcSender;
 use malloc_size_of_derive::MallocSizeOf;
@@ -31,7 +32,7 @@ pub enum StorageThreadMsg {
     ),
 
     /// Gets the available keys in the associated storage data
-    Keys(IpcSender<Vec<String>>, StorageType, WebViewId, ServoUrl),
+    Keys(GenericSender<Vec<String>>, StorageType, WebViewId, ServoUrl),
 
     /// gets the value associated with the given key in the associated storage data
     GetItem(

components/shared/profile/generic_channel.rs

@@ -0,0 +1,53 @@
+/* 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 https://mozilla.org/MPL/2.0/. */
+
+use base::generic_channel;
+use serde::{Deserialize, Serialize};
+
+use crate::time::{ProfilerCategory, ProfilerChan};
+use crate::time_profile;
+
+pub struct GenericReceiver<T>
+where
+    T: for<'de> Deserialize<'de> + Serialize,
+{
+    receiver: generic_channel::GenericReceiver<T>,
+    time_profile_chan: ProfilerChan,
+}
+
+impl<T> GenericReceiver<T>
+where
+    T: for<'de> Deserialize<'de> + Serialize,
+{
+    pub fn recv(&self) -> Result<T, generic_channel::ReceiveError> {
+        time_profile!(
+            ProfilerCategory::IpcReceiver,
+            None,
+            self.time_profile_chan.clone(),
+            move || self.receiver.recv(),
+        )
+    }
+
+    pub fn try_recv(&self) -> Result<T, generic_channel::TryReceiveError> {
+        self.receiver.try_recv()
+    }
+
+    pub fn into_inner(self) -> generic_channel::GenericReceiver<T> {
+        self.receiver
+    }
+}
+
+pub fn channel<T>(
+    time_profile_chan: ProfilerChan,
+) -> Option<(generic_channel::GenericSender<T>, GenericReceiver<T>)>
+where
+    T: for<'de> Deserialize<'de> + Serialize,
+{
+    let (sender, receiver) = generic_channel::channel()?;
+    let profiled_receiver = GenericReceiver {
+        receiver,
+        time_profile_chan,
+    };
+    Some((sender, profiled_receiver))
+}

components/shared/profile/lib.rs

@@ -8,6 +8,7 @@
 
 #![deny(unsafe_code)]
 
+pub mod generic_channel;
 pub mod ipc;
 pub mod mem;
 pub mod time;

components/shared/profile/mem.rs

@@ -11,6 +11,7 @@ use std::collections::HashSet;
 use std::ffi::c_void;
 use std::marker::Send;
 
+use base::generic_channel::GenericSender;
 use crossbeam_channel::Sender;
 use ipc_channel::ipc::{self, IpcSender};
 use ipc_channel::router::ROUTER;
@@ -49,6 +50,20 @@ where
     }
 }
 
+impl<T> OpaqueSender<T> for GenericSender<T>
+where
+    T: serde::Serialize,
+{
+    fn send(&self, message: T) {
+        if let Err(e) = GenericSender::send(self, message) {
+            warn!(
+                "Error communicating with the target thread from the profiler: {}",
+                e
+            );
+        }
+    }
+}
+
 /// Front-end representation of the profiler used to communicate with the
 /// profiler.
 #[derive(Clone, Debug, Deserialize, Serialize)]