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 {
59            fd,
60            events: libc::POLLIN,
61            revents: 0,
62        };
63        let n = unsafe { libc::poll(&mut pfd, 1, timeout.as_millis().max(1) as libc::c_int) };
64        if n < 0 {
65            let err = std::io::Error::last_os_error();
66            if err.raw_os_error() == Some(libc::EINTR) {
67                continue;
68            }
69            return Err(format!("poll: {err}"));
70        }
71        return Ok(n > 0);
72    }
73}
74
75/// Poll two fds for readability at once (wayland socket + a wake pipe); returns
76/// `(a_readable, b_readable)`, both false on timeout. Hangup counts as readable
77/// so a closed wake pipe unblocks its waiter.
78pub(crate) fn wait_readable2(
79    a: RawFd,
80    b: RawFd,
81    timeout: Option<Duration>,
82) -> Result<(bool, bool), String> {
83    loop {
84        let mut pfds = [
85            libc::pollfd {
86                fd: a,
87                events: libc::POLLIN,
88                revents: 0,
89            },
90            libc::pollfd {
91                fd: b,
92                events: libc::POLLIN,
93                revents: 0,
94            },
95        ];
96        let ms = timeout
97            .map(|t| t.as_millis().max(1) as libc::c_int)
98            .unwrap_or(-1);
99        let n = unsafe { libc::poll(pfds.as_mut_ptr(), 2, ms) };
100        if n < 0 {
101            let err = std::io::Error::last_os_error();
102            if err.raw_os_error() == Some(libc::EINTR) {
103                continue;
104            }
105            return Err(format!("poll: {err}"));
106        }
107        let hit = |r: libc::c_short| r & (libc::POLLIN | libc::POLLHUP | libc::POLLERR) != 0;
108        return Ok((hit(pfds[0].revents), hit(pfds[1].revents)));
109    }
110}
111
112/// Drain every pending byte from a wake pipe without blocking.
113pub(crate) fn drain_pipe(fd: RawFd) {
114    let mut buf = [0u8; 64];
115    loop {
116        let n = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut libc::c_void, buf.len()) };
117        if n <= 0 {
118            return;
119        }
120    }
121}
122
123/// One wake byte, non-blocking; a full pipe already guarantees a pending wake.
124pub(crate) fn wake_write(fd: RawFd) {
125    let b = [1u8];
126    unsafe { libc::write(fd, b.as_ptr() as *const libc::c_void, 1) };
127}
128
129/// `EventQueue::roundtrip` with a deadline: a wedged compositor becomes an error
130/// instead of hanging the calling thread (and everything queued behind it).
131pub(crate) fn bounded_roundtrip<S>(
132    conn: &Connection,
133    queue: &mut EventQueue<S>,
134    state: &mut S,
135) -> Result<(), String>
136where
137    S: SyncState + Dispatch<wl_callback::WlCallback, ()> + 'static,
138{
139    *state.sync_done_mut() = false;
140    let _cb = conn.display().sync(&queue.handle(), ());
141    let deadline = Instant::now() + IO_TIMEOUT;
142    loop {
143        queue
144            .dispatch_pending(state)
145            .map_err(|e| format!("dispatch: {e}"))?;
146        if *state.sync_done_mut() {
147            return Ok(());
148        }
149        queue.flush().map_err(|e| format!("flush: {e}"))?;
150        let remaining = deadline
151            .checked_duration_since(Instant::now())
152            .ok_or("compositor round-trip timed out")?;
153        let Some(guard) = conn.prepare_read() else {
154            continue;
155        };
156        if !wait_readable(guard.connection_fd().as_raw_fd(), remaining)? {
157            return Err("compositor round-trip timed out".into());
158        }
159        guard.read().map_err(|e| format!("read: {e}"))?;
160    }
161}
162
163/// Anonymous CLOEXEC memfd holding `data` (keymap uploads, shm-style payloads).
164pub(crate) fn memfd_with(data: &[u8]) -> Result<OwnedFd, String> {
165    let fd = crate::memfd_create(c"pixelflux-wl", libc::MFD_CLOEXEC);
166    if fd < 0 {
167        return Err(format!("memfd_create: {}", std::io::Error::last_os_error()));
168    }
169    let owned = unsafe { OwnedFd::from_raw_fd(fd) };
170    let mut written = 0;
171    while written < data.len() {
172        let n = unsafe {
173            libc::write(
174                owned.as_raw_fd(),
175                data[written..].as_ptr() as *const libc::c_void,
176                data.len() - written,
177            )
178        };
179        if n < 0 {
180            return Err(format!("write memfd: {}", std::io::Error::last_os_error()));
181        }
182        written += n as usize;
183    }
184    Ok(owned)
185}
186
187/// CLOEXEC pipe as (read end, write end).
188pub(crate) fn pipe_cloexec() -> Result<(OwnedFd, OwnedFd), String> {
189    let mut fds = [0i32; 2];
190    if unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC) } < 0 {
191        return Err(format!("pipe2: {}", std::io::Error::last_os_error()));
192    }
193    Ok(unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) })
194}
195
196/// Non-blocking CLOEXEC pipe for wake signaling (read end, write end).
197pub(crate) fn wake_pipe() -> Result<(OwnedFd, OwnedFd), String> {
198    let mut fds = [0i32; 2];
199    if unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) } < 0 {
200        return Err(format!("pipe2: {}", std::io::Error::last_os_error()));
201    }
202    Ok(unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) })
203}
204
205/// Read `fd` to EOF. The deadline is per-chunk (idle), so a large transfer that
206/// keeps flowing is never cut off while a stalled writer still errors out.
207pub(crate) fn read_fd_to_end(fd: &OwnedFd, idle: Duration) -> Result<Vec<u8>, String> {
208    read_fd_to_end_capped(fd, idle, READ_CAP_MAX)
209}
210
211/// Ceiling on an unbounded clipboard source, matching what the selkies clipboard protocol
212/// allows: a hostile or stuck peer could otherwise stream bytes into memory forever.
213const READ_CAP_MAX: usize = 64 * 1024 * 1024;
214
215/// Same as `read_fd_to_end` but refuses a source larger than `cap`.
216///
217/// `cap` is inclusive: a payload of exactly `cap` bytes is accepted. The limit is therefore
218/// tested against the size this chunk *would* produce, before appending — a test on the
219/// already-accumulated length is reached again before EOF can be observed, so it would reject
220/// a payload that exactly fills the cap.
221pub(crate) fn read_fd_to_end_capped(
222    fd: &OwnedFd,
223    idle: Duration,
224    cap: usize,
225) -> Result<Vec<u8>, String> {
226    let mut out = Vec::new();
227    let mut chunk = [0u8; 65536];
228    loop {
229        if !wait_readable(fd.as_raw_fd(), idle)? {
230            return Err("clipboard source stalled".into());
231        }
232        let n = unsafe {
233            libc::read(
234                fd.as_raw_fd(),
235                chunk.as_mut_ptr() as *mut libc::c_void,
236                chunk.len(),
237            )
238        };
239        if n < 0 {
240            let err = std::io::Error::last_os_error();
241            if err.raw_os_error() == Some(libc::EINTR) {
242                continue;
243            }
244            return Err(format!("read: {err}"));
245        }
246        if n == 0 {
247            return Ok(out);
248        }
249        if out.len() + n as usize > cap {
250            return Err(format!("clipboard source exceeds the {cap} byte cap"));
251        }
252        out.extend_from_slice(&chunk[..n as usize]);
253    }
254}
255
256/// Write all of `data` to `fd`, tolerating a slow reader up to `idle` per chunk.
257/// EPIPE is success-shaped: the paster stopped reading, which is its right.
258///
259/// The fd is switched to non-blocking for the transfer (original flags are
260/// restored on every exit path): a single blocking write() of a large payload
261/// would stay parked in the kernel once the peer stops draining mid-transfer,
262/// and the poll deadline could never fire. With O_NONBLOCK, write() returns
263/// EAGAIN instead and the idle deadline bounds the whole transfer.
264pub(crate) fn write_fd_all(fd: &OwnedFd, data: &[u8], idle: Duration) -> Result<(), String> {
265    let raw = fd.as_raw_fd();
266    let old_flags = unsafe { libc::fcntl(raw, libc::F_GETFL) };
267    if old_flags < 0 {
268        return Err(format!(
269            "fcntl F_GETFL: {}",
270            std::io::Error::last_os_error()
271        ));
272    }
273    if unsafe { libc::fcntl(raw, libc::F_SETFL, old_flags | libc::O_NONBLOCK) } < 0 {
274        return Err(format!(
275            "fcntl F_SETFL: {}",
276            std::io::Error::last_os_error()
277        ));
278    }
279    let result = write_fd_all_nb(raw, data, idle);
280    unsafe { libc::fcntl(raw, libc::F_SETFL, old_flags) };
281    result
282}
283
284fn write_fd_all_nb(raw: i32, data: &[u8], idle: Duration) -> Result<(), String> {
285    let mut written = 0;
286    while written < data.len() {
287        let mut pfd = libc::pollfd {
288            fd: raw,
289            events: libc::POLLOUT,
290            revents: 0,
291        };
292        let n = unsafe { libc::poll(&mut pfd, 1, idle.as_millis().max(1) as libc::c_int) };
293        if n < 0 {
294            let err = std::io::Error::last_os_error();
295            if err.raw_os_error() == Some(libc::EINTR) {
296                continue;
297            }
298            return Err(format!("poll: {err}"));
299        }
300        if n == 0 {
301            return Err("clipboard reader stalled".into());
302        }
303        let w = unsafe {
304            libc::write(
305                raw,
306                data[written..].as_ptr() as *const libc::c_void,
307                data.len() - written,
308            )
309        };
310        if w < 0 {
311            let err = std::io::Error::last_os_error();
312            match err.raw_os_error() {
313                Some(libc::EINTR) | Some(libc::EAGAIN) => continue,
314                Some(libc::EPIPE) => return Ok(()),
315                _ => return Err(format!("write: {err}")),
316            }
317        }
318        written += w as usize;
319    }
320    Ok(())
321}
pixelflux/wayland/wlclient.rs

pixelflux/wayland/
wlclient.rs
