aen_window/
worker.rs

1//! Socket progress and deadlines continue independently of application rendering.
2use crate::platform::Wake;
3use crate::state::Session;
4use crate::*;
5use std::collections::VecDeque;
6use std::os::fd::{AsRawFd, RawFd};
7use std::sync::atomic::{AtomicBool, Ordering};
8use std::sync::mpsc::{self, Receiver, SyncSender, TryRecvError, TrySendError};
9use std::sync::{Arc, Mutex, MutexGuard};
10use std::thread::{self, JoinHandle};
11
12const EVENT_CAPACITY: usize = 64;
13#[derive(Default)]
14struct Mailbox {
15    events: VecDeque<Event>,
16    failure: Option<(io::ErrorKind, String)>,
17}
18impl Mailbox {
19    fn push(&mut self, event: Event) -> io::Result<()> {
20        if let (
21            Some(Event::Input(Input {
22                frame_serial: previous_frame,
23                event: InputEvent::RelativeMotion { dx, dy, time_usec },
24            })),
25            Event::Input(Input {
26                frame_serial,
27                event:
28                    InputEvent::RelativeMotion {
29                        dx: next_dx,
30                        dy: next_dy,
31                        time_usec: next_time,
32                    },
33            }),
34        ) = (self.events.back_mut(), &event)
35        {
36            if previous_frame == frame_serial {
37                if let (Some(x), Some(y)) = (dx.checked_add(*next_dx), dy.checked_add(*next_dy)) {
38                    *dx = x;
39                    *dy = y;
40                    *time_usec = *next_time;
41                    return Ok(());
42                }
43                // An unrepresentable aggregate remains two essential events.
44            }
45        }
46        let incoming_motion = coalescible_motion(&event);
47        if self.events.len() == EVENT_CAPACITY {
48            if let Some(key) = incoming_motion {
49                if self.events.back().and_then(coalescible_motion) == Some(key) {
50                    *self.events.back_mut().expect("nonempty mailbox") = event;
51                }
52                // Motion cannot cross a queued lifecycle/configuration fence.
53                return Ok(());
54            }
55            if let Some(index) = self
56                .events
57                .iter()
58                .position(|event| coalescible_motion(event).is_some())
59            {
60                self.events.remove(index);
61            } else {
62                return Err(invalid("native window essential event queue is full"));
63            }
64        }
65        self.events.push_back(event);
66        Ok(())
67    }
68    fn fail(&mut self, error: io::Error) {
69        if self.failure.is_none() {
70            self.failure = Some((error.kind(), error.to_string().chars().take(160).collect()));
71        }
72    }
73}
74fn coalescible_motion(event: &Event) -> Option<(u64, Option<i32>)> {
75    let Event::Input(Input {
76        frame_serial,
77        event,
78    }) = event
79    else {
80        return None;
81    };
82    match event {
83        InputEvent::Motion { .. } => Some((*frame_serial, None)),
84        InputEvent::Touch {
85            id,
86            phase: TouchPhase::Motion,
87            ..
88        } => Some((*frame_serial, Some(*id))),
89        _ => None,
90    }
91}
92
93#[derive(Default)]
94struct Shared {
95    mailbox: Mutex<Mailbox>,
96    stop: AtomicBool,
97    exited: AtomicBool,
98}
99impl Shared {
100    fn mailbox(&self) -> MutexGuard<'_, Mailbox> {
101        self.mailbox.lock().unwrap_or_else(|e| e.into_inner())
102    }
103}
104struct ExitNotice {
105    shared: Arc<Shared>,
106    wake: Arc<Wake>,
107}
108impl Drop for ExitNotice {
109    fn drop(&mut self) {
110        self.shared.exited.store(true, Ordering::Release);
111        let _ = self.wake.signal();
112    }
113}
114
115enum Command {
116    Frame(Frame),
117    Confinement(PointerConfinementRequest),
118}
119
120/// One window connection with one owner-held frame/request credit. Receiving an
121/// ACK on the worker does not release credit until the application drains it.
122/// No connection, frame, input or close request is replayed after an error.
123pub struct WorkerWindow {
124    shared: Arc<Shared>,
125    frames: SyncSender<Command>,
126    owner_wake: Arc<Wake>,
127    worker_wake: Arc<Wake>,
128    worker: Option<JoinHandle<()>>,
129    state: Session,
130    failed: bool,
131    options: ClientOptions,
132}
133impl WorkerWindow {
134    pub fn connect(path: &Path, create: Create) -> io::Result<Self> {
135        Self::connect_with_options(path, create, ClientOptions::default())
136    }
137    /// Connect once with explicit options; unsupported servers are not retried.
138    pub fn connect_with_options(
139        path: &Path,
140        create: Create,
141        options: ClientOptions,
142    ) -> io::Result<Self> {
143        create.validate()?;
144        if !path.is_absolute() {
145            return Err(invalid("native window endpoint must be absolute"));
146        }
147        let shared = Arc::new(Shared::default());
148        let owner_wake = Arc::new(Wake::new()?);
149        let worker_wake = Arc::new(Wake::new()?);
150        let (frames, receiver) = mpsc::sync_channel(1);
151        let thread_shared = Arc::clone(&shared);
152        let thread_owner = Arc::clone(&owner_wake);
153        let thread_wake = Arc::clone(&worker_wake);
154        let path = path.to_owned();
155        let worker = thread::Builder::new()
156            .name("aen-window-client".into())
157            .spawn(move || {
158                let _exit = ExitNotice {
159                    shared: Arc::clone(&thread_shared),
160                    wake: Arc::clone(&thread_owner),
161                };
162                let result =
163                    Client::connect_with_options(&path, create, options).and_then(|client| {
164                        run(
165                            client,
166                            receiver,
167                            &thread_shared,
168                            &thread_owner,
169                            &thread_wake,
170                        )
171                    });
172                if let Err(error) = result {
173                    thread_shared.mailbox().fail(error);
174                }
175            })?;
176        Ok(Self {
177            shared,
178            frames,
179            owner_wake,
180            worker_wake,
181            worker: Some(worker),
182            state: Session::default(),
183            failed: false,
184            options,
185        })
186    }
187    /// Negotiation is visible only after this owner drains the matching configure.
188    pub fn relative_pointer_enabled(&self) -> bool {
189        !self.failed
190            && !self.shared.exited.load(Ordering::Acquire)
191            && self.options.relative_pointer
192            && self.state.configuration().is_some()
193    }
194    pub fn pointer_confinement_enabled(&self) -> bool {
195        !self.failed
196            && !self.shared.exited.load(Ordering::Acquire)
197            && self.options.pointer_confinement
198            && self.state.configuration().is_some()
199    }
200    pub fn pointer_confined(&self) -> bool {
201        !self.failed
202            && !self.shared.exited.load(Ordering::Acquire)
203            && self.state.confinement_active.is_some()
204    }
205    /// Queues one request. The owner must drain the matching reply before
206    /// submitting another request or frame; no grant is implied by this return.
207    pub fn request_pointer_confinement(&mut self, confined: bool) -> io::Result<u64> {
208        if !self.pointer_confinement_enabled() {
209            return Err(invalid("native pointer confinement was not negotiated"));
210        }
211        let request = self.state.next_confinement_request(confined)?;
212        match self.frames.try_send(Command::Confinement(request)) {
213            Ok(()) => {
214                self.state.request_confinement(request)?;
215                self.worker_wake.signal()?;
216                Ok(request.request_serial)
217            }
218            Err(TrySendError::Full(_)) => Err(io::ErrorKind::WouldBlock.into()),
219            Err(TrySendError::Disconnected(_)) => Err(io::ErrorKind::BrokenPipe.into()),
220        }
221    }
222    pub fn configuration(&self) -> Option<Configure> {
223        self.state.configuration()
224    }
225    pub fn window_id(&self) -> Option<u64> {
226        self.configuration().map(|c| c.window_id)
227    }
228    pub fn can_submit(&self) -> bool {
229        !self.failed
230            && !self.shared.exited.load(Ordering::Acquire)
231            && self.configuration().is_some()
232            && !self.state.credit_occupied()
233    }
234    pub fn wants_write(&self) -> bool {
235        false
236    }
237    pub fn submit(&mut self, frame: Frame) -> io::Result<()> {
238        if self.failed || self.shared.exited.load(Ordering::Acquire) {
239            return Err(io::ErrorKind::BrokenPipe.into());
240        }
241        self.state.validate_submit(&frame)?;
242        match self.frames.try_send(Command::Frame(frame.clone())) {
243            Ok(()) => {
244                self.state.submit(&frame)?;
245                self.worker_wake.signal()
246            }
247            Err(TrySendError::Full(_)) => Err(io::ErrorKind::WouldBlock.into()),
248            Err(TrySendError::Disconnected(_)) => Err(io::ErrorKind::BrokenPipe.into()),
249        }
250    }
251    pub fn poll(&mut self) -> io::Result<Vec<Event>> {
252        if self.failed {
253            return Err(io::ErrorKind::BrokenPipe.into());
254        }
255        self.owner_wake.drain()?;
256        let events = {
257            let mut mailbox = self.shared.mailbox();
258            if let Some((kind, message)) = mailbox.failure.take() {
259                self.failed = true;
260                self.shared.stop.store(true, Ordering::Release);
261                let _ = self.worker_wake.signal();
262                return Err(io::Error::new(kind, message));
263            }
264            if self.shared.exited.load(Ordering::Acquire) {
265                self.failed = true;
266                return Err(io::ErrorKind::BrokenPipe.into());
267            }
268            mailbox.events.drain(..).collect::<Vec<_>>()
269        };
270        for event in &events {
271            if let Err(error) = receive(&mut self.state, event.clone()) {
272                self.failed = true;
273                self.shared.stop.store(true, Ordering::Release);
274                let _ = self.worker_wake.signal();
275                return Err(error);
276            }
277        }
278        Ok(events)
279    }
280}
281fn receive(state: &mut Session, event: Event) -> io::Result<()> {
282    match event {
283        Event::Configured(c) => state.configure(c),
284        Event::FrameAccepted {
285            frame_serial,
286            configure_serial,
287        } => state.acknowledge(frame_serial, configure_serial, true),
288        Event::FrameDiscarded {
289            frame_serial,
290            configure_serial,
291        } => state.acknowledge(frame_serial, configure_serial, false),
292        Event::Input(input) => state.input(input),
293        Event::CloseRequested => Ok(()),
294    }
295}
296impl AsRawFd for WorkerWindow {
297    fn as_raw_fd(&self) -> RawFd {
298        self.owner_wake.as_raw_fd()
299    }
300}
301impl Drop for WorkerWindow {
302    fn drop(&mut self) {
303        self.shared.stop.store(true, Ordering::Release);
304        let _ = self.worker_wake.signal();
305        if let Some(worker) = self.worker.take() {
306            let _ = worker.join();
307        }
308    }
309}
310fn run(
311    mut client: Client,
312    frames: Receiver<Command>,
313    shared: &Shared,
314    owner: &Wake,
315    wake: &Wake,
316) -> io::Result<()> {
317    while !shared.stop.load(Ordering::Acquire) {
318        for event in client.poll()? {
319            shared.mailbox().push(event)?;
320            owner.signal()?;
321        }
322        match frames.try_recv() {
323            Ok(Command::Frame(frame)) => client.submit(frame)?,
324            Ok(Command::Confinement(request)) => {
325                let serial = client.request_pointer_confinement(request.confined)?;
326                if serial != request.request_serial {
327                    return Err(invalid("native pointer worker request mismatch"));
328                }
329            }
330            Err(TryRecvError::Empty) => {}
331            Err(TryRecvError::Disconnected) => break,
332        }
333        crate::platform::wait(client.as_raw_fd(), wake.as_raw_fd(), client.wants_write())?;
334        wake.drain()?;
335    }
336    Ok(())
337}
338
339#[cfg(test)]
340mod tests {
341    use super::*;
342    fn configure() -> Configure {
343        Configure {
344            window_id: 1,
345            configure_serial: 1,
346            width: 1,
347            height: 1,
348            scale: 1,
349        }
350    }
351    fn frame(serial: u64) -> Frame {
352        Frame {
353            frame_serial: serial,
354            configure_serial: 1,
355            width: 1,
356            height: 1,
357            pixels: vec![0; 4].into(),
358        }
359    }
360    #[test]
361    fn owner_credit_waits_until_application_drains_exact_ack() {
362        let mut owner = Session::default();
363        receive(&mut owner, Event::Configured(configure())).unwrap();
364        owner.submit(&frame(1)).unwrap();
365        let mut mailbox = Mailbox::default();
366        mailbox
367            .push(Event::FrameAccepted {
368                frame_serial: 1,
369                configure_serial: 1,
370            })
371            .unwrap();
372        assert!(owner.validate_submit(&frame(2)).is_err());
373        receive(&mut owner, mailbox.events.pop_front().unwrap()).unwrap();
374        owner.validate_submit(&frame(2)).unwrap();
375        assert!(receive(
376            &mut owner,
377            Event::FrameAccepted {
378                frame_serial: 1,
379                configure_serial: 1
380            }
381        )
382        .is_err());
383    }
384    #[test]
385    fn bounded_mailbox_never_drops_essential_input_or_crosses_motion_fences() {
386        let mut mailbox = Mailbox::default();
387        for _ in 0..EVENT_CAPACITY - 1 {
388            mailbox.push(Event::CloseRequested).unwrap();
389        }
390        let motion = |serial, x| {
391            Event::Input(Input {
392                frame_serial: serial,
393                event: InputEvent::Motion { x, y: 0 },
394            })
395        };
396        mailbox.push(motion(1, 1)).unwrap();
397        mailbox.push(motion(2, 2)).unwrap();
398        assert_eq!(mailbox.events.back(), Some(&motion(1, 1)));
399        mailbox.push(motion(1, 3)).unwrap();
400        assert_eq!(mailbox.events.back(), Some(&motion(1, 3)));
401        mailbox.push(Event::CloseRequested).unwrap();
402        assert_eq!(mailbox.events.len(), EVENT_CAPACITY);
403        assert!(mailbox.push(Event::CloseRequested).is_err());
404
405        let touch_motion = |id, x| {
406            Event::Input(Input {
407                frame_serial: 9,
408                event: InputEvent::Touch {
409                    id,
410                    phase: TouchPhase::Motion,
411                    x,
412                    y: 0,
413                },
414            })
415        };
416        let mut mailbox = Mailbox::default();
417        for _ in 0..EVENT_CAPACITY - 1 {
418            mailbox.push(Event::CloseRequested).unwrap();
419        }
420        mailbox.push(touch_motion(4, 1)).unwrap();
421        mailbox.push(touch_motion(4, 2)).unwrap();
422        assert_eq!(mailbox.events.back(), Some(&touch_motion(4, 2)));
423        mailbox.push(touch_motion(5, 3)).unwrap();
424        assert_eq!(mailbox.events.len(), EVENT_CAPACITY);
425        assert_eq!(mailbox.events.back(), Some(&touch_motion(4, 2)));
426        mailbox.push(touch_motion(4, 5)).unwrap();
427        assert_eq!(mailbox.events.back(), Some(&touch_motion(4, 5)));
428
429        let mut essential = Mailbox::default();
430        for _ in 0..EVENT_CAPACITY - 1 {
431            essential.push(Event::CloseRequested).unwrap();
432        }
433        essential.push(touch_motion(8, 4)).unwrap();
434        essential.push(Event::CloseRequested).unwrap();
435        assert_eq!(essential.events.len(), EVENT_CAPACITY);
436        assert!(!essential
437            .events
438            .iter()
439            .any(|event| coalescible_motion(event).is_some()));
440        assert!(essential
441            .push(Event::Input(Input {
442                frame_serial: 9,
443                event: InputEvent::Touch {
444                    id: 8,
445                    phase: TouchPhase::Down,
446                    x: 4,
447                    y: 0,
448                },
449            }))
450            .is_err());
451    }
452    #[test]
453    fn one_queued_frame_has_no_implicit_replacement_or_replay() {
454        let (sender, receiver) = mpsc::sync_channel(1);
455        sender.try_send(frame(1)).unwrap();
456        assert!(matches!(
457            sender.try_send(frame(2)),
458            Err(TrySendError::Full(_))
459        ));
460        assert_eq!(receiver.try_recv().unwrap().frame_serial, 1);
461        assert!(receiver.try_recv().is_err());
462    }
463    #[test]
464    fn error_text_is_bounded_and_first_failure_retained() {
465        let mut mailbox = Mailbox::default();
466        mailbox.fail(io::Error::new(io::ErrorKind::TimedOut, "x".repeat(4000)));
467        mailbox.fail(io::ErrorKind::BrokenPipe.into());
468        let (kind, message) = mailbox.failure.unwrap();
469        assert_eq!(kind, io::ErrorKind::TimedOut);
470        assert_eq!(message.len(), 160);
471    }
472}
473
474#[cfg(test)]
475mod relative_tests {
476    use super::*;
477    fn delta(frame: u64, dx: i32, dy: i32, time_usec: u64) -> Event {
478        Event::Input(Input {
479            frame_serial: frame,
480            event: InputEvent::RelativeMotion { dx, dy, time_usec },
481        })
482    }
483    #[test]
484    fn relative_mailbox_preserves_total_motion_and_frame_lifecycle_fences() {
485        let mut mailbox = Mailbox::default();
486        mailbox.push(delta(1, 20, -5, 1)).unwrap();
487        mailbox.push(delta(1, -3, 2, 2)).unwrap();
488        assert_eq!(mailbox.events.pop_front(), Some(delta(1, 17, -3, 2)));
489        mailbox.push(delta(1, 1, 2, 3)).unwrap();
490        mailbox
491            .push(Event::Input(Input {
492                frame_serial: 1,
493                event: InputEvent::Leave,
494            }))
495            .unwrap();
496        mailbox.push(delta(1, 3, 4, 4)).unwrap();
497        mailbox.push(delta(2, 5, 6, 5)).unwrap();
498        assert_eq!(mailbox.events.len(), 4);
499        assert_eq!(mailbox.events[0], delta(1, 1, 2, 3));
500        assert_eq!(mailbox.events[2], delta(1, 3, 4, 4));
501        assert_eq!(mailbox.events[3], delta(2, 5, 6, 5));
502    }
503    #[test]
504    fn relative_mailbox_overflow_is_essential_and_never_wraps_or_discards_deltas() {
505        let mut mailbox = Mailbox::default();
506        mailbox.push(delta(1, i32::MAX, i32::MIN, 1)).unwrap();
507        mailbox.push(delta(1, 1, -1, 2)).unwrap();
508        assert_eq!(mailbox.events.len(), 2);
509        assert_eq!(mailbox.events[0], delta(1, i32::MAX, i32::MIN, 1));
510        assert_eq!(mailbox.events[1], delta(1, 1, -1, 2));
511        for _ in mailbox.events.len()..EVENT_CAPACITY {
512            mailbox.push(Event::CloseRequested).unwrap();
513        }
514        assert!(mailbox.push(delta(1, 1, 1, 3)).is_err());
515        assert!(mailbox.push(Event::CloseRequested).is_err());
516        assert_eq!(mailbox.events[0], delta(1, i32::MAX, i32::MIN, 1));
517        assert_eq!(mailbox.events[1], delta(1, 1, -1, 2));
518    }
519}
aen_window/worker.rs

aen_window/
worker.rs
