From d58dc4670883b0f2a48c84861b837f1b07e7cfdf Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Sat, 18 Jul 2026 12:24:28 +0200 Subject: [PATCH] added parking_lot mutex --- Cargo.lock | 2 + Cargo.toml | 5 + crates/pdfs-core/Cargo.toml | 1 + crates/pdfs-core/src/control.rs | 8 +- crates/pdfs-core/src/db.rs | 136 +++++++++--------- crates/pdfs-core/src/localindex.rs | 6 +- crates/pdfs-fuse/Cargo.toml | 1 + crates/pdfs-fuse/src/lib.rs | 214 ++++++++++++++--------------- crates/pdfs-fuse/src/sync.rs | 6 +- crates/pdfs-fuse/src/transfers.rs | 19 +-- crates/pdfs-gui/src/app.rs | 2 +- 11 files changed, 203 insertions(+), 197 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 85b2820..4661970 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3515,6 +3515,7 @@ dependencies = [ "directories", "ignore", "keyring", + "parking_lot", "proton-drive-rs", "proton-sdk", "rusqlite", @@ -3534,6 +3535,7 @@ dependencies = [ "image", "libc", "notify", + "parking_lot", "pdfs-core", "proton-drive-rs", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 74324c2..4a6b3ee 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,6 +22,11 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" thiserror = "2" anyhow = "1" +# Non-poisoning mutexes. A handler that panics while holding a lock must not +# wedge every later acquisition of it: `Workers` catches the unwind to keep the +# pool alive, and `std::sync::Mutex` would defeat that by poisoning the state +# the recovered worker needs. +parking_lot = "0.12" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } directories = "5" diff --git a/crates/pdfs-core/Cargo.toml b/crates/pdfs-core/Cargo.toml index e273604..4c5880d 100644 --- a/crates/pdfs-core/Cargo.toml +++ b/crates/pdfs-core/Cargo.toml @@ -20,3 +20,4 @@ ignore.workspace = true sha2.workspace = true rusqlite.workspace = true keyring.workspace = true +parking_lot.workspace = true diff --git a/crates/pdfs-core/src/control.rs b/crates/pdfs-core/src/control.rs index 349acf6..de713e7 100644 --- a/crates/pdfs-core/src/control.rs +++ b/crates/pdfs-core/src/control.rs @@ -656,10 +656,10 @@ impl PhotoKind { return PhotoKind::Photo; } } - if let Some(mt) = media_type { - if mt.starts_with("video/") { - return PhotoKind::Video; - } + if let Some(mt) = media_type + && mt.starts_with("video/") + { + return PhotoKind::Video; } PhotoKind::Photo } diff --git a/crates/pdfs-core/src/db.rs b/crates/pdfs-core/src/db.rs index 3068b27..bb77352 100644 --- a/crates/pdfs-core/src/db.rs +++ b/crates/pdfs-core/src/db.rs @@ -12,7 +12,7 @@ use std::collections::HashMap; use std::path::Path; -use std::sync::Mutex; +use parking_lot::Mutex; use proton_drive_rs::proton_sdk::ids::NodeUid; use proton_drive_rs::{Node, NodeKind}; @@ -304,7 +304,7 @@ impl Db { /// Run forward-only migrations from the stored `schema_version` up to /// [`SCHEMA_VERSION`]. Each step is wrapped in its own transaction. fn migrate(&self) -> Result<()> { - let mut conn = self.conn.lock().unwrap(); + let mut conn = self.conn.lock(); // `sync_state` is the key/value table holding `schema_version` and the // event cursor (later). Create it first so we can read the version. @@ -378,7 +378,7 @@ impl Db { /// Run a closure with the locked connection. Escape hatch for callers in /// later phases until typed query methods are added. pub fn with_conn(&self, f: impl FnOnce(&Connection) -> Result) -> Result { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); f(&conn) } @@ -402,7 +402,7 @@ impl Db { if nodes.is_empty() { return Ok(()); } - let mut conn = self.conn.lock().unwrap(); + let mut conn = self.conn.lock(); let tx = conn.transaction()?; for node in nodes { let json = serde_json::to_string(node)?; @@ -449,7 +449,7 @@ impl Db { /// not cascaded here; the daemon forgets a whole subtree node-by-node. pub fn delete_node(&self, uid: &NodeUid) -> Result<()> { let uid = uid.to_string(); - let mut conn = self.conn.lock().unwrap(); + let mut conn = self.conn.lock(); let tx = conn.transaction()?; tx.execute("DELETE FROM nodes WHERE uid = ?1", params![uid])?; tx.execute("DELETE FROM nodes_fts WHERE uid = ?1", params![uid])?; @@ -467,7 +467,7 @@ impl Db { if query.is_empty() { return Ok(Vec::new()); } - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let rows: Vec<(String, String)> = if query.chars().count() < TRIGRAM_MIN { let pat = format!("%{}%", like_escape(query)); let mut stmt = conn.prepare( @@ -506,7 +506,7 @@ impl Db { /// rehydrates its `children` map on mount even when empty; an unlisted one /// re-enumerates from the remote on next access. pub fn set_listed(&self, uid: &NodeUid, listed: bool) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "UPDATE nodes SET listed = ?2 WHERE uid = ?1", params![uid.to_string(), listed as i64], @@ -516,7 +516,7 @@ impl Db { /// Load every persisted node for cold-start hydration of the `State` maps. pub fn load_all(&self) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let mut stmt = conn.prepare("SELECT node_json, listed FROM nodes WHERE node_json IS NOT NULL")?; let rows = stmt.query_map([], |row| { @@ -537,7 +537,7 @@ impl Db { /// when the API is unreachable, so the mount can still serve the cached tree /// (offline.md Phase 1). pub fn node_by_uid(&self, uid: &str) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let json: Option = conn .query_row( "SELECT node_json FROM nodes WHERE uid = ?1 AND node_json IS NOT NULL", @@ -555,7 +555,7 @@ impl Db { /// The daemon resumes from this on restart instead of reseeding to the /// server head, so changes made while unmounted are still applied (P2). pub fn get_event_cursor(&self) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let v = conn .query_row( "SELECT value FROM sync_state WHERE key = 'event_cursor'", @@ -568,7 +568,7 @@ impl Db { /// Persist the incremental-sync cursor after a batch of events is applied. pub fn set_event_cursor(&self, cursor: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "INSERT INTO sync_state (key, value) VALUES ('event_cursor', ?1) ON CONFLICT(key) DO UPDATE SET value = excluded.value", @@ -579,7 +579,7 @@ impl Db { /// Read a `sync_state` value as a string. pub fn state_str(&self, key: &str) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let v = conn .query_row( "SELECT value FROM sync_state WHERE key = ?1", @@ -592,7 +592,7 @@ impl Db { /// Write a `sync_state` string value. pub fn set_state_str(&self, key: &str, value: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "INSERT INTO sync_state (key, value) VALUES (?1, ?2) ON CONFLICT(key) DO UPDATE SET value = excluded.value", @@ -614,7 +614,7 @@ impl Db { /// nothing left to create it. (Writes to a node that is itself still queued /// fold into the create row instead; see [`Db::attach_blob_to_create`].) pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option)> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let superseded: Option = if op_supersedes(&op.kind) { let blob: Option = conn .query_row( @@ -664,7 +664,7 @@ impl Db { blob_path: &str, meta_json: &str, ) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let existing: Option<(i64, Option)> = conn .query_row( "SELECT id, blob_path FROM pending_op WHERE uid = ?1 AND kind = ?2", @@ -693,7 +693,7 @@ impl Db { /// which case the node has a real uid and the caller must rename it there /// instead (offline.md Phase 3b). pub fn rewrite_op_target(&self, uid: &str, parent_uid: &str, name: &str) -> Result { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let n = conn.execute( "UPDATE pending_op SET parent_uid = ?2, name = ?3 WHERE uid = ?1 AND kind IN (?4, ?5)", @@ -725,7 +725,7 @@ impl Db { UNION SELECT p.uid FROM pending_op p JOIN doomed d ON p.parent_uid = d.uid )"; - let mut conn = self.conn.lock().unwrap(); + let mut conn = self.conn.lock(); let tx = conn.transaction()?; let blobs: Vec = { let mut stmt = tx.prepare(&format!( @@ -748,7 +748,7 @@ impl Db { /// parent has drained and has a real uid. Also moves the node rows whose /// parent column still names the placeholder, so listings keep resolving. pub fn remap_local_uid(&self, local: &str, real: &str) -> Result<()> { - let mut conn = self.conn.lock().unwrap(); + let mut conn = self.conn.lock(); let tx = conn.transaction()?; tx.execute( "UPDATE pending_op SET parent_uid = ?2 WHERE parent_uid = ?1", @@ -766,7 +766,7 @@ impl Db { /// Every queued op, oldest first. The drain worker replays them in this order /// so a file's writes land in the order they were made. pub fn pending_ops(&self) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let mut stmt = conn.prepare( "SELECT id, kind, uid, parent_uid, name, blob_path, meta_json, created_at, attempts, last_error, next_attempt_at @@ -799,7 +799,7 @@ impl Db { /// mount owes the server, but it carries no bytes and reporting it as an /// upload is simply untrue. pub fn pending_op_counts(&self) -> Result { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let uploads = conn.query_row( "SELECT COUNT(*) FROM pending_op WHERE kind IN (?1, ?2)", params![OP_REVISION, OP_CREATE], @@ -815,7 +815,7 @@ impl Db { /// Drop a queued op, once its upload has actually landed. pub fn delete_op(&self, id: i64) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute("DELETE FROM pending_op WHERE id = ?1", params![id])?; Ok(()) } @@ -823,7 +823,7 @@ impl Db { /// Record a failed attempt and when to next try. Leaves the row in place — /// the staged bytes are still the only copy of the user's write. pub fn record_op_failure(&self, id: i64, error: &str, next_attempt_at: i64) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "UPDATE pending_op SET attempts = attempts + 1, last_error = ?2, next_attempt_at = ?3 @@ -836,7 +836,7 @@ impl Db { /// Read a `sync_state` value as an integer (the freshness stamps of the /// photos timeline and the trash listing are kept there). pub fn state_i64(&self, key: &str) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let v: Option = conn .query_row( "SELECT value FROM sync_state WHERE key = ?1", @@ -849,7 +849,7 @@ impl Db { /// Write a `sync_state` integer value. pub fn set_state_i64(&self, key: &str, value: i64) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "INSERT INTO sync_state (key, value) VALUES (?1, ?2) ON CONFLICT(key) DO UPDATE SET value = excluded.value", @@ -860,7 +860,7 @@ impl Db { /// Drop a `sync_state` key, so whatever it stamped counts as never fetched. pub fn clear_state(&self, key: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute("DELETE FROM sync_state WHERE key = ?1", params![key])?; Ok(()) } @@ -876,7 +876,7 @@ impl Db { &self, items: &[(String, i64, Option, Option)], ) -> Result<()> { - let mut conn = self.conn.lock().unwrap(); + let mut conn = self.conn.lock(); let tx = conn.transaction()?; // `media_type` is learned-and-kept like the ratio and thumb verdict: the // timeline DTO carries only the uid and capture time, so the daemon may @@ -939,7 +939,7 @@ impl Db { kind: Option, range: Option<(i64, i64)>, ) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); // Built up so any combination of the optional filters is one indexed // query rather than a statement per case. let mut sql = @@ -991,7 +991,7 @@ impl Db { &self, kind: Option, ) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let (filter, binds): (&str, Vec) = match kind { Some(k) => (" WHERE kind = ?1", vec![k.as_i64()]), None => ("", Vec::new()), @@ -1014,7 +1014,7 @@ impl Db { /// Per-tab counts for the Photos page subtitle: `(photos, videos, raw)`. pub fn photos_counts(&self) -> Result<(usize, usize, usize)> { use crate::control::PhotoKind; - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let mut stmt = conn.prepare("SELECT kind, COUNT(*) FROM photos GROUP BY kind")?; let rows = stmt.query_map([], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?)))?; let (mut photos, mut videos, mut raw) = (0usize, 0usize, 0usize); @@ -1036,7 +1036,7 @@ impl Db { if uids.is_empty() { return Ok(Vec::new()); } - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let placeholders = vec!["?"; uids.len()].join(","); let mut stmt = conn.prepare(&format!( "SELECT uid, capture_time, name, ratio, thumb_state, kind FROM photos @@ -1063,7 +1063,7 @@ impl Db { /// ([`THUMB_HAVE`] / [`THUMB_NONE`]), and its aspect ratio if the pixels were /// seen. A `None` ratio leaves any previously learned one alone. pub fn photo_set_thumb(&self, uid: &str, state: i64, ratio: Option) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "UPDATE photos SET thumb_state = ?2, ratio = COALESCE(?3, ratio) WHERE uid = ?1", params![uid, state, ratio], @@ -1073,14 +1073,14 @@ impl Db { /// Number of photos in the persisted timeline. pub fn photos_count(&self) -> Result { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let n: i64 = conn.query_row("SELECT COUNT(*) FROM photos", [], |r| r.get(0))?; Ok(n.max(0) as usize) } /// Replace the persisted trash listing. pub fn trash_replace(&self, items: &[StoredTrash]) -> Result<()> { - let mut conn = self.conn.lock().unwrap(); + let mut conn = self.conn.lock(); let tx = conn.transaction()?; tx.execute("DELETE FROM trash", [])?; { @@ -1104,7 +1104,7 @@ impl Db { /// The persisted trash listing, folders first then by name — the order the /// Trash page shows it in. pub fn trash_list(&self) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let mut stmt = conn.prepare( "SELECT uid, name, is_dir, size, mtime FROM trash ORDER BY is_dir DESC, name COLLATE NOCASE", @@ -1129,7 +1129,7 @@ impl Db { /// The registered device for this machine, if one has been created/cached. pub fn device_get(&self) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.query_row( "SELECT uid, share_id, root_uid, name, created FROM device LIMIT 1", [], @@ -1149,7 +1149,7 @@ impl Db { /// Persist (or replace) this machine's device. The table holds a single row. pub fn device_set(&self, dev: &StoredDevice) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute("DELETE FROM device", [])?; conn.execute( "INSERT INTO device (uid, share_id, root_uid, name, created) @@ -1166,7 +1166,7 @@ impl Db { remote_uid: &str, remote_share_id: &str, ) -> Result { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "INSERT INTO sync_folder (local_path, remote_uid, remote_share_id) VALUES (?1, ?2, ?3)", @@ -1177,7 +1177,7 @@ impl Db { /// Every synced folder, oldest first. pub fn sync_folder_list(&self) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let mut stmt = conn.prepare( "SELECT id, local_path, remote_uid, remote_share_id, mode, pending_mode, state, last_sync FROM sync_folder ORDER BY id", @@ -1203,7 +1203,7 @@ impl Db { /// Look up one synced folder by id. pub fn sync_folder_get(&self, id: i64) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.query_row( "SELECT id, local_path, remote_uid, remote_share_id, mode, pending_mode, state, last_sync FROM sync_folder WHERE id = ?1", @@ -1227,7 +1227,7 @@ impl Db { /// Remove a synced folder and its per-file baseline. pub fn sync_folder_remove(&self, id: i64) -> Result { - let mut conn = self.conn.lock().unwrap(); + let mut conn = self.conn.lock(); let tx = conn.transaction()?; tx.execute("DELETE FROM sync_entry WHERE folder_id = ?1", params![id])?; let n = tx.execute("DELETE FROM sync_folder WHERE id = ?1", params![id])?; @@ -1240,7 +1240,7 @@ impl Db { /// cleared in the same write — a `pending_mode` outliving the switch it asked /// for would have the engine try to apply it again on the next pass. pub fn sync_folder_set_mode(&self, id: i64, mode: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "UPDATE sync_folder SET mode = ?2, pending_mode = NULL WHERE id = ?1", params![id, mode], @@ -1251,7 +1251,7 @@ impl Db { /// Queue (or, with `None`, withdraw) a mode the folder should move to once it /// is safe to switch. pub fn sync_folder_set_pending_mode(&self, id: i64, mode: Option<&str>) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "UPDATE sync_folder SET pending_mode = ?2 WHERE id = ?1", params![id, mode], @@ -1261,7 +1261,7 @@ impl Db { /// Update a synced folder's state and stamp `last_sync` to now. pub fn sync_folder_set_state(&self, id: i64, state: &str, last_sync: i64) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "UPDATE sync_folder SET state = ?2, last_sync = ?3 WHERE id = ?1", params![id, state, last_sync], @@ -1271,7 +1271,7 @@ impl Db { /// The whole per-file sync baseline for a folder, keyed by relative path. pub fn sync_entries(&self, folder_id: i64) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let mut stmt = conn.prepare( "SELECT rel_path, remote_uid, local_mtime, local_size, remote_rev, remote_hash FROM sync_entry WHERE folder_id = ?1", @@ -1296,7 +1296,7 @@ impl Db { /// Insert or replace one baseline row. pub fn sync_entry_upsert(&self, folder_id: i64, e: &StoredSyncEntry) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "INSERT INTO sync_entry (folder_id, rel_path, remote_uid, local_mtime, local_size, remote_rev, remote_hash) @@ -1322,7 +1322,7 @@ impl Db { /// Drop one baseline row (a path that left the sync set on both sides). pub fn sync_entry_remove(&self, folder_id: i64, rel_path: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "DELETE FROM sync_entry WHERE folder_id = ?1 AND rel_path = ?2", params![folder_id, rel_path], @@ -1337,7 +1337,7 @@ impl Db { /// table, so adding an [`ActivityKind`] variant needs no change here. pub fn activity_add(&self, entry: &ActivityEntry) -> Result<()> { let kind = serde_json::to_string(&entry.kind)?; - let mut conn = self.conn.lock().unwrap(); + let mut conn = self.conn.lock(); let tx = conn.transaction()?; tx.execute( "INSERT INTO activity (time, kind, target, detail, ok) VALUES (?1, ?2, ?3, ?4, ?5)", @@ -1357,7 +1357,7 @@ impl Db { /// whose stored `kind` no longer parses (written by an older build) are /// skipped rather than failing the whole read. pub fn activity_list(&self, limit: usize) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let mut stmt = conn.prepare( "SELECT time, kind, target, detail, ok FROM activity ORDER BY id DESC LIMIT ?1", @@ -1393,7 +1393,7 @@ impl Db { /// next reconcile mistake "locally deleted" for "must re-download". Clearing it /// leaves an empty baseline + full remote = pure download (devices.md P3). pub fn sync_entries_clear(&self, folder_id: i64) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "DELETE FROM sync_entry WHERE folder_id = ?1", params![folder_id], @@ -1405,7 +1405,7 @@ impl Db { /// listed folder's `children` map entry was trimmed mid-run. Returns `None` /// when the folder is not marked `listed` (listing unknown, must re-fetch). pub fn children_if_listed(&self, parent: &NodeUid) -> Result>> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let listed: Option = conn .query_row( "SELECT listed FROM nodes WHERE uid = ?1", @@ -1440,7 +1440,7 @@ impl Db { /// Insert or refresh a cache entry: set its on-disk `size` and bump /// `last_accessed` to `now`. Called on every store of a blob or block. pub fn cache_touch(&self, key: &str, kind: &str, size: u64, now: i64) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "INSERT INTO cache_entries (cache_key, kind, size_bytes, last_accessed) VALUES (?1, ?2, ?3, ?4) @@ -1456,7 +1456,7 @@ impl Db { /// Bump only `last_accessed` on a cache hit (LRU ordering). A missing row is /// a no-op — best effort, mirroring the old best-effort mtime touch. pub fn cache_accessed(&self, key: &str, now: i64) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "UPDATE cache_entries SET last_accessed = ?2 WHERE cache_key = ?1", params![key, now], @@ -1466,7 +1466,7 @@ impl Db { /// Drop a single cache-entry row (one evicted blob or block). pub fn cache_remove(&self, key: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "DELETE FROM cache_entries WHERE cache_key = ?1", params![key], @@ -1478,7 +1478,7 @@ impl Db { /// uid (`.b`). Called by `ContentCache::evict`, which removes the /// blob and all of a uid's cached blocks at once. pub fn cache_remove_all(&self, key: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "DELETE FROM cache_entries WHERE cache_key = ?1 OR cache_key LIKE ?2", params![key, format!("{key}.b%")], @@ -1490,7 +1490,7 @@ impl Db { /// `(cache_key, size_bytes)`. The budget enforcer sums the sizes and evicts /// from the front until the cache fits. pub fn cache_entries_by_kind(&self, kind: &str) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let mut stmt = conn.prepare( "SELECT cache_key, size_bytes FROM cache_entries WHERE kind = ?1 ORDER BY last_accessed ASC", @@ -1518,7 +1518,7 @@ impl Db { /// on the open path, so a cache of thousands of blobs used to pay thousands /// of autocommit fsyncs before the mountpoint appeared. pub fn cache_rebuild(&self, entries: &[CacheEntryInput<'_>]) -> Result<()> { - let mut conn = self.conn.lock().unwrap(); + let mut conn = self.conn.lock(); let tx = conn.transaction()?; tx.execute("DELETE FROM cache_entries", [])?; { @@ -1548,7 +1548,7 @@ impl Db { /// Record `uid` as pinned under display `path`. `recursive` keeps the whole /// subtree of a folder. Idempotent — re-pinning refreshes the path/flag. pub fn pin_add(&self, uid: &str, path: &str, recursive: bool) -> Result<()> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); conn.execute( "INSERT INTO pins (uid, path, recursive) VALUES (?1, ?2, ?3) ON CONFLICT(uid) DO UPDATE SET @@ -1560,14 +1560,14 @@ impl Db { /// Drop `uid`'s pin row. Returns whether a row existed. pub fn pin_remove(&self, uid: &str) -> Result { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let n = conn.execute("DELETE FROM pins WHERE uid = ?1", params![uid])?; Ok(n > 0) } /// Every directly-pinned entry `(uid, path, recursive)`, ordered by uid. pub fn pin_list(&self) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let mut stmt = conn.prepare("SELECT uid, path, recursive FROM pins ORDER BY uid")?; let rows = stmt.query_map([], |r| { Ok(( @@ -1588,7 +1588,7 @@ impl Db { /// via a CTE; a direct pin is honoured even when the node has no `nodes` row /// yet (e.g. a CLI that never hydrates the node cache). pub fn is_pinned(&self, uid: &str) -> Result { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let direct: Option = conn .query_row("SELECT 1 FROM pins WHERE uid = ?1", params![uid], |r| { r.get(0) @@ -1618,7 +1618,7 @@ impl Db { /// every descendant of a recursively-pinned folder. Used to build the /// eviction-exempt set (the budget enforcer hashes these into cache keys). pub fn pinned_uids(&self) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let mut stmt = conn.prepare( "WITH RECURSIVE sub(uid) AS ( SELECT uid FROM pins WHERE recursive = 1 @@ -1640,7 +1640,7 @@ impl Db { /// Every descendant uid of `folder` (all depths), via a `parent_uid` CTE. /// Used to evict a recursively-pinned subtree's cached blobs on unpin. pub fn descendants(&self, folder: &str) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let mut stmt = conn.prepare( "WITH RECURSIVE sub(uid) AS ( SELECT uid FROM nodes WHERE parent_uid = ?1 @@ -1663,7 +1663,7 @@ impl Db { /// [`local_finish_scan`](Self::local_finish_scan); rows still carrying an /// older generation are pruned there as "no longer on disk". pub fn local_begin_scan(&self) -> Result { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let current: i64 = conn .query_row( "SELECT value FROM sync_state WHERE key = 'local_scan_gen'", @@ -1686,7 +1686,7 @@ impl Db { /// FTS index is *not* touched here — it is rebuilt once in /// [`local_finish_scan`](Self::local_finish_scan). pub fn local_upsert_batch(&self, generation: i64, entries: &[LocalEntry]) -> Result<()> { - let mut conn = self.conn.lock().unwrap(); + let mut conn = self.conn.lock(); let tx = conn.transaction()?; { let mut stmt = tx.prepare_cached( @@ -1718,7 +1718,7 @@ impl Db { /// (those paths are gone from disk), rebuild the FTS index over what remains, /// and stamp the completion time. Returns the number of indexed entries. pub fn local_finish_scan(&self, generation: i64, finished_at: i64) -> Result { - let mut conn = self.conn.lock().unwrap(); + let mut conn = self.conn.lock(); let tx = conn.transaction()?; tx.execute( "DELETE FROM local_files WHERE scan_gen != ?1", @@ -1739,7 +1739,7 @@ impl Db { /// has never been built. The daemon uses this to decide whether a fresh mount /// needs an immediate rescan or can serve the existing index. pub fn local_indexed_at(&self) -> Result> { - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); Ok(conn .query_row( "SELECT value FROM sync_state WHERE key = 'local_indexed_at'", @@ -1759,7 +1759,7 @@ impl Db { if query.is_empty() { return Ok(Vec::new()); } - let conn = self.conn.lock().unwrap(); + let conn = self.conn.lock(); let mut stmt; let rows = if query.chars().count() < TRIGRAM_MIN { let pat = format!("%{}%", like_escape(query)); @@ -2131,7 +2131,7 @@ mod tests { let path = std::env::temp_dir().join(format!("pdfs-db-wal-{}.db", std::process::id())); let _ = std::fs::remove_file(&path); let db = Db::open(&path).unwrap(); - let conn = db.conn.lock().unwrap(); + let conn = db.conn.lock(); let mode: String = conn .query_row("PRAGMA journal_mode", [], |r| r.get(0)) diff --git a/crates/pdfs-core/src/localindex.rs b/crates/pdfs-core/src/localindex.rs index 229adc2..13324b1 100644 --- a/crates/pdfs-core/src/localindex.rs +++ b/crates/pdfs-core/src/localindex.rs @@ -93,7 +93,7 @@ pub fn scan( !matches!(entry.file_name().to_str(), Some(name) if SKIP_DIRS.contains(&name)) }); - let state = std::sync::Mutex::new(SinkState { + let state = parking_lot::Mutex::new(SinkState { buf: Vec::with_capacity(BATCH), total: 0, sink, @@ -112,7 +112,7 @@ pub fn scan( return WalkState::Continue; }; - let mut state = state.lock().unwrap(); + let mut state = state.lock(); state.push(local); if state.total >= MAX_ENTRIES { return WalkState::Quit; @@ -121,7 +121,7 @@ pub fn scan( }) }); - let mut state = state.into_inner().unwrap(); + let mut state = state.into_inner(); state.flush(); state.total } diff --git a/crates/pdfs-fuse/Cargo.toml b/crates/pdfs-fuse/Cargo.toml index de34a00..c911eee 100644 --- a/crates/pdfs-fuse/Cargo.toml +++ b/crates/pdfs-fuse/Cargo.toml @@ -17,3 +17,4 @@ libc.workspace = true serde_json.workspace = true image.workspace = true notify = "8" +parking_lot.workspace = true diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 7f2dba6..c405274 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -39,7 +39,8 @@ use std::os::unix::fs::FileExt; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::{Component, Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::{Arc, Condvar, Mutex}; +use parking_lot::{Condvar, Mutex}; +use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use fuser::ReplyXattr; @@ -594,12 +595,18 @@ impl Workers { .spawn(move || { loop { // Held only to take the next job, never across it. - let job = rx.lock().unwrap().recv(); + let job = rx.lock().recv(); match job { // A panicking handler must not cost the pool a // worker for the rest of the run. The dropped // `Reply` answers EIO on its own, so the caller of // the failed op is told; the next job is unaffected. + // + // This only holds because the shared state is behind + // `parking_lot` mutexes: a `std` mutex held at the + // point of the panic would come back poisoned, and + // the worker we just rescued would die on its next + // acquisition — as would every other thread. Ok(job) => { let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job)); } @@ -852,13 +859,13 @@ impl Core { // us, so the pending map is populated; snapshot it here rather than hold // both locks at once. let pending_sizes: HashMap = { - let pending = self.pending.lock().unwrap(); + let pending = self.pending.lock(); pending .iter() .map(|(uid, pr)| (uid.clone(), pr.meta.len)) .collect() }; - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); // Pass 1: assign a stable inode to every uid (root is already mapped). for sn in &stored { @@ -936,7 +943,7 @@ impl Core { return; } }; - let mut map = self.pending.lock().unwrap(); + let mut map = self.pending.lock(); let mut restored = 0usize; for op in ops { let Some(uid) = parse_node_uid(&op.uid) else { @@ -1009,7 +1016,7 @@ impl Core { match self.rt.block_on(self.client.get_my_files_folder()) { Ok(root) => { { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); if let Some(e) = st.entries.get_mut(&ROOT_INO) { e.node = root.clone(); } @@ -1037,7 +1044,7 @@ impl Core { /// Lets a handler decide between answering inline and handing off to a /// worker, at the cost of one uncontended map lookup. fn children_cached(&self, ino: u64) -> bool { - self.state.lock().unwrap().children.contains_key(&ino) + self.state.lock().children.contains_key(&ino) } /// Enumerate `ino`'s children from the remote and cache them. No-op if the @@ -1045,7 +1052,7 @@ impl Core { /// held so concurrent metadata reads aren't blocked behind a fetch. fn ensure_children(&self, ino: u64) -> Result<(), Errno> { let folder_uid = { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); if st.children.contains_key(&ino) { return Ok(()); } @@ -1060,7 +1067,7 @@ impl Core { // was trimmed from the hot cache mid-run. match self.db.children_if_listed(&folder_uid) { Ok(Some(nodes)) => { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); if st.children.contains_key(&ino) { return Ok(()); } @@ -1093,7 +1100,7 @@ impl Core { Errno::EIO })?; - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); // Lost the race? Another thread already populated it. if st.children.contains_key(&ino) { return Ok(()); @@ -1118,7 +1125,7 @@ impl Core { /// the parent's listing is cached first. fn lookup_child(&self, parent: u64, name: &str) -> Result<(u64, NodeUid), Errno> { self.ensure_children(parent)?; - let st = self.state.lock().unwrap(); + let st = self.state.lock(); st.children .get(&parent) .and_then(|kids| { @@ -1138,7 +1145,7 @@ impl Core { fn resolve_path(&self, rel: &Path) -> Result<(u64, NodeUid), Errno> { let mut ino = ROOT_INO; let mut uid = { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); st.entries .get(&ROOT_INO) .map(|e| e.uid.clone()) @@ -1217,7 +1224,7 @@ impl Core { ), } let act = { - let mut readers = self.readers.lock().unwrap(); + let mut readers = self.readers.lock(); match readers.get_mut(uid) { Some(ReaderSlot::Ready(entry)) if entry.mtime == mtime && entry.size == fsize => @@ -1275,7 +1282,7 @@ impl Core { } }; { - let mut readers = self.readers.lock().unwrap(); + let mut readers = self.readers.lock(); // Publish only into the slot we claimed. If it is gone // or belongs to a later attempt, an eviction or a newer // revision overtook us: our reader still answers the @@ -1341,7 +1348,7 @@ impl Core { /// to cache what it opened, which is what we want — the eviction says that /// revision is no longer the truth. fn evict_reader(&self, uid: &NodeUid) { - self.readers.lock().unwrap().remove(uid); + self.readers.lock().remove(uid); } /// Serve bytes `[offset, offset + len)` of `uid`'s active revision, hitting @@ -1361,7 +1368,7 @@ impl Core { // A queued write has not reached the remote yet, so the remote's current // revision is stale and the staged blob is the truth. Serve from it until // the drain worker lands the upload (offline.md Phase 3). - if let Some(pending) = self.pending.lock().unwrap().get(uid).cloned() { + if let Some(pending) = self.pending.lock().get(uid).cloned() { return self.read_pending(&pending, offset, len); } // A node created offline and never written has no blob and no remote: it @@ -1439,7 +1446,6 @@ impl Core { .filter(|&bidx| { self.stream_ring .lock() - .unwrap() .get(uid, mtime, fsize, bidx) .is_none() }) @@ -1467,7 +1473,7 @@ impl Core { while let Some(joined) = set.join_next().await { let Ok((bidx, read)) = joined else { continue }; match read { - Ok(bytes) => core.stream_ring.lock().unwrap().insert( + Ok(bytes) => core.stream_ring.lock().insert( &uid, mtime, fsize, @@ -1523,7 +1529,6 @@ impl Core { } else { self.stream_ring .lock() - .unwrap() .get(uid, mtime, fsize, bidx) }; match hit { @@ -1576,7 +1581,6 @@ impl Core { } else { self.stream_ring .lock() - .unwrap() .insert(uid, mtime, fsize, bidx, bytes.clone()); } blocks[(bidx - first) as usize] = Some(bytes); @@ -1769,7 +1773,7 @@ impl Core { if is_local_uid(uid) { return None; } - match self.pending.lock().unwrap().get(uid) { + match self.pending.lock().get(uid) { Some(p) => p.meta.based_on, None => Some(Baseline { mtime: base_mtime, @@ -1800,7 +1804,7 @@ impl Core { // refuse the write and keep the bytes recoverable (Phase 2 behaviour). // Only reachable offline, editing in place a file whose previous edit // has not drained and whose base is not cached. - if !meta.complete && self.pending.lock().unwrap().contains_key(uid) { + if !meta.complete && self.pending.lock().contains_key(uid) { self.stage_orphaned_write(uid, ino, src, &meta); return Err(Errno::EIO); } @@ -1856,13 +1860,12 @@ impl Core { let len = meta.len; self.pending .lock() - .unwrap() .insert(uid.clone(), PendingRevision { path, meta }); // Reflect the write in the tree straight away: `ls` must show the new // size and mtime even though the remote still holds the old revision. let now = now_secs(); { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); st.record_pending_write(ino, len, now); } // Cached blobs and open readers describe the superseded revision. Reads @@ -1892,7 +1895,7 @@ impl Core { /// base, so it is the drain that has to fetch it. fn queue_truncate(&self, ino: u64, size: u64) -> Result<(), Errno> { let (uid, base_mtime, base_size) = { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); match st.entries.get(&ino) { Some(e) if e.node.is_file() => { (e.uid.clone(), e.node.modification_time, node_size(&e.node)) @@ -2017,7 +2020,6 @@ impl Core { })?; self.state .lock() - .unwrap() .rename_in_place(ino, new_parent_ino, new_parent_uid, new_name); self.wake_drain(); debug!(%uid, %new_parent_uid, new_name, "renamed offline; queued"); @@ -2052,7 +2054,7 @@ impl Core { error!(%uid, error = %e, "queueing trash failed"); Errno::EIO })?; - self.state.lock().unwrap().forget(uid); + self.state.lock().forget(uid); self.cache.evict(uid); self.evict_reader(uid); self.wake_drain(); @@ -2070,13 +2072,13 @@ impl Core { } Err(e) => error!(%uid, error = %e, "dropping queued ops failed"), } - self.pending.lock().unwrap().remove(uid); + self.pending.lock().remove(uid); } /// Nudge the drain worker to re-examine the queue now. fn wake_drain(&self) { let (lock, cv) = &*self.drain_wake; - *lock.lock().unwrap() = true; + *lock.lock() = true; cv.notify_all(); } @@ -2091,7 +2093,7 @@ impl Core { "cannot queue write over an undrained edit; bytes kept in staging" ); let name = { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); st.entries .get(&ino) .map(|e| e.node.name.clone()) @@ -2163,10 +2165,9 @@ impl Core { /// the shortest outstanding backoff elapsing. fn wait_for_drain_work(&self) { let (lock, cv) = &*self.drain_wake; - let mut woken = lock.lock().unwrap(); + let mut woken = lock.lock(); if !*woken { - let (guard, _) = cv.wait_timeout(woken, DRAIN_IDLE_POLL).unwrap(); - woken = guard; + cv.wait_for(&mut woken, DRAIN_IDLE_POLL); } *woken = false; } @@ -2309,7 +2310,7 @@ impl Core { /// it away from the one the user asked for. Best effort: the event sync /// would correct the tree anyway, but not before the user has looked at it. fn adopt_drained_name(&self, uid: &NodeUid, name: &str) { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); let Some(&ino) = st.by_uid.get(uid) else { return; }; @@ -2389,7 +2390,7 @@ impl Core { if let Some(blob) = op.blob_path.as_deref() { self.cache.discard_staged(Path::new(blob)); } - self.pending.lock().unwrap().remove(&local); + self.pending.lock().remove(&local); self.log_activity(ActivityKind::Upload, &name, "created", true); info!(%local, %real, name, kind = %op.kind, "pending create landed"); Ok(()) @@ -2472,7 +2473,7 @@ impl Core { self.db .remap_local_uid(&local.to_string(), &real.to_string())?; - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); if let Some(ino) = st.by_uid.remove(local) { st.by_uid.insert(real.clone(), ino); if let Some(e) = st.entries.get_mut(&ino) { @@ -2632,12 +2633,12 @@ impl Core { // Dropping the pending entry hands the node back to the remote's truth: // reads stop coming from the staged blob, and the event sync stops // skipping it as "ahead of the server" (offline.md Phase 3a). - self.pending.lock().unwrap().remove(uid); + self.pending.lock().remove(uid); self.cache.discard_staged(blob); self.cache.evict(uid); self.evict_reader(uid); // The conflict copy is a node the tree has never seen. - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); if let Some(&ino) = st.by_uid.get(&parent) { st.invalidate_listing(ino); } @@ -2670,7 +2671,7 @@ impl Core { error!(%uid, name, reason, staged = %blob.display(), "cannot place a conflicted write; bytes kept in staging"); self.db.delete_op(op.id)?; - self.pending.lock().unwrap().remove(uid); + self.pending.lock().remove(uid); self.cache.evict(uid); self.evict_reader(uid); self.log_activity( @@ -2695,7 +2696,7 @@ impl Core { /// A node's parent uid and name, as the tree currently has them. fn node_place(&self, uid: &NodeUid) -> Option<(NodeUid, String)> { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); let ino = st.by_uid.get(uid)?; let entry = st.entries.get(ino)?; let parent = entry.node.parent_uid.clone()?; @@ -2709,7 +2710,7 @@ impl Core { /// root yet, which is not where the drain runs; the drain must not panic /// over it regardless, since that would stop the queue for good. fn root_uid(&self) -> Option { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); st.entries.get(&ROOT_INO).map(|e| e.uid.clone()) } @@ -2752,7 +2753,7 @@ impl Core { } let name = { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); st.by_uid .get(&uid) .and_then(|ino| st.entries.get(ino)) @@ -2776,7 +2777,7 @@ impl Core { // an orphaned file (harmless), whereas the reverse would leave a queued // op pointing at nothing. self.db.delete_op(op.id)?; - self.pending.lock().unwrap().remove(&uid); + self.pending.lock().remove(&uid); // The staged blob now matches the sealed revision, so a pinned file keeps // it as its cached content rather than re-downloading what we just sent. @@ -2812,7 +2813,7 @@ impl Core { // A write queued while this one was on the wire is ahead of the remote // again, and its optimistic size/mtime must win — the same reason // `apply_event` refuses to overwrite a node with a queued write. - if self.pending.lock().unwrap().contains_key(uid) { + if self.pending.lock().contains_key(uid) { return; } let node = match self.fetch_node_remote(uid) { @@ -2827,7 +2828,7 @@ impl Core { return; } }; - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); let Some(parent) = st .by_uid .get(uid) @@ -2888,7 +2889,7 @@ impl Core { .resolve_path(rel) .map_err(|e| format!("resolve path: {e:?}"))?; let (name, is_folder, mtime, size) = { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); let e = st.entries.get(&ino).ok_or("node vanished")?; ( e.node.name.clone(), @@ -2928,7 +2929,7 @@ impl Core { while let Some(dir) = stack.pop() { self.ensure_children(dir) .map_err(|e| format!("enumerate: {e:?}"))?; - let st = self.state.lock().unwrap(); + let st = self.state.lock(); if let Some(kids) = st.children.get(&dir) { for &k in kids { if let Some(e) = st.entries.get(&k) { @@ -2969,7 +2970,7 @@ impl Core { /// `Ok(None)` when the node is not a file or has no thumbnail of that type. fn thumbnail(&self, ino: u64, ttype: ThumbnailType) -> Result>, Errno> { let (uid, mtime) = { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); match st.entries.get(&ino) { Some(e) if e.node.is_file() => (e.uid.clone(), e.node.modification_time), Some(_) => return Ok(None), @@ -3001,7 +3002,7 @@ impl Core { .resolve_path(rel) .map_err(|e| format!("resolve path: {e:?}"))?; let (name, is_folder) = { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); st.entries .get(&ino) .map(|e| (e.node.name.clone(), e.node.is_folder())) @@ -3044,7 +3045,7 @@ impl Core { // Snapshot the listing, then drop the lock before touching the on-disk // pin registry so a slow disk read doesn't block FUSE metadata ops. let rows: Vec<(String, bool, u64, i64, NodeUid)> = { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); st.children .get(&ino) .map(|kids| { @@ -3260,7 +3261,7 @@ impl Core { } } - let pending = self.thumb_gen.lock().unwrap(); + let pending = self.thumb_gen.lock(); uids.iter() .map(|uid| PhotoThumb { uid: uid.to_string(), @@ -3279,7 +3280,7 @@ impl Core { /// the reply this call is about to send already reports them as pending. fn spawn_generate_thumbs(&self, uids: Vec, tags: &HashMap) { let fresh: Vec = { - let mut inflight = self.thumb_gen.lock().unwrap(); + let mut inflight = self.thumb_gen.lock(); uids.into_iter() .filter(|uid| inflight.insert(uid.clone())) .collect() @@ -3294,7 +3295,7 @@ impl Core { // blocking pool rather than on an async worker. self.rt.spawn_blocking(move || { core.generate_thumbs(&fresh, &tags); - let mut inflight = core.thumb_gen.lock().unwrap(); + let mut inflight = core.thumb_gen.lock(); for uid in &fresh { inflight.remove(uid); } @@ -3490,7 +3491,7 @@ impl Core { .resolve_path(rel) .map_err(|e| format!("resolve path: {e:?}"))?; let (name, mtime, size) = { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); let e = st.entries.get(&ino).ok_or("node vanished")?; if !e.node.is_file() { return Err("not a regular file".into()); @@ -3520,7 +3521,7 @@ impl Core { fn invalidate_parent_listing(&self, rel: &Path) { let parent = rel.parent().unwrap_or_else(|| Path::new("")); if let Ok((pino, _)) = self.resolve_path(parent) { - self.state.lock().unwrap().invalidate_listing(pino); + self.state.lock().invalidate_listing(pino); } } @@ -3538,7 +3539,7 @@ impl Core { self.rt .block_on(self.client.rename_node(&uid, new_name, None)) .map_err(|e| format!("rename: {e}"))?; - self.state.lock().unwrap().forget(&uid); + self.state.lock().forget(&uid); self.invalidate_parent_listing(rel); Ok(new_name.to_string()) } @@ -3559,12 +3560,11 @@ impl Core { let name = self .state .lock() - .unwrap() .forget(&uid) .map(|(_, n)| n) .unwrap_or_default(); self.invalidate_parent_listing(rel); - self.state.lock().unwrap().invalidate_listing(pino); + self.state.lock().invalidate_listing(pino); Ok(name) } @@ -3580,7 +3580,6 @@ impl Core { let name = self .state .lock() - .unwrap() .forget(&uid) .map(|(_, n)| n) .unwrap_or_default(); @@ -3703,7 +3702,7 @@ impl Core { self.db .set_listed(&uid, false) .map_err(|e| format!("db: {e:?}"))?; - self.state.lock().unwrap().children.remove(&ino); + self.state.lock().children.remove(&ino); Ok(()) } @@ -3735,7 +3734,7 @@ impl Core { .block_on(self.client.restore_nodes(&parsed)) .map_err(|e| format!("restore: {e}"))?; { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); for parent in parents { if let Some(&ino) = st.by_uid.get(&parent) { st.invalidate_listing(ino); @@ -3777,7 +3776,7 @@ impl Core { /// Forget every trace of nodes that no longer exist anywhere: their inode and /// DB row, and their cached content. fn drop_local(&self, uids: &[NodeUid]) { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); for uid in uids { st.forget(uid); } @@ -3809,7 +3808,7 @@ impl Core { let node = self .fetch_node(&new_uid) .map_err(|e| format!("fetch node: {e:?}"))?; - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); let ino = st.intern(pino, node); if let Some(kids) = st.children.get_mut(&pino) && !kids.contains(&ino) @@ -3851,7 +3850,7 @@ impl Core { let node = self .fetch_node(&new_uid) .map_err(|e| format!("fetch node: {e:?}"))?; - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); let ino = st.intern(pino, node); if let Some(kids) = st.children.get_mut(&pino) && !kids.contains(&ino) @@ -3928,7 +3927,7 @@ impl Core { stats.bytes += size; match self.fetch_node(&uid) { Ok(node) => { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); let ino = st.intern(parent_ino, node); if let Some(kids) = st.children.get_mut(&parent_ino) && !kids.contains(&ino) @@ -3971,7 +3970,7 @@ impl Core { self.ensure_children(pino) .map_err(|e| format!("enumerate: {e:?}"))?; { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); if let Some(kids) = st.children.get(&pino) { for &ino in kids { if let Some(e) = st.entries.get(&ino) @@ -3993,7 +3992,7 @@ impl Core { let node = self .fetch_node(&new_uid) .map_err(|e| format!("fetch node: {e:?}"))?; - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); let ino = st.intern(pino, node); if let Some(kids) = st.children.get_mut(&pino) && !kids.contains(&ino) @@ -4268,7 +4267,7 @@ impl Core { /// List this device's synced folders for the front-ends, each carrying the /// live progress of its pass when one is running. fn list_sync_folders(&self) -> Result, String> { - let progress = self.sync_progress.lock().unwrap(); + let progress = self.sync_progress.lock(); Ok(self .db .sync_folder_list() @@ -4295,7 +4294,6 @@ impl Core { let mut passes: Vec<(i64, SyncProgress)> = self .sync_progress .lock() - .unwrap() .iter() .map(|(id, p)| (*id, p.clone())) .collect(); @@ -4349,7 +4347,6 @@ impl Core { pub(crate) fn sync_lock(&self, id: i64) -> Arc> { self.sync_locks .lock() - .unwrap() .entry(id) .or_default() .clone() @@ -4368,7 +4365,7 @@ impl Core { // folder the daemon no longer tracks, and nothing would ever unmount it. // Tear it down first — including before trashing the remote tree it // serves, which would otherwise leave it answering for deleted nodes. - if let Some(session) = self.mounts.lock().unwrap().remove(&id) { + if let Some(session) = self.mounts.lock().remove(&id) { if let Err(e) = session.umount_and_join() { warn!(id, error = %e, "unmount on-demand folder failed"); } else { @@ -4537,7 +4534,7 @@ impl Core { // the `state` column is still `idle` in the window between `add_sync_folder` // inserting the row and the engine picking it up. let lock = self.sync_lock(id); - let Ok(_guard) = lock.try_lock() else { + let Some(_guard) = lock.try_lock() else { return Err(SwitchBlocked::NotNow); }; // Re-read under the lock: a pass that finished while we waited may have @@ -4580,7 +4577,7 @@ impl Core { let session = self .spawn_ondemand_mount(&local, root) .map_err(SwitchBlocked::Failed)?; - self.mounts.lock().unwrap().insert(id, session); + self.mounts.lock().insert(id, session); // Persist the mode only now that the mount is actually up. Writing it // first would strand the folder on any failure below: the engine skips // non-mirror folders, so an `ondemand` row with no mount is a folder @@ -4597,7 +4594,7 @@ impl Core { } _ => { // ondemand→mirror: tear down the secondary mount first. - if let Some(session) = self.mounts.lock().unwrap().remove(&id) + if let Some(session) = self.mounts.lock().remove(&id) && let Err(e) = session.umount_and_join() { warn!(id, error = %e, "unmount on-demand folder failed"); @@ -4718,7 +4715,7 @@ impl Core { }; match self.spawn_ondemand_mount(&local, root) { Ok(session) => { - self.mounts.lock().unwrap().insert(folder.id, session); + self.mounts.lock().insert(folder.id, session); let _ = self.db.sync_folder_set_state(folder.id, "idle", now_secs()); info!(id = folder.id, path = %local.display(), "remounted on-demand folder"); } @@ -5111,7 +5108,7 @@ impl Core { /// been seen through the mount (e.g. shared but not browsed to this session) — /// the caller then leaves the path empty. fn rel_path_for_uid(&self, uid: &NodeUid) -> Option { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); let mut ino = *st.by_uid.get(uid)?; let mut parts = Vec::new(); while ino != ROOT_INO { @@ -5163,7 +5160,7 @@ impl Core { /// Start tracking a reconcile pass over `folder_id`, in [`SyncPhase::Scanning`]. pub(crate) fn progress_begin(&self, folder_id: i64) { - self.sync_progress.lock().unwrap().insert( + self.sync_progress.lock().insert( folder_id, SyncProgress { phase: SyncPhase::Scanning, @@ -5176,7 +5173,7 @@ impl Core { /// Apply `f` to a folder's live progress, if a pass is running for it. fn progress_update(&self, folder_id: i64, f: impl FnOnce(&mut SyncProgress)) { - if let Some(p) = self.sync_progress.lock().unwrap().get_mut(&folder_id) { + if let Some(p) = self.sync_progress.lock().get_mut(&folder_id) { f(p); } } @@ -5230,7 +5227,7 @@ impl Core { /// Stop tracking a pass — no progress is reported for the folder until the /// next [`progress_begin`](Self::progress_begin). pub(crate) fn progress_end(&self, folder_id: i64) { - self.sync_progress.lock().unwrap().remove(&folder_id); + self.sync_progress.lock().remove(&folder_id); } } @@ -5298,7 +5295,7 @@ impl ProtonFs { /// Build the filesystem rooted at `root` (the user's My Files folder). fn new(core: Core, root: Node) -> Self { { - let mut st = core.state.lock().unwrap(); + let mut st = core.state.lock(); if let Err(e) = st.db.upsert_node(&root) { warn!(uid = %root.uid, error = %e, "db upsert root failed"); } @@ -5323,7 +5320,7 @@ impl ProtonFs { reply.error(e); return; } - let st = self.core.state.lock().unwrap(); + let st = self.core.state.lock(); let hit = st.children.get(&parent).and_then(|kids| { kids.iter().copied().find_map(|ino| { st.entries @@ -5347,7 +5344,7 @@ impl ProtonFs { reply.error(e); return; } - let st = self.core.state.lock().unwrap(); + let st = self.core.state.lock(); let parent = st.entries.get(&ino).map_or(ROOT_INO, |e| e.parent); // "." and ".." occupy offsets 0 and 1; real children follow. @@ -5421,7 +5418,7 @@ impl ProtonFs { // offline, which the remote path below cannot (offline.md Phase 3b). if is_local_uid(&uid) { self.core.discard_queued_ops(&uid); - self.core.state.lock().unwrap().forget(&uid); + self.core.state.lock().forget(&uid); debug!(%uid, name, "deleted a node that had not been created remotely yet"); reply.ok(); return; @@ -5446,7 +5443,7 @@ impl ProtonFs { return; } self.core.discard_queued_ops(&uid); - self.core.state.lock().unwrap().forget(&uid); + self.core.state.lock().forget(&uid); self.core.cache.evict(&uid); self.core.evict_reader(&uid); self.core.invalidate_trash(); @@ -5796,7 +5793,7 @@ impl Filesystem for ProtonFs { } fn getattr(&self, _req: &Request, ino: INodeNo, _fh: Option, reply: ReplyAttr) { - let st = self.core.state.lock().unwrap(); + let st = self.core.state.lock(); match st.entries.get(&ino.0) { Some(e) => { let attr = self.attr(ino.0, &e.node); @@ -5829,7 +5826,7 @@ impl Filesystem for ProtonFs { fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) { let (uid, base_mtime, base_size) = { - let st = self.core.state.lock().unwrap(); + let st = self.core.state.lock(); match st.entries.get(&ino.0) { Some(e) if e.node.is_file() => { (e.uid.clone(), e.node.modification_time, node_size(&e.node)) @@ -5860,7 +5857,7 @@ impl Filesystem for ProtonFs { return; } }; - let mut st = self.core.state.lock().unwrap(); + let mut st = self.core.state.lock(); let fh = st.next_fh; st.next_fh += 1; st.handles.insert( @@ -5902,7 +5899,7 @@ impl Filesystem for ProtonFs { // has always had between a write and a read of the same handle; a mostly // local read is not worth reasoning about that for. let handle = { - let st = self.core.state.lock().unwrap(); + let st = self.core.state.lock(); st.handles.values().find(|h| h.ino == ino.0).map(|h| { ( h.file.clone(), @@ -5931,7 +5928,7 @@ impl Filesystem for ProtonFs { return; } let (uid, mtime, fsize, is_video) = { - let st = self.core.state.lock().unwrap(); + let st = self.core.state.lock(); match st.entries.get(&ino.0) { Some(e) if e.node.is_file() => { let media_type = match &e.node.kind { @@ -5992,7 +5989,7 @@ impl Filesystem for ProtonFs { } let name = name.to_string_lossy().into_owned(); let parent_uid = { - let st = self.core.state.lock().unwrap(); + let st = self.core.state.lock(); match st.entries.get(&parent) { Some(e) => e.uid.clone(), None => { @@ -6057,7 +6054,7 @@ impl Filesystem for ProtonFs { return; } }; - let mut st = self.core.state.lock().unwrap(); + let mut st = self.core.state.lock(); let ino = st.intern(parent, node); if let Some(kids) = st.children.get_mut(&parent) && !kids.contains(&ino) @@ -6107,7 +6104,7 @@ impl Filesystem for ProtonFs { // Stage the bytes straight into the scratch file (no base download): only // the untouched remainder is pulled from the remote, and only at commit. let file = { - let st = self.core.state.lock().unwrap(); + let st = self.core.state.lock(); match st.handles.get(&fh) { Some(h) => h.file.clone(), None => { @@ -6122,7 +6119,7 @@ impl Filesystem for ProtonFs { return; } let new_len = { - let mut st = self.core.state.lock().unwrap(); + let mut st = self.core.state.lock(); let Some(h) = st.handles.get_mut(&fh) else { reply.error(Errno::EBADF); return; @@ -6170,7 +6167,7 @@ impl Filesystem for ProtonFs { if let Some(size) = size { let fh = fh.map(|f| f.0).filter(|&f| f != 0); let handled = { - let mut st = self.core.state.lock().unwrap(); + let mut st = self.core.state.lock(); // The kernel does not put `O_TRUNC` in the `open` flags unless // the mount enables `atomic_o_trunc`; it opens, then truncates // with a *separate* `setattr` carrying no file handle. So an fh @@ -6218,10 +6215,10 @@ impl Filesystem for ProtonFs { return; } } else { - self.core.state.lock().unwrap().set_size(ino.0, size); + self.core.state.lock().set_size(ino.0, size); } } - let st = self.core.state.lock().unwrap(); + let st = self.core.state.lock(); match st.entries.get(&ino.0) { Some(e) => { let attr = self.attr(ino.0, &e.node); @@ -6261,7 +6258,6 @@ impl Filesystem for ProtonFs { .core .state .lock() - .unwrap() .handles .get(&fh.0) .map(|h| h.file.clone()); @@ -6287,7 +6283,7 @@ impl Filesystem for ProtonFs { _flush: bool, reply: ReplyEmpty, ) { - let handle = self.core.state.lock().unwrap().handles.remove(&fh.0); + let handle = self.core.state.lock().handles.remove(&fh.0); // Hand the bytes to the queue rather than uploading them here: the // scratch file is the only copy of what was just written, and blocking // the caller on the network is what made a copy into the mount run at @@ -6325,7 +6321,7 @@ impl Filesystem for ProtonFs { } let name = name.to_string_lossy().into_owned(); let parent_uid = { - let st = self.core.state.lock().unwrap(); + let st = self.core.state.lock(); match st.entries.get(&parent) { Some(e) => e.uid.clone(), None => { @@ -6371,7 +6367,7 @@ impl Filesystem for ProtonFs { } } }; - let mut st = self.core.state.lock().unwrap(); + let mut st = self.core.state.lock(); let local = is_local_uid(&node.uid); let uid = node.uid.clone(); let ino = st.intern(parent, node); @@ -6424,7 +6420,7 @@ impl Filesystem for ProtonFs { return; } let new_parent_uid = { - let st = self.core.state.lock().unwrap(); + let st = self.core.state.lock(); match st.entries.get(&newparent) { Some(e) => e.uid.clone(), None => { @@ -6444,7 +6440,7 @@ impl Filesystem for ProtonFs { &newname, ) { Ok(true) => { - self.core.state.lock().unwrap().rename_in_place( + self.core.state.lock().rename_in_place( ino, newparent, &new_parent_uid, @@ -6504,7 +6500,7 @@ impl Filesystem for ProtonFs { } // Forget the node so it re-interns under its new parent, and drop the // destination listing so it re-enumerates on next access. - let mut st = self.core.state.lock().unwrap(); + let mut st = self.core.state.lock(); st.forget(&uid); st.children.remove(&newparent); reply.ok(); @@ -6551,7 +6547,7 @@ impl Filesystem for ProtonFs { /// file lacks returns `ENODATA`. fn listxattr(&self, _req: &Request, ino: INodeNo, size: u32, reply: ReplyXattr) { let is_file = { - let st = self.core.state.lock().unwrap(); + let st = self.core.state.lock(); match st.entries.get(&ino.0) { Some(e) => e.node.is_file(), None => { @@ -6599,7 +6595,7 @@ fn apply_event( is_trashed, .. } => { - let mut st = state.lock().unwrap(); + let mut st = state.lock(); if *is_trashed { // Trashing makes a node vanish from its parent listing. let child = st.by_uid.get(node_uid).copied(); @@ -6615,7 +6611,7 @@ fn apply_event( } } } - } else if pending.lock().unwrap().contains_key(node_uid) { + } else if pending.lock().contains_key(node_uid) { // A node we owe an upload for is *ahead* of the remote, not // behind it: this event is almost always the echo of our own // empty-file create, and re-fetching would replace the size and @@ -6641,7 +6637,7 @@ fn apply_event( } } DriveEvent::NodeDeleted { node_uid, .. } => { - let mut st = state.lock().unwrap(); + let mut st = state.lock(); // Capture the inode before `forget` clears the uid mapping. let child = st.by_uid.get(node_uid).copied(); content.evict(node_uid); @@ -6661,7 +6657,7 @@ fn apply_event( // metadata. Inodes stay stable; dirs simply re-enumerate on next access. DriveEvent::ContinuityLost { .. } | DriveEvent::ScopeAccessLost { .. } => { warn!("event continuity lost; dropping all cached listings, resyncing lazily"); - let mut st = state.lock().unwrap(); + let mut st = state.lock(); let dirs: Vec = st.children.keys().copied().collect(); for &ino in &dirs { st.invalidate_listing(ino); @@ -7823,7 +7819,7 @@ pub fn mount( // Unmount every on-demand sync folder too, or the kernel mounts linger as // stale and the next start fails with EBUSY (devices.md Phase 3). - let secondaries: Vec<_> = core.mounts.lock().unwrap().drain().collect(); + let secondaries: Vec<_> = core.mounts.lock().drain().collect(); for (id, session) in secondaries { if let Err(e) = session.umount_and_join() { warn!(id, error = %e, "unmount on-demand folder failed"); diff --git a/crates/pdfs-fuse/src/sync.rs b/crates/pdfs-fuse/src/sync.rs index a7e0f68..6a1124e 100644 --- a/crates/pdfs-fuse/src/sync.rs +++ b/crates/pdfs-fuse/src/sync.rs @@ -16,7 +16,7 @@ use std::collections::{HashMap, HashSet}; use std::io::Write; use std::path::{Path, PathBuf}; use std::sync::Arc; -use std::sync::Mutex; +use parking_lot::Mutex; use std::sync::mpsc::Receiver; use std::time::{Duration, SystemTime}; @@ -209,7 +209,7 @@ fn rewatch( return; } }; - let mut have = watched.lock().unwrap(); + let mut have = watched.lock(); // Drop watches no longer wanted. have.retain(|(path, _)| { if want.iter().any(|(p, _)| p == path) { @@ -420,7 +420,7 @@ impl Core { // Hold the folder's lock for the whole pass so a mode switch can't evict the // local tree (and mount FUSE over it) while we walk and upload it. let lock = self.sync_lock(folder.id); - let _guard = lock.lock().unwrap(); + let _guard = lock.lock(); // `folder` was read before the lock; a switch may have landed in between, so // re-read and re-check the mode rather than trusting the snapshot. let current = match self.db.sync_folder_get(folder.id) { diff --git a/crates/pdfs-fuse/src/transfers.rs b/crates/pdfs-fuse/src/transfers.rs index 9750c4e..2c516e4 100644 --- a/crates/pdfs-fuse/src/transfers.rs +++ b/crates/pdfs-fuse/src/transfers.rs @@ -18,7 +18,8 @@ use std::collections::HashMap; use std::io::{Read, Write}; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, Mutex}; +use parking_lot::Mutex; +use std::sync::Arc; use std::time::Instant; use pdfs_core::control::{JobItem, TransferDirection, TransferItem}; @@ -76,7 +77,7 @@ impl TransferRegistry { done: AtomicU64::new(0), started: Instant::now(), }); - self.inner.lock().unwrap().insert(id, entry.clone()); + self.inner.lock().insert(id, entry.clone()); TransferGuard { reg: self.clone(), id, @@ -95,7 +96,7 @@ impl TransferRegistry { done: AtomicU64::new(0), total: AtomicU64::new(0), }); - self.jobs.lock().unwrap().insert(id, entry.clone()); + self.jobs.lock().insert(id, entry.clone()); JobGuard { reg: self.clone(), id, @@ -108,7 +109,7 @@ impl TransferRegistry { /// /// [`Response::Transfers`]: pdfs_core::control::Response::Transfers pub fn jobs_snapshot(&self) -> Vec { - let map = self.jobs.lock().unwrap(); + let map = self.jobs.lock(); let mut ids: Vec<_> = map.keys().copied().collect(); ids.sort_unstable(); ids.iter() @@ -116,7 +117,7 @@ impl TransferRegistry { let e = &map[id]; JobItem { title: e.title.clone(), - detail: e.detail.lock().unwrap().clone(), + detail: e.detail.lock().clone(), done: e.done.load(Ordering::Relaxed), total: e.total.load(Ordering::Relaxed), } @@ -131,7 +132,7 @@ impl TransferRegistry { /// /// [`Response::Transfers`]: pdfs_core::control::Response::Transfers pub fn snapshot(&self) -> Vec { - let map = self.inner.lock().unwrap(); + let map = self.inner.lock(); let mut ids: Vec<_> = map.keys().copied().collect(); ids.sort_unstable(); ids.iter() @@ -176,7 +177,7 @@ impl TransferGuard { impl Drop for TransferGuard { fn drop(&mut self) { - self.reg.inner.lock().unwrap().remove(&self.id); + self.reg.inner.lock().remove(&self.id); } } @@ -197,7 +198,7 @@ pub struct JobGuard { impl JobGuard { /// Say what the job is working on right now; empty clears it. pub fn detail(&self, what: impl Into) { - *self.entry.detail.lock().unwrap() = what.into(); + *self.entry.detail.lock() = what.into(); } /// Set how many steps the job now knows it has. Safe to raise mid-job as more @@ -214,7 +215,7 @@ impl JobGuard { impl Drop for JobGuard { fn drop(&mut self) { - self.reg.jobs.lock().unwrap().remove(&self.id); + self.reg.jobs.lock().remove(&self.id); } } diff --git a/crates/pdfs-gui/src/app.rs b/crates/pdfs-gui/src/app.rs index 4fca10f..4d40a69 100644 --- a/crates/pdfs-gui/src/app.rs +++ b/crates/pdfs-gui/src/app.rs @@ -2955,7 +2955,7 @@ fn stream_entry(ui: &Rc, entry: &DirEntry) { let mountpoint = ui.dirs.resolved_mountpoint(&ui.dirs.load_config()); let abs = mountpoint.join(&rel); let Some(path) = abs.to_str() else { - toast_error(&ui, "Couldn't play video", "The file path isn't valid UTF-8."); + toast_error(ui, "Couldn't play video", "The file path isn't valid UTF-8."); return; }; toast(ui, &format!("Streaming “{}”…", entry.name));