diff --git a/crates/tty7-core/src/core/threads.rs b/crates/tty7-core/src/core/threads.rs index f61f7e86..ed546967 100644 --- a/crates/tty7-core/src/core/threads.rs +++ b/crates/tty7-core/src/core/threads.rs @@ -7,3 +7,118 @@ pub fn promote_to_user_interactive() { libc::pthread_set_qos_class_self_np(libc::qos_class_t::QOS_CLASS_USER_INTERACTIVE, 0); } } + +/// Takes a lock, and goes on taking it after another thread has panicked +/// holding it. +/// +/// `Mutex::lock().unwrap()` is the wrong default for anything the daemon owns. +/// Poisoning protects nothing: whatever inconsistency a panic left behind is +/// there either way, and the flag only decides whether the *next* thread to +/// ask also dies. In a process that holds every shell on the machine, that +/// turns one bug in one pane into a daemon that can no longer serve any of +/// them — the pty master, the child handle, the writer and the pane state all +/// sit behind mutexes, and a panic in any critical section takes the lot. +/// +/// So the policy is to carry on. A garbled write to one pane is recoverable; +/// losing every session on the box is not. This is what most of the code +/// already did by hand, spelled `unwrap_or_else(|e| e.into_inner())` in about +/// seventy places, against about eighty that panicked instead — the drift is +/// what this exists to stop. +/// +/// Where the inconsistency genuinely cannot be tolerated, take the poison +/// explicitly with `lock()` and decide there; the point is that it be a +/// decision rather than a default. +pub trait Locked { + /// The guard, whether or not the lock is poisoned. + fn locked(&self) -> std::sync::MutexGuard<'_, T>; +} + +impl Locked for std::sync::Mutex { + fn locked(&self) -> std::sync::MutexGuard<'_, T> { + self.lock().unwrap_or_else(|e| e.into_inner()) + } +} + +#[cfg(test)] +mod tests { + use super::Locked as _; + + /// A mutex another thread panicked under is still usable, and still holds + /// what that thread had written before it went. + #[test] + fn a_poisoned_lock_is_still_a_lock() { + let shared = std::sync::Arc::new(std::sync::Mutex::new(vec![1, 2, 3])); + let poisoner = std::sync::Arc::clone(&shared); + let died = std::thread::spawn(move || { + let mut held = poisoner.lock().expect("the first lock is clean"); + held.push(4); + panic!("the thread goes down holding it"); + }) + .join(); + assert!(died.is_err(), "the thread did panic"); + assert!(shared.lock().is_err(), "and the mutex is poisoned"); + + assert_eq!( + *shared.locked(), + vec![1, 2, 3, 4], + "the lock still opens, and the write that got through is still there" + ); + shared.locked().push(5); + assert_eq!(*shared.locked(), vec![1, 2, 3, 4, 5], "and it stays usable"); + } + + /// Nothing the daemon owns takes a lock that can panic on poison. + /// + /// The policy above is only worth stating if it holds, and it drifted + /// once already: about seventy places carried on and about eighty died, + /// with nothing to say which was intended. A guard is cheaper than + /// re-deciding it per review. + /// + /// Scoped to `daemon/` deliberately. That is the process holding every + /// shell on the machine, where the cascade costs the most; a tool that + /// panics takes only itself with it. + #[test] + fn the_daemon_takes_no_lock_that_dies_of_poison() { + fn walk(dir: &std::path::Path, found: &mut Vec) { + let Ok(entries) = std::fs::read_dir(dir) else { + return; + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + walk(&path, found); + continue; + } + if path.extension().is_none_or(|e| e != "rs") + || path.file_name().is_some_and(|n| n == "tests.rs") + { + continue; + } + let Ok(text) = std::fs::read_to_string(&path) else { + continue; + }; + let mut in_tests = false; + for (n, line) in text.lines().enumerate() { + in_tests |= line.contains("#[cfg(test)]"); + if !in_tests && line.contains(".lock().unwrap()") { + found.push(format!("{}:{}", path.display(), n + 1)); + } + } + } + } + + let daemon = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("src") + .join("daemon"); + assert!(daemon.is_dir(), "the daemon sources moved: {daemon:?}"); + let mut found = Vec::new(); + walk(&daemon, &mut found); + assert!( + found.is_empty(), + "these take a lock that panics once another thread has poisoned it, \ + which is how one pane's bug becomes every pane's; use `Locked::locked` \ + or take the poison deliberately:\n{}", + found.join("\n") + ); + } +} diff --git a/crates/tty7-core/src/daemon/pane.rs b/crates/tty7-core/src/daemon/pane.rs index 3e4d580d..8107f869 100644 --- a/crates/tty7-core/src/daemon/pane.rs +++ b/crates/tty7-core/src/daemon/pane.rs @@ -17,6 +17,7 @@ use crate::daemon::protocol::{ ShellSpec, WinSize, }; use crate::daemon::shell_integration; +use crate::core::threads::Locked as _; #[cfg(windows)] fn default_prog() -> CommandBuilder { @@ -644,14 +645,14 @@ impl OutputGate { pub fn sub(&self, n: usize) { let prev = self.queued.fetch_sub(n as i64, Ordering::Relaxed); if prev >= Self::HIGH_WATER && prev - (n as i64) < Self::HIGH_WATER { - let _park = self.park.lock().unwrap(); + let _park = self.park.locked(); self.drained.notify_all(); } } fn reset(&self) { self.queued.store(0, Ordering::Relaxed); - let _park = self.park.lock().unwrap(); + let _park = self.park.locked(); self.drained.notify_all(); } @@ -668,7 +669,7 @@ impl OutputGate { return; } let deadline = std::time::Instant::now() + Self::MAX_WAIT; - let mut park = self.park.lock().unwrap(); + let mut park = self.park.locked(); while self.queued.load(Ordering::Relaxed) >= Self::HIGH_WATER { let left = deadline.saturating_duration_since(std::time::Instant::now()); if left.is_zero() { @@ -973,7 +974,7 @@ impl DeathReporter { } fn probe_exit_code(&self, probe: impl FnMut() -> Option + Send + 'static) { - *self.exit_code.lock().unwrap() = Some(Box::new(probe)); + *self.exit_code.locked() = Some(Box::new(probe)); } #[cfg(windows)] @@ -991,7 +992,7 @@ impl DeathReporter { .unwrap() .as_mut() .and_then(|probe| probe()); - let mut st = state.lock().unwrap(); + let mut st = state.locked(); st.alive = false; st.exit_code = code; let pane = st.id; @@ -1007,7 +1008,7 @@ impl DeathReporter { if subscribed { return; } - if let Some(on_dead) = self.on_dead.lock().unwrap().take() { + if let Some(on_dead) = self.on_dead.locked().take() { on_dead(); } } @@ -1599,7 +1600,7 @@ impl DaemonPane { }, death, ); - *pane.reader.lock().unwrap() = Some(reader); + *pane.reader.locked() = Some(reader); pane } @@ -1625,7 +1626,7 @@ impl DaemonPane { .ok()? .as_ref() .and_then(|master| master.as_raw_fd())?; - let st = self.state.lock().unwrap(); + let st = self.state.locked(); Some(Carried { id: self.id, owner: self.owner.clone(), @@ -1769,7 +1770,7 @@ impl DaemonPane { let broker = { let state = state.clone(); crate::daemon::ssh::PromptBroker::new(Box::new(move |msg: DaemonMsg| { - match &state.lock().unwrap().subscriber { + match &state.locked().subscriber { Some(sub) => sub.send(msg).is_ok(), None => false, } @@ -1807,7 +1808,7 @@ impl DaemonPane { }, death, ); - *pane.reader.lock().unwrap() = Some(reader); + *pane.reader.locked() = Some(reader); crate::daemon::ssh::SshManager::global().spawn_native_session( id, @@ -1830,7 +1831,7 @@ impl DaemonPane { pub(crate) fn ssh_connection(&self) -> Option> { match &self.backend { - PaneBackend::NativeSsh(b) => b.connection.lock().unwrap().upgrade(), + PaneBackend::NativeSsh(b) => b.connection.locked().upgrade(), PaneBackend::Pty(_) => None, } } @@ -1870,7 +1871,7 @@ impl DaemonPane { // local can `ssh` out mid-session, so the flag is refreshed from // the same remote-context poll below. Seed it from the pane's // current context so the first probe answers correctly. - let starts_local = state.lock().unwrap().remote.is_none(); + let starts_local = state.locked().remote.is_none(); let mut graphics = GraphicsSniffer::new_local(starts_local); let mut buf = [0u8; 65536]; @@ -1993,7 +1994,7 @@ impl DaemonPane { } let remote = if poll_now { let managed = { - let st = state.lock().unwrap(); + let st = state.locked(); st.remote .as_ref() .is_some_and(|remote| remote.kind != RemoteKind::Ssh) @@ -2014,7 +2015,7 @@ impl DaemonPane { || remote.is_some() || agent.is_some() || probed_cwd.is_some(); - let mut st = state.lock().unwrap(); + let mut st = state.locked(); let facts_before = may_change_facts.then(|| observed_facts(&st)); st.ring.append(bytes); fan_out_output(&mut st, bytes, frames, &gate); @@ -2070,14 +2071,14 @@ impl DaemonPane { } pub fn attach(&self, subscriber: Sender) -> u64 { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.locked(); let epoch = attach_subscriber(&mut st, subscriber); self.gate.reset(); epoch } pub fn detach(&self, epoch: u64) -> bool { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.locked(); if st.subscriber_epoch == epoch { st.subscriber = None; self.gate.reset(); @@ -2086,21 +2087,21 @@ impl DaemonPane { } pub fn observe(&self, observer: Sender, gate: Arc) -> u64 { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.locked(); observe_subscriber(&mut st, observer, gate) } pub(crate) fn unobserve(&self, observer_id: u64) { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.locked(); st.observers.retain(|obs| obs.id != observer_id); } pub fn controls(&self, epoch: u64) -> bool { - self.state.lock().unwrap().subscriber_epoch == epoch + self.state.locked().subscriber_epoch == epoch } pub fn agent_state(&self) -> Option { - agent_state_snapshot(&self.state.lock().unwrap()) + agent_state_snapshot(&self.state.locked()) } pub fn gate(&self) -> Arc { @@ -2135,7 +2136,7 @@ impl DaemonPane { } pub fn resize(&self, size: WinSize) { - resize_state(&mut self.state.lock().unwrap(), size); + resize_state(&mut self.state.locked(), size); match &self.backend { PaneBackend::Pty(p) => { if let Ok(master) = p.master.lock() @@ -2149,12 +2150,12 @@ impl DaemonPane { } pub fn alive(&self) -> bool { - self.state.lock().unwrap().alive + self.state.locked().alive } pub fn info(&self) -> PaneInfo { let (cwd, osc_title, alive) = { - let st = self.state.lock().unwrap(); + let st = self.state.locked(); (st.cwd.clone(), st.osc_title.clone(), st.alive) }; PaneInfo { @@ -2168,7 +2169,7 @@ impl DaemonPane { } pub(crate) fn remote_context(&self) -> Option { - let cached = self.state.lock().unwrap().remote.clone(); + let cached = self.state.locked().remote.clone(); cached.or_else(|| self.foreground_remote_context()) } @@ -2179,7 +2180,7 @@ impl DaemonPane { /// snapshot is fine — the snapshot returns its own mark, and that pair is /// what gets recorded. pub(crate) fn scrollback_mark(&self) -> u64 { - self.state.lock().unwrap().ring.appended + self.state.locked().ring.appended } /// The pane's screen, capped for storage, with the mark that says how much @@ -2191,7 +2192,7 @@ impl DaemonPane { /// that stopped producing at that moment would keep the stale copy for /// good, because its mark would never move again. pub(crate) fn scrollback_snapshot(&self) -> (Vec, u64) { - let st = self.state.lock().unwrap(); + let st = self.state.locked(); let mut segments = st.ring.snapshot(); crate::daemon::scrollback::trim_to(&mut segments, crate::daemon::scrollback::SNAPSHOT_CAP); (segments, st.ring.appended) @@ -2345,7 +2346,7 @@ impl Drop for DaemonPane { { let _ = child.wait(); } - if let Some(handle) = self.reader.lock().unwrap().take() { + if let Some(handle) = self.reader.locked().take() { join_bounded(handle, Duration::from_secs(2)); } if let PaneBackend::Pty(p) = &mut self.backend diff --git a/crates/tty7-core/src/daemon/server.rs b/crates/tty7-core/src/daemon/server.rs index da895dde..dd45ecdf 100644 --- a/crates/tty7-core/src/daemon/server.rs +++ b/crates/tty7-core/src/daemon/server.rs @@ -8,6 +8,7 @@ use crate::daemon::pane::DaemonPane; use crate::daemon::protocol::{ClientMsg, DaemonMsg, DaemonVersion, RemoteKind}; use crate::daemon::ssh::SshConnection; use crate::daemon::transport::{self, Stream}; +use crate::core::threads::Locked as _; struct Registry { panes: Mutex>>, @@ -57,20 +58,20 @@ impl Registry { } fn insert(&self, pane: Arc) { - self.panes.lock().unwrap().insert(pane.id, pane); + self.panes.locked().insert(pane.id, pane); } fn get(&self, id: u64) -> Option> { - self.panes.lock().unwrap().get(&id).cloned() + self.panes.locked().get(&id).cloned() } fn remove(&self, id: u64) -> Option> { - self.panes.lock().unwrap().remove(&id) + self.panes.locked().remove(&id) } fn drain_and_kill(&self) { let panes: Vec> = { - let mut guard = self.panes.lock().unwrap(); + let mut guard = self.panes.locked(); guard.drain().map(|(_, p)| p).collect() }; for pane in panes { @@ -79,7 +80,7 @@ impl Registry { } fn all(&self) -> Vec> { - self.panes.lock().unwrap().values().cloned().collect() + self.panes.locked().values().cloned().collect() } fn list(&self) -> Vec { @@ -94,7 +95,7 @@ impl Registry { impl crate::host::server::PaneDirectory for Registry { fn pane_count(&self) -> u64 { - self.panes.lock().unwrap().len() as u64 + self.panes.locked().len() as u64 } fn panes(&self) -> Vec { @@ -102,7 +103,7 @@ impl crate::host::server::PaneDirectory for Registry { } fn agent_states(&self) -> Vec { - let panes: Vec> = self.panes.lock().unwrap().values().cloned().collect(); + let panes: Vec> = self.panes.locked().values().cloned().collect(); let mut states: Vec<_> = panes.iter().filter_map(|p| p.agent_state()).collect(); states.sort_by_key(|s| s.pane_id); states diff --git a/crates/tty7-core/src/daemon/ssh/broker.rs b/crates/tty7-core/src/daemon/ssh/broker.rs index 1bd80c81..3e7da93d 100644 --- a/crates/tty7-core/src/daemon/ssh/broker.rs +++ b/crates/tty7-core/src/daemon/ssh/broker.rs @@ -6,6 +6,7 @@ use std::time::Duration; use tokio::sync::oneshot; use crate::daemon::protocol::{AuthPromptKind, AuthResponse, DaemonMsg, SshPhase}; +use crate::core::threads::Locked as _; const PROMPT_TIMEOUT: Duration = Duration::from_secs(120); const DELIVERY_WINDOW: Duration = Duration::from_secs(15); @@ -27,27 +28,27 @@ impl PromptBroker { } pub fn has_pending(&self) -> bool { - !self.pending.lock().unwrap().is_empty() + !self.pending.locked().is_empty() } pub async fn prompt(&self, kind: AuthPromptKind) -> AuthResponse { let id = self.next_id.fetch_add(1, Ordering::Relaxed); let (tx, rx) = oneshot::channel(); - self.pending.lock().unwrap().insert(id, tx); + self.pending.locked().insert(id, tx); let frame = DaemonMsg::AuthPrompt { request_id: id, prompt: kind, }; if !self.deliver_with_retry(frame).await { - self.pending.lock().unwrap().remove(&id); + self.pending.locked().remove(&id); return AuthResponse::Cancelled; } match tokio::time::timeout(PROMPT_TIMEOUT, rx).await { Ok(Ok(resp)) => resp, _ => { - self.pending.lock().unwrap().remove(&id); + self.pending.locked().remove(&id); AuthResponse::Cancelled } } @@ -78,7 +79,7 @@ impl PromptBroker { } pub fn deliver(&self, request_id: u64, response: AuthResponse) { - if let Some(tx) = self.pending.lock().unwrap().remove(&request_id) { + if let Some(tx) = self.pending.locked().remove(&request_id) { let _ = tx.send(response); } } diff --git a/crates/tty7-core/src/daemon/ssh/forward.rs b/crates/tty7-core/src/daemon/ssh/forward.rs index 2759a8fc..bfaef1fc 100644 --- a/crates/tty7-core/src/daemon/ssh/forward.rs +++ b/crates/tty7-core/src/daemon/ssh/forward.rs @@ -15,6 +15,7 @@ use crate::daemon::protocol::{ use super::session::SshConnection; use super::{ConnectionKey, SshManager}; +use crate::core::threads::Locked as _; async fn accept_retrying(listener: &TcpListener) -> Option<(TcpStream, std::net::SocketAddr)> { let mut failures = 0u32; @@ -166,7 +167,7 @@ impl RemoteForwardTable { } pub(super) fn rekey(&self, bind_host: &str, from_port: u16, to_port: u16) { - let mut map = self.inner.lock().unwrap(); + let mut map = self.inner.locked(); if let Some(target) = map.remove(&(bind_host.to_string(), from_port)) { map.insert((bind_host.to_string(), to_port), target); } @@ -177,7 +178,7 @@ impl RemoteForwardTable { connected_address: &str, connected_port: u16, ) -> Option<(String, u16)> { - let map = self.inner.lock().unwrap(); + let map = self.inner.locked(); if let Some(t) = map.get(&(connected_address.to_string(), connected_port)) { return Some(t.clone()); } @@ -276,7 +277,7 @@ impl ForwardEntry { target_host: self.target_host.clone(), target_port: self.target_port, description: self.description.clone(), - status: self.status.lock().unwrap().clone(), + status: self.status.locked().clone(), } } } @@ -381,7 +382,7 @@ impl SshForwardRegistry { } fn list_owned(&self, owner: &ForwardOwner, view_pane: u64) -> Vec { - let owners = self.owners.lock().unwrap(); + let owners = self.owners.locked(); let mut list: Vec<_> = owners .get(owner) .into_iter() @@ -399,7 +400,7 @@ impl SshForwardRegistry { forward_id: u64, ) -> Vec { let removed = { - let mut owners = self.owners.lock().unwrap(); + let mut owners = self.owners.locked(); owners.get_mut(owner).and_then(|entries| { let pos = entries.iter().position(|e| e.id == forward_id)?; Some(entries.remove(pos)) @@ -412,7 +413,7 @@ impl SshForwardRegistry { } async fn teardown_owned(&self, owner: &ForwardOwner) { - let entries = self.owners.lock().unwrap().remove(owner); + let entries = self.owners.locked().remove(owner); for entry in entries.into_iter().flatten() { Self::cancel_entry(entry).await; } @@ -479,7 +480,7 @@ impl SshForwardRegistry { } }); }; - *task_status.lock().unwrap() = loop_exit_status(exit); + *task_status.locked() = loop_exit_status(exit); }); (bound, status, ForwardCancel::Task(handle)) } @@ -535,7 +536,7 @@ impl SshForwardRegistry { } }); }; - *task_status.lock().unwrap() = loop_exit_status(exit); + *task_status.locked() = loop_exit_status(exit); }); (bound, status, ForwardCancel::Task(handle)) } @@ -625,7 +626,7 @@ impl SshForwardRegistry { }; let id = self.next_id.fetch_add(1, Ordering::Relaxed); let (bind_port, status, cancel) = self.start_local(&conn, &rule).await; - if let ForwardStatus::Error(e) = &*status.lock().unwrap() { + if let ForwardStatus::Error(e) = &*status.locked() { return Err(io::Error::other(e.clone())); } let entry = ForwardEntry { @@ -657,7 +658,7 @@ impl SshForwardRegistry { remote_host: &str, remote_port: u16, ) -> Option { - let owners = self.owners.lock().unwrap(); + let owners = self.owners.locked(); owners .get(owner)? .iter() @@ -668,7 +669,7 @@ impl SshForwardRegistry { && e.target_port == remote_port // Now that a dead loop says so, this stops handing out the // port of a forward that no longer serves anything. - && matches!(*e.status.lock().unwrap(), ForwardStatus::Listening) + && matches!(*e.status.locked(), ForwardStatus::Listening) }) .map(|e| e.bind_port) } @@ -677,7 +678,7 @@ impl SshForwardRegistry { impl SshManager { pub(crate) fn existing_connection(&self, spec: &NativeSshSpec) -> Option> { let key = ConnectionKey::from_spec(spec); - let slot = self.conns.lock().unwrap().get(&key).cloned()?; + let slot = self.conns.locked().get(&key).cloned()?; let guard = slot.try_lock().ok()?; let conn = guard.upgrade()?; conn.is_alive().then_some(conn) diff --git a/crates/tty7-core/src/daemon/ssh/mod.rs b/crates/tty7-core/src/daemon/ssh/mod.rs index 17287f82..15becbca 100644 --- a/crates/tty7-core/src/daemon/ssh/mod.rs +++ b/crates/tty7-core/src/daemon/ssh/mod.rs @@ -34,6 +34,7 @@ use crate::daemon::shell_integration::remote; use forward::RemoteForwardTable; use handler::ClientHandler; use session::drive_channel; +use crate::core::threads::Locked as _; const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(30); @@ -233,7 +234,7 @@ impl SshManager { .await .map_err(|e| format!("{e}"))?; - *conn_slot.lock().unwrap() = Arc::downgrade(&conn); + *conn_slot.locked() = Arc::downgrade(&conn); broker.status(SshPhase::Connected); @@ -252,7 +253,7 @@ impl SshManager { .await .map_err(|e| format!("{e}"))?; conn = fresh; - *conn_slot.lock().unwrap() = Arc::downgrade(&conn); + *conn_slot.locked() = Arc::downgrade(&conn); conn.open_session_channel() .await .map_err(|e| format!("open shell channel failed: {e}"))? @@ -411,11 +412,11 @@ impl SshManager { } fn evict_connection(&self, key: &ConnectionKey) { - self.conns.lock().unwrap().remove(key); + self.conns.locked().remove(key); } pub fn routes(&self) -> Vec { - let conns = self.conns.lock().unwrap(); + let conns = self.conns.locked(); let mut routes: Vec<_> = conns .iter() .map(|(key, slot)| { @@ -442,7 +443,7 @@ impl SshManager { async fn remote_bootstrap(&self, conn: &Arc) -> Option { let key = conn.key().clone(); - let cached = { self.probes.lock().unwrap().get(&key).cloned() }; + let cached = { self.probes.locked().get(&key).cloned() }; let probed = match cached { Some(hit) => hit, // Only an answer is remembered. This map is on the process-wide @@ -459,7 +460,7 @@ impl SshManager { } None => log::debug!("ssh {key:?}: no remote shell integration"), } - self.probes.lock().unwrap().insert(key, answer.clone()); + self.probes.locked().insert(key, answer.clone()); answer } None => { @@ -502,7 +503,7 @@ impl SshManager { let mut guard = match reuse { true => { let slot: ConnSlot = { - let mut map = self.conns.lock().unwrap(); + let mut map = self.conns.locked(); map.entry(key.clone()) .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(Weak::new()))) .clone() diff --git a/crates/tty7-core/src/daemon/ssh/sftp.rs b/crates/tty7-core/src/daemon/ssh/sftp.rs index dec1a2b5..799c2176 100644 --- a/crates/tty7-core/src/daemon/ssh/sftp.rs +++ b/crates/tty7-core/src/daemon/ssh/sftp.rs @@ -15,6 +15,7 @@ use crate::daemon::protocol::{ }; use super::{ConnectionKey, SshConnection, SshManager}; +use crate::core::threads::Locked as _; const CHUNK: usize = 256 * 1024; @@ -196,34 +197,34 @@ impl Job { } fn set_total(&self, total: u64) { - self.progress.lock().unwrap().set_total(total); + self.progress.locked().set_total(total); } fn set_current(&self, path: impl Into) { - self.progress.lock().unwrap().set_current(path); + self.progress.locked().set_current(path); } fn add_bytes(&self, n: u64) { - self.progress.lock().unwrap().add_bytes(n); + self.progress.locked().add_bytes(n); } fn finish(&self) { - self.progress.lock().unwrap().finish(); - *self.done_at.lock().unwrap() = Some(Instant::now()); + self.progress.locked().finish(); + *self.done_at.locked() = Some(Instant::now()); } fn fail(&self, reason: impl Into) { - self.progress.lock().unwrap().fail(reason); - *self.done_at.lock().unwrap() = Some(Instant::now()); + self.progress.locked().fail(reason); + *self.done_at.locked() = Some(Instant::now()); } fn mark_cancelled(&self) { - self.progress.lock().unwrap().cancel(); - *self.done_at.lock().unwrap() = Some(Instant::now()); + self.progress.locked().cancel(); + *self.done_at.locked() = Some(Instant::now()); } fn snapshot(&self) -> SftpJobProgress { - let p = self.progress.lock().unwrap(); + let p = self.progress.locked(); SftpJobProgress { job_id: self.id, pane_id: self.pane_id, @@ -240,7 +241,7 @@ impl Job { fn is_expired(&self) -> bool { matches!( - *self.done_at.lock().unwrap(), + *self.done_at.locked(), Some(t) if t.elapsed() > JOB_RETENTION ) } @@ -337,7 +338,7 @@ impl SftpManager { progress: Mutex::new(JobProgress::new()), done_at: Mutex::new(None), }); - self.jobs.lock().unwrap().insert(id, job.clone()); + self.jobs.locked().insert(id, job.clone()); SshManager::global().handle().spawn(async move { run_transfer(sftp, spec, job).await; @@ -347,7 +348,7 @@ impl SftpManager { pub fn cancel(&self, job_id: u64) -> Vec { let pane = { - let jobs = self.jobs.lock().unwrap(); + let jobs = self.jobs.locked(); if let Some(job) = jobs.get(&job_id) { job.cancel.store(true, Ordering::SeqCst); Some(job.pane_id) @@ -362,7 +363,7 @@ impl SftpManager { } pub(crate) fn list_jobs(&self, pane_id: u64) -> Vec { - let mut jobs = self.jobs.lock().unwrap(); + let mut jobs = self.jobs.locked(); jobs.retain(|_, job| !job.is_expired()); let mut out: Vec = jobs .values() @@ -392,7 +393,7 @@ impl SftpManager { async fn session_for(&self, conn: &Arc) -> Result, String> { let slot = { - let mut map = self.sessions.lock().unwrap(); + let mut map = self.sessions.locked(); map.entry(conn.key().clone()) .or_insert_with(|| { Arc::new(SessionSlot { @@ -417,7 +418,7 @@ impl SftpManager { } fn invalidate(&self, key: &ConnectionKey) { - self.sessions.lock().unwrap().remove(key); + self.sessions.locked().remove(key); } }