From 4720b2a5fbb2b04289693da3dbd58e7808f9f84e Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Fri, 17 Jul 2026 13:49:16 +0200 Subject: [PATCH] concurrent reads --- Cargo.lock | 8 +- Cargo.toml | 2 +- crates/pdfs-core/src/cache.rs | 20 +- crates/pdfs-core/src/db.rs | 128 +++++++----- crates/pdfs-fuse/src/lib.rs | 371 +++++++++++++++++++++++----------- packaging/PKGBUILD | 2 +- 6 files changed, 355 insertions(+), 176 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 36d1ed8..a4b5527 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3463,7 +3463,7 @@ checksum = "35fb2e5f958ec131621fdd531e9fc186ed768cbe395337403ae56c17a74c68ec" [[package]] name = "pdfs-cli" -version = "0.4.0" +version = "0.5.0" dependencies = [ "anyhow", "clap 4.6.1", @@ -3478,7 +3478,7 @@ dependencies = [ [[package]] name = "pdfs-core" -version = "0.4.0" +version = "0.5.0" dependencies = [ "directories", "ignore", @@ -3496,7 +3496,7 @@ dependencies = [ [[package]] name = "pdfs-fuse" -version = "0.4.0" +version = "0.5.0" dependencies = [ "fuser", "image", @@ -3511,7 +3511,7 @@ dependencies = [ [[package]] name = "pdfs-gui" -version = "0.4.0" +version = "0.5.0" dependencies = [ "anyhow", "async-channel", diff --git a/Cargo.toml b/Cargo.toml index 394de30..1704561 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "2" members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli", "crates/pdfs-gui"] [workspace.package] -version = "0.4.0" +version = "0.5.0" edition = "2024" license = "MIT" authors = ["Nils Pukropp "] diff --git a/crates/pdfs-core/src/cache.rs b/crates/pdfs-core/src/cache.rs index a93fde3..5c2c20b 100644 --- a/crates/pdfs-core/src/cache.rs +++ b/crates/pdfs-core/src/cache.rs @@ -27,7 +27,7 @@ use proton_drive_rs::proton_sdk::ids::NodeUid; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; -use crate::db::Db; +use crate::db::{CacheEntryInput, Db}; use crate::error::Result; /// `cache_entries.kind` tag for whole-file content blobs. @@ -233,7 +233,11 @@ impl ContentCache { /// crash or an external file deletion, and picks up caches written by builds /// predating the index. In-run accesses then refine the ordering. fn reconcile(&self) -> Result<()> { - self.db.cache_clear()?; + // Both directories are walked before anything is written: the index is + // replaced in a single transaction (see [`Db::cache_rebuild`]), so the + // rows have to be in hand first. Names are owned for the same reason — + // `DirEntry::file_name` does not outlive the iteration. + let mut names: Vec<(String, &str, u64, i64)> = Vec::new(); for (dir, kind) in [ (&self.content_dir, KIND_BLOB), (&self.block_dir, KIND_BLOCK), @@ -257,9 +261,19 @@ impl ContentCache { .and_then(|t| t.duration_since(UNIX_EPOCH).ok()) .map(|d| d.as_millis() as i64) .unwrap_or(0); - self.db.cache_touch(name, kind, meta.len(), at)?; + names.push((name.to_string(), kind, meta.len(), at)); } } + let entries: Vec> = names + .iter() + .map(|(key, kind, size, at)| CacheEntryInput { + key, + kind, + size: *size, + last_accessed: *at, + }) + .collect(); + self.db.cache_rebuild(&entries)?; Ok(()) } diff --git a/crates/pdfs-core/src/db.rs b/crates/pdfs-core/src/db.rs index 24c6d5b..d3a0028 100644 --- a/crates/pdfs-core/src/db.rs +++ b/crates/pdfs-core/src/db.rs @@ -129,6 +129,19 @@ pub struct PendingCounts { pub changes: i64, } +/// One cached blob or block as [`Db::cache_rebuild`] wants it: borrowed, because +/// the caller builds the whole list from a directory walk before handing it over. +#[derive(Debug, Clone, Copy)] +pub struct CacheEntryInput<'a> { + /// On-disk filename of the blob or block. + pub key: &'a str, + /// `'blob'` or `'block'` — the two byte budgets are counted separately. + pub kind: &'a str, + pub size: u64, + /// Initial LRU key, in unix milliseconds. + pub last_accessed: i64, +} + /// The outcome of folding freshly written bytes into a queued create. #[derive(Debug, Clone)] pub struct AttachedBlob { @@ -368,48 +381,17 @@ impl Db { /// hot-cache maps rehydrate losslessly on the next mount. `listed` is never /// changed here — it is owned by [`set_listed`](Self::set_listed). pub fn upsert_node(&self, node: &Node) -> Result<()> { - let json = serde_json::to_string(node)?; - let uid = node.uid.to_string(); - let mut conn = self.conn.lock().unwrap(); - let tx = conn.transaction()?; - tx.execute( - "INSERT INTO nodes - (uid, parent_uid, name, is_dir, size, mtime, trashed, node_json) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) - ON CONFLICT(uid) DO UPDATE SET - parent_uid = excluded.parent_uid, - name = excluded.name, - is_dir = excluded.is_dir, - size = excluded.size, - mtime = excluded.mtime, - trashed = excluded.trashed, - node_json = excluded.node_json", - params![ - uid, - node.parent_uid.as_ref().map(|u| u.to_string()), - node.name, - node.is_folder() as i64, - node_size(node), - node.modification_time, - node.trashed as i64, - json, - ], - )?; - // FTS5 has no UPSERT, so refresh the row by delete-then-insert. Trashed - // nodes are kept out of the index entirely so they never surface in - // search results. - tx.execute("DELETE FROM nodes_fts WHERE uid = ?1", params![uid])?; - if !node.trashed { - tx.execute( - "INSERT INTO nodes_fts (uid, name) VALUES (?1, ?2)", - params![uid, node.name], - )?; - } - tx.commit()?; - Ok(()) + self.upsert_nodes(std::slice::from_ref(node)) } - /// Write-through multiple nodes in a single database transaction for performance. + /// Write-through a batch of nodes as one transaction — a whole directory + /// listing, typically. Otherwise identical to [`upsert_node`](Self::upsert_node), + /// which is the single-node case of it. + /// + /// The commit count is the point: SQLite autocommits every statement that is + /// not in an explicit transaction, so interning a folder of a thousand + /// children row-by-row cost a thousand fsyncs, and `ls` waited for all of + /// them. pub fn upsert_nodes(&self, nodes: &[Node]) -> Result<()> { if nodes.is_empty() { return Ok(()); @@ -442,6 +424,9 @@ impl Db { json, ], )?; + // FTS5 has no UPSERT, so refresh the row by delete-then-insert. + // Trashed nodes are kept out of the index entirely so they never + // surface in search results. tx.execute("DELETE FROM nodes_fts WHERE uid = ?1", params![uid])?; if !node.trashed { tx.execute( @@ -1412,12 +1397,36 @@ impl Db { Ok(out) } - /// Wipe the entire cache index. The daemon calls this on open before - /// rebuilding the index from the on-disk cache, so a crash or external file - /// deletion can never leave a phantom row inflating the budget total. - pub fn cache_clear(&self) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute("DELETE FROM cache_entries", [])?; + /// Replace the whole cache index with `entries`, as one transaction. The + /// daemon calls this on open, so a crash or an external file deletion can + /// never leave a phantom row inflating the budget total. + /// + /// The clear and the refill are the same commit deliberately: they are one + /// operation — "the index now describes what is on disk" — and a crash + /// between them would otherwise leave an empty index against a full cache + /// directory, i.e. a budget total of zero and nothing eviction can see. + /// + /// One commit also matters for speed. A mount is the only caller and it runs + /// 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 tx = conn.transaction()?; + tx.execute("DELETE FROM cache_entries", [])?; + { + let mut stmt = tx.prepare( + "INSERT INTO cache_entries (cache_key, kind, size_bytes, last_accessed) + VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(cache_key) DO UPDATE SET + kind = excluded.kind, + size_bytes = excluded.size_bytes, + last_accessed = excluded.last_accessed", + )?; + for e in entries { + stmt.execute(params![e.key, e.kind, e.size as i64, e.last_accessed])?; + } + } + tx.commit()?; Ok(()) } @@ -2893,14 +2902,31 @@ mod tests { assert_eq!(blobs, vec![("xyz".into(), 1)]); } + /// A rebuild is a replacement, not a merge: whatever the index said before + /// is what a stale or externally-deleted cache file would leave behind. #[test] - fn cache_index_clear() { + fn cache_index_rebuild_replaces_every_row() { let db = Db::open_in_memory().unwrap(); - db.cache_touch("k1", "blob", 1, 1).unwrap(); - db.cache_touch("k2", "block", 1, 1).unwrap(); - db.cache_clear().unwrap(); - assert!(db.cache_entries_by_kind("blob").unwrap().is_empty()); + db.cache_touch("gone", "blob", 1, 1).unwrap(); + db.cache_touch("also-gone", "block", 1, 1).unwrap(); + + db.cache_rebuild(&[CacheEntryInput { + key: "kept", + kind: "blob", + size: 7, + last_accessed: 42, + }]) + .unwrap(); + + assert_eq!( + db.cache_entries_by_kind("blob").unwrap(), + vec![("kept".to_string(), 7)] + ); assert!(db.cache_entries_by_kind("block").unwrap().is_empty()); + + // An empty rebuild is how a cache directory that vanished reports itself. + db.cache_rebuild(&[]).unwrap(); + assert!(db.cache_entries_by_kind("blob").unwrap().is_empty()); } #[test] diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 1ec810a..30222ba 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -282,6 +282,13 @@ impl State { if let Err(e) = self.db.upsert_node(&node) { warn!(uid = %node.uid, error = %e, "db upsert_node failed"); } + self.intern_mem(parent, node) + } + + /// Allocate (or reuse) a stable inode for a node, updating the hot-cache maps + /// only. Every caller owes the DB a write-through — see the callers below; + /// the split exists so a batch can pay for one transaction instead of `n`. + fn intern_mem(&mut self, parent: u64, node: Node) -> u64 { if let Some(&ino) = self.by_uid.get(&node.uid) { if let Some(e) = self.entries.get_mut(&ino) { e.node = node; @@ -303,60 +310,25 @@ impl State { ino } - /// Allocate (or reuse) a stable inode for a node loaded from the database - /// without writing it back. + /// Allocate (or reuse) a stable inode for a node that came *from* the + /// database, which is why nothing is written back. fn intern_from_db(&mut self, parent: u64, node: Node) -> u64 { - if let Some(&ino) = self.by_uid.get(&node.uid) { - if let Some(e) = self.entries.get_mut(&ino) { - e.node = node; - e.parent = parent; - } - return ino; - } - let ino = self.next_ino; - self.next_ino += 1; - self.by_uid.insert(node.uid.clone(), ino); - self.entries.insert( - ino, - Entry { - uid: node.uid.clone(), - parent, - node, - }, - ); - ino + self.intern_mem(parent, node) } - /// Allocate (or reuse) stable inodes for a batch of nodes, writing all of them - /// to the DB in a single transaction. + /// Allocate (or reuse) stable inodes for a whole listing, writing every node + /// through in a single transaction. + /// + /// This is what keeps `ls` on a large folder quick: one commit for the + /// listing, rather than one autocommit — and one fsync — per child. fn intern_batch(&mut self, parent: u64, nodes: Vec) -> Vec { if let Err(e) = self.db.upsert_nodes(&nodes) { warn!(error = %e, "db upsert_nodes failed"); } - let mut inos = Vec::with_capacity(nodes.len()); - for node in nodes { - if let Some(&ino) = self.by_uid.get(&node.uid) { - if let Some(e) = self.entries.get_mut(&ino) { - e.node = node; - e.parent = parent; - } - inos.push(ino); - continue; - } - let ino = self.next_ino; - self.next_ino += 1; - self.by_uid.insert(node.uid.clone(), ino); - self.entries.insert( - ino, - Entry { - uid: node.uid.clone(), - parent, - node, - }, - ); - inos.push(ino); - } - inos + nodes + .into_iter() + .map(|node| self.intern_mem(parent, node)) + .collect() } /// Forget a node entirely: drop its inode, its uid mapping, its own cached @@ -561,6 +533,68 @@ async fn run_uploads( /// (two API calls and a node-key unlock) the next time that file is read. const MAX_OPEN_READERS: usize = 64; +/// How many FUSE handlers may run off the dispatch loop at once. +/// +/// `fuser`'s session loop is non-concurrent: it reads one request, calls the +/// handler, and only then reads the next. A handler that touches the network +/// therefore stalls every `getattr`/`lookup` on the mount behind it. The slow +/// handlers hand their `Reply` to this pool instead and answer from a worker, +/// which frees the loop immediately. +/// +/// Bounded on purpose: one worker can hold a 4 MiB block in flight, and an +/// unbounded pool would let read-ahead on a big file spawn threads without +/// limit. Sized just above the SDK's in-flight block cap so the semaphore +/// there, not thread count, is what bounds download memory. +const FUSE_WORKERS: usize = 8; + +/// Bounded thread pool behind the network-touching FUSE handlers. +/// +/// Shared by every session forked off one [`Core`] (the main mount plus each +/// on-demand sync folder), so the bound is per daemon rather than per mount. +struct Workers { + tx: std::sync::mpsc::Sender>, +} + +impl Workers { + fn new(n: usize) -> std::io::Result { + let (tx, rx) = std::sync::mpsc::channel::>(); + let rx = Arc::new(Mutex::new(rx)); + for i in 0..n { + let rx = rx.clone(); + std::thread::Builder::new() + .name(format!("pdfs-fuse-{i}")) + .spawn(move || { + loop { + // Held only to take the next job, never across it. + let job = rx.lock().unwrap().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. + Ok(job) => { + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job)); + } + // Every sender is gone: the daemon is shutting down. + Err(_) => break, + } + } + })?; + } + Ok(Self { tx }) + } + + /// Run `job` on a worker. Falls back to running it inline — the pre-pool + /// behaviour — if no worker is left to take it, so a shut-down pool degrades + /// to a slow mount rather than a mount that answers every read with EIO. + fn run(&self, job: impl FnOnce() + Send + 'static) { + if let Err(e) = self.tx.send(Box::new(job)) { + warn!("fuse worker pool is gone; serving inline"); + (e.0)(); + } + } +} + /// An open reader plus the node metadata it was opened against. /// /// The SDK pins a reader to the revision that was active at `open_revision`, so @@ -591,6 +625,10 @@ struct Core { /// Validated by `(mtime, size)` exactly like the content cache, and bounded /// by [`MAX_OPEN_READERS`]. readers: Arc>>, + /// Threads that serve the FUSE handlers which touch the network, so the + /// session's dispatch loop stays free to answer cheap metadata calls while a + /// cold read is on the wire. See [`Workers`]. + workers: Arc, /// Unified SQLite metadata cache: the persistence layer behind the in-memory /// `State` maps. Every mutation writes through here, and the maps rehydrate /// from it on mount (plan.md P1). @@ -843,6 +881,14 @@ impl Core { } } + /// Whether `ino`'s listing is already in memory, i.e. whether + /// [`Core::ensure_children`] would return without touching the network. + /// 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) + } + /// Enumerate `ino`'s children from the remote and cache them. No-op if the /// directory has already been listed. Network I/O happens without the lock /// held so concurrent metadata reads aren't blocked behind a fetch. @@ -1760,6 +1806,12 @@ impl Core { /// retried with doubling backoff and *never* dropped on failure — the staged /// blob is the only copy of the user's bytes, so a failed op stays queued /// until it lands or the user deletes the file. + /// + /// A failure does not pause the queue. Recording it pushes that op's + /// `next_attempt_at` past `now`, so the next pass simply picks the next op + /// that is due — one file wedged against a folder that no longer exists must + /// not hold up an unrelated upload behind it. The worker only blocks once + /// nothing is due at all. fn run_pending_drain(&self) { loop { let now = now_millis(); @@ -1775,19 +1827,22 @@ impl Core { continue; }; if let Err(e) = self.drain_op(&op) { - // Backoff is per-op, so one wedged file cannot starve the rest: - // the next pass picks whichever op is due. let attempts = op.attempts + 1; let backoff = DRAIN_BACKOFF_MIN .saturating_mul(1u32 << attempts.min(6)) .min(DRAIN_BACKOFF_MAX); warn!(uid = %op.uid, attempts, error = %e, "pending upload failed; will retry"); - let _ = self.db.record_op_failure( + if let Err(e) = self.db.record_op_failure( op.id, &e.to_string(), now_millis() + backoff.as_millis() as i64, - ); - self.wait_for_drain_work(); + ) { + // The backoff is the only thing keeping a failing op from + // being picked again immediately, so without it the loop + // would spin on this op as fast as the API can refuse it. + error!(uid = %op.uid, error = %e, "recording a drain failure failed"); + self.wait_for_drain_work(); + } } } } @@ -1842,12 +1897,13 @@ impl Core { return Ok(()); } }; - if node.parent_uid.as_ref() != Some(&parent) { - self.rt.block_on(self.client.move_node(&uid, &parent))?; - } - if node.name != name { + // The name half goes first, so that a collision on the move half below is + // about the name the node will actually land under rather than the one it + // is about to lose. + let mut landed = node.name.clone(); + if landed != name { match self.rt.block_on(self.client.rename_node(&uid, &name, None)) { - Ok(()) => {} + Ok(()) => landed = name.clone(), // Someone took the name while we were offline. Renaming to a // *different* name is the non-destructive resolution: it neither // clobbers their file nor drops ours, and it is visible. @@ -1856,6 +1912,7 @@ impl Core { warn!(%uid, name, alt, "rename target name is taken; using a conflict name"); self.rt .block_on(self.client.rename_node(&uid, &alt, None))?; + landed = alt.clone(); self.adopt_drained_name(&uid, &alt); self.log_activity( ActivityKind::Rename, @@ -1867,8 +1924,46 @@ impl Core { Err(e) => return Err(e.into()), } } + if node.parent_uid.as_ref() != Some(&parent) { + match self.rt.block_on(self.client.move_node(&uid, &parent)) { + Ok(()) => {} + // The destination holds a `landed` of its own. Same resolution as + // a name collision, and it has to happen before the move: the API + // has no move-and-rename, so the node is renamed out of the way + // here and moved second. + Err(e) if is_already_exists(&e) => { + let alt = conflict_name(&landed, now_secs()); + warn!(%uid, name = %landed, alt, "destination already holds that name; using a conflict name"); + self.rt + .block_on(self.client.rename_node(&uid, &alt, None))?; + self.rt.block_on(self.client.move_node(&uid, &parent))?; + self.adopt_drained_name(&uid, &alt); + self.log_activity( + ActivityKind::Rename, + &landed, + format!("destination already had that name; moved as {alt}"), + false, + ); + } + // The destination folder is gone. Leaving the node in its current + // parent is the honest outcome: it is not where the user asked for + // it, but it exists, it is where it has always been, and the + // rename half above still applied. Retrying could only fail again + // — the folder is not coming back — and would wedge the queue. + Err(e) if is_gone(&e) => { + warn!(%uid, name = %landed, %parent, "move destination is gone; leaving the node where it is"); + self.log_activity( + ActivityKind::Rename, + &landed, + "destination folder no longer exists; the file was left in place".to_string(), + false, + ); + } + Err(e) => return Err(e.into()), + } + } self.db.delete_op(op.id)?; - info!(%uid, name, "pending rename landed"); + info!(%uid, name = %landed, "pending rename landed"); Ok(()) } @@ -4799,6 +4894,9 @@ fn parse_uid(s: &str) -> Option { /// The Proton Drive VFS. FUSE callbacks are synchronous, so the Tokio handle /// bridges each one to the async SDK via [`Handle::block_on`]; the fuser /// session thread is not a runtime worker, so blocking on it is sound. +/// Cloneable so a handler can move a copy onto a [`Workers`] thread and answer +/// from there; every field is a handle or a plain id. +#[derive(Clone)] pub struct ProtonFs { core: Core, uid: u32, @@ -4828,6 +4926,67 @@ impl ProtonFs { Self { core, uid, gid } } + /// The body of [`Filesystem::lookup`], on whichever thread ends up serving it. + fn serve_lookup(&self, parent: u64, name: &str, reply: ReplyEntry) { + if let Err(e) = self.core.ensure_children(parent) { + reply.error(e); + return; + } + let st = self.core.state.lock().unwrap(); + let hit = st.children.get(&parent).and_then(|kids| { + kids.iter().copied().find_map(|ino| { + st.entries + .get(&ino) + .filter(|e| e.node.name == name) + .map(|e| (ino, &e.node)) + }) + }); + match hit { + Some((ino, node)) => { + let attr = self.attr(ino, node); + reply.entry(&TTL, &attr, Generation(0)); + } + None => reply.error(Errno::ENOENT), + } + } + + /// The body of [`Filesystem::readdir`], on whichever thread ends up serving it. + fn serve_readdir(&self, ino: u64, offset: u64, mut reply: ReplyDirectory) { + if let Err(e) = self.core.ensure_children(ino) { + reply.error(e); + return; + } + let st = self.core.state.lock().unwrap(); + let parent = st.entries.get(&ino).map_or(ROOT_INO, |e| e.parent); + + // "." and ".." occupy offsets 0 and 1; real children follow. + let mut listing: Vec<(u64, FileType, String)> = vec![ + (ino, FileType::Directory, ".".to_string()), + (parent, FileType::Directory, "..".to_string()), + ]; + if let Some(kids) = st.children.get(&ino) { + for &kid in kids { + if let Some(e) = st.entries.get(&kid) { + let ft = if e.node.is_folder() { + FileType::Directory + } else { + FileType::RegularFile + }; + listing.push((kid, ft, e.node.name.clone())); + } + } + } + drop(st); + + for (i, (ino, ft, name)) in listing.into_iter().enumerate().skip(offset as usize) { + // The stored offset is that of the *next* entry to resume at. + if reply.add(INodeNo(ino), (i + 1) as u64, ft, &name) { + break; + } + } + reply.ok(); + } + fn attr(&self, ino: u64, node: &Node) -> FileAttr { let (kind, perm) = match node.kind { NodeKind::Folder => (FileType::Directory, 0o755), @@ -5230,27 +5389,19 @@ fn unix_secs(secs: i64) -> SystemTime { impl Filesystem for ProtonFs { fn lookup(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) { let parent = parent.0; - if let Err(e) = self.core.ensure_children(parent) { - reply.error(e); + let name = name.to_string_lossy().into_owned(); + // A folder that has not been listed yet is enumerated from the remote, + // so serve it from a worker rather than stalling the dispatch loop. A + // listed folder — the common case — is a map hit, and answering it + // inline costs less than the handoff would. + if self.core.children_cached(parent) { + self.serve_lookup(parent, &name, reply); return; } - let name = name.to_string_lossy(); - let st = self.core.state.lock().unwrap(); - let hit = st.children.get(&parent).and_then(|kids| { - kids.iter().copied().find_map(|ino| { - st.entries - .get(&ino) - .filter(|e| e.node.name == name) - .map(|e| (ino, &e.node)) - }) - }); - match hit { - Some((ino, node)) => { - let attr = self.attr(ino, node); - reply.entry(&TTL, &attr, Generation(0)); - } - None => reply.error(Errno::ENOENT), - } + let fs = self.clone(); + self.core + .workers + .run(move || fs.serve_lookup(parent, &name, reply)); } fn getattr(&self, _req: &Request, ino: INodeNo, _fh: Option, reply: ReplyAttr) { @@ -5270,42 +5421,19 @@ impl Filesystem for ProtonFs { ino: INodeNo, _fh: FileHandle, offset: u64, - mut reply: ReplyDirectory, + reply: ReplyDirectory, ) { let ino = ino.0; - if let Err(e) = self.core.ensure_children(ino) { - reply.error(e); + // Same split as `lookup`: only the cold, remote-enumerating path pays + // for a worker. + if self.core.children_cached(ino) { + self.serve_readdir(ino, offset, reply); return; } - let st = self.core.state.lock().unwrap(); - let parent = st.entries.get(&ino).map_or(ROOT_INO, |e| e.parent); - - // "." and ".." occupy offsets 0 and 1; real children follow. - let mut listing: Vec<(u64, FileType, String)> = vec![ - (ino, FileType::Directory, ".".to_string()), - (parent, FileType::Directory, "..".to_string()), - ]; - if let Some(kids) = st.children.get(&ino) { - for &kid in kids { - if let Some(e) = st.entries.get(&kid) { - let ft = if e.node.is_folder() { - FileType::Directory - } else { - FileType::RegularFile - }; - listing.push((kid, ft, e.node.name.clone())); - } - } - } - drop(st); - - for (i, (ino, ft, name)) in listing.into_iter().enumerate().skip(offset as usize) { - // The stored offset is that of the *next* entry to resume at. - if reply.add(INodeNo(ino), (i + 1) as u64, ft, &name) { - break; - } - } - reply.ok(); + let fs = self.clone(); + self.core + .workers + .run(move || fs.serve_readdir(ino, offset, reply)); } fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) { @@ -5376,6 +5504,11 @@ impl Filesystem for ProtonFs { // A file open for writing is served from its handle so reads see the // in-flight (possibly unsaved) content: authored bytes come from the // scratch file, untouched bytes from the base. + // + // This path stays on the dispatch loop. `write` runs there too, so + // keeping its reads there as well preserves the ordering the write path + // 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(); st.handles.values().find(|h| h.ino == ino.0).map(|h| { @@ -5421,13 +5554,18 @@ impl Filesystem for ProtonFs { } } }; - match self - .core - .read_range(&uid, mtime, fsize, offset, size as u64) - { - Ok(bytes) => reply.data(&bytes), - Err(e) => reply.error(e), - } + // Off the dispatch loop: this is the one handler that routinely goes to + // the network (block fetch + decrypt, hundreds of ms on a cold file), + // and until it returns the kernel's next request for this mount is not + // even read off the FUSE device. It only reads `state`, so moving it + // races with nothing. FUSE does not require replies in request order. + let core = self.core.clone(); + self.core.workers.run(move || { + match core.read_range(&uid, mtime, fsize, offset, size as u64) { + Ok(bytes) => reply.data(&bytes), + Err(e) => reply.error(e), + } + }); } fn create( @@ -7056,6 +7194,7 @@ pub fn mount( })), cache: Arc::new(cache), readers: Arc::new(Mutex::new(HashMap::new())), + workers: Arc::new(Workers::new(FUSE_WORKERS)?), db, online: Arc::new(AtomicBool::new(online)), pending: Arc::new(Mutex::new(HashMap::new())), diff --git a/packaging/PKGBUILD b/packaging/PKGBUILD index ff25d20..64f8695 100644 --- a/packaging/PKGBUILD +++ b/packaging/PKGBUILD @@ -9,7 +9,7 @@ # Build + install: cd packaging && makepkg -fi pkgname=proton-drive-linux -pkgver=0.4.0 +pkgver=0.5.0 pkgrel=1 pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray" arch=('x86_64')