nvrl/proton-drive-linux
1
0 Fork 0
mirror of https://github.com/narrrl/proton-drive-linux.git synced 2026-11-03 16:41:26 +01:00
Code Issues Packages Projects Releases Wiki Activity

added parking_lot mutex

Browse Source
This commit is contained in:
nvrl committed 2026-07-18 12:24:28 +02:00
1 parent 38aca3cd48
commit d58dc46708
11 files changed
+203 -197
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Cargo.lock
Generated
+2
View File
@@ -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",
Cargo.toml
+5
View File
@@ -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"
crates/pdfs-core/Cargo.toml
+1
View File
@@ -20,3 +20,4 @@ ignore.workspace = true
sha2.workspace = true
rusqlite.workspace = true
keyring.workspace = true
parking_lot.workspace = true
crates/pdfs-core/src/control.rs
+4 -4
View File
@@ -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
}
crates/pdfs-core/src/db.rs
+68 -68
View File
@@ -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<T>(&self, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
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<Vec<StoredNode>> {
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<Option<Node>> {
let conn = self.conn.lock().unwrap();
let conn = self.conn.lock();
let json: Option<String> = 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<Option<String>> {
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<Option<String>> {
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<String>)> {
let conn = self.conn.lock().unwrap();
let conn = self.conn.lock();
let superseded: Option<String> = if op_supersedes(&op.kind) {
let blob: Option<String> = conn
.query_row(
@@ -664,7 +664,7 @@ impl Db {
blob_path: &str,
meta_json: &str,
) -> Result<Option<AttachedBlob>> {
let conn = self.conn.lock().unwrap();
let conn = self.conn.lock();
let existing: Option<(i64, Option<String>)> = 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<bool> {
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<String> = {
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<Vec<PendingOp>> {
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<PendingCounts> {
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<Option<i64>> {
let conn = self.conn.lock().unwrap();
let conn = self.conn.lock();
let v: Option<String> = 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<String>, Option<String>)],
) -> 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<crate::control::PhotoKind>,
range: Option<(i64, i64)>,
) -> Result<Vec<StoredPhoto>> {
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<crate::control::PhotoKind>,
) -> Result<Vec<(i32, i32, usize)>> {
let conn = self.conn.lock().unwrap();
let conn = self.conn.lock();
let (filter, binds): (&str, Vec<i64>) = 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<f64>) -> 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<usize> {
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<Vec<StoredTrash>> {
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<Option<StoredDevice>> {
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<i64> {
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<Vec<StoredSyncFolder>> {
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<Option<StoredSyncFolder>> {
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<bool> {
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<HashMap<String, StoredSyncEntry>> {
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<Vec<ActivityEntry>> {
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<Option<Vec<Node>>> {
let conn = self.conn.lock().unwrap();
let conn = self.conn.lock();
let listed: Option<i64> = 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 (`<key>.b<idx>`). 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<Vec<(String, u64)>> {
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<bool> {
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<Vec<(String, String, bool)>> {
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<bool> {
let conn = self.conn.lock().unwrap();
let conn = self.conn.lock();
let direct: Option<i64> = 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<Vec<String>> {
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<Vec<String>> {
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<i64> {
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<i64> {
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<Option<i64>> {
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))
crates/pdfs-core/src/localindex.rs
+3 -3
View File
@@ -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
}
crates/pdfs-fuse/Cargo.toml
+1
View File
@@ -17,3 +17,4 @@ libc.workspace = true
serde_json.workspace = true
image.workspace = true
notify = "8"
parking_lot.workspace = true
crates/pdfs-fuse/src/lib.rs
+105 -109
View File
@@ -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<NodeUid, u64> = {
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<NodeUid> {
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<Option<Vec<u8>>, 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<NodeUid>, tags: &HashMap<String, i64>) {
let fresh: Vec<NodeUid> = {
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<Vec<SyncFolderInfo>, 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<Mutex<()>> {
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<String> {
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<FileHandle>, 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<u64> = 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");
crates/pdfs-fuse/src/sync.rs
+3 -3
View File
@@ -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) {
crates/pdfs-fuse/src/transfers.rs
+10 -9
View File
@@ -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<JobItem> {
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<TransferItem> {
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<String>) {
*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);
}
}
crates/pdfs-gui/src/app.rs
+1 -1
View File
@@ -2955,7 +2955,7 @@ fn stream_entry(ui: &Rc<Ui>, 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));