diff --git a/crates/pdfs-core/src/cache.rs b/crates/pdfs-core/src/cache.rs
index e9e92fa..7cac1e6 100644
--- a/crates/pdfs-core/src/cache.rs
+++ b/crates/pdfs-core/src/cache.rs
@@ -611,15 +611,26 @@ impl ContentCache {
.join(format!("{}.b{idx}.meta", Self::key(uid)))
}
+ /// Whether block `idx` of `uid` is cached and fresh for `(mtime, size)`,
+ /// without reading it. For a caller deciding whether a block is worth
+ /// *fetching* — reading 4 MiB to answer that would defeat the point.
+ pub fn has_block(&self, uid: &NodeUid, mtime: i64, size: u64, idx: u64) -> bool {
+ self.block_tag(uid, idx) == Some(Meta { mtime, size })
+ && std::fs::metadata(self.block_blob(uid, idx))
+ .is_ok_and(|m| m.len() == block_len(size, idx))
+ }
+
+ /// The `(mtime, size)` a cached block claims to belong to, from its sidecar.
+ fn block_tag(&self, uid: &NodeUid, idx: u64) -> Option {
+ serde_json::from_slice(&std::fs::read(self.block_meta(uid, idx)).ok()?).ok()
+ }
+
/// Serve cached block `idx` (a [`BLOCK_SIZE`]-aligned chunk) of `uid`, or
/// `None` on miss/stale. Validated against `(mtime, size)` like a whole-file
/// blob, so a new revision (which bumps the mtime) is detected. Bumps the
/// block's mtime for LRU, best effort.
pub fn cached_block(&self, uid: &NodeUid, mtime: i64, size: u64, idx: u64) -> Option> {
- let want = Meta { mtime, size };
- let meta: Meta =
- serde_json::from_slice(&std::fs::read(self.block_meta(uid, idx)).ok()?).ok()?;
- if meta != want {
+ if self.block_tag(uid, idx)? != (Meta { mtime, size }) {
return None;
}
let mut f = std::fs::File::open(self.block_blob(uid, idx)).ok()?;
diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs
index 34f0ee7..5fe28e5 100644
--- a/crates/pdfs-fuse/src/filesystem.rs
+++ b/crates/pdfs-fuse/src/filesystem.rs
@@ -136,23 +136,30 @@ impl Filesystem for ProtonFs {
// single upgrade. This is why B12's split is still worth having — the
// plain listing never comes here at all.
//
- // Off the dispatch loop, because it goes to the network: blocking there
- // would stall every other operation on the mount (the B5 lesson).
+ // Starting the batch is local work — a `state` read and a thread spawn —
+ // so it happens here; what goes to the network is the batch itself, and
+ // this reply is parked on it rather than waiting on any thread of the
+ // mount's (audit F4).
+ let ino = ino.0;
let fs = self.clone();
- self.core.workers.run(Lane::Meta, move || {
- fs.core.upgrade_sizes_for_parent(ino.0, &uid, parent);
+ let respond = move || {
// Re-read: the upgrade writes through `state`, and on timeout or
// failure this is simply the provisional attr we already had.
let attr = {
let st = fs.core.state.lock();
- match st.entries.get(&ino.0) {
- Some(e) => fs.attr(ino.0, &e.node, e.access),
+ match st.entries.get(&ino) {
+ Some(e) => fs.attr(ino, &e.node, e.access),
// Forgotten while we waited.
None => attr,
}
};
reply.attr(&TTL, &attr);
- });
+ };
+ match self.core.upgrade_sizes_for_parent(ino, &uid, parent) {
+ Some(slot) => self.core.await_size(ino, slot, respond),
+ // Nothing to upgrade, so nothing to wait for.
+ None => respond(),
+ }
}
fn readdir(
@@ -606,10 +613,11 @@ impl Filesystem for ProtonFs {
// acceptance suite's concurrency case failed with "concurrent file 1
// mismatch" on all three runs the moment this moved.
//
- // The cost is real — `queue_revision` gap-fills a partial write from the
- // remote — so moving it off still wants doing, but it needs the ordering
- // made explicit first (a per-node "staging in flight" barrier that reads
- // wait on) rather than inherited from the dispatch loop.
+ // That inherited ordering is now affordable because the network came out
+ // of this path: `queue_revision` no longer downloads the ranges an edit
+ // did not touch, it stages an incomplete blob and lets the drain thread
+ // finish it (audit F1). What is left is a rename, two fsyncs, a database
+ // insert, and at most a bounded copy out of the local cache.
self.finish_release(handle, unlinked_uid, reply);
}
@@ -1000,16 +1008,22 @@ impl ProtonFs {
});
return;
}
- self.core.upgrade_sizes_for_parent(ino, &uid, grandparent);
- // Re-read: on timeout or failure this is the provisional attr again,
- // which is no worse than what this path used to always send.
- let attr = {
- let st = self.core.state.lock();
- st.entries
- .get(&ino)
- .map_or(attr, |e| self.attr(ino, &e.node, e.access))
+ let fs = self.clone();
+ let respond = move || {
+ // Re-read: on timeout or failure this is the provisional attr again,
+ // which is no worse than what this path used to always send.
+ let attr = {
+ let st = fs.core.state.lock();
+ st.entries
+ .get(&ino)
+ .map_or(attr, |e| fs.attr(ino, &e.node, e.access))
+ };
+ reply.entry(&TTL, &attr, Generation(0));
};
- reply.entry(&TTL, &attr, Generation(0));
+ match self.core.upgrade_sizes_for_parent(ino, &uid, grandparent) {
+ Some(slot) => self.core.await_size(ino, slot, respond),
+ None => respond(),
+ }
}
/// The body of [`Filesystem::readdir`], on whichever thread ends up serving it.
diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs
index 69f1204..155036d 100644
--- a/crates/pdfs-fuse/src/lib.rs
+++ b/crates/pdfs-fuse/src/lib.rs
@@ -105,7 +105,7 @@ pub use mount::{MountOptions, MountOutcome, mount};
use mount::{
SecondaryInsertRejection, SecondaryMount, clear_stale_mount, fuse_connection_id, spawn_session,
};
-use reads::{ReaderSlot, STREAM_BYPASS_MIN, StreamRing};
+use reads::{BlockFlight, BlockRing, PREFETCH_BUDGET, Prefetch, ReaderSlot, STREAM_BYPASS_MIN};
use state::{Entry, Intervals, PendingRevision, State, WriteHandle};
use tracing::{debug, error, info, warn};
use transfers::{CountingWriter, JobGuard, OwnedCountingReader, TransferRegistry};
@@ -253,6 +253,15 @@ const MAX_THUMBNAIL_MISSES: usize = 8192;
/// delay the waiters this chunking exists to release (bugs.md B14).
const SIZE_UPGRADE_CHUNK: usize = 150;
+/// How many folders may have a size upgrade in flight at once.
+///
+/// Each one owns a thread and an API request stream, and the single-flight only
+/// deduplicates *within* a folder — so without this a recursive listing scaled
+/// both with the number of folders it walked. Eight keeps an interactive `ls -l`
+/// (one folder, sometimes a couple) entirely unaffected while putting a ceiling
+/// on the recursive case.
+const MAX_SIZE_UPGRADES: usize = 8;
+
#[derive(Clone, Debug, Eq, PartialEq)]
struct RootListingSnapshot {
children: Vec,
@@ -330,10 +339,19 @@ struct Core {
/// Validated by `(mtime, size)` exactly like the content cache, and bounded
/// by [`MAX_OPEN_READERS`].
readers: Arc>>,
- /// Blocks of files streaming past the on-disk cache, held in memory so a
- /// player's small sequential reads are served from the last few 4 MiB blocks
- /// instead of re-downloading each one. See [`StreamRing`].
- stream_ring: Arc>,
+ /// Decrypted blocks held in memory, so the ~32 kernel reads a 4 MiB block is
+ /// delivered in cost the block once — a download for a streaming file, a
+ /// disk read for a cached one. See [`BlockRing`].
+ block_ring: Arc>,
+ /// Block fetches in flight, so concurrent readers of one block (demand read,
+ /// kernel read-ahead, our own prefetch) share a single download and decrypt.
+ /// See [`BlockFlight`].
+ block_flight: Arc,
+ /// Per-file sequential-read detection driving [`Core::prefetch`].
+ prefetch: Arc,
+ /// Permits bounding speculative block fetches, so prefetch cannot queue in
+ /// front of a read someone is waiting on.
+ prefetch_budget: 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`].
@@ -441,6 +459,9 @@ struct Core {
/// `ls -l` of a folder is one `getattr` per file, and they must collapse
/// onto a single batch (bugs.md B14).
size_upgrades: Arc>>>,
+ /// Replies parked on those upgrades. Shared across every mount for the same
+ /// reason the pool is: the bound that matters is per daemon.
+ size_waiters: Arc,
/// This mount's kernel notification channel, for telling the kernel to drop
/// metadata it has cached. Set once the session exists — which is *after*
/// the `Core` it is built from, hence the cell.
@@ -1009,6 +1030,59 @@ fn preserve_on_access_denied(
Ok(())
}
+/// Fold the staged blob of an undrained write into the blob at `src` that is
+/// about to supersede it, so the newer write inherits the older one's bytes
+/// instead of losing them.
+///
+/// Only the ranges the older write authored and the newer one did not are
+/// copied, so a small edit over a small edit moves almost nothing. What neither
+/// wrote is untouched remote content, which is why `base_size`, `base_mtime` and
+/// the baseline all come from the write being superseded: that is the revision
+/// those ranges were last observed against, and the one the drain must gap-fill
+/// and conflict-check them against.
+fn merge_over_pending(
+ meta: &mut StagedWrite,
+ src: &Path,
+ previous: &PendingRevision,
+) -> std::io::Result<()> {
+ let mut written = Intervals::default();
+ for &(s, e) in &meta.authored {
+ written.add(s, e);
+ }
+ let mut earlier = Intervals::default();
+ for &(s, e) in &previous.meta.authored {
+ earlier.add(s, e);
+ }
+ let blob = File::open(&previous.path)?;
+ let dst = std::fs::OpenOptions::new().write(true).open(src)?;
+ for (s, e, authored) in written.clone().segments(0, meta.len) {
+ if authored {
+ continue;
+ }
+ for (ps, pe, have) in earlier.segments(s, e.min(previous.meta.len)) {
+ if !have {
+ continue;
+ }
+ let mut buf = vec![0u8; (pe - ps) as usize];
+ blob.read_exact_at(&mut buf, ps)?;
+ dst.write_all_at(&buf, ps)?;
+ written.add(ps, pe);
+ }
+ }
+ dst.sync_all()?;
+ meta.base_size = previous.meta.base_size;
+ meta.base_mtime = previous.meta.base_mtime;
+ meta.based_on = previous.meta.based_on.clone();
+ meta.authored = written
+ .segments(0, meta.len)
+ .into_iter()
+ .filter(|&(_, _, authored)| authored)
+ .map(|(s, e, _)| (s, e))
+ .collect();
+ meta.complete = meta.authored == [(0, meta.len)];
+ Ok(())
+}
+
/// Retire the last resident reference to an unlinked inode without touching its
/// persisted node row. The namespace path already chose whether that row is
/// disposable or retained as queued-trash authority.
@@ -2008,7 +2082,12 @@ impl Core {
/// an earlier upgrade had a chance to run. Gathers the whole folder rather
/// than the one file asked for, because a `stat` of one entry in a listing
/// almost always means a `stat` of all of them.
- fn upgrade_sizes_for_parent(&self, ino: u64, uid: &NodeUid, parent: u64) {
+ fn upgrade_sizes_for_parent(
+ &self,
+ ino: u64,
+ uid: &NodeUid,
+ parent: u64,
+ ) -> Option> {
let provisional = |e: &Entry| {
matches!(
&e.node.kind,
@@ -2046,9 +2125,10 @@ impl Core {
if !missing.iter().any(|u| u == uid) {
missing.push(uid.clone());
}
- // Blocking, not spawning: the caller is a `getattr` that must not answer
- // with a provisional size (bugs.md B14).
- self.upgrade_sizes(key, missing, Some(ino));
+ // Returned, not awaited here: the caller is a `getattr` that must not
+ // answer with a provisional size (bugs.md B14), and it parks its reply
+ // on this batch rather than holding a thread until it lands.
+ self.upgrade_sizes(key, missing)
}
/// Fill in the true sizes of files a `Light` enumeration returned without
@@ -2069,11 +2149,11 @@ impl Core {
/// Single-flight per folder: a `stat` of every entry in a fresh listing is
/// the normal case, and each one must not start its own upgrade.
fn spawn_size_upgrade(&self, folder_ino: u64, uids: Vec) {
- self.upgrade_sizes(folder_ino, uids, None);
+ self.upgrade_sizes(folder_ino, uids);
}
- /// Start (or join) the size upgrade for `key`, and — when `waiting_for` names
- /// an inode — block until that node's real size has landed.
+ /// Start (or join) the size upgrade for `key`, returning the batch a caller
+ /// can wait on — `None` when there was nothing to upgrade.
///
/// Single-flight per `key`. The fetch runs on its own thread rather than on
/// the [`Workers`] pool: callers wait on `Lane::Meta`, so a batch queued onto
@@ -2081,9 +2161,9 @@ impl Core {
/// for a job that can never be scheduled. `Lane::Transfer` would swap that
/// for starvation behind bulk reads. One short-lived thread per folder,
/// bounded by the single-flight, avoids both.
- fn upgrade_sizes(&self, key: u64, uids: Vec, waiting_for: Option) {
+ fn upgrade_sizes(&self, key: u64, uids: Vec) -> Option> {
if uids.is_empty() {
- return;
+ return None;
}
let slot = {
let mut in_flight = self.size_upgrades.lock();
@@ -2091,6 +2171,13 @@ impl Core {
// Someone else is already fetching this folder; their batch
// covers us, so just wait on it.
Some(existing) => existing.clone(),
+ // Nothing bounded how many of these could be in flight: the
+ // single-flight is per folder, so a recursive listing across
+ // hundreds of cold folders started a thread and a batch for
+ // each (audit F4). Past the cap, answer with the provisional
+ // size — the same fallback the deadline already takes, and the
+ // upgrade is retried by the next `stat` once a slot frees.
+ None if in_flight.len() >= MAX_SIZE_UPGRADES => return None,
None => {
let slot = Arc::new(SizeUpgrade::default());
in_flight.insert(key, slot.clone());
@@ -2103,10 +2190,41 @@ impl Core {
}
}
};
- let Some(ino) = waiting_for else {
- return;
+ Some(slot)
+ }
+
+ /// Hold `respond` back until `ino` has a real size, `slot`'s batch ends, or
+ /// [`SizeUpgrade::WAIT`] elapses — then run it, on the queue's thread.
+ ///
+ /// The caller returns immediately. That is the whole point: the thread it
+ /// was running on is a FUSE dispatch thread or a [`Workers`] one, and both
+ /// are needed by everything else on the mount (audit F4).
+ fn await_size(
+ &self,
+ ino: u64,
+ slot: Arc,
+ respond: impl FnOnce() + Send + 'static,
+ ) {
+ let core = self.clone();
+ let waiter = SizeWaiter {
+ slot,
+ resolved: Box::new(move || core.size_is_real(ino)),
+ deadline: Instant::now() + SizeUpgrade::WAIT,
+ respond: Box::new(respond),
};
- slot.wait_for(|| self.size_is_real(ino));
+ if !self.size_waiters.park(waiter) {
+ return;
+ }
+ let queue = self.size_waiters.clone();
+ if let Err(error) = std::thread::Builder::new()
+ .name("pdfs-size-waiters".into())
+ .spawn(move || queue.serve())
+ {
+ // Without a thread nothing would ever answer these. A provisional
+ // size is wrong; a `stat` that never returns wedges the caller.
+ warn!(%error, "no thread for parked size waiters; answering provisionally");
+ self.size_waiters.answer_all();
+ }
}
/// Fetch `uids` in chunks, applying and announcing each one as it lands.
@@ -2393,41 +2511,47 @@ impl Core {
let _ = std::fs::remove_file(&h.path);
return Ok(());
}
- // Materialize the full content now if we can. A complete blob is
- // uploadable without the network and, crucially, lets a later write to
- // the same file supersede this one safely.
+ // Materialize as much of the full content as is free to materialize. A
+ // complete blob is uploadable without the network and lets a later write
+ // to the same file supersede this one without any reconstruction at all.
+ //
+ // What is deliberately *not* done here is fetching the untouched ranges
+ // from the remote: that is a download of everything the edit did not
+ // touch, and this runs on the FUSE dispatch loop, so closing a large
+ // file after a small edit stopped every other request on the mount for
+ // the length of it (audit F1). An incomplete blob is first-class —
+ // `read_pending` resolves its gaps against the base, and `drain_revision`
+ // fills them on the drain thread — so the network half of the fill
+ // belongs there, and only there.
//
// A node that exists only locally has no remote base to fill from: its
- // untouched ranges live in the blob queued by an earlier write, not on
- // any server. Filling is only safe — and only needed — while it is still
- // empty, which is exactly when `fill_gaps` skips the network anyway.
- let filled = if is_local_uid(&h.uid) && h.base_size > 0 {
- if let Err(e) = h.file.set_len(h.len) {
- error!(uid = %h.uid, error = %e, "resize scratch file failed");
- return Err(Errno::EIO);
- }
- false
- } else {
- self.fill_gaps(
+ // untouched ranges live in the blob queued by an earlier write, which
+ // `merge_over_pending` folds in below.
+ if let Err(e) = h.file.set_len(h.len) {
+ error!(uid = %h.uid, error = %e, "resize scratch file failed");
+ return Err(Errno::EIO);
+ }
+ let mut written = h.written.clone();
+ if !(is_local_uid(&h.uid) && h.base_size > 0) {
+ // A local node has no remote base, so nothing of it can be fetched
+ // or cached: passing zero says so, and leaves the fill doing only
+ // the part that still applies — claiming the zeroed tail.
+ let base_size = if is_local_uid(&h.uid) { 0 } else { h.base_size };
+ self.fill_gaps_cached(
&h.uid,
&h.file,
h.len,
h.base_mtime,
- h.base_size,
- &h.written,
- )
- .is_ok()
- };
- let authored: Vec<(u64, u64)> = if filled {
- vec![(0, h.len)]
- } else {
- h.written
- .segments(0, h.len)
- .into_iter()
- .filter(|&(_, _, authored)| authored)
- .map(|(s, e, _)| (s, e))
- .collect()
- };
+ base_size,
+ &mut written,
+ );
+ }
+ let authored: Vec<(u64, u64)> = written
+ .segments(0, h.len)
+ .into_iter()
+ .filter(|&(_, _, authored)| authored)
+ .map(|(s, e, _)| (s, e))
+ .collect();
let meta = StagedWrite {
uid: h.uid.to_string(),
len: h.len,
@@ -2442,8 +2566,9 @@ impl Core {
h.base_revision_id.clone(),
),
};
+ let complete = meta.complete;
self.enqueue_staged_write(&h.uid, h.ino, &h.path, meta)?;
- debug!(uid = %h.uid, len = h.len, complete = filled, "queued revision upload");
+ debug!(uid = %h.uid, len = h.len, complete, "queued revision upload");
Ok(())
}
@@ -2524,7 +2649,7 @@ impl Core {
uid: &NodeUid,
ino: u64,
src: &Path,
- meta: StagedWrite,
+ mut meta: StagedWrite,
) -> Result<(), Errno> {
preserve_on_access_denied(self.require_uid_writable(uid), true, || {
self.stage_orphaned_write(
@@ -2537,12 +2662,21 @@ impl Core {
})?;
// An incomplete blob's gaps refer to the *remote* base. If an earlier
// write to this file is still queued, the remote no longer holds that
- // base — the previous staged blob does — so superseding it would fill
- // those gaps from the wrong revision. Rather than corrupt the file,
- // 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().contains_key(uid) {
+ // base — the previous staged blob does — so taking those gaps at face
+ // value would fill them from the wrong revision. Fold the blob being
+ // superseded into this one instead: its authored bytes are a local copy
+ // of exactly the ranges this write did not touch, and what neither wrote
+ // is still the remote content that the earlier write was based on, which
+ // is why the baseline is inherited along with them.
+ let previous = self.pending.lock().get(uid).cloned();
+ if !meta.complete
+ && let Some(previous) = previous
+ && let Err(error) = merge_over_pending(&mut meta, src, &previous)
+ {
+ // Without the earlier bytes the gaps cannot be resolved against
+ // anything, so this is where the write stops being uploadable.
+ // Keep it recoverable rather than corrupting the file.
+ warn!(%uid, %error, "folding in the undrained edit failed");
self.stage_orphaned_write(
uid,
ino,
@@ -4619,37 +4753,10 @@ impl SizeUpgrade {
/// wedging whatever is listing the directory.
const WAIT: Duration = Duration::from_secs(10);
- /// Block until `resolved` reports the caller's own node has a real size, the
- /// batch finishes without producing one, or [`SizeUpgrade::WAIT`] elapses.
- ///
- /// `resolved` is called with **no lock of ours held**. It reaches into
- /// `state`, and the thread applying a chunk holds `state` before it signals
- /// here — taking them in the other order would close the cycle.
- fn wait_for(&self, resolved: impl Fn() -> bool) {
- let deadline = Instant::now() + Self::WAIT;
- loop {
- let seen = {
- let progress = self.inner.lock();
- if progress.done {
- return;
- }
- progress.generation
- };
- if resolved() {
- return;
- }
- let mut progress = self.inner.lock();
- if progress.done {
- return;
- }
- // A chunk landed while we were checking, so re-check rather than
- // sleeping through the answer we were waiting for.
- if progress.generation == seen
- && self.ready.wait_until(&mut progress, deadline).timed_out()
- {
- return;
- }
- }
+ /// Whether the batch has ended, however it ended. A waiter that sees this
+ /// has nothing left to wait for.
+ fn is_finished(&self) -> bool {
+ self.inner.lock().done
}
/// Announce that a chunk has been applied. Every waiter re-checks its own
@@ -4669,6 +4776,121 @@ impl SizeUpgrade {
}
}
+/// How often [`SizeWaitQueue::serve`] re-tests its parked waiters.
+///
+/// The predicates read `state`, which the applying thread writes under its own
+/// lock, so this polls rather than waiting on each slot's condvar: one thread
+/// cannot wait on a hundred condvars, and the whole point of the queue is that
+/// there is only one thread. Twenty milliseconds is far below the round trip a
+/// waiter is waiting on and only ticks while something is actually parked.
+const SIZE_WAIT_POLL: Duration = Duration::from_millis(20);
+
+/// A reply held back until a size upgrade resolves the node it describes.
+struct SizeWaiter {
+ /// The batch this reply is waiting on. When it ends, so does the wait.
+ slot: Arc,
+ /// Whether *this* waiter's node now has a real size.
+ ///
+ /// Called with no queue lock held: it reaches into `state`, and the thread
+ /// applying a chunk holds `state` before it announces the chunk — taking
+ /// them in the other order would close the cycle.
+ resolved: Box bool + Send>,
+ /// When to answer with the provisional size anyway. A `stat` that never
+ /// returns is far worse than one that is briefly wrong.
+ deadline: Instant,
+ /// Sends the FUSE reply. Runs on the serving thread.
+ respond: Box,
+}
+
+#[derive(Default)]
+struct Parked {
+ waiters: Vec,
+ /// Whether a thread is currently serving `waiters`. The thread clears this
+ /// as it retires, so the next park starts a new one.
+ running: bool,
+}
+
+/// Every reply parked on a size upgrade, served by at most one thread.
+///
+/// This exists because waiting used to happen on the [`Workers`] pool. Sizes
+/// are single-flighted per folder, so an `ls -l` across N cold folders parks a
+/// waiter per entry — and each of those held a pool thread asleep on a condvar
+/// for up to [`SizeUpgrade::WAIT`]. With enough folders that is every thread in
+/// both lanes, including the ones reserved for metadata precisely so that this
+/// could not happen (audit F4). Parking the `Reply` instead costs one thread
+/// for the whole mount, and only while something is parked.
+#[derive(Default)]
+struct SizeWaitQueue {
+ parked: Mutex,
+ /// Woken when a waiter is added, so a serving thread re-computes how long
+ /// it may sleep.
+ added: Condvar,
+}
+
+impl SizeWaitQueue {
+ /// Park `waiter`. Returns whether the caller must start a serving thread.
+ fn park(&self, waiter: SizeWaiter) -> bool {
+ let mut parked = self.parked.lock();
+ parked.waiters.push(waiter);
+ if parked.running {
+ drop(parked);
+ self.added.notify_all();
+ return false;
+ }
+ parked.running = true;
+ true
+ }
+
+ /// Answer every waiter whose node has resolved, whose batch has ended, or
+ /// whose deadline has passed; sleep until the next one might. Returns when
+ /// nothing is parked any more.
+ fn serve(&self) {
+ loop {
+ // Taken out of the lock wholesale: the predicates and the replies
+ // both take locks of their own, and none of them may be called
+ // while this one is held.
+ let taken = std::mem::take(&mut self.parked.lock().waiters);
+ let now = Instant::now();
+ let mut keep = Vec::with_capacity(taken.len());
+ for waiter in taken {
+ if now >= waiter.deadline || waiter.slot.is_finished() || (waiter.resolved)() {
+ (waiter.respond)();
+ } else {
+ keep.push(waiter);
+ }
+ }
+ let mut parked = self.parked.lock();
+ parked.waiters.append(&mut keep);
+ if parked.waiters.is_empty() {
+ parked.running = false;
+ return;
+ }
+ let until = parked
+ .waiters
+ .iter()
+ .map(|w| w.deadline)
+ .min()
+ .unwrap_or(now)
+ .min(Instant::now() + SIZE_WAIT_POLL);
+ self.added.wait_until(&mut parked, until);
+ }
+ }
+
+ /// Answer everything parked, immediately, and retire the serving thread.
+ /// Used when there is no thread to serve with: a provisional size now beats
+ /// a reply that never comes.
+ fn answer_all(&self) {
+ let taken = {
+ let mut parked = self.parked.lock();
+ parked.running = false;
+ std::mem::take(&mut parked.waiters)
+ };
+ for waiter in taken {
+ (waiter.respond)();
+ }
+ }
+}
+
/// Whether a `rename` may replace an existing destination, per POSIX.
///
/// Split out from the handler because it is the part that is pure and the part
@@ -4827,19 +5049,220 @@ fn unix_secs(secs: i64) -> SystemTime {
}
}
+#[cfg(test)]
+mod merge_over_pending_tests {
+ use super::{PendingRevision, merge_over_pending};
+ use pdfs_core::cache::{Baseline, StagedWrite};
+ use std::io::Write as _;
+ use std::path::PathBuf;
+
+ /// A unique temp directory removed on drop; avoids a dev-dependency, as in
+ /// `state.rs`.
+ struct TempDir(PathBuf);
+ impl TempDir {
+ fn new() -> Self {
+ use std::sync::atomic::{AtomicU64, Ordering};
+ static N: AtomicU64 = AtomicU64::new(0);
+ let p = std::env::temp_dir().join(format!(
+ "pdfs-merge-{}-{}",
+ std::process::id(),
+ N.fetch_add(1, Ordering::Relaxed)
+ ));
+ std::fs::create_dir_all(&p).unwrap();
+ TempDir(p)
+ }
+ }
+ impl Drop for TempDir {
+ fn drop(&mut self) {
+ let _ = std::fs::remove_dir_all(&self.0);
+ }
+ }
+
+ fn staged(len: u64, base_size: u64, authored: &[(u64, u64)]) -> StagedWrite {
+ StagedWrite {
+ uid: "vol~link".into(),
+ len,
+ base_size,
+ base_mtime: 100,
+ complete: authored == [(0, len)],
+ authored: authored.to_vec(),
+ based_on: Some(Baseline {
+ mtime: 100,
+ size: base_size,
+ hash: None,
+ revision_id: Some("r1".into()),
+ }),
+ }
+ }
+
+ /// Write a file of `len` bytes with `fill` at every offset in `ranges`.
+ fn blob(
+ dir: &std::path::Path,
+ name: &str,
+ len: u64,
+ ranges: &[(u64, u64)],
+ fill: u8,
+ ) -> std::path::PathBuf {
+ let path = dir.join(name);
+ let mut bytes = vec![0u8; len as usize];
+ for &(s, e) in ranges {
+ bytes[s as usize..e as usize].fill(fill);
+ }
+ let mut f = std::fs::File::create(&path).unwrap();
+ f.write_all(&bytes).unwrap();
+ path
+ }
+
+ /// The case the merge exists for: two small edits to different parts of a
+ /// file, the second closing before the first has drained. The second blob
+ /// must come out holding both, or the first edit is silently lost.
+ #[test]
+ fn the_superseding_write_inherits_the_earlier_ones_bytes() {
+ let dir = TempDir::new();
+ let previous = PendingRevision {
+ path: blob(&dir.0, "old", 64, &[(0, 16)], 0xAA),
+ meta: staged(64, 64, &[(0, 16)]),
+ };
+ let src = blob(&dir.0, "new", 64, &[(32, 48)], 0xBB);
+ let mut meta = staged(64, 64, &[(32, 48)]);
+
+ merge_over_pending(&mut meta, &src, &previous).unwrap();
+
+ assert_eq!(meta.authored, vec![(0, 16), (32, 48)]);
+ assert!(
+ !meta.complete,
+ "the untouched middle still comes from the base"
+ );
+ let merged = std::fs::read(&src).unwrap();
+ assert!(
+ merged[0..16].iter().all(|&b| b == 0xAA),
+ "earlier edit kept"
+ );
+ assert!(merged[32..48].iter().all(|&b| b == 0xBB), "newer edit kept");
+ }
+
+ /// Where they overlap, the newer write wins — it is the later state of the
+ /// file, and the older bytes were never visible to anyone after it landed.
+ #[test]
+ fn the_newer_write_wins_the_ranges_it_authored() {
+ let dir = TempDir::new();
+ let previous = PendingRevision {
+ path: blob(&dir.0, "old", 32, &[(0, 32)], 0xAA),
+ meta: staged(32, 32, &[(0, 32)]),
+ };
+ let src = blob(&dir.0, "new", 32, &[(8, 16)], 0xBB);
+ let mut meta = staged(32, 32, &[(8, 16)]);
+
+ merge_over_pending(&mut meta, &src, &previous).unwrap();
+
+ assert_eq!(meta.authored, vec![(0, 32)]);
+ assert!(meta.complete, "between them they cover the file");
+ let merged = std::fs::read(&src).unwrap();
+ assert!(
+ merged[8..16].iter().all(|&b| b == 0xBB),
+ "newer edit survives"
+ );
+ assert!(merged[16..32].iter().all(|&b| b == 0xAA));
+ }
+
+ /// The baseline has to come from the write being superseded. This one's own
+ /// "base" is the earlier *staged blob*, whose size and mtime are ours, not
+ /// the server's — filling or conflict-checking against that would compare
+ /// the remote to a revision it never had.
+ #[test]
+ fn the_baseline_is_inherited_from_the_write_being_superseded() {
+ let dir = TempDir::new();
+ let mut earlier = staged(64, 64, &[(0, 16)]);
+ earlier.base_mtime = 42;
+ earlier.base_size = 50;
+ earlier.based_on = Some(Baseline {
+ mtime: 42,
+ size: 50,
+ hash: None,
+ revision_id: Some("original".into()),
+ });
+ let previous = PendingRevision {
+ path: blob(&dir.0, "old", 64, &[(0, 16)], 0xAA),
+ meta: earlier,
+ };
+ let src = blob(&dir.0, "new", 64, &[(32, 48)], 0xBB);
+ // Opened over the staged blob, so this write thinks its base is 64/100.
+ let mut meta = staged(64, 64, &[(32, 48)]);
+
+ merge_over_pending(&mut meta, &src, &previous).unwrap();
+
+ assert_eq!(meta.base_mtime, 42);
+ assert_eq!(meta.base_size, 50);
+ assert_eq!(
+ meta.based_on.unwrap().revision_id.as_deref(),
+ Some("original")
+ );
+ }
+
+ /// A shrink must not drag the earlier write's tail back in.
+ #[test]
+ fn a_truncating_write_does_not_inherit_past_its_own_length() {
+ let dir = TempDir::new();
+ let previous = PendingRevision {
+ path: blob(&dir.0, "old", 64, &[(0, 64)], 0xAA),
+ meta: staged(64, 64, &[(0, 64)]),
+ };
+ let src = blob(&dir.0, "new", 16, &[(8, 16)], 0xBB);
+ let mut meta = staged(16, 64, &[(8, 16)]);
+
+ merge_over_pending(&mut meta, &src, &previous).unwrap();
+
+ assert_eq!(meta.authored, vec![(0, 16)]);
+ assert!(meta.complete);
+ assert_eq!(std::fs::metadata(&src).unwrap().len(), 16);
+ }
+}
+
#[cfg(test)]
mod size_upgrade_tests {
- use super::SizeUpgrade;
+ use super::{SizeUpgrade, SizeWaitQueue, SizeWaiter};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
+ use std::sync::mpsc;
use std::time::{Duration, Instant};
- /// The point of the per-chunk design: a waiter returns as soon as *its own*
- /// file is resolved, without waiting for the rest of the folder. A single
- /// batch over 793 nodes took ~80 s, which outran the timeout and handed back
- /// the provisional size this is all meant to prevent.
+ /// Park a waiter the way [`Core::await_size`] does, and hand back the
+ /// channel its reply is sent down, so a test can time the release.
+ fn park(
+ queue: &Arc,
+ slot: Arc,
+ resolved: impl Fn() -> bool + Send + 'static,
+ deadline: Instant,
+ ) -> mpsc::Receiver<()> {
+ let (tx, rx) = mpsc::channel();
+ let waiter = SizeWaiter {
+ slot,
+ resolved: Box::new(resolved),
+ deadline,
+ respond: Box::new(move || {
+ let _ = tx.send(());
+ }),
+ };
+ if queue.park(waiter) {
+ let queue = queue.clone();
+ std::thread::spawn(move || queue.serve());
+ }
+ rx
+ }
+
+ /// Long enough that a test never trips the deadline by accident, short
+ /// enough that one which is *meant* to does not take ten seconds.
+ fn far_off() -> Instant {
+ Instant::now() + Duration::from_secs(30)
+ }
+
+ /// The point of the per-chunk design: a waiter is answered as soon as *its
+ /// own* file is resolved, without waiting for the rest of the folder. A
+ /// single batch over 793 nodes took ~80 s, which outran the timeout and
+ /// handed back the provisional size this is all meant to prevent.
#[test]
fn a_waiter_returns_on_the_chunk_that_resolves_it() {
+ let queue = Arc::new(SizeWaitQueue::default());
let slot = Arc::new(SizeUpgrade::default());
let mine = Arc::new(AtomicBool::new(false));
let worker = slot.clone();
@@ -4849,39 +5272,43 @@ mod size_upgrade_tests {
std::thread::sleep(Duration::from_millis(30));
flag.store(true, Ordering::SeqCst);
worker.chunk_done();
- std::thread::sleep(Duration::from_millis(200));
+ std::thread::sleep(Duration::from_millis(300));
worker.chunk_done();
worker.finish();
});
let started = Instant::now();
- slot.wait_for(|| mine.load(Ordering::SeqCst));
+ let rx = park(&queue, slot, move || mine.load(Ordering::SeqCst), far_off());
+ rx.recv().expect("the waiter is answered");
// Released by the first chunk, not the last.
- assert!(started.elapsed() < Duration::from_millis(150));
+ assert!(started.elapsed() < Duration::from_millis(200));
t.join().unwrap();
}
- /// A chunk that does not resolve this waiter must not wake it for good.
+ /// A chunk that does not resolve this waiter must not answer it.
#[test]
fn an_unrelated_chunk_does_not_release_a_waiter() {
+ let queue = Arc::new(SizeWaitQueue::default());
let slot = Arc::new(SizeUpgrade::default());
let worker = slot.clone();
let t = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(30));
worker.chunk_done();
- std::thread::sleep(Duration::from_millis(60));
+ std::thread::sleep(Duration::from_millis(120));
worker.finish();
});
let started = Instant::now();
// Never resolved: only `finish` can end this wait.
- slot.wait_for(|| false);
- assert!(started.elapsed() >= Duration::from_millis(80));
+ let rx = park(&queue, slot, || false, far_off());
+ rx.recv().expect("the waiter is answered");
+ assert!(started.elapsed() >= Duration::from_millis(140));
t.join().unwrap();
}
/// A batch that ends without resolving the node — a failed fetch — still
- /// releases its waiters, who fall back to the provisional size.
+ /// answers its waiters, who fall back to the provisional size.
#[test]
fn finish_releases_a_waiter_that_was_never_resolved() {
+ let queue = Arc::new(SizeWaitQueue::default());
let slot = Arc::new(SizeUpgrade::default());
let worker = slot.clone();
let t = std::thread::spawn(move || {
@@ -4889,59 +5316,101 @@ mod size_upgrade_tests {
worker.finish();
});
let started = Instant::now();
- slot.wait_for(|| false);
+ let rx = park(&queue, slot, || false, far_off());
+ rx.recv().expect("the waiter is answered");
assert!(started.elapsed() < SizeUpgrade::WAIT);
t.join().unwrap();
}
- /// Already resolved before waiting: must not block at all. This is the
+ /// Already resolved before parking: answered on the first pass. This is the
/// follower that arrives after the chunk it needed has landed.
#[test]
fn an_already_resolved_waiter_does_not_block() {
- let slot = SizeUpgrade::default();
+ let queue = Arc::new(SizeWaitQueue::default());
let started = Instant::now();
- slot.wait_for(|| true);
+ let rx = park(&queue, Arc::new(SizeUpgrade::default()), || true, far_off());
+ rx.recv().expect("the waiter is answered");
assert!(started.elapsed() < Duration::from_millis(100));
}
- /// Finishing before anyone waits must not strand the late arrival — the
+ /// Finishing before anyone parks must not strand the late arrival — the
/// flag is what is checked, not the notification, which it would miss.
#[test]
fn waiting_after_finish_returns_at_once() {
- let slot = SizeUpgrade::default();
+ let queue = Arc::new(SizeWaitQueue::default());
+ let slot = Arc::new(SizeUpgrade::default());
slot.finish();
let started = Instant::now();
- slot.wait_for(|| false);
+ let rx = park(&queue, slot, || false, far_off());
+ rx.recv().expect("the waiter is answered");
assert!(started.elapsed() < Duration::from_millis(100));
}
- /// Every waiter is released, not just the first: one `ls -l` puts one
+ /// Every waiter is answered, not just the first: one `ls -l` puts one
/// waiter per file on the same folder.
#[test]
fn all_waiters_are_released() {
+ let queue = Arc::new(SizeWaitQueue::default());
let slot = Arc::new(SizeUpgrade::default());
let waiters: Vec<_> = (0..8)
- .map(|_| {
- let slot = slot.clone();
- std::thread::spawn(move || slot.wait_for(|| false))
- })
+ .map(|_| park(&queue, slot.clone(), || false, far_off()))
.collect();
std::thread::sleep(Duration::from_millis(20));
slot.finish();
- for w in waiters {
- w.join().expect("every waiter returns");
+ for rx in waiters {
+ rx.recv().expect("every waiter is answered");
}
}
- /// The timeout is the backstop: a batch that never finishes must not wedge
+ /// The deadline is the backstop: a batch that never finishes must not wedge
/// the caller's `stat` forever.
#[test]
- fn a_waiter_gives_up_at_the_timeout() {
- let slot = SizeUpgrade::default();
+ fn a_waiter_gives_up_at_the_deadline() {
+ let queue = Arc::new(SizeWaitQueue::default());
let started = Instant::now();
// Nothing will ever resolve or finish this.
- slot.wait_for(|| false);
- assert!(started.elapsed() >= SizeUpgrade::WAIT);
+ let rx = park(
+ &queue,
+ Arc::new(SizeUpgrade::default()),
+ || false,
+ Instant::now() + Duration::from_millis(120),
+ );
+ rx.recv().expect("the waiter is answered");
+ assert!(started.elapsed() >= Duration::from_millis(120));
+ }
+
+ /// The serving thread retires when nothing is parked, and a later park
+ /// starts a new one — otherwise the second `ls -l` of the session would
+ /// hang forever.
+ #[test]
+ fn the_queue_serves_again_after_it_has_drained() {
+ let queue = Arc::new(SizeWaitQueue::default());
+ for _ in 0..3 {
+ let rx = park(&queue, Arc::new(SizeUpgrade::default()), || true, far_off());
+ rx.recv().expect("the waiter is answered");
+ }
+ }
+
+ /// With no thread to serve them, waiters are answered on the spot rather
+ /// than never: a provisional size beats a `stat` that does not return.
+ #[test]
+ fn answer_all_releases_everything_parked() {
+ let queue = SizeWaitQueue::default();
+ let (tx, rx) = mpsc::channel();
+ for _ in 0..4 {
+ let tx = tx.clone();
+ queue.park(SizeWaiter {
+ slot: Arc::new(SizeUpgrade::default()),
+ resolved: Box::new(|| false),
+ deadline: far_off(),
+ respond: Box::new(move || {
+ let _ = tx.send(());
+ }),
+ });
+ }
+ drop(tx);
+ queue.answer_all();
+ assert_eq!(rx.into_iter().count(), 4);
}
}
diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs
index 64a2eed..302941c 100644
--- a/crates/pdfs-fuse/src/mount.rs
+++ b/crates/pdfs-fuse/src/mount.rs
@@ -374,7 +374,10 @@ pub fn mount(
})),
cache: Arc::new(cache),
readers: Arc::new(Mutex::new(HashMap::new())),
- stream_ring: Arc::new(Mutex::new(StreamRing::default())),
+ block_ring: Arc::new(Mutex::new(BlockRing::default())),
+ block_flight: Arc::new(BlockFlight::default()),
+ prefetch: Arc::new(Prefetch::default()),
+ prefetch_budget: Arc::new(tokio::sync::Semaphore::new(PREFETCH_BUDGET)),
workers: Arc::new(Workers::new(FUSE_WORKERS)?),
db,
shared_publication: Arc::new(Mutex::new(())),
@@ -396,6 +399,7 @@ pub fn mount(
no_thumbnail: Arc::new(Mutex::new(HashMap::new())),
quota: Arc::new(Mutex::new(None)),
size_upgrades: Arc::new(Mutex::new(HashMap::new())),
+ size_waiters: Arc::new(Default::default()),
notifier: Arc::new(OnceLock::new()),
session_live: Arc::new(AtomicBool::new(false)),
transfers: TransferRegistry::new(),
diff --git a/crates/pdfs-fuse/src/reads.rs b/crates/pdfs-fuse/src/reads.rs
index 71a1ca0..c0f24dd 100644
--- a/crates/pdfs-fuse/src/reads.rs
+++ b/crates/pdfs-fuse/src/reads.rs
@@ -10,18 +10,30 @@ use super::*;
/// usual.
pub(super) const STREAM_BYPASS_MIN: u64 = 256 * 1024 * 1024;
-/// How many bytes of streamed blocks the in-memory ring keeps. Bypassing the
-/// on-disk cache must not mean re-fetching: the kernel asks for a streamed file
-/// in reads far smaller than [`BLOCK_SIZE`], so without a ring every 128 KiB the
-/// player consumes would download and decrypt a whole 4 MiB block again. Sized
-/// for a handful of blocks per concurrently-streamed file.
-const STREAM_RING_BYTES: u64 = 128 * 1024 * 1024;
+/// How many bytes of decrypted blocks the in-memory ring keeps.
+///
+/// The kernel asks for a file in reads far smaller than [`BLOCK_SIZE`], so
+/// without a ring every 128 KiB it consumes costs a whole 4 MiB block again —
+/// downloaded and decrypted for a streaming file, or re-read off disk and
+/// re-allocated for a cached one (audit T3: 128 MiB of I/O to deliver 4 MiB).
+/// Sized for a handful of blocks per concurrently-read file. Bounded and only
+/// ever filled by active reads.
+const RING_BYTES: u64 = 128 * 1024 * 1024;
-/// Blocks fetched *past* the one a sequential streaming read needed, warmed into
-/// the ring in the background so the player's next read is already in memory
-/// instead of waiting a round-trip. Kept small so it cannot crowd out the
-/// in-flight block governor.
-const STREAM_READAHEAD: u64 = 4;
+/// Deepest read-ahead window, in blocks, for a read that has proved sequential.
+const PREFETCH_MAX: u64 = 8;
+
+/// Window a sequential read starts at, and returns to after a seek.
+const PREFETCH_MIN: u64 = 2;
+
+/// Blocks that may be prefetching at once, across every file. Prefetch takes a
+/// permit and gives up rather than queueing, so it can never sit in front of a
+/// demand read in the SDK's flat block semaphore.
+pub(super) const PREFETCH_BUDGET: usize = 8;
+
+/// Cap on tracked sequential streams. Only reads in progress matter, so the map
+/// is cleared wholesale rather than aged — losing the ramp costs one read.
+const MAX_PREFETCH_STREAMS: usize = 256;
/// How many [`RevisionReader`]s stay open at once.
///
@@ -76,35 +88,39 @@ pub(super) struct PendingOpen {
/// does; ids are never compared for order.
static NEXT_OPEN_ID: AtomicU64 = AtomicU64::new(0);
-/// Drive client, a Tokio handle to bridge the synchronous FUSE/socket threads
-/// to the async SDK, the inode bookkeeping, and the on-disk content cache.
-///
-/// Cheaply cloneable (every field is a handle/`Arc`), so the control-socket task
-/// gets its own copy while the FUSE session keeps another.
-/// In-memory LRU of blocks belonging to a file streaming past the on-disk cache
-/// (see [`STREAM_BYPASS_MIN`]). Bypassing the disk stops one film evicting the
-/// block LRU; it must not also mean the same 4 MiB block is downloaded once per
-/// 128 KiB the player reads out of it.
+/// In-memory LRU of decrypted 4 MiB blocks, in front of both block paths: the
+/// on-disk block cache, and the disk-bypassing stream of a large unpinned video
+/// (see [`STREAM_BYPASS_MIN`], which stops one film evicting the whole disk
+/// LRU). Either way the point is the same — the kernel reads a block out in ~32
+/// pieces, and none of them should cost the block again.
///
/// Validated by the same `(mtime, size)` pair the on-disk caches use: a node
/// whose tag no longer matches has its blocks dropped rather than served, so a
/// new revision can never be stitched together from the old one.
#[derive(Default)]
-pub(super) struct StreamRing {
+pub(super) struct BlockRing {
blocks: HashMap<(NodeUid, u64), Arc>>,
- /// Keys oldest-first, for eviction.
+ /// Keys least-recently-used first, for eviction.
order: VecDeque<(NodeUid, u64)>,
/// Per-node validity tag, `(mtime, size)`.
tags: HashMap,
bytes: u64,
}
-impl StreamRing {
+impl BlockRing {
fn get(&mut self, uid: &NodeUid, mtime: i64, size: u64, idx: u64) -> Option>> {
if self.tags.get(uid) != Some(&(mtime, size)) {
return None;
}
- self.blocks.get(&(uid.clone(), idx)).cloned()
+ let key = (uid.clone(), idx);
+ let hit = self.blocks.get(&key).cloned()?;
+ // Move to the young end: a block being read repeatedly (the whole reason
+ // this exists) must not age out under a sequential read passing through.
+ if let Some(at) = self.order.iter().position(|k| *k == key) {
+ self.order.remove(at);
+ self.order.push_back(key);
+ }
+ Some(hit)
}
fn insert(&mut self, uid: &NodeUid, mtime: i64, size: u64, idx: u64, bytes: Arc>) {
@@ -121,13 +137,18 @@ impl StreamRing {
self.bytes += bytes.len() as u64;
self.blocks.insert(key.clone(), bytes);
self.order.push_back(key);
- while self.bytes > STREAM_RING_BYTES {
+ while self.bytes > RING_BYTES {
let Some(victim) = self.order.pop_front() else {
break;
};
if let Some(dropped) = self.blocks.remove(&victim) {
self.bytes -= dropped.len() as u64;
}
+ // A node whose last block just aged out keeps no tag: otherwise the
+ // tag map is the one thing here that grows with every file ever read.
+ if !self.blocks.keys().any(|(u, _)| *u == victim.0) {
+ self.tags.remove(&victim.0);
+ }
}
}
@@ -147,6 +168,87 @@ impl StreamRing {
}
}
+/// The result of fetching one block, shared between everyone who wanted it.
+type BlockResult = Result>, Errno>;
+
+/// Identifies a block *of a revision*: node, validity tag, index.
+type BlockKey = (NodeUid, i64, u64, u64);
+
+/// A claim on a block key: who owns the fetch, and where its result is published.
+type Claim = (u64, tokio::sync::watch::Receiver>);
+
+/// Block fetches in flight, so a block is downloaded and decrypted once however
+/// many readers want it at that moment (audit T1).
+///
+/// With `FUSE_ASYNC_READ` the demand read and the kernel's read-ahead land in
+/// the same 4 MiB block for 31 of every 32 reads, and read-ahead of our own
+/// (see [`Core::prefetch`]) aims at blocks a demand read may be about to ask
+/// for. Without this each of them pays the full round-trip and decrypt and all
+/// but one result is discarded — roughly double the bandwidth and CPU in steady
+/// state. Same shape as [`ReaderSlot::Pending`] one level up.
+#[derive(Default)]
+pub(super) struct BlockFlight {
+ inflight: Mutex>,
+}
+
+/// Distinguishes attempts at the same key, so a leader only retires its own.
+static NEXT_FLIGHT_ID: AtomicU64 = AtomicU64::new(0);
+
+/// The leader's claim on a key. Retires it on drop — including when the task is
+/// cancelled mid-fetch, which is what keeps a dropped `JoinSet` from leaving a
+/// key nobody will ever publish.
+struct Flight {
+ flight: Arc,
+ key: BlockKey,
+ id: u64,
+}
+
+impl Drop for Flight {
+ fn drop(&mut self) {
+ let mut inflight = self.flight.inflight.lock();
+ if inflight
+ .get(&self.key)
+ .is_some_and(|(id, _)| *id == self.id)
+ {
+ inflight.remove(&self.key);
+ }
+ }
+}
+
+/// Where each file's reader is expected to read next, so sequential reads can be
+/// told from random ones and only the former prefetched (audit T2).
+#[derive(Default)]
+pub(super) struct Prefetch {
+ /// Per node: the block index a sequential reader should ask for next, and
+ /// the window depth it has earned.
+ streams: Mutex>,
+}
+
+impl Prefetch {
+ /// How far past `last` to read ahead for a read of blocks `[first, last]`,
+ /// and record where that read leaves the file's cursor.
+ ///
+ /// A read that starts where the last one stopped doubles the window (from
+ /// [`PREFETCH_MIN`], up to [`PREFETCH_MAX`]); anything else is a seek and
+ /// earns nothing, so random access — a database file, a linker — never pays
+ /// for speculative blocks it will not read. The first read of a file counts
+ /// as sequential: `(0, 0)` is both the initial cursor and block zero.
+ fn window(&self, uid: &NodeUid, first: u64, last: u64) -> u64 {
+ let mut streams = self.streams.lock();
+ if streams.len() > MAX_PREFETCH_STREAMS {
+ streams.clear();
+ }
+ let stream = streams.entry(uid.clone()).or_insert((0, 0));
+ let depth = if stream.0 == first {
+ (stream.1 * 2).clamp(PREFETCH_MIN, PREFETCH_MAX)
+ } else {
+ 0
+ };
+ *stream = (last + 1, depth);
+ depth
+ }
+}
+
impl Core {
/// An open [`RevisionReader`] for `uid`, reusing the cached one when it is
/// still valid for `(mtime, fsize)` and opening a fresh one otherwise.
@@ -304,6 +406,10 @@ impl Core {
/// revision is no longer the truth.
pub(super) fn evict_reader(&self, uid: &NodeUid) {
self.readers.lock().remove(uid);
+ // The ring holds decrypted bytes of the same revision, and an eviction is
+ // exactly the statement that those are no longer the file's content.
+ self.block_ring.lock().drop_node(uid);
+ self.prefetch.streams.lock().remove(uid);
}
/// Serve bytes `[offset, offset + len)` of `uid`'s active revision, hitting
@@ -389,20 +495,141 @@ impl Core {
Ok(out)
}
- /// Warm the [`STREAM_READAHEAD`] blocks after `last` into the ring, in the
- /// background, so a player reading forward finds its next block already in
- /// memory rather than paying a fetch + decrypt round-trip at each boundary.
+ /// Fetch block `bidx`, joining a fetch already in flight for it rather than
+ /// issuing a second one, and publish the result to everyone who joined.
///
- /// Fire-and-forget: a failure here is not an error, it only means the read
- /// that needs those bytes will fetch them itself.
- fn stream_readahead(&self, uid: &NodeUid, mtime: i64, fsize: u64, last: u64) {
- let last_block = (fsize.saturating_sub(1)) / BLOCK_SIZE;
- let wanted: Vec = ((last + 1)..=(last + STREAM_READAHEAD).min(last_block))
+ /// See [`BlockFlight`]. A follower whose leader disappears (a panic, or a
+ /// cancelled `JoinSet`) fetches the block itself rather than failing — the
+ /// leader's claim is gone by then, so this cannot recurse.
+ async fn fetch_block(
+ &self,
+ reader: &Arc,
+ uid: &NodeUid,
+ mtime: i64,
+ fsize: u64,
+ bidx: u64,
+ cache_blocks: bool,
+ ) -> BlockResult {
+ let key: BlockKey = (uid.clone(), mtime, fsize, bidx);
+ let claim = {
+ let mut inflight = self.block_flight.inflight.lock();
+ match inflight.get(&key) {
+ Some((_, rx)) => Err(rx.clone()),
+ None => {
+ let id = NEXT_FLIGHT_ID.fetch_add(1, Ordering::Relaxed);
+ let (tx, rx) = tokio::sync::watch::channel(None);
+ inflight.insert(key.clone(), (id, rx));
+ Ok((
+ tx,
+ Flight {
+ flight: self.block_flight.clone(),
+ key: key.clone(),
+ id,
+ },
+ ))
+ }
+ }
+ };
+ let (tx, flight) = match claim {
+ Ok(lead) => lead,
+ Err(mut rx) => {
+ loop {
+ if let Some(result) = rx.borrow_and_update().clone() {
+ return result;
+ }
+ if rx.changed().await.is_err() {
+ break;
+ }
+ }
+ return self
+ .read_block(reader, uid, mtime, fsize, bidx, cache_blocks)
+ .await;
+ }
+ };
+ let result = self
+ .read_block(reader, uid, mtime, fsize, bidx, cache_blocks)
+ .await;
+ // Retire the claim before publishing, so a racer that wakes and re-enters
+ // finds no stale entry to join.
+ drop(flight);
+ let _ = tx.send(Some(result.clone()));
+ result
+ }
+
+ /// Download and decrypt one block, then put it where the next reader will
+ /// find it: the in-memory ring always, the on-disk block cache too unless
+ /// this is a disk-bypassing stream.
+ ///
+ /// The disk store is handed to a blocking thread — it is a 4 MiB write plus
+ /// a cache-index update, and neither the reactor nor the kernel read waiting
+ /// on these bytes should be held up by it.
+ async fn read_block(
+ &self,
+ reader: &Arc,
+ uid: &NodeUid,
+ mtime: i64,
+ fsize: u64,
+ bidx: u64,
+ cache_blocks: bool,
+ ) -> BlockResult {
+ let bstart = bidx * BLOCK_SIZE;
+ let blen = BLOCK_SIZE.min(fsize - bstart);
+ let bytes = reader.read_at(bstart, blen).await.map_err(|e| {
+ warn!(%uid, bstart, blen, error = %e, "block read failed");
+ Errno::EIO
+ })?;
+ let bytes = Arc::new(bytes);
+ if cache_blocks {
+ let cache = self.cache.clone();
+ let uid = uid.clone();
+ let bytes = bytes.clone();
+ tokio::task::spawn_blocking(move || {
+ let _ = cache.store_block(&uid, mtime, fsize, bidx, &bytes);
+ });
+ }
+ self.block_ring
+ .lock()
+ .insert(uid, mtime, fsize, bidx, bytes.clone());
+ Ok(bytes)
+ }
+
+ /// Warm the blocks after a sequential read into the caches, in the
+ /// background, so the next read finds them instead of paying a round-trip.
+ ///
+ /// A read that continues where the last one stopped doubles the window, up
+ /// to [`PREFETCH_MAX`]; a seek drops it back to nothing and one more
+ /// sequential read is needed to earn it again. Without this every read is a
+ /// serial fetch/serve/stall chain: on a 100 ms link that ceilings a file at
+ /// ~24 MB/s of the ~125 MB/s available, and the SDK's block concurrency is
+ /// never exercised at all.
+ ///
+ /// Fire-and-forget, and permit-bounded ([`PREFETCH_BUDGET`]) so speculative
+ /// work never sits in front of a read someone is waiting on. A failure here
+ /// is not an error: the read that needs those bytes will fetch them itself.
+ fn prefetch_ahead(
+ &self,
+ uid: &NodeUid,
+ mtime: i64,
+ fsize: u64,
+ first: u64,
+ last: u64,
+ cache_blocks: bool,
+ ) {
+ let depth = self.prefetch.window(uid, first, last);
+ if depth == 0 {
+ return;
+ }
+ let last_block = fsize.saturating_sub(1) / BLOCK_SIZE;
+ let wanted: Vec = ((last + 1)..=(last + depth).min(last_block))
.filter(|&bidx| {
- self.stream_ring
+ // Blocks already in memory need nothing; blocks already on disk
+ // are cheap enough to serve that spending a round-trip to move
+ // them into the ring would be the wrong trade.
+ self.block_ring
.lock()
.get(uid, mtime, fsize, bidx)
.is_none()
+ && !(cache_blocks && self.cache.has_block(uid, mtime, fsize, bidx))
})
.collect();
if wanted.is_empty() {
@@ -416,26 +643,25 @@ impl Core {
};
// Concurrently, for the same reason the demand path fetches its
// misses concurrently: serially awaited blocks cost the sum of their
- // round-trips, and the player catches up with the readahead before it
- // finishes. The SDK's in-flight block governor bounds the fan-out.
+ // round-trips, and the reader catches up with the window before it
+ // finishes.
let mut set = tokio::task::JoinSet::new();
for bidx in wanted {
- let reader = reader.clone();
- let bstart = bidx * BLOCK_SIZE;
- let blen = BLOCK_SIZE.min(fsize - bstart);
- set.spawn(async move { (bidx, reader.read_at(bstart, blen).await) });
- }
- while let Some(joined) = set.join_next().await {
- let Ok((bidx, read)) = joined else { continue };
- match read {
- Ok(bytes) => {
- core.stream_ring
- .lock()
- .insert(&uid, mtime, fsize, bidx, Arc::new(bytes))
+ let Ok(permit) = core.prefetch_budget.clone().try_acquire_owned() else {
+ break; // the budget is spent; the demand path can have it
+ };
+ let (core, reader, uid) = (core.clone(), reader.clone(), uid.clone());
+ set.spawn(async move {
+ let _permit = permit;
+ if let Err(error) = core
+ .fetch_block(&reader, &uid, mtime, fsize, bidx, cache_blocks)
+ .await
+ {
+ debug!(%uid, bidx, ?error, "prefetch failed");
}
- Err(e) => debug!(%uid, bidx, error = %e, "stream readahead failed"),
- }
+ });
}
+ while set.join_next().await.is_some() {}
});
}
@@ -473,17 +699,23 @@ impl Core {
let mut blocks: Vec>>> = Vec::with_capacity((last - first + 1) as usize);
let mut misses: Vec = Vec::new();
for bidx in first..=last {
- // A streaming read keeps its blocks only in memory, so the ring is
- // the only cache it has — check it before the disk (which will not
- // hold them) and before the network (which would refetch the same
- // 4 MiB block for every 128 KiB the player consumes).
- let hit = if cache_blocks {
- self.cache
- .cached_block(uid, mtime, fsize, bidx)
- .map(Arc::new)
- } else {
- self.stream_ring.lock().get(uid, mtime, fsize, bidx)
- };
+ // The in-memory ring comes first on both paths: a streaming read has
+ // no other cache at all, and a cached one would otherwise pay two
+ // opens, a JSON parse and a 4 MiB read+alloc for every 128 KiB the
+ // kernel asks for (audit T3).
+ let hit = self.block_ring.lock().get(uid, mtime, fsize, bidx);
+ let hit = hit.or_else(|| {
+ if !cache_blocks {
+ return None;
+ }
+ let bytes = Arc::new(self.cache.cached_block(uid, mtime, fsize, bidx)?);
+ // Promote it, so the other ~31 reads of this block are answered
+ // from memory.
+ self.block_ring
+ .lock()
+ .insert(uid, mtime, fsize, bidx, bytes.clone());
+ Some(bytes)
+ });
match hit {
Some(b) => blocks.push(Some(b)),
None => {
@@ -502,19 +734,11 @@ impl Core {
let mut set = tokio::task::JoinSet::new();
for &bidx in &misses {
- let reader = reader.clone();
- let uid = uid.clone();
- let bstart = bidx * BLOCK_SIZE;
- let blen = BLOCK_SIZE.min(fsize - bstart);
+ let (core, reader, uid) = (self.clone(), reader.clone(), uid.clone());
set.spawn(async move {
- reader
- .read_at(bstart, blen)
+ core.fetch_block(&reader, &uid, mtime, fsize, bidx, cache_blocks)
.await
.map(|bytes| (bidx, bytes))
- .map_err(|e| {
- warn!(%uid, bstart, blen, error = %e, "block read failed");
- Errno::EIO
- })
});
}
let mut out = Vec::with_capacity(misses.len());
@@ -525,23 +749,12 @@ impl Core {
Ok::<_, Errno>(out)
})?;
for (bidx, bytes) in fetched {
- // A streaming read (large unpinned video) skips the on-disk cache
- // so it can't evict the rest of it; it keeps the block in the
- // in-memory ring instead, which is bounded and dies with the read.
- let bytes = Arc::new(bytes);
- if cache_blocks {
- let _ = self.cache.store_block(uid, mtime, fsize, bidx, &bytes);
- } else {
- self.stream_ring
- .lock()
- .insert(uid, mtime, fsize, bidx, bytes.clone());
- }
blocks[(bidx - first) as usize] = Some(bytes);
}
- if !cache_blocks {
- self.stream_readahead(uid, mtime, fsize, last);
- }
}
+ // Also after a wholly-cached read: that is what keeps the window sliding
+ // ahead of a sequential reader instead of stalling once it catches up.
+ self.prefetch_ahead(uid, mtime, fsize, first, last, cache_blocks);
for (i, block) in blocks.into_iter().enumerate() {
let bidx = first + i as u64;
@@ -605,6 +818,106 @@ impl Core {
Ok(out)
}
+ /// The largest amount [`Core::fill_gaps_cached`] will copy out of the local
+ /// caches while a `release(2)` is in progress.
+ ///
+ /// The point of the cached fill is that editing a file the mount already
+ /// holds costs nothing extra; it is not to reconstruct arbitrarily large
+ /// files on the dispatch loop. Past this budget the blob stays incomplete
+ /// and the drain finishes it, which is the same work in a place where it
+ /// blocks nobody.
+ const CACHED_FILL_BUDGET: u64 = 64 * 1024 * 1024;
+
+ /// Bytes `[offset, offset + len)` of `uid`'s base revision, but only if the
+ /// local caches can answer in full — the whole-file blob first, then every
+ /// overlapping block. `None` the moment anything would have to be fetched.
+ fn cached_range(
+ &self,
+ uid: &NodeUid,
+ mtime: i64,
+ fsize: u64,
+ offset: u64,
+ len: u64,
+ ) -> Option> {
+ if len == 0 {
+ return Some(Vec::new());
+ }
+ if let Some(bytes) = self.cache.read_range(uid, mtime, fsize, offset, len) {
+ return Some(bytes);
+ }
+ let end = offset.checked_add(len)?.min(fsize);
+ if offset >= end {
+ return None;
+ }
+ let mut out = Vec::with_capacity((end - offset) as usize);
+ for bidx in (offset / BLOCK_SIZE)..=((end - 1) / BLOCK_SIZE) {
+ let ring = self.block_ring.lock().get(uid, mtime, fsize, bidx);
+ let block = match ring {
+ Some(bytes) => bytes,
+ None => Arc::new(self.cache.cached_block(uid, mtime, fsize, bidx)?),
+ };
+ let bstart = bidx * BLOCK_SIZE;
+ let s = (offset.max(bstart) - bstart) as usize;
+ let e = (end.min(bstart + block.len() as u64) - bstart) as usize;
+ if s >= e {
+ return None;
+ }
+ out.extend_from_slice(&block[s..e]);
+ }
+ (out.len() as u64 == end - offset).then_some(out)
+ }
+
+ /// Fill what [`Core::fill_gaps`] would, but only from bytes already on this
+ /// machine, and only up to [`Core::CACHED_FILL_BUDGET`]. Ranges past the
+ /// base are zeros the caller's `set_len` already wrote, so they count as
+ /// filled. `written` is extended with everything that ends up resolved.
+ ///
+ /// This is the release-time half of gap filling. Doing the network half
+ /// there as well is what froze the whole mount for the length of a download
+ /// (audit F1): the closing file's untouched ranges are allowed to stay
+ /// unresolved in the staged blob — `StagedWrite::complete` is false, reads
+ /// resolve the gaps against the base, and `drain_revision` fills them on the
+ /// drain thread before it uploads. What is left here is the case where
+ /// filling is free, which is also the common one: a file that was read
+ /// before it was edited.
+ pub(super) fn fill_gaps_cached(
+ &self,
+ uid: &NodeUid,
+ file: &File,
+ len: u64,
+ base_mtime: i64,
+ base_size: u64,
+ written: &mut Intervals,
+ ) {
+ let mut budget = Self::CACHED_FILL_BUDGET;
+ for (s, e, authored) in written.clone().segments(0, len) {
+ if authored {
+ continue;
+ }
+ let bend = e.min(base_size);
+ if s < bend {
+ let want = bend - s;
+ if want > budget {
+ continue;
+ }
+ let Some(bytes) = self.cached_range(uid, base_mtime, base_size, s, want) else {
+ continue;
+ };
+ if let Err(error) = file.write_all_at(&bytes, s) {
+ warn!(%uid, %error, "cached gap-fill write failed");
+ continue;
+ }
+ budget -= want;
+ written.add(s, bend);
+ }
+ // Past the base there is nothing to fetch: the resize zeroed it, so
+ // those bytes are already the file's content.
+ if e > base_size {
+ written.add(s.max(base_size), e);
+ }
+ }
+ }
+
/// Fill every unauthored range of a scratch/staged file that overlaps its
/// base with the base's bytes, so the file becomes the complete new content.
///
@@ -643,3 +956,145 @@ impl Core {
Ok(())
}
}
+
+#[cfg(test)]
+mod block_ring_tests {
+ use super::*;
+
+ fn uid(link: &str) -> NodeUid {
+ NodeUid::new(VolumeId::from("vol"), LinkId::from(link))
+ }
+
+ /// One block's worth of bytes, so byte accounting is exercised at the scale
+ /// the ring actually sees.
+ fn block(byte: u8) -> Arc> {
+ Arc::new(vec![byte; BLOCK_SIZE as usize])
+ }
+
+ /// How many whole blocks fit before the ring starts evicting.
+ fn capacity() -> u64 {
+ RING_BYTES / BLOCK_SIZE
+ }
+
+ #[test]
+ fn a_stored_block_is_served_back() {
+ let mut ring = BlockRing::default();
+ let node = uid("a");
+ ring.insert(&node, 7, 100, 3, block(0xab));
+ assert_eq!(ring.get(&node, 7, 100, 3).map(|b| b[0]), Some(0xab));
+ }
+
+ #[test]
+ fn a_block_of_another_revision_is_never_served() {
+ let mut ring = BlockRing::default();
+ let node = uid("a");
+ ring.insert(&node, 7, 100, 0, block(1));
+ // Same node, newer revision: the tag moved, so the old bytes go.
+ assert!(ring.get(&node, 8, 100, 0).is_none());
+ assert!(ring.get(&node, 7, 101, 0).is_none());
+ }
+
+ #[test]
+ fn a_new_revision_drops_what_the_old_one_left() {
+ let mut ring = BlockRing::default();
+ let node = uid("a");
+ ring.insert(&node, 7, 100, 0, block(1));
+ ring.insert(&node, 8, 100, 0, block(2));
+ assert_eq!(ring.get(&node, 8, 100, 0).map(|b| b[0]), Some(2));
+ assert_eq!(ring.bytes, BLOCK_SIZE);
+ }
+
+ #[test]
+ fn the_ring_evicts_the_least_recently_used_block() {
+ let mut ring = BlockRing::default();
+ let node = uid("a");
+ for idx in 0..capacity() {
+ ring.insert(&node, 1, u64::MAX, idx, block(idx as u8));
+ }
+ // Re-reading block 0 makes block 1 the oldest instead.
+ assert!(ring.get(&node, 1, u64::MAX, 0).is_some());
+ ring.insert(&node, 1, u64::MAX, capacity(), block(0xff));
+ assert!(ring.get(&node, 1, u64::MAX, 0).is_some());
+ assert!(ring.get(&node, 1, u64::MAX, 1).is_none());
+ assert_eq!(ring.bytes, RING_BYTES);
+ }
+
+ #[test]
+ fn a_node_whose_last_block_ages_out_keeps_no_tag() {
+ let mut ring = BlockRing::default();
+ let (old, new) = (uid("old"), uid("new"));
+ ring.insert(&old, 1, u64::MAX, 0, block(1));
+ for idx in 0..capacity() {
+ ring.insert(&new, 1, u64::MAX, idx, block(2));
+ }
+ assert!(!ring.tags.contains_key(&old));
+ assert!(ring.tags.contains_key(&new));
+ }
+
+ #[test]
+ fn dropping_a_node_frees_only_its_own_bytes() {
+ let mut ring = BlockRing::default();
+ let (a, b) = (uid("a"), uid("b"));
+ ring.insert(&a, 1, 100, 0, block(1));
+ ring.insert(&b, 1, 100, 0, block(2));
+ ring.drop_node(&a);
+ assert!(ring.get(&a, 1, 100, 0).is_none());
+ assert_eq!(ring.get(&b, 1, 100, 0).map(|x| x[0]), Some(2));
+ assert_eq!(ring.bytes, BLOCK_SIZE);
+ }
+}
+
+#[cfg(test)]
+mod prefetch_tests {
+ use super::*;
+
+ fn uid(link: &str) -> NodeUid {
+ NodeUid::new(VolumeId::from("vol"), LinkId::from(link))
+ }
+
+ #[test]
+ fn a_sequential_reader_ramps_up_to_the_cap() {
+ let prefetch = Prefetch::default();
+ let node = uid("a");
+ let depths: Vec = (0..6)
+ .map(|next| prefetch.window(&node, next, next))
+ .collect();
+ assert_eq!(depths, vec![2, 4, 8, 8, 8, 8]);
+ }
+
+ #[test]
+ fn a_seek_forfeits_the_window() {
+ let prefetch = Prefetch::default();
+ let node = uid("a");
+ assert_eq!(prefetch.window(&node, 0, 0), PREFETCH_MIN);
+ assert_eq!(prefetch.window(&node, 40, 40), 0);
+ // And has to be earned again from the bottom.
+ assert_eq!(prefetch.window(&node, 41, 41), PREFETCH_MIN);
+ }
+
+ #[test]
+ fn a_multi_block_read_advances_the_cursor_past_all_of_it() {
+ let prefetch = Prefetch::default();
+ let node = uid("a");
+ assert_eq!(prefetch.window(&node, 0, 3), PREFETCH_MIN);
+ assert_eq!(prefetch.window(&node, 4, 7), 4);
+ }
+
+ #[test]
+ fn files_are_tracked_independently() {
+ let prefetch = Prefetch::default();
+ let (a, b) = (uid("a"), uid("b"));
+ assert_eq!(prefetch.window(&a, 0, 0), PREFETCH_MIN);
+ assert_eq!(prefetch.window(&b, 9, 9), 0);
+ assert_eq!(prefetch.window(&a, 1, 1), 4);
+ }
+
+ #[test]
+ fn the_stream_map_cannot_grow_without_bound() {
+ let prefetch = Prefetch::default();
+ for n in 0..=MAX_PREFETCH_STREAMS + 1 {
+ prefetch.window(&uid(&n.to_string()), 0, 0);
+ }
+ assert!(prefetch.streams.lock().len() <= MAX_PREFETCH_STREAMS + 1);
+ }
+}