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 name = b"pixelflux-wl\0";
166    let fd = unsafe { libc::memfd_create(name.as_ptr() as *const libc::c_char, libc::MFD_CLOEXEC) };
167    if fd < 0 {
168        return Err(format!("memfd_create: {}", std::io::Error::last_os_error()));
169    }
170    let owned = unsafe { OwnedFd::from_raw_fd(fd) };
171    let mut written = 0;
172    while written < data.len() {
173        let n = unsafe {
174            libc::write(
175                owned.as_raw_fd(),
176                data[written..].as_ptr() as *const libc::c_void,
177                data.len() - written,
178            )
179        };
180        if n < 0 {
181            return Err(format!("write memfd: {}", std::io::Error::last_os_error()));
182        }
183        written += n as usize;
184    }
185    Ok(owned)
186}
187
188/// CLOEXEC pipe as (read end, write end).
189pub(crate) fn pipe_cloexec() -> Result<(OwnedFd, OwnedFd), String> {
190    let mut fds = [0i32; 2];
191    if unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC) } < 0 {
192        return Err(format!("pipe2: {}", std::io::Error::last_os_error()));
193    }
194    Ok(unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) })
195}
196
197/// Non-blocking CLOEXEC pipe for wake signaling (read end, write end).
198pub(crate) fn wake_pipe() -> Result<(OwnedFd, OwnedFd), String> {
199    let mut fds = [0i32; 2];
200    if unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) } < 0 {
201        return Err(format!("pipe2: {}", std::io::Error::last_os_error()));
202    }
203    Ok(unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) })
204}
205
206/// Read `fd` to EOF. The deadline is per-chunk (idle), so a large transfer that
207/// keeps flowing is never cut off while a stalled writer still errors out.
208pub(crate) fn read_fd_to_end(fd: &OwnedFd, idle: Duration) -> Result<Vec<u8>, String> {
209    read_fd_to_end_capped(fd, idle, READ_CAP_MAX)
210}
211
212/// Ceiling on an unbounded clipboard source, matching what the selkies clipboard protocol
213/// allows: a hostile or stuck peer could otherwise stream bytes into memory forever.
214const READ_CAP_MAX: usize = 64 * 1024 * 1024;
215
216/// Same as `read_fd_to_end` but refuses a source larger than `cap`.
217///
218/// `cap` is inclusive: a payload of exactly `cap` bytes is accepted. The limit is therefore
219/// tested against the size this chunk *would* produce, before appending — a test on the
220/// already-accumulated length is reached again before EOF can be observed, so it would reject
221/// a payload that exactly fills the cap.
222pub(crate) fn read_fd_to_end_capped(
223    fd: &OwnedFd,
224    idle: Duration,
225    cap: usize,
226) -> Result<Vec<u8>, String> {
227    let mut out = Vec::new();
228    let mut chunk = [0u8; 65536];
229    loop {
230        if !wait_readable(fd.as_raw_fd(), idle)? {
231            return Err("clipboard source stalled".into());
232        }
233        let n = unsafe {
234            libc::read(
235                fd.as_raw_fd(),
236                chunk.as_mut_ptr() as *mut libc::c_void,
237                chunk.len(),
238            )
239        };
240        if n < 0 {
241            let err = std::io::Error::last_os_error();
242            if err.raw_os_error() == Some(libc::EINTR) {
243                continue;
244            }
245            return Err(format!("read: {err}"));
246        }
247        if n == 0 {
248            return Ok(out);
249        }
250        if out.len() + n as usize > cap {
251            return Err(format!("clipboard source exceeds the {cap} byte cap"));
252        }
253        out.extend_from_slice(&chunk[..n as usize]);
254    }
255}
256
257/// Write all of `data` to `fd`, tolerating a slow reader up to `idle` per chunk.
258/// EPIPE is success-shaped: the paster stopped reading, which is its right.
259///
260/// The fd is switched to non-blocking for the transfer (original flags are
261/// restored on every exit path): a single blocking write() of a large payload
262/// would stay parked in the kernel once the peer stops draining mid-transfer,
263/// and the poll deadline could never fire. With O_NONBLOCK, write() returns
264/// EAGAIN instead and the idle deadline bounds the whole transfer.
265pub(crate) fn write_fd_all(fd: &OwnedFd, data: &[u8], idle: Duration) -> Result<(), String> {
266    let raw = fd.as_raw_fd();
267    let old_flags = unsafe { libc::fcntl(raw, libc::F_GETFL) };
268    if old_flags < 0 {
269        return Err(format!(
270            "fcntl F_GETFL: {}",
271            std::io::Error::last_os_error()
272        ));
273    }
274    if unsafe { libc::fcntl(raw, libc::F_SETFL, old_flags | libc::O_NONBLOCK) } < 0 {
275        return Err(format!(
276            "fcntl F_SETFL: {}",
277            std::io::Error::last_os_error()
278        ));
279    }
280    let result = write_fd_all_nb(raw, data, idle);
281    unsafe { libc::fcntl(raw, libc::F_SETFL, old_flags) };
282    result
283}
284
285fn write_fd_all_nb(raw: i32, data: &[u8], idle: Duration) -> Result<(), String> {
286    let mut written = 0;
287    while written < data.len() {
288        let mut pfd = libc::pollfd {
289            fd: raw,
290            events: libc::POLLOUT,
291            revents: 0,
292        };
293        let n = unsafe { libc::poll(&mut pfd, 1, idle.as_millis().max(1) as libc::c_int) };
294        if n < 0 {
295            let err = std::io::Error::last_os_error();
296            if err.raw_os_error() == Some(libc::EINTR) {
297                continue;
298            }
299            return Err(format!("poll: {err}"));
300        }
301        if n == 0 {
302            return Err("clipboard reader stalled".into());
303        }
304        let w = unsafe {
305            libc::write(
306                raw,
307                data[written..].as_ptr() as *const libc::c_void,
308                data.len() - written,
309            )
310        };
311        if w < 0 {
312            let err = std::io::Error::last_os_error();
313            match err.raw_os_error() {
314                Some(libc::EINTR) | Some(libc::EAGAIN) => continue,
315                Some(libc::EPIPE) => return Ok(()),
316                _ => return Err(format!("write: {err}")),
317            }
318        }
319        written += w as usize;
320    }
321    Ok(())
322}
pixelflux/wayland/wlclient.rs

pixelflux/wayland/
wlclient.rs
