Wayland compositor (wlroots)
git clone https://git.lucas.co/cce-compositor.git
src/server/ipc_server.rs (15.5K)
1 // Monolithic IPC Server socket listener for CCE
2 use std::io::{Read, Write};
3 use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
4 use std::os::unix::net::{UnixListener, UnixStream};
5 use std::sync::mpsc;
6 use std::sync::Arc;
7 use std::thread;
8
9 pub struct IpcRequest {
10 pub command: String,
11 pub reply_tx: mpsc::Sender<String>,
12 /// PID of the process on the other end of the socket, from SO_PEERCRED.
13 /// A command that acts on "whoever is asking" (`fade-out`) resolves its
14 /// target with this instead of trusting a name the caller supplies: the
15 /// kernel vouches for it, and a client always knows its own pid even
16 /// when it does not know its app_id. 0 when the credentials were
17 /// unreadable, which every such command treats as no target.
18 pub peer_pid: i32,
19 }
20
21 /// The server-thread end of the request channel. Every `send` is followed by
22 /// a write to the wake eventfd, which the compositor has registered with its
23 /// wl_event_loop — that is what gets a request dispatched. The drain used to
24 /// be a 10 ms timer polling `try_recv` forever, ~100 wakeups/s on an idle
25 /// desktop; now the main thread sleeps until a command actually arrives.
26 #[derive(Clone)]
27 struct IpcSender {
28 tx: mpsc::Sender<IpcRequest>,
29 wake: Arc<OwnedFd>,
30 }
31
32 impl IpcSender {
33 fn send(&self, req: IpcRequest) -> bool {
34 if self.tx.send(req).is_err() {
35 return false;
36 }
37 wake_fd(&self.wake);
38 true
39 }
40 }
41
42 /// Bump an eventfd. Errors are ignored on purpose: EAGAIN means the counter
43 /// is already saturated (the reader is about to run anyway), and EBADF only
44 /// happens at shutdown.
45 pub fn wake_fd(fd: &OwnedFd) {
46 let one: u64 = 1;
47 unsafe {
48 libc::write(fd.as_raw_fd(), &one as *const u64 as *const libc::c_void, 8);
49 }
50 }
51
52 /// Clear an eventfd after its readable event fired.
53 pub fn drain_wake_fd(fd: std::os::raw::c_int) {
54 let mut v: u64 = 0;
55 unsafe {
56 libc::read(fd, &mut v as *mut u64 as *mut libc::c_void, 8);
57 }
58 }
59
60 /// A non-blocking, close-on-exec eventfd for cross-thread wakeups into the
61 /// wl_event_loop.
62 pub fn new_wake_fd() -> std::io::Result<Arc<OwnedFd>> {
63 let raw = unsafe { libc::eventfd(0, libc::EFD_CLOEXEC | libc::EFD_NONBLOCK) };
64 if raw < 0 {
65 return Err(std::io::Error::last_os_error());
66 }
67 Ok(Arc::new(unsafe { OwnedFd::from_raw_fd(raw) }))
68 }
69
70 pub fn get_ipc_socket_path(display_socket: Option<&str>) -> String {
71 if let Some(display) = display_socket {
72 format!("/tmp/cce-{}.sock", display)
73 } else {
74 "/tmp/cce.sock".to_string()
75 }
76 }
77
78 /// Spawn the IPC listener thread. Returns the request receiver and the
79 /// eventfd that is bumped after every request is queued; the caller adds the
80 /// fd to its event loop and drains the receiver when it fires.
81 pub fn spawn_ipc_server(display_socket: Option<String>) -> (mpsc::Receiver<IpcRequest>, Arc<OwnedFd>) {
82 let (tx, rx) = mpsc::channel::<IpcRequest>();
83 let wake = new_wake_fd().expect("Failed to create IPC wake eventfd");
84 let sender = IpcSender { tx, wake: wake.clone() };
85
86 thread::Builder::new()
87 .name("cce-ipc-server".to_string())
88 .spawn(move || {
89 ipc_server_main(sender, display_socket);
90 })
91 .expect("Failed to spawn CCE IPC server thread");
92
93 (rx, wake)
94 }
95
96 fn ipc_server_main(tx: IpcSender, display_socket: Option<String>) {
97 let socket_path = get_ipc_socket_path(display_socket.as_deref());
98 let _ = std::fs::remove_file(&socket_path);
99
100 let listener = match UnixListener::bind(&socket_path) {
101 Ok(l) => l,
102 Err(e) => {
103 log::error!("[ipc] failed to bind IPC socket {}: {}", socket_path, e);
104 return;
105 }
106 };
107
108 log::info!("[ipc] Listening on UNIX socket: {}", socket_path);
109
110 for stream in listener.incoming() {
111 match stream {
112 Ok(s) => {
113 let tx_clone = tx.clone();
114 thread::spawn(move || {
115 handle_client(s, tx_clone);
116 });
117 }
118 Err(e) => {
119 // EMFILE and friends leave the socket readable, so a bare
120 // continue spins this thread at 100% and floods the log
121 // (167GB observed under fd exhaustion). Back off instead —
122 // the session is degraded but stays diagnosable.
123 log::error!("[ipc] accept error: {}", e);
124 thread::sleep(std::time::Duration::from_millis(100));
125 }
126 }
127 }
128 }
129
130 /// PID of the process on the other end of a Unix socket, via SO_PEERCRED.
131 /// 0 when the credentials cannot be read — the kernel supplies them for every
132 /// AF_UNIX peer, so that only happens on a socket already going away.
133 /// (`UnixStream::peer_cred` is still nightly-only, hence the raw getsockopt.)
134 fn socket_peer_pid(stream: &UnixStream) -> i32 {
135 let mut cred: libc::ucred = unsafe { std::mem::zeroed() };
136 let mut len = std::mem::size_of::<libc::ucred>() as libc::socklen_t;
137 let rc = unsafe {
138 libc::getsockopt(
139 stream.as_raw_fd(),
140 libc::SOL_SOCKET,
141 libc::SO_PEERCRED,
142 &mut cred as *mut libc::ucred as *mut libc::c_void,
143 &mut len,
144 )
145 };
146 if rc == 0 {
147 cred.pid
148 } else {
149 0
150 }
151 }
152
153 /// One subscription line from a socket client, within `limit` bytes and an
154 /// overall `deadline`. The status and stream sockets read their client's
155 /// first line on a thread that serves everyone else too, with `read_line`
156 /// and a per-read timeout only: a client trickling a byte per timeout held
157 /// that thread (every status-bar update, or every new stream subscriber)
158 /// indefinitely, and grew the line without bound inside the compositor.
159 /// None on timeout, overflow, EOF before any byte, or a read error.
160 pub fn read_line_bounded(stream: &UnixStream, limit: usize, deadline: std::time::Duration) -> Option<String> {
161 let until = std::time::Instant::now() + deadline;
162 let mut buf: Vec<u8> = Vec::new();
163 let mut chunk = [0u8; 512];
164 let mut reader = stream;
165 loop {
166 let left = until.checked_duration_since(std::time::Instant::now())?;
167 if left.is_zero() {
168 return None;
169 }
170 stream.set_read_timeout(Some(left)).ok()?;
171 match reader.read(&mut chunk) {
172 Ok(0) if buf.is_empty() => return None,
173 Ok(0) => break,
174 Ok(n) => {
175 buf.extend_from_slice(&chunk[..n]);
176 if let Some(end) = buf.iter().position(|&b| b == b'\n') {
177 buf.truncate(end);
178 break;
179 }
180 if buf.len() > limit {
181 return None;
182 }
183 }
184 Err(_) => return None,
185 }
186 }
187 if buf.len() > limit {
188 return None;
189 }
190 Some(String::from_utf8_lossy(&buf).into_owned())
191 }
192
193 /// Longest command accepted. The old single `read` into 4096 bytes cut a
194 /// longer command — or one written in pieces — short and RAN the prefix
195 /// (`spawn` included); now an overlong command is refused whole.
196 const MAX_COMMAND: usize = 64 * 1024;
197 /// How long a connection may sit without sending anything. Each one holds a
198 /// thread, and the old read had no timeout, so idle connections leaked them.
199 const FIRST_BYTE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
200 /// Once some bytes are in, a pause this long ends the command. Clients send
201 /// one line and then wait for the reply, but not all of them send the
202 /// newline or close their side, and those must not stall.
203 const QUIET_GAP: std::time::Duration = std::time::Duration::from_millis(50);
204
205 #[derive(Debug, PartialEq)]
206 enum Request {
207 Command(String),
208 /// Nothing arrived (a probe, an idle connection, EOF at once).
209 Empty,
210 TooLong,
211 /// A NUL byte: no command contains one, and it reached a CString
212 /// conversion (`shortcut bind`'s keysym lookup) whose panic aborted the
213 /// compositor.
214 Nul,
215 }
216
217 /// Read one command: up to its newline, the client closing its side, or a
218 /// pause of `QUIET_GAP` after the first bytes — never more than
219 /// `MAX_COMMAND`, never waiting more than `FIRST_BYTE_TIMEOUT` for a start.
220 fn read_command(stream: &mut UnixStream) -> Request {
221 let _ = stream.set_read_timeout(Some(FIRST_BYTE_TIMEOUT));
222 let mut buf: Vec<u8> = Vec::new();
223 let mut chunk = [0u8; 4096];
224 loop {
225 match stream.read(&mut chunk) {
226 Ok(0) => break,
227 Ok(n) => {
228 buf.extend_from_slice(&chunk[..n]);
229 if buf.contains(&b'\n') {
230 break;
231 }
232 if buf.len() > MAX_COMMAND {
233 return Request::TooLong;
234 }
235 let _ = stream.set_read_timeout(Some(QUIET_GAP));
236 }
237 Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => break,
238 Err(_) => return Request::Empty,
239 }
240 }
241 let line = match buf.iter().position(|&b| b == b'\n') {
242 Some(end) => &buf[..end],
243 None => &buf[..],
244 };
245 if line.len() > MAX_COMMAND {
246 return Request::TooLong;
247 }
248 if line.contains(&0) {
249 return Request::Nul;
250 }
251 let cmd = String::from_utf8_lossy(line).trim().to_string();
252 if cmd.is_empty() {
253 Request::Empty
254 } else {
255 Request::Command(cmd)
256 }
257 }
258
259 fn handle_client(mut stream: UnixStream, tx: IpcSender) {
260 let peer_pid = socket_peer_pid(&stream);
261 match read_command(&mut stream) {
262 Request::Empty => {}
263 Request::TooLong => {
264 let _ = stream.write_all(format!("error: command longer than {MAX_COMMAND} bytes, not run\n").as_bytes());
265 }
266 Request::Nul => {
267 let _ = stream.write_all(b"error: command contains a NUL byte, not run\n");
268 }
269 Request::Command(cmd) => {
270 {
271 // Commands answer from the IPC drain and so are quick; a
272 // second is a generous leash that still surfaces a wedged
273 // compositor. `screenshot` is the exception: its reply now
274 // waits for the capture, which happens on the next composited
275 // frame, and a cold readback (first capture after an idle
276 // spell — NVIDIA recompiles shaders on the way) has been
277 // measured over a second. Timing that out would report
278 // failure for a capture that lands.
279 // `lock` answers once the session IS locked: a frame on every
280 // output plus a locker starting, and the lock-before-sleep
281 // thread holds logind's sleep (5s at most) on that reply.
282 // `focus-window --wait` answers once the window holds still,
283 // `SETTLE_TIMEOUT_MS` (3s) at most.
284 let timeout = if cmd.starts_with("screenshot") {
285 std::time::Duration::from_secs(5)
286 } else if cmd == "lock" || cmd.starts_with("focus-window --wait") {
287 std::time::Duration::from_secs(4)
288 } else {
289 std::time::Duration::from_millis(1000)
290 };
291 let (reply_tx, reply_rx) = mpsc::channel();
292 if tx.send(IpcRequest { command: cmd, reply_tx, peer_pid }) {
293 if let Ok(reply) = reply_rx.recv_timeout(timeout) {
294 let _ = stream.write_all(reply.as_bytes());
295 } else {
296 let _ = stream.write_all(b"error: timeout processing command\n");
297 }
298 }
299 }
300 }
301 }
302 }
303
304 #[cfg(test)]
305 mod framing_tests {
306 use super::*;
307 use std::time::{Duration, Instant};
308
309 fn send(bytes: &[u8], close: bool) -> Request {
310 let (mut client, mut server) = UnixStream::pair().unwrap();
311 client.write_all(bytes).unwrap();
312 if close {
313 client.shutdown(std::net::Shutdown::Write).unwrap();
314 }
315 let got = read_command(&mut server);
316 drop(client);
317 got
318 }
319
320 #[test]
321 fn a_command_ends_at_its_newline_its_close_or_a_pause() {
322 assert_eq!(send(b"windows --json\n", false), Request::Command("windows --json".into()));
323 assert_eq!(send(b"lock", true), Request::Command("lock".into()));
324 // No newline and no close (the old sleep_lock request did this): the
325 // quiet gap ends it rather than the 5s first-byte timeout.
326 let t = Instant::now();
327 assert_eq!(send(b"lock", false), Request::Command("lock".into()));
328 assert!(t.elapsed() < Duration::from_secs(1));
329 // Only the first line is the command.
330 assert_eq!(send(b"spawn foo\nspawn bar\n", false), Request::Command("spawn foo".into()));
331 }
332
333 #[test]
334 fn a_command_written_in_pieces_arrives_whole() {
335 let (mut client, mut server) = UnixStream::pair().unwrap();
336 let writer = std::thread::spawn(move || {
337 client.write_all(b"spawn ").unwrap();
338 std::thread::sleep(Duration::from_millis(10));
339 client.write_all(b"cce-terminal\n").unwrap();
340 client
341 });
342 assert_eq!(read_command(&mut server), Request::Command("spawn cce-terminal".into()));
343 drop(writer.join().unwrap());
344 }
345
346 #[test]
347 fn an_overlong_command_is_refused_not_cut_short() {
348 // The old reader ran the first 4096 bytes of this.
349 let mut long = b"spawn ".to_vec();
350 long.extend(std::iter::repeat(b'x').take(MAX_COMMAND + 10));
351 long.push(b'\n');
352 let (mut client, mut server) = UnixStream::pair().unwrap();
353 let writer = std::thread::spawn(move || {
354 let _ = client.write_all(&long);
355 client
356 });
357 assert_eq!(read_command(&mut server), Request::TooLong);
358 drop(server);
359 drop(writer.join().unwrap());
360 // 5000 bytes, over the old 4096, is now one whole command.
361 let mid = format!("spawn {}\n", "y".repeat(5000));
362 assert_eq!(send(mid.as_bytes(), false), Request::Command(mid.trim().to_string()));
363 }
364
365 #[test]
366 fn a_nul_byte_is_refused() {
367 assert_eq!(send(b"shortcut bind /s/1 x CTRL+a\0b\n", false), Request::Nul);
368 }
369
370 #[test]
371 fn a_silent_or_trickling_subscriber_is_cut_off_on_time() {
372 let (_client, server) = UnixStream::pair().unwrap();
373 let t = Instant::now();
374 assert_eq!(read_line_bounded(&server, 256, Duration::from_millis(100)), None);
375 assert!(t.elapsed() < Duration::from_millis(500));
376
377 // A byte every 30ms never finishes a line; the TOTAL deadline ends it
378 // (the old per-read timeout never fired, each read being on time).
379 let (mut client, server) = UnixStream::pair().unwrap();
380 let trickle = std::thread::spawn(move || {
381 for _ in 0..40 {
382 if client.write_all(b"a").is_err() {
383 break;
384 }
385 std::thread::sleep(Duration::from_millis(30));
386 }
387 });
388 let t = Instant::now();
389 assert_eq!(read_line_bounded(&server, 256, Duration::from_millis(200)), None);
390 assert!(t.elapsed() < Duration::from_millis(600), "took {:?}", t.elapsed());
391 drop(server);
392 trickle.join().unwrap();
393
394 let (mut client, server) = UnixStream::pair().unwrap();
395 client.write_all(b"layout\n").unwrap();
396 assert_eq!(read_line_bounded(&server, 256, Duration::from_millis(200)).as_deref(), Some("layout"));
397 let (mut client, server) = UnixStream::pair().unwrap();
398 client.write_all(&[b'z'; 300]).unwrap();
399 assert_eq!(read_line_bounded(&server, 256, Duration::from_millis(200)), None, "over the size cap");
400 }
401 }