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
115/// One window connection with one owner-held frame credit. Receiving an ACK on
116/// the transport worker does not release credit until the application drains it.
117/// No connection, frame, input or close request is replayed after an error.
118pub struct WorkerWindow {
119    shared: Arc<Shared>,
120    frames: SyncSender<Frame>,
121    owner_wake: Arc<Wake>,
122    worker_wake: Arc<Wake>,
123    worker: Option<JoinHandle<()>>,
124    state: Session,
125    failed: bool,
126    options: ClientOptions,
127}
128impl WorkerWindow {
129    pub fn connect(path: &Path, create: Create) -> io::Result<Self> {
130        Self::connect_with_options(path, create, ClientOptions::default())
131    }
132    /// Connect once with explicit options; unsupported servers are not retried.
133    pub fn connect_with_options(
134        path: &Path,
135        create: Create,
136        options: ClientOptions,
137    ) -> io::Result<Self> {
138        create.validate()?;
139        if !path.is_absolute() {
140            return Err(invalid("native window endpoint must be absolute"));
141        }
142        let shared = Arc::new(Shared::default());
143        let owner_wake = Arc::new(Wake::new()?);
144        let worker_wake = Arc::new(Wake::new()?);
145        let (frames, receiver) = mpsc::sync_channel(1);
146        let thread_shared = Arc::clone(&shared);
147        let thread_owner = Arc::clone(&owner_wake);
148        let thread_wake = Arc::clone(&worker_wake);
149        let path = path.to_owned();
150        let worker = thread::Builder::new()
151            .name("aen-window-client".into())
152            .spawn(move || {
153                let _exit = ExitNotice {
154                    shared: Arc::clone(&thread_shared),
155                    wake: Arc::clone(&thread_owner),
156                };
157                let result =
158                    Client::connect_with_options(&path, create, options).and_then(|client| {
159                        run(
160                            client,
161                            receiver,
162                            &thread_shared,
163                            &thread_owner,
164                            &thread_wake,
165                        )
166                    });
167                if let Err(error) = result {
168                    thread_shared.mailbox().fail(error);
169                }
170            })?;
171        Ok(Self {
172            shared,
173            frames,
174            owner_wake,
175            worker_wake,
176            worker: Some(worker),
177            state: Session::default(),
178            failed: false,
179            options,
180        })
181    }
182    /// Negotiation is visible only after this owner drains the matching configure.
183    pub fn relative_pointer_enabled(&self) -> bool {
184        !self.failed
185            && !self.shared.exited.load(Ordering::Acquire)
186            && self.options.relative_pointer
187            && self.state.configuration().is_some()
188    }
189    pub fn configuration(&self) -> Option<Configure> {
190        self.state.configuration()
191    }
192    pub fn window_id(&self) -> Option<u64> {
193        self.configuration().map(|c| c.window_id)
194    }
195    pub fn can_submit(&self) -> bool {
196        !self.failed
197            && !self.shared.exited.load(Ordering::Acquire)
198            && self.configuration().is_some()
199            && self.state.inflight.is_none()
200    }
201    pub fn wants_write(&self) -> bool {
202        false
203    }
204    pub fn submit(&mut self, frame: Frame) -> io::Result<()> {
205        if self.failed || self.shared.exited.load(Ordering::Acquire) {
206            return Err(io::ErrorKind::BrokenPipe.into());
207        }
208        self.state.validate_submit(&frame)?;
209        match self.frames.try_send(frame.clone()) {
210            Ok(()) => {
211                self.state.submit(&frame)?;
212                self.worker_wake.signal()
213            }
214            Err(TrySendError::Full(_)) => Err(io::ErrorKind::WouldBlock.into()),
215            Err(TrySendError::Disconnected(_)) => Err(io::ErrorKind::BrokenPipe.into()),
216        }
217    }
218    pub fn poll(&mut self) -> io::Result<Vec<Event>> {
219        if self.failed {
220            return Err(io::ErrorKind::BrokenPipe.into());
221        }
222        self.owner_wake.drain()?;
223        let events = {
224            let mut mailbox = self.shared.mailbox();
225            if let Some((kind, message)) = mailbox.failure.take() {
226                self.failed = true;
227                self.shared.stop.store(true, Ordering::Release);
228                let _ = self.worker_wake.signal();
229                return Err(io::Error::new(kind, message));
230            }
231            if self.shared.exited.load(Ordering::Acquire) {
232                self.failed = true;
233                return Err(io::ErrorKind::BrokenPipe.into());
234            }
235            mailbox.events.drain(..).collect::<Vec<_>>()
236        };
237        for event in &events {
238            if let Err(error) = receive(&mut self.state, event.clone()) {
239                self.failed = true;
240                self.shared.stop.store(true, Ordering::Release);
241                let _ = self.worker_wake.signal();
242                return Err(error);
243            }
244        }
245        Ok(events)
246    }
247}
248fn receive(state: &mut Session, event: Event) -> io::Result<()> {
249    match event {
250        Event::Configured(c) => state.configure(c),
251        Event::FrameAccepted {
252            frame_serial,
253            configure_serial,
254        } => state.acknowledge(frame_serial, configure_serial, true),
255        Event::FrameDiscarded {
256            frame_serial,
257            configure_serial,
258        } => state.acknowledge(frame_serial, configure_serial, false),
259        Event::Input(input) => state.input(input),
260        Event::CloseRequested => Ok(()),
261    }
262}
263impl AsRawFd for WorkerWindow {
264    fn as_raw_fd(&self) -> RawFd {
265        self.owner_wake.as_raw_fd()
266    }
267}
268impl Drop for WorkerWindow {
269    fn drop(&mut self) {
270        self.shared.stop.store(true, Ordering::Release);
271        let _ = self.worker_wake.signal();
272        if let Some(worker) = self.worker.take() {
273            let _ = worker.join();
274        }
275    }
276}
277fn run(
278    mut client: Client,
279    frames: Receiver<Frame>,
280    shared: &Shared,
281    owner: &Wake,
282    wake: &Wake,
283) -> io::Result<()> {
284    while !shared.stop.load(Ordering::Acquire) {
285        for event in client.poll()? {
286            shared.mailbox().push(event)?;
287            owner.signal()?;
288        }
289        match frames.try_recv() {
290            Ok(frame) => client.submit(frame)?,
291            Err(TryRecvError::Empty) => {}
292            Err(TryRecvError::Disconnected) => break,
293        }
294        crate::platform::wait(client.as_raw_fd(), wake.as_raw_fd(), client.wants_write())?;
295        wake.drain()?;
296    }
297    Ok(())
298}
299
300#[cfg(test)]
301mod tests {
302    use super::*;
303    fn configure() -> Configure {
304        Configure {
305            window_id: 1,
306            configure_serial: 1,
307            width: 1,
308            height: 1,
309            scale: 1,
310        }
311    }
312    fn frame(serial: u64) -> Frame {
313        Frame {
314            frame_serial: serial,
315            configure_serial: 1,
316            width: 1,
317            height: 1,
318            pixels: vec![0; 4].into(),
319        }
320    }
321    #[test]
322    fn owner_credit_waits_until_application_drains_exact_ack() {
323        let mut owner = Session::default();
324        receive(&mut owner, Event::Configured(configure())).unwrap();
325        owner.submit(&frame(1)).unwrap();
326        let mut mailbox = Mailbox::default();
327        mailbox
328            .push(Event::FrameAccepted {
329                frame_serial: 1,
330                configure_serial: 1,
331            })
332            .unwrap();
333        assert!(owner.validate_submit(&frame(2)).is_err());
334        receive(&mut owner, mailbox.events.pop_front().unwrap()).unwrap();
335        owner.validate_submit(&frame(2)).unwrap();
336        assert!(receive(
337            &mut owner,
338            Event::FrameAccepted {
339                frame_serial: 1,
340                configure_serial: 1
341            }
342        )
343        .is_err());
344    }
345    #[test]
346    fn bounded_mailbox_never_drops_essential_input_or_crosses_motion_fences() {
347        let mut mailbox = Mailbox::default();
348        for _ in 0..EVENT_CAPACITY - 1 {
349            mailbox.push(Event::CloseRequested).unwrap();
350        }
351        let motion = |serial, x| {
352            Event::Input(Input {
353                frame_serial: serial,
354                event: InputEvent::Motion { x, y: 0 },
355            })
356        };
357        mailbox.push(motion(1, 1)).unwrap();
358        mailbox.push(motion(2, 2)).unwrap();
359        assert_eq!(mailbox.events.back(), Some(&motion(1, 1)));
360        mailbox.push(motion(1, 3)).unwrap();
361        assert_eq!(mailbox.events.back(), Some(&motion(1, 3)));
362        mailbox.push(Event::CloseRequested).unwrap();
363        assert_eq!(mailbox.events.len(), EVENT_CAPACITY);
364        assert!(mailbox.push(Event::CloseRequested).is_err());
365
366        let touch_motion = |id, x| {
367            Event::Input(Input {
368                frame_serial: 9,
369                event: InputEvent::Touch {
370                    id,
371                    phase: TouchPhase::Motion,
372                    x,
373                    y: 0,
374                },
375            })
376        };
377        let mut mailbox = Mailbox::default();
378        for _ in 0..EVENT_CAPACITY - 1 {
379            mailbox.push(Event::CloseRequested).unwrap();
380        }
381        mailbox.push(touch_motion(4, 1)).unwrap();
382        mailbox.push(touch_motion(4, 2)).unwrap();
383        assert_eq!(mailbox.events.back(), Some(&touch_motion(4, 2)));
384        mailbox.push(touch_motion(5, 3)).unwrap();
385        assert_eq!(mailbox.events.len(), EVENT_CAPACITY);
386        assert_eq!(mailbox.events.back(), Some(&touch_motion(4, 2)));
387        mailbox.push(touch_motion(4, 5)).unwrap();
388        assert_eq!(mailbox.events.back(), Some(&touch_motion(4, 5)));
389
390        let mut essential = Mailbox::default();
391        for _ in 0..EVENT_CAPACITY - 1 {
392            essential.push(Event::CloseRequested).unwrap();
393        }
394        essential.push(touch_motion(8, 4)).unwrap();
395        essential.push(Event::CloseRequested).unwrap();
396        assert_eq!(essential.events.len(), EVENT_CAPACITY);
397        assert!(!essential
398            .events
399            .iter()
400            .any(|event| coalescible_motion(event).is_some()));
401        assert!(essential
402            .push(Event::Input(Input {
403                frame_serial: 9,
404                event: InputEvent::Touch {
405                    id: 8,
406                    phase: TouchPhase::Down,
407                    x: 4,
408                    y: 0,
409                },
410            }))
411            .is_err());
412    }
413    #[test]
414    fn one_queued_frame_has_no_implicit_replacement_or_replay() {
415        let (sender, receiver) = mpsc::sync_channel(1);
416        sender.try_send(frame(1)).unwrap();
417        assert!(matches!(
418            sender.try_send(frame(2)),
419            Err(TrySendError::Full(_))
420        ));
421        assert_eq!(receiver.try_recv().unwrap().frame_serial, 1);
422        assert!(receiver.try_recv().is_err());
423    }
424    #[test]
425    fn error_text_is_bounded_and_first_failure_retained() {
426        let mut mailbox = Mailbox::default();
427        mailbox.fail(io::Error::new(io::ErrorKind::TimedOut, "x".repeat(4000)));
428        mailbox.fail(io::ErrorKind::BrokenPipe.into());
429        let (kind, message) = mailbox.failure.unwrap();
430        assert_eq!(kind, io::ErrorKind::TimedOut);
431        assert_eq!(message.len(), 160);
432    }
433}
434
435#[cfg(test)]
436mod relative_tests {
437    use super::*;
438    fn delta(frame: u64, dx: i32, dy: i32, time_usec: u64) -> Event {
439        Event::Input(Input {
440            frame_serial: frame,
441            event: InputEvent::RelativeMotion { dx, dy, time_usec },
442        })
443    }
444    #[test]
445    fn relative_mailbox_preserves_total_motion_and_frame_lifecycle_fences() {
446        let mut mailbox = Mailbox::default();
447        mailbox.push(delta(1, 20, -5, 1)).unwrap();
448        mailbox.push(delta(1, -3, 2, 2)).unwrap();
449        assert_eq!(mailbox.events.pop_front(), Some(delta(1, 17, -3, 2)));
450        mailbox.push(delta(1, 1, 2, 3)).unwrap();
451        mailbox
452            .push(Event::Input(Input {
453                frame_serial: 1,
454                event: InputEvent::Leave,
455            }))
456            .unwrap();
457        mailbox.push(delta(1, 3, 4, 4)).unwrap();
458        mailbox.push(delta(2, 5, 6, 5)).unwrap();
459        assert_eq!(mailbox.events.len(), 4);
460        assert_eq!(mailbox.events[0], delta(1, 1, 2, 3));
461        assert_eq!(mailbox.events[2], delta(1, 3, 4, 4));
462        assert_eq!(mailbox.events[3], delta(2, 5, 6, 5));
463    }
464    #[test]
465    fn relative_mailbox_overflow_is_essential_and_never_wraps_or_discards_deltas() {
466        let mut mailbox = Mailbox::default();
467        mailbox.push(delta(1, i32::MAX, i32::MIN, 1)).unwrap();
468        mailbox.push(delta(1, 1, -1, 2)).unwrap();
469        assert_eq!(mailbox.events.len(), 2);
470        assert_eq!(mailbox.events[0], delta(1, i32::MAX, i32::MIN, 1));
471        assert_eq!(mailbox.events[1], delta(1, 1, -1, 2));
472        for _ in mailbox.events.len()..EVENT_CAPACITY {
473            mailbox.push(Event::CloseRequested).unwrap();
474        }
475        assert!(mailbox.push(delta(1, 1, 1, 3)).is_err());
476        assert!(mailbox.push(Event::CloseRequested).is_err());
477        assert_eq!(mailbox.events[0], delta(1, i32::MAX, i32::MIN, 1));
478        assert_eq!(mailbox.events[1], delta(1, 1, -1, 2));
479    }
480}
aen_window/worker.rs

aen_window/
worker.rs
