Wayland compositor (wlroots)
git clone https://git.lucas.co/cce-compositor.git
src/server/ipc_server.rs (6.9K)
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 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 fn handle_client(mut stream: UnixStream, tx: IpcSender) {
154 let peer_pid = socket_peer_pid(&stream);
155 let mut buf = [0u8; 4096];
156 match stream.read(&mut buf) {
157 Ok(0) => {}
158 Ok(n) => {
159 let s = String::from_utf8_lossy(&buf[..n]);
160 let cmd = s.trim().to_string();
161 if !cmd.is_empty() {
162 // Commands answer from the IPC drain and so are quick; a
163 // second is a generous leash that still surfaces a wedged
164 // compositor. `screenshot` is the exception: its reply now
165 // waits for the capture, which happens on the next composited
166 // frame, and a cold readback (first capture after an idle
167 // spell — NVIDIA recompiles shaders on the way) has been
168 // measured over a second. Timing that out would report
169 // failure for a capture that lands.
170 let timeout = if cmd.starts_with("screenshot") {
171 std::time::Duration::from_secs(5)
172 } else {
173 std::time::Duration::from_millis(1000)
174 };
175 let (reply_tx, reply_rx) = mpsc::channel();
176 if tx.send(IpcRequest { command: cmd, reply_tx, peer_pid }) {
177 if let Ok(reply) = reply_rx.recv_timeout(timeout) {
178 let _ = stream.write_all(reply.as_bytes());
179 } else {
180 let _ = stream.write_all(b"error: timeout processing command\n");
181 }
182 }
183 }
184 }
185 Err(e) => {
186 log::error!("[ipc] stream read error: {}", e);
187 }
188 }
189 }