zng_view_api/
ipc.rs

Help
1//! IPC types.
2
3use std::{fmt, ops::Deref, time::Duration};
4
5use crate::{AnyResult, Event, Request, Response};
6
7#[cfg(ipc)]
8use ipc_channel::ipc::{IpcOneShotServer, IpcReceiver, IpcSender, channel};
9
10#[cfg(not(ipc))]
11use flume::unbounded as channel;
12
13use parking_lot::Mutex;
14use serde::{Deserialize, Serialize};
15use zng_txt::Txt;
16
17pub(crate) type IpcResult<T> = std::result::Result<T, Disconnected>;
18
19/// Bytes sender.
20///
21/// Use [`bytes_channel`] to create.
22#[cfg_attr(ipc, derive(serde::Serialize, serde::Deserialize))]
23pub struct IpcBytesSender {
24    #[cfg(ipc)]
25    sender: ipc_channel::ipc::IpcBytesSender,
26    #[cfg(not(ipc))]
27    sender: flume::Sender<Vec<u8>>,
28}
29impl IpcBytesSender {
30    /// Send a byte package.
31    pub fn send(&self, bytes: Vec<u8>) -> IpcResult<()> {
32        #[cfg(ipc)]
33        {
34            self.sender.send(&bytes).map_err(handle_io_error)
35        }
36
37        #[cfg(not(ipc))]
38        self.sender.send(bytes).map_err(handle_send_error)
39    }
40}
41impl fmt::Debug for IpcBytesSender {
42    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
43        write!(f, "IpcBytesSender")
44    }
45}
46
47/// Bytes receiver.
48///
49/// Use [`bytes_channel`] to create.
50#[cfg_attr(ipc, derive(serde::Serialize, serde::Deserialize))]
51pub struct IpcBytesReceiver {
52    #[cfg(ipc)]
53    recv: ipc_channel::ipc::IpcBytesReceiver,
54    #[cfg(not(ipc))]
55    recv: flume::Receiver<Vec<u8>>,
56}
57impl IpcBytesReceiver {
58    /// Receive a bytes package.
59    pub fn recv(&self) -> IpcResult<Vec<u8>> {
60        self.recv.recv().map_err(handle_recv_error)
61    }
62}
63impl fmt::Debug for IpcBytesReceiver {
64    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
65        write!(f, "IpcBytesReceiver")
66    }
67}
68
69/// Create a bytes channel.
70#[cfg(ipc)]
71pub fn bytes_channel() -> (IpcBytesSender, IpcBytesReceiver) {
72    let (sender, recv) = ipc_channel::ipc::bytes_channel().unwrap();
73    (IpcBytesSender { sender }, IpcBytesReceiver { recv })
74}
75
76/// Create a bytes channel.
77#[cfg(not(ipc))]
78pub fn bytes_channel() -> (IpcBytesSender, IpcBytesReceiver) {
79    let (sender, recv) = flume::unbounded();
80    (IpcBytesSender { sender }, IpcBytesReceiver { recv })
81}
82
83#[cfg(not(ipc))]
84mod arc_bytes {
85    pub fn serialize<S>(bytes: &std::sync::Arc<Vec<u8>>, serializer: S) -> Result<S::Ok, S::Error>
86    where
87        S: serde::Serializer,
88    {
89        serde_bytes::serialize(&bytes[..], serializer)
90    }
91    pub fn deserialize<'de, D>(deserializer: D) -> Result<std::sync::Arc<Vec<u8>>, D::Error>
92    where
93        D: serde::Deserializer<'de>,
94    {
95        Ok(std::sync::Arc::new(serde_bytes::deserialize(deserializer)?))
96    }
97}
98
99/// Immutable shared memory that can be send fast over IPC.
100///
101/// # `not(feature="ipc")`
102///
103/// If the default `"ipc"` feature is disabled this is only a `Vec<u8>`.
104#[derive(Clone, Serialize, Deserialize)]
105pub struct IpcBytes {
106    // `IpcSharedMemory` cannot have zero length, we use `None` in this case.
107    #[cfg(ipc)]
108    bytes: Option<ipc_channel::ipc::IpcSharedMemory>,
109    // `IpcSharedMemory` only clones a pointer.
110    #[cfg(not(ipc))]
111    #[serde(with = "arc_bytes")]
112    bytes: std::sync::Arc<Vec<u8>>,
113}
114/// Pointer equal.
115impl PartialEq for IpcBytes {
116    #[cfg(not(ipc))]
117    fn eq(&self, other: &Self) -> bool {
118        std::sync::Arc::ptr_eq(&self.bytes, &other.bytes)
119    }
120
121    #[cfg(ipc)]
122    fn eq(&self, other: &Self) -> bool {
123        match (&self.bytes, &other.bytes) {
124            (None, None) => true,
125            (Some(a), Some(b)) => a.as_ptr() == b.as_ptr(),
126            _ => false,
127        }
128    }
129}
130impl IpcBytes {
131    /// Copy the `bytes` to a new shared memory allocation.
132    pub fn from_slice(bytes: &[u8]) -> Self {
133        IpcBytes {
134            #[cfg(ipc)]
135            bytes: {
136                if bytes.is_empty() {
137                    None
138                } else {
139                    Some(ipc_channel::ipc::IpcSharedMemory::from_bytes(bytes))
140                }
141            },
142            #[cfg(not(ipc))]
143            bytes: std::sync::Arc::new(bytes.to_vec()),
144        }
145    }
146
147    /// If the `"ipc"` feature is enabled copy the bytes to a new shared memory region, if not
148    /// just wraps the `bytes` in a shared pointer.
149    pub fn from_vec(bytes: Vec<u8>) -> Self {
150        #[cfg(ipc)]
151        {
152            Self::from_slice(&bytes)
153        }
154
155        #[cfg(not(ipc))]
156        IpcBytes {
157            bytes: std::sync::Arc::new(bytes),
158        }
159    }
160
161    /// Copy the shared bytes to a new vec.
162    ///
163    /// If the `"ipc"` feature is not enabled and `self` is the only reference this operation is zero-cost.
164    pub fn to_vec(self) -> Vec<u8> {
165        #[cfg(ipc)]
166        {
167            self.bytes.map(|s| s.to_vec()).unwrap_or_default()
168        }
169        #[cfg(not(ipc))]
170        {
171            match std::sync::Arc::try_unwrap(self.bytes) {
172                Ok(d) => d,
173                Err(a) => a.as_ref().to_vec(),
174            }
175        }
176    }
177
178    /// Returns the underlying shared memory reference, if the bytes are not zero-length.
179    #[cfg(ipc)]
180    pub fn ipc_shared_memory(&self) -> Option<ipc_channel::ipc::IpcSharedMemory> {
181        self.bytes.clone()
182    }
183
184    /// Returns the underlying shared reference.
185    #[cfg(not(ipc))]
186    pub fn arc(&self) -> std::sync::Arc<Vec<u8>> {
187        self.bytes.clone()
188    }
189}
190impl Deref for IpcBytes {
191    type Target = [u8];
192
193    fn deref(&self) -> &Self::Target {
194        #[cfg(ipc)]
195        return if let Some(bytes) = &self.bytes { bytes } else { &[] };
196
197        #[cfg(not(ipc))]
198        &self.bytes
199    }
200}
201impl fmt::Debug for IpcBytes {
202    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
203        write!(f, "IpcBytes(<{} bytes>)", self.len())
204    }
205}
206
207#[cfg(not(ipc))]
208type IpcSender<T> = flume::Sender<T>;
209#[cfg(not(ipc))]
210type IpcReceiver<T> = flume::Receiver<T>;
211
212/// Channel disconnected error.
213#[derive(Debug)]
214pub struct Disconnected;
215impl fmt::Display for Disconnected {
216    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
217        write!(f, "ipc channel disconnected")
218    }
219}
220impl std::error::Error for Disconnected {}
221
222/// Call `new`, then spawn the view-process using the `name` then call `connect`.
223#[cfg(ipc)]
224pub(crate) struct AppInit {
225    server: IpcOneShotServer<AppInitMsg>,
226    name: Txt,
227}
228#[cfg(ipc)]
229impl AppInit {
230    pub fn new() -> Self {
231        let (server, name) = IpcOneShotServer::new().expect("failed to create init channel");
232        AppInit {
233            server,
234            name: Txt::from_str(&name),
235        }
236    }
237
238    /// Unique name for the view-process to find this channel.
239    pub fn name(&self) -> &str {
240        &self.name
241    }
242
243    /// Tries to connect to the view-process and receive the actual channels.
244    pub fn connect(self) -> AnyResult<(RequestSender, ResponseReceiver, EventReceiver)> {
245        let (init_sender, init_recv) = flume::bounded(1);
246        let handle = std::thread::spawn(move || {
247            let r = self.server.accept();
248            let _ = init_sender.send(r);
249        });
250
251        let (_, (req_sender, chan_sender)) = init_recv.recv_timeout(Duration::from_secs(10)).map_err(|e| match e {
252            flume::RecvTimeoutError::Timeout => "timeout, did not connect in 10 seconds",
253            flume::RecvTimeoutError::Disconnected => {
254                std::panic::resume_unwind(handle.join().unwrap_err());
255            }
256        })??;
257        let (rsp_sender, rsp_recv) = channel()?;
258        let (evt_sender, evt_recv) = channel()?;
259        chan_sender.send((rsp_sender, evt_sender))?;
260        Ok((
261            RequestSender(Mutex::new(req_sender)),
262            ResponseReceiver(Mutex::new(rsp_recv)),
263            EventReceiver(Mutex::new(evt_recv)),
264        ))
265    }
266}
267
268/// Start the view-process server and waits for `(request, response, event)`.
269#[cfg(ipc)]
270pub fn connect_view_process(server_name: Txt) -> IpcResult<ViewChannels> {
271    let _s = tracing::trace_span!("connect_view_process").entered();
272
273    let app_init_sender = IpcSender::connect(server_name.into_owned()).expect("failed to connect to init channel");
274
275    let (req_sender, req_recv) = channel().map_err(handle_io_error)?;
276    // Large messages can only be received in a receiver created in the same process that is receiving (on Windows)
277    // so we create a channel to transfer the response and event senders.
278    // See issue: https://github.com/servo/ipc-channel/issues/277
279    let (chan_sender, chan_recv) = channel().map_err(handle_io_error)?;
280
281    app_init_sender.send((req_sender, chan_sender)).map_err(handle_send_error)?;
282    let (rsp_sender, evt_sender) = chan_recv.recv().map_err(handle_recv_error)?;
283
284    Ok(ViewChannels {
285        request_receiver: RequestReceiver(Mutex::new(req_recv)),
286        response_sender: ResponseSender(Mutex::new(rsp_sender)),
287        event_sender: EventSender(Mutex::new(evt_sender)),
288    })
289}
290
291/// (
292///    RequestSender,
293///    Workaround-sender-for-response-channel,
294///    EventReceiver,
295/// )
296type AppInitMsg = (IpcSender<Request>, IpcSender<(IpcSender<Response>, IpcSender<Event>)>);
297
298#[cfg(not(ipc))]
299pub(crate) struct AppInit {
300    // (
301    //    RequestSender,
302    //    Workaround-sender-for-response-channel,
303    //    EventReceiver,
304    // )
305    init: flume::Receiver<AppInitMsg>,
306    name: Txt,
307}
308#[cfg(not(ipc))]
309mod name_map {
310    use std::{
311        collections::HashMap,
312        sync::{Mutex, OnceLock},
313    };
314
315    use zng_txt::Txt;
316
317    use super::AppInitMsg;
318
319    type Map = Mutex<HashMap<Txt, flume::Sender<AppInitMsg>>>;
320
321    pub fn get() -> &'static Map {
322        static MAP: OnceLock<Map> = OnceLock::new();
323        MAP.get_or_init(Map::default)
324    }
325}
326#[cfg(not(ipc))]
327impl AppInit {
328    pub fn new() -> Self {
329        use std::sync::atomic::{AtomicU32, Ordering};
330        use zng_txt::formatx;
331
332        static NAME_COUNT: AtomicU32 = AtomicU32::new(0);
333
334        let name = formatx!("<not-ipc-{}>", NAME_COUNT.fetch_add(1, Ordering::Relaxed));
335
336        let (init_sender, init_recv) = flume::bounded(1);
337
338        name_map::get().lock().unwrap().insert(name.clone(), init_sender);
339
340        AppInit { name, init: init_recv }
341    }
342
343    pub fn name(&self) -> &str {
344        &self.name
345    }
346
347    /// Tries to connect to the view-process and receive the actual channels.
348    pub fn connect(self) -> AnyResult<(RequestSender, ResponseReceiver, EventReceiver)> {
349        let (req_sender, chan_sender) = self.init.recv_timeout(Duration::from_secs(5)).map_err(|e| match e {
350            flume::RecvTimeoutError::Timeout => "timeout, did not connect in 5 seconds",
351            flume::RecvTimeoutError::Disconnected => panic!("disconnected"),
352        })?;
353        let (rsp_sender, rsp_recv) = flume::unbounded();
354        let (evt_sender, evt_recv) = flume::unbounded();
355        chan_sender.send((rsp_sender, evt_sender))?;
356        Ok((
357            RequestSender(Mutex::new(req_sender)),
358            ResponseReceiver(Mutex::new(rsp_recv)),
359            EventReceiver(Mutex::new(evt_recv)),
360        ))
361    }
362}
363
364/// Start the view-process server and waits for `(request, response, event)`.
365#[cfg(not(ipc))]
366pub fn connect_view_process(server_name: Txt) -> IpcResult<ViewChannels> {
367    let app_init_sender = name_map::get().lock().unwrap().remove(&server_name).unwrap();
368
369    let (req_sender, req_recv) = channel();
370    let (chan_sender, chan_recv) = channel();
371
372    app_init_sender.send((req_sender, chan_sender)).map_err(handle_send_error)?;
373    let (rsp_sender, evt_sender) = chan_recv.recv().map_err(handle_recv_error)?;
374
375    Ok(ViewChannels {
376        request_receiver: RequestReceiver(Mutex::new(req_recv)),
377        response_sender: ResponseSender(Mutex::new(rsp_sender)),
378        event_sender: EventSender(Mutex::new(evt_sender)),
379    })
380}
381
382/// Channels that must be used for implementing a view-process.
383pub struct ViewChannels {
384    /// View implementers must receive requests from this channel, call [`Api::respond`] and then
385    /// return the response using the `response_sender`.
386    ///
387    /// [`Api::respond`]: crate::Api::respond
388    pub request_receiver: RequestReceiver,
389
390    /// View implementers must synchronously send one response per request received in `request_receiver`.
391    pub response_sender: ResponseSender,
392
393    /// View implements must send events using this channel. Events can be asynchronous.
394    pub event_sender: EventSender,
395}
396
397pub(crate) struct RequestSender(Mutex<IpcSender<Request>>);
398impl RequestSender {
399    pub fn send(&mut self, req: Request) -> IpcResult<()> {
400        self.0.get_mut().send(req).map_err(handle_send_error)
401    }
402}
403
404/// Requests channel end-point.
405///
406/// View-process implementers must receive [`Request`], call [`Api::respond`] and then use a [`ResponseSender`]
407/// to send back the response.
408///
409/// [`Api::respond`]: crate::Api::respond
410pub struct RequestReceiver(Mutex<IpcReceiver<Request>>); // Mutex for Sync
411impl RequestReceiver {
412    /// Receive one [`Request`].
413    pub fn recv(&mut self) -> IpcResult<Request> {
414        self.0.get_mut().recv().map_err(handle_recv_error)
415    }
416}
417
418/// Responses channel entry-point.
419///
420/// View-process implementers must send [`Response`] returned by [`Api::respond`] using this sender.
421///
422/// Requests are received using [`RequestReceiver`] a response must be send for each request, synchronously.
423///
424/// [`Api::respond`]: crate::Api::respond
425pub struct ResponseSender(Mutex<IpcSender<Response>>); // Mutex for Sync
426impl ResponseSender {
427    /// Send a response.
428    ///
429    /// # Panics
430    ///
431    /// If the `rsp` is not [`must_be_send`].
432    ///
433    /// [`must_be_send`]: Response::must_be_send
434    pub fn send(&mut self, rsp: Response) -> IpcResult<()> {
435        assert!(rsp.must_be_send());
436        self.0.get_mut().send(rsp).map_err(handle_send_error)
437    }
438}
439pub(crate) struct ResponseReceiver(Mutex<IpcReceiver<Response>>);
440impl ResponseReceiver {
441    pub fn recv(&mut self) -> IpcResult<Response> {
442        self.0.get_mut().recv().map_err(handle_recv_error)
443    }
444}
445
446/// Event channel entry-point.
447///
448/// View-process implementers must send [`Event`] messages using this sender. The events
449/// can be asynchronous, not related to the [`Api::respond`] calls.
450///
451/// [`Api::respond`]: crate::Api::respond
452pub struct EventSender(Mutex<IpcSender<Event>>);
453impl EventSender {
454    /// Send an event notification.
455    pub fn send(&mut self, ev: Event) -> IpcResult<()> {
456        self.0.get_mut().send(ev).map_err(handle_send_error)
457    }
458}
459pub(crate) struct EventReceiver(Mutex<IpcReceiver<Event>>);
460impl EventReceiver {
461    pub fn recv(&mut self) -> IpcResult<Event> {
462        self.0.get_mut().recv().map_err(handle_recv_error)
463    }
464}
465
466#[cfg(ipc)]
467fn handle_recv_error(e: ipc_channel::ipc::IpcError) -> Disconnected {
468    match e {
469        ipc_channel::ipc::IpcError::Disconnected => Disconnected,
470        e => {
471            tracing::error!("IO or bincode error: {e:?}");
472            Disconnected
473        }
474    }
475}
476#[cfg(not(ipc))]
477fn handle_recv_error(e: flume::RecvError) -> Disconnected {
478    match e {
479        flume::RecvError::Disconnected => Disconnected,
480    }
481}
482
483#[cfg(ipc)]
484#[expect(clippy::boxed_local)]
485fn handle_send_error(e: ipc_channel::Error) -> Disconnected {
486    match *e {
487        ipc_channel::ErrorKind::Io(e) => {
488            if e.kind() == std::io::ErrorKind::BrokenPipe {
489                return Disconnected;
490            }
491            #[cfg(windows)]
492            if e.raw_os_error() == Some(-2147024664) {
493                // 0x800700E8 - "The pipe is being closed."
494                return Disconnected;
495            }
496            #[cfg(target_os = "macos")]
497            if e.kind() == std::io::ErrorKind::NotFound && format!("{e:?}") == "Custom { kind: NotFound, error: SendInvalidDest }" {
498                // this error happens in the same test that on Windows is 0x800700E8 and on Ubuntu is BrokenPipe
499                return Disconnected;
500            }
501            panic!("unexpected IO error: {e:?}")
502        }
503        e => panic!("serialization error: {e:?}"),
504    }
505}
506
507#[cfg(not(ipc))]
508fn handle_send_error<T>(_: flume::SendError<T>) -> Disconnected {
509    Disconnected
510}
511
512#[cfg(ipc)]
513fn handle_io_error(e: std::io::Error) -> Disconnected {
514    match e.kind() {
515        std::io::ErrorKind::BrokenPipe => Disconnected,
516        e => panic!("unexpected IO error: {e:?}"),
517    }
518}
519
520#[cfg(all(test, ipc))]
521mod tests {
522    use std::thread;
523
524    use zng_txt::ToTxt;
525
526    use super::*;
527    use crate::RequestData;
528
529    #[test]
530    fn disconnect_recv() {
531        let app = AppInit::new();
532
533        let name = app.name().to_txt();
534
535        let view = thread::spawn(move || {
536            let _channels = connect_view_process(name);
537        });
538
539        let (_request_sender, mut response_recv, _event_recv) = app.connect().unwrap();
540
541        view.join().unwrap();
542
543        let _ = response_recv.recv().unwrap_err();
544    }
545
546    #[test]
547    fn disconnect_send() {
548        let app = AppInit::new();
549
550        let name = app.name().to_txt();
551
552        let view = thread::spawn(move || {
553            let _channels = connect_view_process(name);
554        });
555
556        let (mut request_sender, _response_recv, _event_recv) = app.connect().unwrap();
557
558        view.join().unwrap();
559
560        let _ = request_sender
561            .send(Request(RequestData::close {
562                id: crate::window::WindowId::INVALID,
563            }))
564            .unwrap_err();
565    }
566}
zng_view_api/ipc.rs

zng_view_api/
ipc.rs
