all: fix memory leaks

jay-config/src/_private/client.rs

@@ -588,6 +588,9 @@ impl Client {
                     handler();
                 }
             }
+            ServerMessage::Clear => {
+                // only used by test config
+            }
         }
     }
 

jay-config/src/_private/ipc.rs

@@ -45,6 +45,7 @@ pub enum ServerMessage {
     TimerExpired {
         timer: Timer,
     },
+    Clear,
 }
 
 #[derive(Encode, BorrowDecode, Debug)]

src/async_engine.rs

@@ -82,6 +82,14 @@ impl AsyncEngine {
         Timer::new(self, clock_id)
     }
 
+    pub fn clear(&self) {
+        for (_, fd) in self.fds.lock().drain() {
+            fd.readers.take();
+            fd.writers.take();
+        }
+        self.queue.clear();
+    }
+
     pub fn spawn<T, F: Future<Output = T> + 'static>(&self, f: F) -> SpawnedFuture<T> {
         self.queue.spawn(Phase::EventHandling, f)
     }
@@ -607,6 +615,13 @@ mod queue {
     }
 
     impl DispatchQueue {
+        pub fn clear(&self) {
+            self.yields.take();
+            for queue in &self.queues {
+                queue.take();
+            }
+        }
+
         pub fn push(&self, runnable: Runnable, phase: Phase) {
             self.queues[phase as usize].push(runnable);
             self.num_queued.fetch_add(1);

src/compositor.rs

@@ -10,7 +10,7 @@ use {
             metal, x,
         },
         cli::{CliBackend, GlobalArgs, RunArgs},
-        client::Clients,
+        client::{ClientId, Clients},
         clientmem::{self, ClientMemError},
         config::ConfigProxy,
         dbus::Dbus,
@@ -176,7 +176,9 @@ fn start_compositor2(
         run_toplevel,
         config_dir: config_dir(),
         config_file_id: NumCell::new(1),
+        tracker: Default::default(),
     });
+    state.tracker.register(ClientId::from_raw(0));
     create_dummy_output(&state);
     let acceptor = Acceptor::install(&state)?;
     forker.install(&state);
@@ -190,10 +192,14 @@ fn start_compositor2(
     let compositor = engine.spawn(start_compositor3(state.clone(), test_future));
     el.run()?;
     drop(compositor);
+    drop(acceptor);
+    drop(forker);
+    engine.clear();
     state.clear();
     for (_, seat) in state.globals.seats.lock().deref() {
         seat.clear();
     }
+    drop(state);
     leaks::log_leaked();
     Ok(())
 }

src/config.rs

@@ -108,6 +108,10 @@ impl ConfigProxy {
     pub fn graphics_initialized(&self) {
         self.send(&ServerMessage::GraphicsInitialized);
     }
+
+    pub fn clear(&self) {
+        self.send(&ServerMessage::Clear);
+    }
 }
 
 impl Drop for ConfigProxy {

src/dbus.rs

@@ -153,6 +153,11 @@ impl Dbus {
         }
     }
 
+    pub fn clear(&self) {
+        self.system.clear();
+        self.session.clear();
+    }
+
     pub fn system(&self) -> Result<Rc<DbusSocket>, DbusError> {
         self.system
             .get(&self.eng, "/var/run/dbus/system_bus_socket", "System bus")
@@ -261,6 +266,12 @@ impl DbusHolder {
             run_toplevel: run_toplevel.clone(),
         }
     }
+
+    pub fn clear(&self) {
+        if let Some(socket) = self.socket.take() {
+            socket.clear();
+        }
+    }
 }
 
 impl Drop for DbusHolder {

src/dbus/socket.rs

@@ -20,6 +20,14 @@ use {
 };
 
 impl DbusSocket {
+    pub fn clear(&self) {
+        self.auth.take();
+        self.incoming.take();
+        self.outgoing_.take();
+        self.reply_handlers.clear();
+        self.signal_handlers.borrow_mut().clear();
+    }
+
     pub(super) fn kill(self: &Rc<Self>) {
         self.dead.set(true);
         self.auth.take();

src/forker.rs

@@ -74,6 +74,12 @@ pub enum ForkerError {
 }
 
 impl ForkerProxy {
+    pub fn clear(&self) {
+        self.task_in.take();
+        self.task_out.take();
+        self.task_proc.take();
+    }
+
     pub fn create() -> Result<Self, ForkerError> {
         let (parent, child) = match uapi::socketpair(
             c::AF_UNIX,
@@ -240,8 +246,8 @@ impl ForkerProxy {
             }
             if let Err(e) = io.write_msg(msg).await {
                 log::error!("Could not write to the ol' forker: {}", ErrorFmt(e));
+                self.clear();
                 state.forker.set(None);
-                self.task_out.take();
                 return;
             }
         }
@@ -258,9 +264,8 @@ impl ForkerProxy {
             let _ = uapi::waitpid(self.pid, 0);
         }
         log::error!("The ol' forker died. Cannot spawn further processes.");
+        self.clear();
         state.forker.set(None);
-        self.task_out.take();
-        self.task_proc.take();
     }
 }
 

src/globals.rs

@@ -116,6 +116,12 @@ impl Globals {
         }
     }
 
+    pub fn clear(&self) {
+        self.registry.clear();
+        self.outputs.clear();
+        self.seats.clear();
+    }
+
     pub fn add_singletons(&self, backend: &Rc<dyn Backend>) {
         macro_rules! add_singleton {
             ($name:ident) => {

src/ifs/wl_output.rs

@@ -82,6 +82,11 @@ pub struct WlOutputGlobal {
 }
 
 impl WlOutputGlobal {
+    pub fn clear(&self) {
+        self.node.take();
+        self.bindings.borrow_mut().clear();
+    }
+
     pub fn new(
         name: GlobalName,
         state: &Rc<State>,

src/it/test_config.rs

@@ -61,7 +61,7 @@ unsafe extern "C" fn unref(data: *const u8) {
 }
 
 unsafe extern "C" fn handle_msg(data: *const u8, msg: *const u8, size: usize) {
-    let _tc = &*data.cast::<TestConfig>();
+    let tc = &*data.cast::<TestConfig>();
     let msg = std::slice::from_raw_parts(msg, size);
     let res = bincode::decode_from_slice::<ServerMessage, _>(msg, bincode_ops());
     let (msg, _) = match res {
@@ -83,6 +83,7 @@ unsafe extern "C" fn handle_msg(data: *const u8, msg: *const u8, size: usize) {
         ServerMessage::DelConnector { .. } => {}
         ServerMessage::TimerExpired { .. } => {}
         ServerMessage::GraphicsInitialized => {}
+        ServerMessage::Clear => tc.clear(),
     }
 }
 
@@ -114,10 +115,8 @@ impl TestConfig {
     pub fn quit(&self) -> Result<(), TestError> {
         self.send(ClientMessage::Quit)
     }
-}
 
-impl Drop for TestConfig {
-    fn drop(&mut self) {
+    fn clear(&self) {
         unsafe {
             if let Some(srv) = self.srv.take() {
                 (srv.srv_unref)(srv.srv_data);
@@ -125,3 +124,9 @@ impl Drop for TestConfig {
         }
     }
 }
+
+impl Drop for TestConfig {
+    fn drop(&mut self) {
+        self.clear();
+    }
+}

src/state.rs

@@ -22,6 +22,7 @@ use {
                 NoneSurfaceExt,
             },
         },
+        leaks::Tracker,
         logger::Logger,
         rect::Rect,
         render::RenderContext,
@@ -99,8 +100,15 @@ pub struct State {
     pub run_toplevel: Rc<RunToplevel>,
     pub config_dir: Option<String>,
     pub config_file_id: NumCell<u64>,
+    pub tracker: Tracker<Self>,
 }
 
+// impl Drop for State {
+//     fn drop(&mut self) {
+//         log::info!("drop state");
+//     }
+// }
+
 impl Debug for State {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("State").finish_non_exhaustive()
@@ -408,8 +416,39 @@ impl State {
     pub fn clear(&self) {
         self.xwayland.handler.borrow_mut().take();
         self.clients.clear();
-        self.config.set(None);
+        if let Some(config) = self.config.set(None) {
+            config.clear();
+        }
+        if let Some(forker) = self.forker.set(None) {
+            forker.clear();
+        }
+        if let Some(output) = self.dummy_output.set(None) {
+            output.clear();
+        }
+        self.acceptor.set(None);
         self.backend.set(Rc::new(DummyBackend));
+        self.run_toplevel.clear();
+        self.xwayland.handler.borrow_mut().take();
+        self.xwayland.queue.clear();
+        self.idle.inhibitors.clear();
+        for (_, connector) in self.connectors.lock().drain() {
+            connector.handler.take();
+        }
+        for (_, output) in self.outputs.lock().drain() {
+            output.node.clear();
+        }
+        self.dbus.clear();
+        self.pending_container_layout.clear();
+        self.pending_container_render_data.clear();
+        self.pending_float_layout.clear();
+        self.pending_float_titles.clear();
+        self.slow_clients.clear();
+        self.input_device_handlers.borrow_mut().clear();
+        self.backend_events.clear();
+        self.workspaces.clear();
+        self.globals.clear();
+        self.render_ctx.set(None);
+        self.root.clear();
         {
             let seats = mem::take(self.globals.seats.lock().deref_mut());
             for seat in seats.values() {

src/tasks/connector.rs

@@ -27,7 +27,9 @@ pub fn handle(state: &Rc<State>, connector: &Rc<dyn Connector>) {
     };
     let future = state.eng.spawn(oh.handle());
     data.handler.set(Some(future));
-    state.connectors.set(id, data);
+    if state.connectors.set(id, data).is_some() {
+        panic!("Connector id has been reused");
+    }
 }
 
 struct ConnectorHandler {

src/tree/display.rs

@@ -32,6 +32,10 @@ impl DisplayNode {
         }
     }
 
+    pub fn clear(&self) {
+        self.outputs.clear();
+    }
+
     pub fn update_extents(&self) {
         let outputs = self.outputs.lock();
         let mut x1 = i32::MAX;

src/tree/output.rs

@@ -41,6 +41,15 @@ pub struct OutputNode {
 }
 
 impl OutputNode {
+    pub fn clear(&self) {
+        self.global.clear();
+        self.workspace.set(None);
+        let workspaces: Vec<_> = self.workspaces.iter().collect();
+        for workspace in workspaces {
+            workspace.clear();
+        }
+    }
+
     pub fn update_render_data(&self) {
         let mut rd = self.render_data.borrow_mut();
         rd.titles.clear();

src/tree/workspace.rs

@@ -33,6 +33,12 @@ pub struct WorkspaceNode {
 }
 
 impl WorkspaceNode {
+    pub fn clear(&self) {
+        self.container.set(None);
+        self.output_link.set(None);
+        self.fullscreen.set(None);
+    }
+
     pub fn set_container(self: &Rc<Self>, container: &Rc<ContainerNode>) {
         let pos = self.position.get();
         container.clone().tl_change_extents(&pos);

src/utils/run_toplevel.rs

@@ -36,6 +36,10 @@ impl RunToplevel {
         self.schedule_dyn(Box::new(f));
     }
 
+    pub fn clear(&self) {
+        self.queue.clear();
+    }
+
     fn schedule_dyn(&self, f: Box<dyn FnOnce()>) {
         self.queue.push(f);
     }

src/utils/syncqueue.rs

@@ -36,4 +36,10 @@ impl<T> SyncQueue<T> {
             mem::swap(self.el.get().deref_mut(), queue);
         }
     }
+
+    pub fn take(&self) -> VecDeque<T> {
+        let mut res = VecDeque::new();
+        self.swap(&mut res);
+        res
+    }
 }