Wayland compositor (wlroots)
git clone https://git.lucas.co/cce-compositor.git
src/server/status_server.rs (23.4K)
1 // Status socket server for monolithic cce server
2 //
3 // Runs in a dedicated thread. cce-status connects to
4 // /tmp/cce-status-{WAYLAND_DISPLAY}.sock, sends a subscription line
5 // ("layout", "title", "modifiers", "adjust", "dismiss", or "shortcuts") and receives lines whenever the status changes.
6 //
7 // The main loop sends updates through an mpsc channel. The server thread
8 // owns the socket and handles all I/O independently of the Wayland event loop.
9
10 use std::io::{BufRead, Write};
11 use std::os::fd::{AsRawFd, OwnedFd};
12 use std::os::unix::net::{UnixListener, UnixStream};
13 use std::sync::mpsc;
14 use std::sync::Arc;
15
16 use crate::ipc_server::{drain_wake_fd, new_wake_fd, wake_fd};
17
18 /// A status update sent from the main loop to the server thread.
19 #[derive(Debug, Clone, PartialEq, Eq)]
20 pub struct StatusUpdate {
21 /// Plain text for layout module subscribers
22 pub layout_text: String,
23 /// Plain text for title module subscribers
24 pub title_text: String,
25 /// Plain text for modifiers subscriber
26 pub modifiers_text: String,
27 /// "on" while window-adjust mode is active (overview, or Super held —
28 /// `WindowManager::window_adjust_active`), else "off". The desktop grid
29 /// subscribes to show its own resize handles on the pinned images in
30 /// step with the windows' handles; it never holds keyboard focus, so
31 /// it cannot read the modifier state for itself.
32 pub adjust_text: String,
33 /// What each status segment is composited OVER, by app_id — see
34 /// [`crate::backdrop`]. Unlike the other topics this one is
35 /// per-subscriber: a segment gets only its own entry, since the whole
36 /// point is that the far ends of a bar sit over different things.
37 /// Quantized to whole percent, which is what keeps this struct `Eq` and
38 /// therefore keeps `update_status`'s resend gate working while the
39 /// camera pans.
40 pub backdrops: Vec<(String, u8, u8)>,
41 }
42
43 /// A message from the main loop to the server thread: either a new state
44 /// snapshot for the state topics, or a one-shot menu-dismiss event.
45 #[derive(Debug, Clone)]
46 pub enum StatusMsg {
47 State(StatusUpdate),
48 /// Click-away-close for in-surface status menus: every `dismiss`
49 /// subscriber EXCEPT the segment whose app_id is carried here should
50 /// close its open menu (the exempt segment saw the press itself).
51 MenuDismiss { except_app_id: String },
52 /// A portal-bound chord went down or up (`global_shortcuts`): one line,
53 /// `activated|deactivated <session> <id> <time_msec>`, for every
54 /// `shortcuts` subscriber — in practice the one portal backend.
55 Shortcut(String),
56 }
57
58 /// Subscription types that the status bar script can request.
59 ///
60 /// Not `Copy`: `Backdrop` names the segment doing the asking, because it is
61 /// the one topic whose value differs per subscriber.
62 #[derive(Debug, Clone, PartialEq, Eq)]
63 enum Subscription {
64 Layout,
65 Title,
66 Modifiers,
67 /// `adjust` — "on"/"off" as window-adjust mode comes and goes.
68 Adjust,
69 /// One-shot menu-dismiss events only — never receives state pushes.
70 Dismiss,
71 /// `shortcuts` — one-shot portal shortcut press/release lines only
72 /// (see `StatusMsg::Shortcut`); never receives state pushes.
73 Shortcuts,
74 /// `backdrop <app_id>` — what THIS segment is composited over, so it can
75 /// adapt its own text contrast. Lines are `<luma> <spread>`, both 0-100.
76 Backdrop(String),
77 Unknown,
78 }
79
80 impl Subscription {
81 fn from_str(s: &str) -> Self {
82 let s = s.trim();
83 // The one topic that takes an argument. A bare `backdrop` is
84 // accepted and simply never matches a segment, which reads as a
85 // permanently unknown backdrop rather than as an error.
86 if let Some(app_id) = s.strip_prefix("backdrop") {
87 return Subscription::Backdrop(app_id.trim().to_string());
88 }
89 match s {
90 "layout" => Subscription::Layout,
91 "title" => Subscription::Title,
92 "modifiers" => Subscription::Modifiers,
93 "adjust" => Subscription::Adjust,
94 "dismiss" => Subscription::Dismiss,
95 "shortcuts" => Subscription::Shortcuts,
96 _ => Subscription::Unknown,
97 }
98 }
99 }
100
101 /// A connected client with a known subscription.
102 struct Client {
103 subscription: Subscription,
104 stream: UnixStream,
105 /// The last line actually written to this client. A state push only
106 /// re-sends a topic whose formatted line CHANGED — a StatusUpdate is
107 /// one struct, so e.g. a camera animation (viewport text embeds pan/
108 /// zoom) used to re-broadcast identical layout/title lines at frame
109 /// rate, and every subscriber rebuilt its segment per frame.
110 last_line: Option<String>,
111 }
112
113 /// Handle to the status server for sending updates from the main loop.
114 ///
115 /// Every send bumps `wake`, the eventfd the server thread `poll()`s on
116 /// alongside its sockets. The thread used to spin on `try_recv` with a 20 ms
117 /// sleep — 50 wakeups/s forever, subscribers or not; now it blocks until a
118 /// socket or the main loop has something for it.
119 #[derive(Debug, Clone)]
120 pub struct StatusSender {
121 tx: mpsc::Sender<StatusMsg>,
122 wake: Arc<OwnedFd>,
123 }
124
125 impl StatusSender {
126 pub fn send(&self, update: StatusUpdate) {
127 // If the channel is full or the receiver is gone, just drop it.
128 if self.tx.send(StatusMsg::State(update)).is_ok() {
129 wake_fd(&self.wake);
130 }
131 }
132
133 /// Fire a one-shot menu-dismiss at every `dismiss` subscriber except the
134 /// segment with this app_id (pass "-" to exempt nobody).
135 pub fn send_menu_dismiss(&self, except_app_id: &str) {
136 if self.tx.send(StatusMsg::MenuDismiss { except_app_id: except_app_id.to_string() }).is_ok() {
137 wake_fd(&self.wake);
138 }
139 }
140 }
141
142 impl StatusSender {
143 /// Report a portal shortcut edge to every `shortcuts` subscriber.
144 pub fn send_shortcut_event(&self, line: &str) {
145 if self.tx.send(StatusMsg::Shortcut(line.to_string())).is_ok() {
146 wake_fd(&self.wake);
147 }
148 }
149 }
150
151 impl Drop for StatusSender {
152 /// Dropping the last handle disconnects the channel; the thread only
153 /// notices when it next wakes, so give it one.
154 fn drop(&mut self) {
155 wake_fd(&self.wake);
156 }
157 }
158
159 pub fn get_status_socket_path(display_socket: Option<&str>) -> String {
160 if let Some(display) = display_socket {
161 format!("/tmp/cce-status-interface-{}.sock", display)
162 } else {
163 "/tmp/cce-status-interface.sock".to_string()
164 }
165 }
166
167 /// Spawn the status server thread. Returns a StatusSender for the main loop.
168 pub fn spawn_status_server(display_socket: Option<String>) -> StatusSender {
169 let (tx, rx) = mpsc::channel::<StatusMsg>();
170 let wake = new_wake_fd().expect("failed to create status wake eventfd");
171 let thread_wake = wake.clone();
172
173 std::thread::Builder::new()
174 .name("cce-status-server".into())
175 .spawn(move || {
176 status_server_main(rx, thread_wake, display_socket);
177 })
178 .expect("failed to spawn status server thread");
179
180 StatusSender { tx, wake }
181 }
182
183 /// Block until the wake eventfd, the listener, or any subscriber socket is
184 /// readable. Returns `(wake, accept, per-client readiness)`; a client is
185 /// "ready" on data, hangup or error alike, since all three are handled by
186 /// reading it.
187 fn wait_for_activity(wake: &OwnedFd, listener: &UnixListener, clients: &[Client]) -> Option<(bool, bool, Vec<bool>)> {
188 let mut fds: Vec<libc::pollfd> = Vec::with_capacity(2 + clients.len());
189 for fd in [wake.as_raw_fd(), listener.as_raw_fd()] {
190 fds.push(libc::pollfd { fd, events: libc::POLLIN, revents: 0 });
191 }
192 for client in clients {
193 fds.push(libc::pollfd { fd: client.stream.as_raw_fd(), events: libc::POLLIN, revents: 0 });
194 }
195 let n = unsafe { libc::poll(fds.as_mut_ptr(), fds.len() as libc::nfds_t, -1) };
196 if n < 0 {
197 let err = std::io::Error::last_os_error();
198 if err.kind() == std::io::ErrorKind::Interrupted {
199 return Some((false, false, vec![false; clients.len()]));
200 }
201 log::error!("[status] poll failed: {}", err);
202 return None;
203 }
204 let ready = |f: &libc::pollfd| f.revents != 0;
205 Some((ready(&fds[0]), ready(&fds[1]), fds[2..].iter().map(ready).collect()))
206 }
207
208 fn status_server_main(rx: mpsc::Receiver<StatusMsg>, wake: Arc<OwnedFd>, display_socket: Option<String>) {
209 let socket_path = get_status_socket_path(display_socket.as_deref());
210 // Remove stale socket
211 let _ = std::fs::remove_file(&socket_path);
212
213 let listener = match UnixListener::bind(&socket_path) {
214 Ok(l) => l,
215 Err(e) => {
216 log::error!("[status] failed to bind {}: {}", socket_path, e);
217 return;
218 }
219 };
220
221 // Set non-blocking so accept() doesn't hang the thread
222 if let Err(e) = listener.set_nonblocking(true) {
223 log::error!("[status] failed to set non-blocking: {}", e);
224 return;
225 }
226
227 log::info!("[status] listening on {}", socket_path);
228
229 let mut clients: Vec<Client> = Vec::new();
230 let mut latest: Option<StatusUpdate> = None;
231
232 loop {
233 let Some((wake_ready, accept_ready, client_ready)) = wait_for_activity(&wake, &listener, &clients) else {
234 // poll() itself failing is not something a retry fixes fast;
235 // back off so the error line cannot flood the log.
236 std::thread::sleep(std::time::Duration::from_millis(100));
237 continue;
238 };
239 if wake_ready {
240 drain_wake_fd(wake.as_raw_fd());
241 }
242
243 let mut has_new_update = false;
244
245 // Accept new connections (the listener is non-blocking)
246 for _ in 0..5 {
247 if !accept_ready {
248 break;
249 }
250 match listener.accept() {
251 Ok((stream, _addr)) => {
252 // Read the subscription line while the socket is still
253 // blocking (bounded by a read timeout): poll() hands us
254 // the connection the instant it lands, which can be
255 // before the client's first line is in the buffer.
256 let sub = read_subscription(&stream);
257 if let Err(e) = stream.set_nonblocking(true) {
258 log::error!("[status] failed to set non-blocking on client: {}", e);
259 continue;
260 }
261 if sub != Subscription::Unknown {
262 log::info!("[status] new subscriber for {:?}", sub);
263 let client = Client {
264 subscription: sub,
265 stream,
266 last_line: None,
267 };
268 clients.push(client);
269 has_new_update = true; // push the latest status to the new client
270 }
271 }
272 Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
273 break;
274 }
275 Err(e) => {
276 // EMFILE and friends: the socket stays readable, so
277 // without a pause this loop (and its log line) spins the
278 // thread at 100% — observed as a 167GB log once dead
279 // subscribers had exhausted the fd table.
280 log::error!("[status] accept error: {}", e);
281 std::thread::sleep(std::time::Duration::from_millis(100));
282 break;
283 }
284 }
285 }
286
287 // Reap dead subscribers by reading: a subscriber never sends after
288 // its subscription line, so a successful zero-byte read is EOF (the
289 // client vanished). Waiting for a WRITE to fail leaked them instead
290 // — last_line dedup means a quiet topic may never write again, and
291 // every bar restart stranded its whole subscriber set. Enough
292 // restarts exhausted the fd table and took the session down.
293 {
294 let mut buf = [0u8; 64];
295 let mut dead_clients = Vec::new();
296 for (i, client) in clients.iter_mut().enumerate() {
297 // Only sockets poll() flagged; the rest are quiet, not dead.
298 if !client_ready.get(i).copied().unwrap_or(false) {
299 continue;
300 }
301 loop {
302 use std::io::Read;
303 match client.stream.read(&mut buf) {
304 Ok(0) => {
305 log::info!(
306 "[status] client {:?} disconnected (eof)",
307 client.subscription
308 );
309 dead_clients.push(i);
310 break;
311 }
312 // Unexpected chatter: drain and keep the client.
313 Ok(_) => continue,
314 Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
315 Err(_) => {
316 dead_clients.push(i);
317 break;
318 }
319 }
320 }
321 }
322 for i in dead_clients.into_iter().rev() {
323 clients.remove(i);
324 }
325 }
326
327 // Process incoming updates from the main loop
328 let mut dismiss_events: Vec<String> = Vec::new();
329 let mut shortcut_events: Vec<String> = Vec::new();
330 loop {
331 match rx.try_recv() {
332 Ok(StatusMsg::State(update)) => {
333 latest = Some(update);
334 has_new_update = true;
335 }
336 Ok(StatusMsg::MenuDismiss { except_app_id }) => {
337 dismiss_events.push(except_app_id);
338 }
339 Ok(StatusMsg::Shortcut(line)) => {
340 shortcut_events.push(line);
341 }
342 Err(mpsc::TryRecvError::Empty) => break,
343 Err(mpsc::TryRecvError::Disconnected) => {
344 log::info!("[status] channel disconnected, exiting");
345 let _ = std::fs::remove_file(&socket_path);
346 return;
347 }
348 }
349 }
350
351 // One-shot dismiss lines go only to `dismiss` subscribers; the line
352 // payload is the exempt app_id.
353 if !dismiss_events.is_empty() {
354 let mut dead_clients = Vec::new();
355 for (i, client) in clients.iter_mut().enumerate() {
356 if client.subscription != Subscription::Dismiss {
357 continue;
358 }
359 for except in &dismiss_events {
360 match client
361 .stream
362 .write_all(except.as_bytes())
363 .and_then(|_| client.stream.write_all(b"\n"))
364 {
365 Ok(_) => {}
366 Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {}
367 Err(_) => {
368 dead_clients.push(i);
369 break;
370 }
371 }
372 }
373 }
374 dead_clients.dedup();
375 for i in dead_clients.into_iter().rev() {
376 clients.remove(i);
377 }
378 }
379
380 // Portal shortcut edges go only to `shortcuts` subscribers, in order.
381 if !shortcut_events.is_empty() {
382 let mut dead_clients = Vec::new();
383 for (i, client) in clients.iter_mut().enumerate() {
384 if client.subscription != Subscription::Shortcuts {
385 continue;
386 }
387 for line in &shortcut_events {
388 match client
389 .stream
390 .write_all(line.as_bytes())
391 .and_then(|_| client.stream.write_all(b"\n"))
392 {
393 Ok(_) => {}
394 Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {}
395 Err(_) => {
396 dead_clients.push(i);
397 break;
398 }
399 }
400 }
401 }
402 dead_clients.dedup();
403 for i in dead_clients.into_iter().rev() {
404 clients.remove(i);
405 }
406 }
407
408 // If we got a new update, push it to all clients
409 if has_new_update {
410 if let Some(ref update) = latest {
411 let mut dead_clients = Vec::new();
412
413 for (i, client) in clients.iter_mut().enumerate() {
414 // Dismiss and shortcuts subscribers get one-shot events
415 // only, never state pushes.
416 if matches!(client.subscription, Subscription::Dismiss | Subscription::Shortcuts) {
417 continue;
418 }
419 let msg = format_for_subscription(&client.subscription, update);
420 // Only lines that changed for THIS topic go out (see
421 // Client::last_line); a fresh client always gets one.
422 if client.last_line.as_deref() == Some(msg.as_str()) {
423 continue;
424 }
425 match client
426 .stream
427 .write_all(msg.as_bytes())
428 .and_then(|_| client.stream.write_all(b"\n"))
429 {
430 Ok(_) => {
431 client.last_line = Some(msg);
432 }
433 Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
434 // Client not ready to receive — skip for now
435 }
436 Err(ref e) if e.kind() == std::io::ErrorKind::BrokenPipe => {
437 log::info!(
438 "[status] client {:?} disconnected (broken pipe)",
439 client.subscription
440 );
441 dead_clients.push(i);
442 }
443 Err(e) => {
444 log::error!(
445 "[status] write error to client {:?}: {}",
446 client.subscription, e
447 );
448 dead_clients.push(i);
449 }
450 }
451 }
452
453 // Remove dead clients (iterate in reverse to preserve indices)
454 for i in dead_clients.into_iter().rev() {
455 clients.remove(i);
456 }
457 }
458 }
459 }
460 }
461
462 fn read_subscription(stream: &UnixStream) -> Subscription {
463 let mut reader = std::io::BufReader::new(stream);
464 let mut line = String::new();
465 // Bounded blocking read: a subscriber writes its one line right after
466 // connecting, so this returns at once in practice; the timeout is for a
467 // client that connects and says nothing.
468 stream
469 .set_read_timeout(Some(std::time::Duration::from_millis(200)))
470 .ok();
471 match reader.read_line(&mut line) {
472 Ok(_) => Subscription::from_str(&line),
473 Err(e) => {
474 log::error!("[status] failed to read subscription: {}", e);
475 Subscription::Unknown
476 }
477 }
478 }
479
480 fn format_for_subscription(sub: &Subscription, update: &StatusUpdate) -> String {
481 match sub {
482 Subscription::Layout => update.layout_text.clone(),
483 Subscription::Title => update.title_text.clone(),
484 Subscription::Modifiers => update.modifiers_text.clone(),
485 Subscription::Adjust => update.adjust_text.clone(),
486 Subscription::Backdrop(app_id) => {
487 // A segment the compositor has no sample for (not mapped yet, or
488 // its app_id does not match a window) is told so explicitly
489 // rather than left to time out: "unknown" is a state the bar
490 // renders for, not an absence.
491 match update.backdrops.iter().find(|(id, _, _)| id == app_id) {
492 Some((_, luma, spread)) => format!("{} {}", luma, spread),
493 None => "unknown".to_string(),
494 }
495 }
496 Subscription::Dismiss | Subscription::Shortcuts | Subscription::Unknown => String::new(),
497 }
498 }
499
500 pub unsafe fn build_status_update(wm: &crate::window_manager::WindowManager) -> StatusUpdate {
501 // `focused_window()` falls back to the most recent real window so the
502 // bar doesn't flash while overlay UI (the launcher) briefly holds
503 // focus. But an explicit Focus::None (desktop click) is a real,
504 // user-visible state — keystrokes go nowhere — and the status feed
505 // must report it honestly instead of showing the last window as if it
506 // still had focus.
507 let seat_focus_is_none = wm
508 .first_seat()
509 .map(|s| matches!((*s).focused, crate::seat::Focus::None))
510 .unwrap_or(false);
511 let focused_window = if seat_focus_is_none {
512 std::ptr::null_mut()
513 } else {
514 wm.focused_window()
515 };
516
517 let layout_text = if !focused_window.is_null() {
518 (*focused_window).tiling_mode.as_str().to_string()
519 } else {
520 let focused_layer = wm.focused_layer_surface();
521 let mut is_cce_cloud = false;
522 if !focused_layer.is_null() {
523 let wlr_layer_surface = crate::ffi::wlr_layer_surface_v1_try_from_wlr_surface(focused_layer);
524 if !wlr_layer_surface.is_null() && !(*wlr_layer_surface).namespace.is_null() {
525 let ns = std::ffi::CStr::from_ptr((*wlr_layer_surface).namespace).to_string_lossy();
526 if ns.starts_with("cce-cloud") {
527 is_cce_cloud = true;
528 }
529 }
530 }
531 if is_cce_cloud {
532 "Overlay".to_string()
533 } else {
534 "---".to_string()
535 }
536 };
537
538 let title_text = if !focused_window.is_null() {
539 let title_ptr = (*focused_window).get_title();
540 if !title_ptr.is_null() {
541 std::ffi::CStr::from_ptr(title_ptr).to_string_lossy().into_owned()
542 } else {
543 "(none)".to_string()
544 }
545 } else {
546 let focused_layer = wm.focused_layer_surface();
547 if !focused_layer.is_null() {
548 let wlr_layer_surface = crate::ffi::wlr_layer_surface_v1_try_from_wlr_surface(focused_layer);
549 if !wlr_layer_surface.is_null() && !(*wlr_layer_surface).namespace.is_null() {
550 std::ffi::CStr::from_ptr((*wlr_layer_surface).namespace).to_string_lossy().into_owned()
551 } else {
552 "(none)".to_string()
553 }
554 } else {
555 "(none)".to_string()
556 }
557 };
558
559 let seat_ptr = wm.first_seat().unwrap_or(std::ptr::null_mut());
560 let mut super_pressed = false;
561 if !seat_ptr.is_null() {
562 let wlr_keyboard = crate::ffi::river_wlr_seat_get_keyboard((*seat_ptr).wlr_seat);
563 if !wlr_keyboard.is_null() {
564 let modifiers = crate::ffi::wlr_keyboard_get_modifiers(wlr_keyboard);
565 super_pressed = modifiers & 0x40 != 0;
566 }
567 }
568 let modifiers_text = if super_pressed { "super" } else { "none" }.to_string();
569
570 StatusUpdate {
571 layout_text,
572 title_text,
573 modifiers_text,
574 adjust_text: if wm.window_adjust_active() { "on" } else { "off" }.to_string(),
575 // Measured in the render pass (see `Output::measure_status_backdrops`)
576 // because that is where the frame's grid geometry already lives;
577 // here it is only carried.
578 backdrops: wm.status_backdrops.borrow().clone(),
579 }
580 }