zng_view_api/
ipc.rs

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, ViewChannelError>;
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/// IPC channel with view-process error.
213#[derive(Debug, Clone, PartialEq)]
214#[non_exhaustive]
215pub enum ViewChannelError {
216    /// IPC channel disconnected.
217    Disconnected,
218}
219impl fmt::Display for ViewChannelError {
220    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
221        write!(f, "ipc channel disconnected")
222    }
223}
224impl std::error::Error for ViewChannelError {}
225
226/// Call `new`, then spawn the view-process using the `name` then call `connect`.
227#[cfg(ipc)]
228pub(crate) struct AppInit {
229    server: IpcOneShotServer<AppInitMsg>,
230    name: Txt,
231}
232#[cfg(ipc)]
233impl AppInit {
234    pub fn new() -> Self {
235        let (server, name) = IpcOneShotServer::new().expect("failed to create init channel");
236        AppInit {
237            server,
238            name: Txt::from_str(&name),
239        }
240    }
241
242    /// Unique name for the view-process to find this channel.
243    pub fn name(&self) -> &str {
244        &self.name
245    }
246
247    /// Tries to connect to the view-process and receive the actual channels.
248    pub fn connect(self) -> AnyResult<(RequestSender, ResponseReceiver, EventReceiver)> {
249        let (init_sender, init_recv) = flume::bounded(1);
250        let handle = std::thread::spawn(move || {
251            let r = self.server.accept();
252            let _ = init_sender.send(r);
253        });
254
255        let (_, (req_sender, chan_sender)) = init_recv.recv_timeout(Duration::from_secs(10)).map_err(|e| match e {
256            flume::RecvTimeoutError::Timeout => "timeout, did not connect in 10 seconds",
257            flume::RecvTimeoutError::Disconnected => {
258                std::panic::resume_unwind(handle.join().unwrap_err());
259            }
260        })??;
261        let (rsp_sender, rsp_recv) = channel()?;
262        let (evt_sender, evt_recv) = channel()?;
263        chan_sender.send((rsp_sender, evt_sender))?;
264        Ok((
265            RequestSender(Mutex::new(req_sender)),
266            ResponseReceiver(Mutex::new(rsp_recv)),
267            EventReceiver(Mutex::new(evt_recv)),
268        ))
269    }
270}
271
272/// Start the view-process server and waits for `(request, response, event)`.
273#[cfg(ipc)]
274pub fn connect_view_process(server_name: Txt) -> IpcResult<ViewChannels> {
275    let _s = tracing::trace_span!("connect_view_process").entered();
276
277    let app_init_sender = IpcSender::connect(server_name.into_owned()).expect("failed to connect to init channel");
278
279    let (req_sender, req_recv) = channel().map_err(handle_io_error)?;
280    // Large messages can only be received in a receiver created in the same process that is receiving (on Windows)
281    // so we create a channel to transfer the response and event senders.
282    // See issue: https://github.com/servo/ipc-channel/issues/277
283    let (chan_sender, chan_recv) = channel().map_err(handle_io_error)?;
284
285    app_init_sender.send((req_sender, chan_sender)).map_err(handle_send_error)?;
286    let (rsp_sender, evt_sender) = chan_recv.recv().map_err(handle_recv_error)?;
287
288    Ok(ViewChannels {
289        request_receiver: RequestReceiver(Mutex::new(req_recv)),
290        response_sender: ResponseSender(Mutex::new(rsp_sender)),
291        event_sender: EventSender(Mutex::new(evt_sender)),
292    })
293}
294
295/// (
296///    RequestSender,
297///    Workaround-sender-for-response-channel,
298///    EventReceiver,
299/// )
300type AppInitMsg = (IpcSender<Request>, IpcSender<(IpcSender<Response>, IpcSender<Event>)>);
301
302#[cfg(not(ipc))]
303pub(crate) struct AppInit {
304    // (
305    //    RequestSender,
306    //    Workaround-sender-for-response-channel,
307    //    EventReceiver,
308    // )
309    init: flume::Receiver<AppInitMsg>,
310    name: Txt,
311}
312#[cfg(not(ipc))]
313mod name_map {
314    use std::{
315        collections::HashMap,
316        sync::{Mutex, OnceLock},
317    };
318
319    use zng_txt::Txt;
320
321    use super::AppInitMsg;
322
323    type Map = Mutex<HashMap<Txt, flume::Sender<AppInitMsg>>>;
324
325    pub fn get() -> &'static Map {
326        static MAP: OnceLock<Map> = OnceLock::new();
327        MAP.get_or_init(Map::default)
328    }
329}
330#[cfg(not(ipc))]
331impl AppInit {
332    pub fn new() -> Self {
333        use std::sync::atomic::{AtomicU32, Ordering};
334        use zng_txt::formatx;
335
336        static NAME_COUNT: AtomicU32 = AtomicU32::new(0);
337
338        let name = formatx!("<not-ipc-{}>", NAME_COUNT.fetch_add(1, Ordering::Relaxed));
339
340        let (init_sender, init_recv) = flume::bounded(1);
341
342        name_map::get().lock().unwrap().insert(name.clone(), init_sender);
343
344        AppInit { name, init: init_recv }
345    }
346
347    pub fn name(&self) -> &str {
348        &self.name
349    }
350
351    /// Tries to connect to the view-process and receive the actual channels.
352    pub fn connect(self) -> AnyResult<(RequestSender, ResponseReceiver, EventReceiver)> {
353        let (req_sender, chan_sender) = self.init.recv_timeout(Duration::from_secs(5)).map_err(|e| match e {
354            flume::RecvTimeoutError::Timeout => "timeout, did not connect in 5 seconds",
355            flume::RecvTimeoutError::Disconnected => panic!("disconnected"),
356        })?;
357        let (rsp_sender, rsp_recv) = flume::unbounded();
358        let (evt_sender, evt_recv) = flume::unbounded();
359        chan_sender.send((rsp_sender, evt_sender))?;
360        Ok((
361            RequestSender(Mutex::new(req_sender)),
362            ResponseReceiver(Mutex::new(rsp_recv)),
363            EventReceiver(Mutex::new(evt_recv)),
364        ))
365    }
366}
367
368/// Start the view-process server and waits for `(request, response, event)`.
369#[cfg(not(ipc))]
370pub fn connect_view_process(server_name: Txt) -> IpcResult<ViewChannels> {
371    let app_init_sender = name_map::get().lock().unwrap().remove(&server_name).unwrap();
372
373    let (req_sender, req_recv) = channel();
374    let (chan_sender, chan_recv) = channel();
375
376    app_init_sender.send((req_sender, chan_sender)).map_err(handle_send_error)?;
377    let (rsp_sender, evt_sender) = chan_recv.recv().map_err(handle_recv_error)?;
378
379    Ok(ViewChannels {
380        request_receiver: RequestReceiver(Mutex::new(req_recv)),
381        response_sender: ResponseSender(Mutex::new(rsp_sender)),
382        event_sender: EventSender(Mutex::new(evt_sender)),
383    })
384}
385
386/// Channels that must be used for implementing a view-process.
387pub struct ViewChannels {
388    /// View implementers must receive requests from this channel, call [`Api::respond`] and then
389    /// return the response using the `response_sender`.
390    ///
391    /// [`Api::respond`]: crate::Api::respond
392    pub request_receiver: RequestReceiver,
393
394    /// View implementers must synchronously send one response per request received in `request_receiver`.
395    pub response_sender: ResponseSender,
396
397    /// View implements must send events using this channel. Events can be asynchronous.
398    pub event_sender: EventSender,
399}
400
401pub(crate) struct RequestSender(Mutex<IpcSender<Request>>);
402impl RequestSender {
403    pub fn send(&mut self, req: Request) -> IpcResult<()> {
404        self.0.get_mut().send(req).map_err(handle_send_error)
405    }
406}
407
408/// Requests channel end-point.
409///
410/// View-process implementers must receive [`Request`], call [`Api::respond`] and then use a [`ResponseSender`]
411/// to send back the response.
412///
413/// [`Api::respond`]: crate::Api::respond
414pub struct RequestReceiver(Mutex<IpcReceiver<Request>>); // Mutex for Sync
415impl RequestReceiver {
416    /// Receive one [`Request`].
417    pub fn recv(&mut self) -> IpcResult<Request> {
418        self.0.get_mut().recv().map_err(handle_recv_error)
419    }
420}
421
422/// Responses channel entry-point.
423///
424/// View-process implementers must send [`Response`] returned by [`Api::respond`] using this sender.
425///
426/// Requests are received using [`RequestReceiver`] a response must be send for each request, synchronously.
427///
428/// [`Api::respond`]: crate::Api::respond
429pub struct ResponseSender(Mutex<IpcSender<Response>>); // Mutex for Sync
430impl ResponseSender {
431    /// Send a response.
432    ///
433    /// # Panics
434    ///
435    /// If the `rsp` is not [`must_be_send`].
436    ///
437    /// [`must_be_send`]: Response::must_be_send
438    pub fn send(&mut self, rsp: Response) -> IpcResult<()> {
439        assert!(rsp.must_be_send());
440        self.0.get_mut().send(rsp).map_err(handle_send_error)
441    }
442}
443pub(crate) struct ResponseReceiver(Mutex<IpcReceiver<Response>>);
444impl ResponseReceiver {
445    pub fn recv(&mut self) -> IpcResult<Response> {
446        self.0.get_mut().recv().map_err(handle_recv_error)
447    }
448}
449
450/// Event channel entry-point.
451///
452/// View-process implementers must send [`Event`] messages using this sender. The events
453/// can be asynchronous, not related to the [`Api::respond`] calls.
454///
455/// [`Api::respond`]: crate::Api::respond
456pub struct EventSender(Mutex<IpcSender<Event>>);
457impl EventSender {
458    /// Send an event notification.
459    pub fn send(&mut self, ev: Event) -> IpcResult<()> {
460        self.0.get_mut().send(ev).map_err(handle_send_error)
461    }
462}
463pub(crate) struct EventReceiver(Mutex<IpcReceiver<Event>>);
464impl EventReceiver {
465    pub fn recv(&mut self) -> IpcResult<Event> {
466        self.0.get_mut().recv().map_err(handle_recv_error)
467    }
468
469    #[cfg(ipc)]
470    pub fn recv_timeout(&mut self, duration: Duration) -> IpcResult<Option<Event>> {
471        match self.0.get_mut().try_recv_timeout(duration) {
472            Ok(ev) => Ok(Some(ev)),
473            Err(e) => match e {
474                ipc_channel::ipc::TryRecvError::IpcError(ipc_error) => Err(handle_recv_error(ipc_error)),
475                ipc_channel::ipc::TryRecvError::Empty => Ok(None),
476            },
477        }
478    }
479
480    #[cfg(not(ipc))]
481    pub fn recv_timeout(&mut self, duration: Duration) -> IpcResult<Option<Event>> {
482        match self.0.get_mut().recv_timeout(duration) {
483            Ok(ev) => Ok(Some(ev)),
484            Err(e) => match e {
485                flume::RecvTimeoutError::Timeout => Ok(None),
486                flume::RecvTimeoutError::Disconnected => Err(ViewChannelError::Disconnected),
487            },
488        }
489    }
490}
491
492#[cfg(ipc)]
493fn handle_recv_error(e: ipc_channel::ipc::IpcError) -> ViewChannelError {
494    match e {
495        ipc_channel::ipc::IpcError::Disconnected => ViewChannelError::Disconnected,
496        e => {
497            tracing::error!("IO or bincode error: {e:?}");
498            ViewChannelError::Disconnected
499        }
500    }
501}
502#[cfg(not(ipc))]
503fn handle_recv_error(e: flume::RecvError) -> ViewChannelError {
504    match e {
505        flume::RecvError::Disconnected => ViewChannelError::Disconnected,
506    }
507}
508
509#[cfg(ipc)]
510#[expect(clippy::boxed_local)]
511fn handle_send_error(e: ipc_channel::Error) -> ViewChannelError {
512    match *e {
513        ipc_channel::ErrorKind::Io(e) => {
514            if e.kind() == std::io::ErrorKind::BrokenPipe {
515                return ViewChannelError::Disconnected;
516            }
517            #[cfg(windows)]
518            if e.raw_os_error() == Some(-2147024664) {
519                // 0x800700E8 - "The pipe is being closed."
520                return ViewChannelError::Disconnected;
521            }
522            #[cfg(target_os = "macos")]
523            if e.kind() == std::io::ErrorKind::NotFound && format!("{e:?}") == "Custom { kind: NotFound, error: SendInvalidDest }" {
524                // this error happens in the same test that on Windows is 0x800700E8 and on Ubuntu is BrokenPipe
525                return ViewChannelError::Disconnected;
526            }
527            panic!("unexpected IO error: {e:?}")
528        }
529        e => panic!("serialization error: {e:?}"),
530    }
531}
532
533#[cfg(not(ipc))]
534fn handle_send_error<T>(_: flume::SendError<T>) -> ViewChannelError {
535    ViewChannelError::Disconnected
536}
537
538#[cfg(ipc)]
539fn handle_io_error(e: std::io::Error) -> ViewChannelError {
540    match e.kind() {
541        std::io::ErrorKind::BrokenPipe => ViewChannelError::Disconnected,
542        e => panic!("unexpected IO error: {e:?}"),
543    }
544}
zng_view_api/ipc.rs

zng_view_api/
ipc.rs
