servo_base/generic_channel/
lazy_callback.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//! # Lazy Callbacks
6//!
7//! When constructing callbacks we sometimes have a large distance between where the channel for the callback
8//! is created and where the initial callback will be created. Refactoring of this code is sometimes not possible.
9//! Here we provide [LazyCallback]. We use 'lazy_callback()' to generate a [LazyCallback] and a [CallbackSetter].
10//! The [LazyCallback] works like a [GenericCallback] and can be used to execute callbacks in the receiver process.
11//! The [CallbackSetter] has a single consuming method of 'set_callback' which will set the callback that the [LazyCallback]
12//! will then execute on messages send to it.
13//!
14//! This is achieved with having the LazyCallback having a back channel in single process mode that sets the [GenericCallback].
15//! Hence, this is slightly less efficient than a [GenericCallback]
16
17use std::cell::{OnceCell, RefCell};
18use std::fmt;
19use std::marker::PhantomData;
20
21use ipc_channel::ipc::{IpcReceiver, IpcSender};
22use ipc_channel::router::ROUTER;
23use malloc_size_of::{MallocSizeOf as MallocSizeOfTrait, MallocSizeOfOps};
24use malloc_size_of_derive::MallocSizeOf;
25use serde::de::VariantAccess;
26use serde::{Deserialize, Deserializer, Serialize, Serializer};
27
28use crate::generic_channel::{GenericCallback, SendError, SendResult, use_ipc};
29
30/// Basic struct for [LazyCallback]
31#[derive(MallocSizeOf)]
32pub struct LazyCallback<T: Serialize + for<'de> Deserialize<'de> + Send + 'static>(
33    LazyCallbackVariants<T>,
34);
35
36enum LazyCallbackVariants<T>
37where
38    T: Serialize + Send + 'static,
39{
40    InProcess {
41        callback_receiver: RefCell<Option<crossbeam_channel::Receiver<GenericCallback<T>>>>,
42        callback: OnceCell<GenericCallback<T>>,
43    },
44    Ipc(IpcSender<T>),
45}
46
47impl<T> MallocSizeOfTrait for LazyCallbackVariants<T>
48where
49    T: Serialize + Send + 'static,
50{
51    fn size_of(&self, ops: &mut MallocSizeOfOps) -> usize {
52        match self {
53            LazyCallbackVariants::InProcess {
54                callback_receiver,
55                callback,
56            } => callback_receiver.size_of(ops) + callback.size_of(ops),
57            LazyCallbackVariants::Ipc(_) => 0,
58        }
59    }
60}
61
62impl<T> LazyCallback<T>
63where
64    T: Serialize + for<'de> Deserialize<'de> + Send + 'static,
65{
66    /// Send messages to the callback. This might block until the callback is set via the 'CallbackSetter'
67    pub fn send(&self, value: T) -> SendResult {
68        match &self.0 {
69            LazyCallbackVariants::InProcess {
70                callback_receiver,
71                callback,
72            } => {
73                if let Some(cb) = callback.get() {
74                    cb.send(value)
75                } else {
76                    // Init callback
77                    if let Ok(cb) = callback_receiver.borrow_mut().take().unwrap().recv() {
78                        let _ = callback.set(cb);
79                        callback.get().unwrap().send(value)
80                    } else {
81                        log::error!("Could not get callback. Callback_receiver already dropped");
82                        SendResult::Err(SendError::Disconnected)
83                    }
84                }
85            },
86            LazyCallbackVariants::Ipc(ipc_sender) => {
87                ipc_sender.send(value).map_err(|error| match error {
88                    ipc_channel::IpcError::SerializationError(ser_de_error) => {
89                        SendError::SerializationError(ser_de_error.to_string())
90                    },
91                    ipc_channel::IpcError::Io(_) | ipc_channel::IpcError::Disconnected => {
92                        SendError::Disconnected
93                    },
94                })
95            },
96        }
97    }
98}
99
100pub struct CallbackSetter<T: Serialize + Send + 'static>(CallbackSetterVariants<T>);
101
102impl<T: Serialize + Send> fmt::Debug for CallbackSetter<T> {
103    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
104        f.debug_tuple("CallbackSetter").finish()
105    }
106}
107
108impl<T> Serialize for CallbackSetter<T>
109where
110    T: Serialize + Send + 'static,
111{
112    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
113        match &self.0 {
114            CallbackSetterVariants::Ipc(sender) => {
115                s.serialize_newtype_variant("CallbackSetter", 0, "Ipc", sender)
116            },
117            // The only reason we need / want serialization in single-process mode is to support
118            // sending GenericCallbacks over existing IPC channels. This allows us to
119            // incrementally port IPC channels to the GenericChannel, without needing to follow a
120            // top-to-bottom approach.
121            // Long-term we can remove this branch in the code again and replace it with
122            // unreachable, since likely all IPC channels would be GenericChannels.
123            CallbackSetterVariants::InProcess(wrapped_callback) => {
124                if use_ipc() {
125                    return Err(serde::ser::Error::custom(
126                        "InProcess callback setter can't be serialized in multiprocess mode",
127                    ));
128                }
129                // Due to the signature of `serialize` we need to clone the Arc to get an owned
130                // pointer we can leak.
131                // We additionally need to Box to get a thin pointer.
132                let cloned_callback = Box::new(wrapped_callback.clone());
133                let sender_clone_addr = Box::leak(cloned_callback) as *mut _ as usize;
134                s.serialize_newtype_variant("CallbackSetter", 1, "InProcess", &sender_clone_addr)
135            },
136        }
137    }
138}
139
140struct LazyCallbackSetterVisitor<T> {
141    marker: PhantomData<T>,
142}
143
144impl<'de, T> serde::de::Visitor<'de> for LazyCallbackSetterVisitor<T>
145where
146    T: Serialize + Deserialize<'de> + Send + 'static,
147{
148    type Value = CallbackSetter<T>;
149
150    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
151        formatter.write_str("a GenericCallback variant")
152    }
153
154    fn visit_enum<A>(self, data: A) -> Result<Self::Value, A::Error>
155    where
156        A: serde::de::EnumAccess<'de>,
157    {
158        #[derive(Deserialize)]
159        enum LazyCallbackSetterVariantNames {
160            Ipc,
161            InProcess,
162        }
163
164        let (variant_name, variant_data): (LazyCallbackSetterVariantNames, _) = data.variant()?;
165
166        match variant_name {
167            LazyCallbackSetterVariantNames::Ipc => variant_data
168                .newtype_variant::<IpcReceiver<T>>()
169                .map(|receiver| CallbackSetter(CallbackSetterVariants::Ipc(receiver))),
170            LazyCallbackSetterVariantNames::InProcess => {
171                if use_ipc() {
172                    return Err(serde::de::Error::custom(
173                        "InProcess callback found in multiprocess mode",
174                    ));
175                }
176                let addr = variant_data.newtype_variant::<usize>()?;
177                let ptr = addr as *mut _;
178                // SAFETY: We know we are in the same address space as the sender, so we can safely
179                // reconstruct the Box, that we previously leaked with `into_raw` during
180                // serialization.
181                // Attention: Code reviewers should carefully compare the deserialization here
182                // with the serialization above.
183                #[expect(unsafe_code)]
184                let callback = unsafe { Box::from_raw(ptr) };
185                Ok(CallbackSetter(CallbackSetterVariants::InProcess(*callback)))
186            },
187        }
188    }
189}
190
191impl<'a, T> Deserialize<'a> for CallbackSetter<T>
192where
193    T: Serialize + Deserialize<'a> + Send + 'static,
194{
195    fn deserialize<D>(d: D) -> Result<CallbackSetter<T>, D::Error>
196    where
197        D: Deserializer<'a>,
198    {
199        d.deserialize_enum(
200            "GenericCallback",
201            &["CrossProcess", "InProcess"],
202            LazyCallbackSetterVisitor {
203                marker: PhantomData,
204            },
205        )
206    }
207}
208
209enum CallbackSetterVariants<T>
210where
211    T: Serialize + Send + 'static,
212{
213    InProcess(crossbeam_channel::Sender<GenericCallback<T>>),
214    Ipc(IpcReceiver<T>),
215}
216
217impl<T> CallbackSetter<T>
218where
219    T: Serialize + for<'de> Deserialize<'de> + Send + 'static,
220{
221    /// This sets the callback.
222    pub fn set_callback<F: FnMut(Result<T, SendError>) + Send + 'static>(self, mut callback: F) {
223        match self.0 {
224            CallbackSetterVariants::InProcess(sender) => {
225                let callback = GenericCallback::new(callback).expect("Could not create callback");
226                if sender.send(callback).is_err() {
227                    log::error!("Could not send callback, sender was already dropped");
228                }
229            },
230            CallbackSetterVariants::Ipc(ipc_receiver) => {
231                let new_callback = move |msg: Result<T, ipc_channel::SerDeError>| {
232                    callback(msg.map_err(|error| error.into()))
233                };
234                ROUTER.add_typed_route(ipc_receiver, Box::new(new_callback));
235            },
236        }
237    }
238}
239
240/// This function should never be exported.
241fn lazy_callback_inprocess<T>() -> (LazyCallback<T>, CallbackSetter<T>)
242where
243    T: Serialize + for<'de> Deserialize<'de> + Send + 'static,
244{
245    let (callback_sender, callback_receiver) = crossbeam_channel::bounded(1);
246    let lazycallback = LazyCallback(LazyCallbackVariants::InProcess {
247        callback_receiver: RefCell::new(Some(callback_receiver)),
248        callback: OnceCell::new(),
249    });
250
251    let callback_setter = CallbackSetter(CallbackSetterVariants::InProcess(callback_sender));
252
253    (lazycallback, callback_setter)
254}
255
256/// This function should never be exported.
257fn lazy_callback_ipc<T>() -> (LazyCallback<T>, CallbackSetter<T>)
258where
259    T: Serialize + for<'de> Deserialize<'de> + Send + 'static,
260{
261    let (sender, receiver) = ipc_channel::ipc::channel().expect("Could not create channel");
262    let callback = LazyCallback(LazyCallbackVariants::Ipc(sender));
263    let callback_setter = CallbackSetter(CallbackSetterVariants::Ipc(receiver));
264    (callback, callback_setter)
265}
266
267/// A LazyCallback is a Callback that will be initialized at a later date.
268/// We return the 'LazyCallback' which is a GenericCallback.
269/// We also return a 'CallbackSetter' where the callback can be set at a later date.
270pub fn lazy_callback<T>() -> (LazyCallback<T>, CallbackSetter<T>)
271where
272    T: Serialize + for<'de> Deserialize<'de> + Send + 'static,
273{
274    if use_ipc() {
275        lazy_callback_ipc()
276    } else {
277        lazy_callback_inprocess()
278    }
279}
280
281#[cfg(test)]
282mod single_process_callback_test {
283    use crate::generic_channel::lazy_callback::{lazy_callback_inprocess, lazy_callback_ipc};
284    use crate::generic_channel::{CallbackSetter, LazyCallback};
285    fn test_lazy_callback(callback: LazyCallback<bool>, callback_setter: CallbackSetter<bool>) {
286        let t1 = std::thread::spawn(move || {
287            callback.send(true).expect("Could not send");
288        });
289
290        let (sender, receiver) = crossbeam_channel::bounded(1);
291        let t2 = std::thread::spawn(move || {
292            std::thread::sleep(std::time::Duration::from_secs(1));
293            callback_setter.set_callback(move |value| {
294                sender.send(value).expect("Could not send");
295            });
296        });
297
298        t1.join().expect("error joining thread");
299        t2.join().expect("error joining thread");
300        assert_eq!(receiver.recv().unwrap().unwrap(), true);
301    }
302
303    #[test]
304    fn lazy_callback_simple_inprocess() {
305        let (callback, callback_setter) = lazy_callback_inprocess();
306        test_lazy_callback(callback, callback_setter);
307    }
308
309    #[test]
310    fn lazy_callback_simple_ipc() {
311        let (callback, callback_setter) = lazy_callback_ipc();
312        test_lazy_callback(callback, callback_setter);
313    }
314}
servo_base/generic_channel/lazy_callback.rs

servo_base/generic_channel/
lazy_callback.rs
