pixelflux/wayland/
wlclient.rs

1//! Shared plumbing for pixelflux's outbound Wayland CLIENT connections (the
2//! virtual-keyboard typer and the data-control clipboard bridge, both talking to
3//! a nested app compositor): socket resolution, deadline-bounded round-trips so a
4//! wedged compositor turns into an error instead of a hang, and fd read/write
5//! helpers for clipboard pipes.
6
7use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
8use std::time::{Duration, Instant};
9
10use wayland_client::protocol::wl_callback;
11use wayland_client::{Connection, Dispatch, EventQueue};
12
13pub(crate) const IO_TIMEOUT: Duration = Duration::from_secs(5);
14
15/// Absolute socket path for a Wayland display name (absolute paths pass through,
16/// names join XDG_RUNTIME_DIR).
17pub(crate) fn socket_path(name: &str) -> Option<String> {
18    if name.starts_with('/') {
19        return Some(name.to_string());
20    }
21    let rt = std::env::var("XDG_RUNTIME_DIR").ok()?;
22    Some(format!("{}/{}", rt.trim_end_matches('/'), name))
23}
24
25/// Implemented by every client state so [`bounded_roundtrip`] can flag the sync
26/// callback's completion; pair with [`impl_sync_callback`].
27pub(crate) trait SyncState {
28    fn sync_done_mut(&mut self) -> &mut bool;
29}
30
31/// `Dispatch<WlCallback>` for a [`SyncState`] type (a blanket impl would violate
32/// the orphan rule, so each state stamps its own).
33macro_rules! impl_sync_callback {
34    ($t:ty) => {
35        impl wayland_client::Dispatch<wayland_client::protocol::wl_callback::WlCallback, ()>
36            for $t
37        {
38            fn event(
39                state: &mut Self,
40                _: &wayland_client::protocol::wl_callback::WlCallback,
41                event: wayland_client::protocol::wl_callback::Event,
42                _: &(),
43                _: &wayland_client::Connection,
44                _: &wayland_client::QueueHandle<Self>,
45            ) {
46                if let wayland_client::protocol::wl_callback::Event::Done { .. } = event {
47                    *crate::wayland::wlclient::SyncState::sync_done_mut(state) = true;
48                }
49            }
50        }
51    };
52}
53pub(crate) use impl_sync_callback;
54
55/// Block until `fd` is readable or `timeout` passes (false = timed out).
56pub(crate) fn wait_readable(fd: RawFd, timeout: Duration) -> Result<bool, String> {
57    loop {
58        let mut pfd = libc::pollfd { fd, events: libc::POLLIN, revents: 0 };
59        let n = unsafe { libc::poll(&mut pfd, 1, timeout.as_millis().max(1) as libc::c_int) };
60        if n < 0 {
61            let err = std::io::Error::last_os_error();
62            if err.raw_os_error() == Some(libc::EINTR) {
63                continue;
64            }
65            return Err(format!("poll: {err}"));
66        }
67        return Ok(n > 0);
68    }
69}
70
71/// Poll two fds for readability at once (wayland socket + a wake pipe); returns
72/// `(a_readable, b_readable)`, both false on timeout. Hangup counts as readable
73/// so a closed wake pipe unblocks its waiter.
74pub(crate) fn wait_readable2(
75    a: RawFd,
76    b: RawFd,
77    timeout: Option<Duration>,
78) -> Result<(bool, bool), String> {
79    loop {
80        let mut pfds = [
81            libc::pollfd { fd: a, events: libc::POLLIN, revents: 0 },
82            libc::pollfd { fd: b, events: libc::POLLIN, revents: 0 },
83        ];
84        let ms = timeout.map(|t| t.as_millis().max(1) as libc::c_int).unwrap_or(-1);
85        let n = unsafe { libc::poll(pfds.as_mut_ptr(), 2, ms) };
86        if n < 0 {
87            let err = std::io::Error::last_os_error();
88            if err.raw_os_error() == Some(libc::EINTR) {
89                continue;
90            }
91            return Err(format!("poll: {err}"));
92        }
93        let hit = |r: libc::c_short| r & (libc::POLLIN | libc::POLLHUP | libc::POLLERR) != 0;
94        return Ok((hit(pfds[0].revents), hit(pfds[1].revents)));
95    }
96}
97
98/// Drain every pending byte from a wake pipe without blocking.
99pub(crate) fn drain_pipe(fd: RawFd) {
100    let mut buf = [0u8; 64];
101    loop {
102        let n = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut libc::c_void, buf.len()) };
103        if n <= 0 {
104            return;
105        }
106    }
107}
108
109/// One wake byte, non-blocking; a full pipe already guarantees a pending wake.
110pub(crate) fn wake_write(fd: RawFd) {
111    let b = [1u8];
112    unsafe { libc::write(fd, b.as_ptr() as *const libc::c_void, 1) };
113}
114
115/// `EventQueue::roundtrip` with a deadline: a wedged compositor becomes an error
116/// instead of hanging the calling thread (and everything queued behind it).
117pub(crate) fn bounded_roundtrip<S>(
118    conn: &Connection,
119    queue: &mut EventQueue<S>,
120    state: &mut S,
121) -> Result<(), String>
122where
123    S: SyncState + Dispatch<wl_callback::WlCallback, ()> + 'static,
124{
125    *state.sync_done_mut() = false;
126    let _cb = conn.display().sync(&queue.handle(), ());
127    let deadline = Instant::now() + IO_TIMEOUT;
128    loop {
129        queue
130            .dispatch_pending(state)
131            .map_err(|e| format!("dispatch: {e}"))?;
132        if *state.sync_done_mut() {
133            return Ok(());
134        }
135        queue.flush().map_err(|e| format!("flush: {e}"))?;
136        let remaining = deadline
137            .checked_duration_since(Instant::now())
138            .ok_or("compositor round-trip timed out")?;
139        let Some(guard) = conn.prepare_read() else {
140            continue;
141        };
142        if !wait_readable(guard.connection_fd().as_raw_fd(), remaining)? {
143            return Err("compositor round-trip timed out".into());
144        }
145        guard.read().map_err(|e| format!("read: {e}"))?;
146    }
147}
148
149/// Anonymous CLOEXEC memfd holding `data` (keymap uploads, shm-style payloads).
150pub(crate) fn memfd_with(data: &[u8]) -> Result<OwnedFd, String> {
151    let name = b"pixelflux-wl\0";
152    let fd =
153        unsafe { libc::memfd_create(name.as_ptr() as *const libc::c_char, libc::MFD_CLOEXEC) };
154    if fd < 0 {
155        return Err(format!("memfd_create: {}", std::io::Error::last_os_error()));
156    }
157    let owned = unsafe { OwnedFd::from_raw_fd(fd) };
158    let mut written = 0;
159    while written < data.len() {
160        let n = unsafe {
161            libc::write(
162                owned.as_raw_fd(),
163                data[written..].as_ptr() as *const libc::c_void,
164                data.len() - written,
165            )
166        };
167        if n < 0 {
168            return Err(format!("write memfd: {}", std::io::Error::last_os_error()));
169        }
170        written += n as usize;
171    }
172    Ok(owned)
173}
174
175/// CLOEXEC pipe as (read end, write end).
176pub(crate) fn pipe_cloexec() -> Result<(OwnedFd, OwnedFd), String> {
177    let mut fds = [0i32; 2];
178    if unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC) } < 0 {
179        return Err(format!("pipe2: {}", std::io::Error::last_os_error()));
180    }
181    Ok(unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) })
182}
183
184/// Non-blocking CLOEXEC pipe for wake signalling (read end, write end).
185pub(crate) fn wake_pipe() -> Result<(OwnedFd, OwnedFd), String> {
186    let mut fds = [0i32; 2];
187    if unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) } < 0 {
188        return Err(format!("pipe2: {}", std::io::Error::last_os_error()));
189    }
190    Ok(unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) })
191}
192
193/// Read `fd` to EOF. The deadline is per-chunk (idle), so a large transfer that
194/// keeps flowing is never cut off while a stalled writer still errors out.
195pub(crate) fn read_fd_to_end(fd: &OwnedFd, idle: Duration) -> Result<Vec<u8>, String> {
196    read_fd_to_end_capped(fd, idle, READ_CAP_MAX)
197}
198
199/// Ceiling on an unbounded clipboard source, matching what the selkies clipboard protocol
200/// allows: a hostile or stuck peer could otherwise stream bytes into memory forever.
201const READ_CAP_MAX: usize = 64 * 1024 * 1024;
202
203/// Same as `read_fd_to_end` but refuses a source larger than `cap`.
204///
205/// `cap` is inclusive: a payload of exactly `cap` bytes is accepted. The limit is therefore
206/// tested against the size this chunk *would* produce, before appending — a test on the
207/// already-accumulated length is reached again before EOF can be observed, so it would reject
208/// a payload that exactly fills the cap.
209pub(crate) fn read_fd_to_end_capped(
210    fd: &OwnedFd,
211    idle: Duration,
212    cap: usize,
213) -> Result<Vec<u8>, String> {
214    let mut out = Vec::new();
215    let mut chunk = [0u8; 65536];
216    loop {
217        if !wait_readable(fd.as_raw_fd(), idle)? {
218            return Err("clipboard source stalled".into());
219        }
220        let n = unsafe {
221            libc::read(fd.as_raw_fd(), chunk.as_mut_ptr() as *mut libc::c_void, chunk.len())
222        };
223        if n < 0 {
224            let err = std::io::Error::last_os_error();
225            if err.raw_os_error() == Some(libc::EINTR) {
226                continue;
227            }
228            return Err(format!("read: {err}"));
229        }
230        if n == 0 {
231            return Ok(out);
232        }
233        if out.len() + n as usize > cap {
234            return Err(format!("clipboard source exceeds the {cap} byte cap"));
235        }
236        out.extend_from_slice(&chunk[..n as usize]);
237    }
238}
239
240/// Write all of `data` to `fd`, tolerating a slow reader up to `idle` per chunk.
241/// EPIPE is success-shaped: the paster stopped reading, which is its right.
242///
243/// The fd is switched to non-blocking for the transfer (original flags are
244/// restored on every exit path): a single blocking write() of a large payload
245/// would stay parked in the kernel once the peer stops draining mid-transfer,
246/// and the poll deadline could never fire. With O_NONBLOCK, write() returns
247/// EAGAIN instead and the idle deadline bounds the whole transfer.
248pub(crate) fn write_fd_all(fd: &OwnedFd, data: &[u8], idle: Duration) -> Result<(), String> {
249    let raw = fd.as_raw_fd();
250    let old_flags = unsafe { libc::fcntl(raw, libc::F_GETFL) };
251    if old_flags < 0 {
252        return Err(format!("fcntl F_GETFL: {}", std::io::Error::last_os_error()));
253    }
254    if unsafe { libc::fcntl(raw, libc::F_SETFL, old_flags | libc::O_NONBLOCK) } < 0 {
255        return Err(format!("fcntl F_SETFL: {}", std::io::Error::last_os_error()));
256    }
257    let result = write_fd_all_nb(raw, data, idle);
258    unsafe { libc::fcntl(raw, libc::F_SETFL, old_flags) };
259    result
260}
261
262fn write_fd_all_nb(raw: i32, data: &[u8], idle: Duration) -> Result<(), String> {
263    let mut written = 0;
264    while written < data.len() {
265        let mut pfd = libc::pollfd { fd: raw, events: libc::POLLOUT, revents: 0 };
266        let n = unsafe { libc::poll(&mut pfd, 1, idle.as_millis().max(1) as libc::c_int) };
267        if n < 0 {
268            let err = std::io::Error::last_os_error();
269            if err.raw_os_error() == Some(libc::EINTR) {
270                continue;
271            }
272            return Err(format!("poll: {err}"));
273        }
274        if n == 0 {
275            return Err("clipboard reader stalled".into());
276        }
277        let w = unsafe {
278            libc::write(
279                raw,
280                data[written..].as_ptr() as *const libc::c_void,
281                data.len() - written,
282            )
283        };
284        if w < 0 {
285            let err = std::io::Error::last_os_error();
286            match err.raw_os_error() {
287                Some(libc::EINTR) | Some(libc::EAGAIN) => continue,
288                Some(libc::EPIPE) => return Ok(()),
289                _ => return Err(format!("write: {err}")),
290            }
291        }
292        written += w as usize;
293    }
294    Ok(())
295}
pixelflux/wayland/wlclient.rs

pixelflux/wayland/
wlclient.rs
