mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-04 00:47:00 +01:00
nvrl
9089c8d13c
A copy inside the mount went through the kernel's read/write loop, one FUSE round trip per page, because copy_file_range answered ENOSYS. The daemon now copies a range whose source bytes are local (an open write handle, a queued write, the cache) into the destination's scratch file, the same way write stages them. A range it would have to download gets EOPNOTSUPP, which makes the kernel copy that one call itself and keeps the operation enabled. The copy still uploads: Drive cannot copy into a file cp has already created. In the simulation, a warm cp of a 256 MiB file went from about 100 MiB/s to about 450 MiB/s (debug build).
1575 lines
65 KiB
Rust
1575 lines
65 KiB
Rust
//! Revision-reader cache and streamed range-read subsystem.
|
|
|
|
use super::*;
|
|
|
|
/// A read of an unpinned video at least this large streams *without* persisting
|
|
/// its blocks to the on-disk cache. Playing a 2 GB film would otherwise pour it
|
|
/// through the block LRU and evict everything else the user actually wants kept —
|
|
/// and it re-streams cheaply enough on a rewatch that keeping it was never worth
|
|
/// that. Pinned videos (kept offline on purpose) and anything smaller cache as
|
|
/// usual.
|
|
pub(super) const STREAM_BYPASS_MIN: u64 = 256 * 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;
|
|
|
|
/// 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;
|
|
|
|
/// Largest file [`Core::repair_block`] will pull down whole to work around a
|
|
/// revision whose block table understates it. Past this a single read would cost
|
|
/// gigabytes of download, so the read fails instead and says to pin the file.
|
|
pub(super) const REPAIR_MAX: u64 = 512 * 1024 * 1024;
|
|
|
|
/// 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;
|
|
|
|
/// How long one read may wait on the network before it fails with `EIO`.
|
|
///
|
|
/// The SDK bounds each request, not a read: a block that times out is retried,
|
|
/// and a read can need several blocks, a block table and a repair. On a link
|
|
/// that stalled without dropping, one `read` held a FUSE worker for 34 minutes
|
|
/// and the application behind it with it (`docs/BUGS.md` B107). An error the
|
|
/// caller can retry beats a process stuck in `D` state.
|
|
pub(super) const READ_FETCH_TIMEOUT: Duration = Duration::from_secs(120);
|
|
|
|
/// How long a read of blocks nothing has cached waits for a lost link to come
|
|
/// back before failing. Long enough to ride out a Wi-Fi blip, short enough
|
|
/// that a program reading while the laptop is really offline gets its `EIO`
|
|
/// before the user gives up on it.
|
|
const READ_OFFLINE_WAIT: Duration = Duration::from_secs(30);
|
|
|
|
/// 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.
|
|
///
|
|
/// A reader holds its revision's content key and block table — a few KB even for
|
|
/// a large file, so this is bounded for tidiness and staleness rather than
|
|
/// memory. Evicted least-recently-used; a dropped reader costs one re-open
|
|
/// (two API calls and a node-key unlock) the next time that file is read.
|
|
const MAX_OPEN_READERS: usize = 64;
|
|
|
|
/// 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
|
|
/// a reader is only reusable while the node still reports the same
|
|
/// `(mtime, size)` — the same validity pair the content cache uses (a new
|
|
/// revision bumps mtime). On a mismatch the reader is dropped and reopened.
|
|
pub(super) struct CachedReader {
|
|
reader: Arc<dyn RevisionRead>,
|
|
mtime: i64,
|
|
size: u64,
|
|
/// For LRU eviction.
|
|
last_used: Instant,
|
|
}
|
|
|
|
/// What the `readers` map holds for a node: an open reader, or the fact that
|
|
/// someone is already opening one.
|
|
///
|
|
/// The `Pending` arm is what makes the map single-flight rather than merely a
|
|
/// cache. Opening resolves link details, ancestor keys, an S2K node-key unlock
|
|
/// and the block table — five round-trips — and read-ahead on a cold file puts
|
|
/// several workers into the open at the same moment, so without this each
|
|
/// racer would redo all of it and discard every result but one.
|
|
pub(super) enum ReaderSlot {
|
|
Ready(CachedReader),
|
|
Pending(PendingOpen),
|
|
}
|
|
|
|
/// An `open_revision` in flight. Racers clone `rx` and await the leader's result
|
|
/// instead of opening their own.
|
|
pub(super) struct PendingOpen {
|
|
/// Identifies this attempt so the leader can tell "my slot is still there"
|
|
/// from "someone evicted or replaced it while I was on the network", and
|
|
/// only publish its reader in the first case.
|
|
id: u64,
|
|
/// What the leader is opening against. A racer wanting a different revision
|
|
/// must not join this open — it would get a reader for the wrong bytes.
|
|
mtime: i64,
|
|
size: u64,
|
|
rx: tokio::sync::watch::Receiver<Option<Result<Arc<dyn RevisionRead>, Errno>>>,
|
|
}
|
|
|
|
/// Distinguishes `PendingOpen`s. Only uniqueness matters, so a plain counter
|
|
/// does; ids are never compared for order.
|
|
static NEXT_OPEN_ID: AtomicU64 = AtomicU64::new(0);
|
|
|
|
/// 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.
|
|
/// A ring entry's identity: the node, and the plaintext range the block covers.
|
|
///
|
|
/// The range, rather than the block index, because the index alone stopped
|
|
/// naming one range when the geometry started coming from the revision — a
|
|
/// block cached under the [`BlockGeometry::uniform`] fallback and the block at
|
|
/// the same index under the real geometry are different bytes
|
|
/// (`docs/BUGS.md` B85). A range is the same identity under either plan, so a
|
|
/// re-planned read misses rather than being served the wrong bytes.
|
|
type RingKey = (NodeUid, BlockSpan);
|
|
|
|
#[derive(Default)]
|
|
pub(super) struct BlockRing {
|
|
blocks: HashMap<RingKey, Arc<Vec<u8>>>,
|
|
/// Keys least-recently-used first, for eviction.
|
|
order: VecDeque<RingKey>,
|
|
/// Per-node validity tag, `(mtime, size)`.
|
|
tags: HashMap<NodeUid, (i64, u64)>,
|
|
bytes: u64,
|
|
}
|
|
|
|
impl BlockRing {
|
|
fn get(
|
|
&mut self,
|
|
uid: &NodeUid,
|
|
mtime: i64,
|
|
size: u64,
|
|
span: BlockSpan,
|
|
) -> Option<Arc<Vec<u8>>> {
|
|
if self.tags.get(uid) != Some(&(mtime, size)) {
|
|
return None;
|
|
}
|
|
let key = (uid.clone(), span);
|
|
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,
|
|
span: BlockSpan,
|
|
bytes: Arc<Vec<u8>>,
|
|
) {
|
|
if self.tags.get(uid) != Some(&(mtime, size)) {
|
|
// Revision changed under us (or first sight): anything held for this
|
|
// node describes the old one.
|
|
self.drop_node(uid);
|
|
self.tags.insert(uid.clone(), (mtime, size));
|
|
}
|
|
let key = (uid.clone(), span);
|
|
if self.blocks.contains_key(&key) {
|
|
return;
|
|
}
|
|
self.bytes += bytes.len() as u64;
|
|
self.blocks.insert(key.clone(), bytes);
|
|
self.order.push_back(key);
|
|
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);
|
|
}
|
|
}
|
|
}
|
|
|
|
fn drop_node(&mut self, uid: &NodeUid) {
|
|
self.order.retain(|(u, _)| u != uid);
|
|
let mut freed = 0u64;
|
|
self.blocks.retain(|(u, _), bytes| {
|
|
if u == uid {
|
|
freed += bytes.len() as u64;
|
|
false
|
|
} else {
|
|
true
|
|
}
|
|
});
|
|
self.bytes -= freed;
|
|
self.tags.remove(uid);
|
|
}
|
|
}
|
|
|
|
/// The result of fetching one block, shared between everyone who wanted it.
|
|
type BlockResult = Result<Arc<Vec<u8>>, Errno>;
|
|
|
|
/// Identifies a block *of a revision*: node, validity tag, and the plaintext
|
|
/// range the block covers — see [`RingKey`] for why it is the range and not the
|
|
/// index.
|
|
type BlockKey = (NodeUid, i64, u64, BlockSpan);
|
|
|
|
/// A claim on a block key: who owns the fetch, and where its result is published.
|
|
type Claim = (u64, tokio::sync::watch::Receiver<Option<BlockResult>>);
|
|
|
|
/// 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<HashMap<BlockKey, Claim>>,
|
|
}
|
|
|
|
/// 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<BlockFlight>,
|
|
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<HashMap<NodeUid, (u64, u64)>>,
|
|
}
|
|
|
|
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.
|
|
///
|
|
/// Opening resolves the file's link details, ancestor keys, node key (an S2K
|
|
/// unlock) and block table. Doing that once per file rather than once per
|
|
/// block is the whole point: a cold 100 MB read is 25 block misses, which
|
|
/// used to mean 25 full resolutions (50 API calls, 25 unlocks) and now means
|
|
/// one.
|
|
///
|
|
/// Racing reads of one cold file open it **once**: the first caller claims
|
|
/// the slot and the rest await its result. That matters now that `read` is
|
|
/// served off the dispatch loop, because read-ahead on a first touch issues
|
|
/// several parallel reads of the same file as a matter of course.
|
|
async fn revision_reader(
|
|
&self,
|
|
uid: &NodeUid,
|
|
mtime: i64,
|
|
fsize: u64,
|
|
) -> Result<Arc<dyn RevisionRead>, Errno> {
|
|
loop {
|
|
// Decide under the lock, act outside it — network I/O must never run
|
|
// under a std `Mutex`, and an await must never hold one.
|
|
enum Act {
|
|
Hit(Arc<dyn RevisionRead>),
|
|
Join(tokio::sync::watch::Receiver<Option<Result<Arc<dyn RevisionRead>, Errno>>>),
|
|
Lead(
|
|
u64,
|
|
tokio::sync::watch::Sender<Option<Result<Arc<dyn RevisionRead>, Errno>>>,
|
|
),
|
|
}
|
|
let act = {
|
|
let mut readers = self.readers.lock();
|
|
match readers.get_mut(uid) {
|
|
Some(ReaderSlot::Ready(entry))
|
|
if entry.mtime == mtime && entry.size == fsize =>
|
|
{
|
|
entry.last_used = Instant::now();
|
|
Act::Hit(entry.reader.clone())
|
|
}
|
|
Some(ReaderSlot::Pending(p)) if p.mtime == mtime && p.size == fsize => {
|
|
Act::Join(p.rx.clone())
|
|
}
|
|
// Either nothing here, or something for a revision we no
|
|
// longer want (a stale reader, or an open for one). Ours
|
|
// replaces it: the newer `(mtime, size)` is the truth.
|
|
_ => {
|
|
let id = NEXT_OPEN_ID.fetch_add(1, Ordering::Relaxed);
|
|
let (tx, rx) = tokio::sync::watch::channel(None);
|
|
readers.insert(
|
|
uid.clone(),
|
|
ReaderSlot::Pending(PendingOpen {
|
|
id,
|
|
mtime,
|
|
size: fsize,
|
|
rx,
|
|
}),
|
|
);
|
|
Act::Lead(id, tx)
|
|
}
|
|
}
|
|
};
|
|
|
|
match act {
|
|
Act::Hit(reader) => return Ok(reader),
|
|
Act::Join(mut rx) => {
|
|
// `changed()` errors only if the leader dropped its sender
|
|
// without publishing — it panicked. Retry from the top
|
|
// rather than fail: the slot it left is gone, so this pass
|
|
// leads its own open.
|
|
loop {
|
|
if let Some(result) = rx.borrow_and_update().clone() {
|
|
return result;
|
|
}
|
|
if rx.changed().await.is_err() {
|
|
break;
|
|
}
|
|
}
|
|
continue;
|
|
}
|
|
Act::Lead(id, tx) => {
|
|
debug!(%uid, mtime, fsize, "opening revision");
|
|
let result = match self.drive.open_revision(uid).await {
|
|
Ok(reader) => Ok(reader),
|
|
Err(e) => {
|
|
self.lost_link(&e, "open a revision");
|
|
warn!(%uid, error = %e, "open_revision failed");
|
|
Err(Errno::EIO)
|
|
}
|
|
};
|
|
{
|
|
let mut readers = self.readers.lock();
|
|
// Publish only into the slot we claimed. If it is gone
|
|
// or belongs to a later attempt, an eviction or a newer
|
|
// revision overtook us: our reader still answers the
|
|
// callers waiting on it, but it must not be cached.
|
|
let ours = matches!(
|
|
readers.get(uid),
|
|
Some(ReaderSlot::Pending(p)) if p.id == id
|
|
);
|
|
if ours {
|
|
match &result {
|
|
Ok(reader) => {
|
|
readers.insert(
|
|
uid.clone(),
|
|
ReaderSlot::Ready(CachedReader {
|
|
reader: reader.clone(),
|
|
mtime,
|
|
size: fsize,
|
|
last_used: Instant::now(),
|
|
}),
|
|
);
|
|
}
|
|
// Never cache a failure: the next read retries.
|
|
Err(_) => {
|
|
readers.remove(uid);
|
|
}
|
|
}
|
|
}
|
|
Self::trim_readers(&mut readers);
|
|
}
|
|
// After the map is updated, so a racer that wakes and
|
|
// re-enters finds the finished slot rather than racing us.
|
|
let _ = tx.send(Some(result.clone()));
|
|
return result;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Bound the reader map: drop least-recently-used entries until it is back
|
|
/// at [`MAX_OPEN_READERS`]. Only `Ready` slots are eviction candidates — a
|
|
/// `Pending` has callers waiting on it and no reader to drop yet.
|
|
fn trim_readers(readers: &mut HashMap<NodeUid, ReaderSlot>) {
|
|
while readers.len() > MAX_OPEN_READERS {
|
|
let victim = readers
|
|
.iter()
|
|
.filter_map(|(uid, slot)| match slot {
|
|
ReaderSlot::Ready(entry) => Some((uid, entry.last_used)),
|
|
ReaderSlot::Pending(_) => None,
|
|
})
|
|
.min_by_key(|(_, last_used)| *last_used)
|
|
.map(|(uid, _)| uid.clone());
|
|
let Some(victim) = victim else {
|
|
break; // every slot is an open in flight; nothing to drop
|
|
};
|
|
readers.remove(&victim);
|
|
}
|
|
}
|
|
|
|
/// Drop any open reader for `uid`, so the next read reopens against the
|
|
/// current revision. Called wherever cached content is evicted.
|
|
///
|
|
/// Drops an in-flight open too: its leader finds the slot gone and declines
|
|
/// to cache what it opened, which is what we want — the eviction says that
|
|
/// 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);
|
|
}
|
|
|
|
/// How this revision's plaintext is divided into content blocks.
|
|
///
|
|
/// The revision's own geometry when a previous read has learned and recorded
|
|
/// it, and the [`BLOCK_SIZE`] assumption when it has not — which is every
|
|
/// first read, because learning it means opening a `RevisionReader`, and
|
|
/// opening one per read is precisely the per-file key derivation the block
|
|
/// cache exists to avoid (`docs/BUGS.md` B12, B85).
|
|
///
|
|
/// Being wrong here is not a correctness problem: `read_at` plans over the
|
|
/// real sizes and clamps to what was asked for either way. It costs a
|
|
/// straddling fetch and a cached block the next read will not ask for, and
|
|
/// both settle once the geometry is known.
|
|
fn block_geometry(&self, uid: &NodeUid, mtime: i64, fsize: u64) -> BlockGeometry {
|
|
self.cache
|
|
.block_geometry(uid, mtime, fsize)
|
|
.unwrap_or_else(|| BlockGeometry::uniform(fsize))
|
|
}
|
|
|
|
/// Serve bytes `[offset, offset + len)` of `uid`'s active revision, hitting
|
|
/// the on-disk caches before the network: a whole-file blob (pinned files)
|
|
/// first, then the block cache — fetching only the blocks that overlap the
|
|
/// request, on the revision's own boundaries ([`Core::block_geometry`]), and
|
|
/// caching each. `mtime`/`fsize` validate both caches. Network I/O runs
|
|
/// without any lock held.
|
|
pub(super) fn read_range(
|
|
&self,
|
|
uid: &NodeUid,
|
|
mtime: i64,
|
|
fsize: u64,
|
|
offset: u64,
|
|
len: u64,
|
|
cache_blocks: bool,
|
|
) -> Result<Vec<u8>, Errno> {
|
|
// A queued write has not reached the remote yet, so the remote's current
|
|
// revision is stale and the staged blob is the truth. Serve from it until
|
|
// the drain worker lands the upload.
|
|
if let Some(pending) = self.pending.lock().get(uid).cloned() {
|
|
return self.read_pending(&pending, offset, len);
|
|
}
|
|
// A node created offline and never written has no blob and no remote: it
|
|
// is an empty file, and asking the API about a `local~` uid would only
|
|
// earn a 404. Unless its create landed since the
|
|
// read looked: the landing moved the queued write to the uid Drive
|
|
// gave it, and the row says which (`docs/BUGS.md` B179).
|
|
if is_local_uid(uid) {
|
|
return match self.landed_stand_in_uid(uid) {
|
|
Some(real) => self.read_range(&real, mtime, fsize, offset, len, cache_blocks),
|
|
None => Ok(Vec::new()),
|
|
};
|
|
}
|
|
self.read_range_remote(uid, mtime, fsize, offset, len, cache_blocks)
|
|
}
|
|
|
|
/// Serve a read from a staged blob, falling back to the remote base for any
|
|
/// range the write did not author (an incomplete [`StagedWrite`] holds zeros
|
|
/// there, which must never be handed out as content).
|
|
fn read_pending(
|
|
&self,
|
|
pending: &PendingRevision,
|
|
offset: u64,
|
|
len: u64,
|
|
) -> Result<Vec<u8>, Errno> {
|
|
let m = &pending.meta;
|
|
if offset >= m.len || len == 0 {
|
|
return Ok(Vec::new());
|
|
}
|
|
let uid = parse_node_uid(&m.uid).ok_or(Errno::EIO)?;
|
|
let file = File::open(&pending.path).map_err(|e| {
|
|
error!(%uid, path = %pending.path.display(), error = %e, "open staged blob failed");
|
|
Errno::EIO
|
|
})?;
|
|
let mut written = Intervals::default();
|
|
for &(s, e) in &m.authored {
|
|
written.add(s, e);
|
|
}
|
|
// Same merge as `serve_open_read`, but resolving gaps against the remote
|
|
// rather than through `read_range` — going through `read_range` would find
|
|
// this very pending op and recurse.
|
|
let end = offset.saturating_add(len).min(m.len);
|
|
let mut out = Vec::with_capacity((end - offset) as usize);
|
|
for (s, e, authored) in written.segments(offset, end) {
|
|
if authored {
|
|
let mut buf = vec![0u8; (e - s) as usize];
|
|
file.read_exact_at(&mut buf, s).map_err(|err| {
|
|
warn!(%uid, error = %err, "staged blob read failed");
|
|
Errno::EIO
|
|
})?;
|
|
out.extend_from_slice(&buf);
|
|
continue;
|
|
}
|
|
let bend = e.min(m.base_size);
|
|
if s < bend {
|
|
out.extend_from_slice(&self.read_range_remote(
|
|
&uid,
|
|
m.base_mtime,
|
|
m.base_size,
|
|
s,
|
|
bend - s,
|
|
true,
|
|
)?);
|
|
}
|
|
// Past the base: a hole the write extended over.
|
|
out.resize(out.len() + e.saturating_sub(s.max(m.base_size)) as usize, 0);
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
/// Fetch the block at `span`, joining a fetch already in flight rather than
|
|
/// issuing a second one, and publish the result to everyone who joined.
|
|
///
|
|
/// 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<dyn RevisionRead>,
|
|
uid: &NodeUid,
|
|
mtime: i64,
|
|
fsize: u64,
|
|
span: BlockSpan,
|
|
cache_blocks: bool,
|
|
) -> BlockResult {
|
|
let key: BlockKey = (uid.clone(), mtime, fsize, span);
|
|
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, span, cache_blocks)
|
|
.await;
|
|
}
|
|
};
|
|
let result = self
|
|
.read_block(reader, uid, mtime, fsize, span, 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.
|
|
///
|
|
/// A block that comes back shorter than the file's size says it must be is
|
|
/// never served, cached or promoted: it is repaired from the whole-file
|
|
/// download, or the read fails. See [`Core::repair_block`] (`docs/BUGS.md` B84).
|
|
async fn read_block(
|
|
&self,
|
|
reader: &Arc<dyn RevisionRead>,
|
|
uid: &NodeUid,
|
|
mtime: i64,
|
|
fsize: u64,
|
|
span: BlockSpan,
|
|
cache_blocks: bool,
|
|
) -> BlockResult {
|
|
// The reader derives every block boundary from the revision's recorded
|
|
// block sizes. If they do not add up to the size the node reports, this
|
|
// file's boundaries are wrong, and the failure is not confined to short
|
|
// reads: sizes that *overstate* a block shift every block after it and
|
|
// return full-length reads of the wrong bytes. So a disagreement of any
|
|
// kind takes the file off the range path entirely (`docs/BUGS.md` B84).
|
|
let bytes = if reader.size() != fsize {
|
|
warn!(
|
|
%uid, bidx = span.idx, fsize,
|
|
reader_size = reader.size(),
|
|
reader_blocks = reader.block_sizes().len(),
|
|
"the revision's block table disagrees with the node size"
|
|
);
|
|
self.repair_block(uid, mtime, fsize, span).await?
|
|
} else {
|
|
// Learned here rather than on a path of its own: this is where a
|
|
// reader is open anyway, and recording the real geometry is what
|
|
// lets the *next* read plan its blocks on the revision's own
|
|
// boundaries instead of the 4 MiB fallback (`docs/BUGS.md` B85).
|
|
if cache_blocks {
|
|
self.cache
|
|
.store_block_geometry(uid, mtime, fsize, reader.block_sizes());
|
|
}
|
|
let bytes = reader.read_at(span.start, span.len).await.map_err(|e| {
|
|
self.lost_link(&e, "read a block");
|
|
warn!(%uid, start = span.start, len = span.len, error = %e, "block read failed");
|
|
Errno::EIO
|
|
})?;
|
|
if bytes.len() as u64 == span.len {
|
|
bytes
|
|
} else {
|
|
warn!(
|
|
%uid, bidx = span.idx, blen = span.len, got = bytes.len(), fsize,
|
|
"block came back shorter than the range asked for"
|
|
);
|
|
self.repair_block(uid, mtime, fsize, span).await?
|
|
}
|
|
};
|
|
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, span, &bytes);
|
|
});
|
|
}
|
|
self.block_ring
|
|
.lock()
|
|
.insert(uid, mtime, fsize, span, bytes.clone());
|
|
Ok(bytes)
|
|
}
|
|
|
|
/// Recover the block at `span` of a file whose per-block reads cannot be
|
|
/// by fetching the whole revision through the manifest-verified download and
|
|
/// serving the block out of that.
|
|
///
|
|
/// The range path derives every block boundary from the revision's recorded
|
|
/// block sizes. When those understate the file, reads come back short with
|
|
/// no error anywhere, and the caller hands userspace a truncated file that
|
|
/// looks like a clean EOF (`docs/BUGS.md` B84). `download_file_to` does not do that
|
|
/// arithmetic — it streams the block list — so it returns the whole file
|
|
/// where the range path cannot, which is why pinning was the workaround.
|
|
/// This does the same thing on the user's behalf, once, and the resulting
|
|
/// blob answers every later read of the file before the block path is
|
|
/// consulted at all.
|
|
///
|
|
/// Bounded by [`REPAIR_MAX`]: past that, one unlucky read would pull down
|
|
/// gigabytes, and a clear `EIO` plus a `pdfs pin` is the better trade.
|
|
async fn repair_block(
|
|
&self,
|
|
uid: &NodeUid,
|
|
mtime: i64,
|
|
fsize: u64,
|
|
span: BlockSpan,
|
|
) -> Result<Vec<u8>, Errno> {
|
|
let (bstart, blen) = (span.start, span.len);
|
|
if fsize > REPAIR_MAX {
|
|
warn!(
|
|
%uid, fsize,
|
|
"file is too large to repair from a whole-file download; pin it to read it"
|
|
);
|
|
return Err(Errno::EIO);
|
|
}
|
|
// One download per file, not per block: every block of a file with a bad
|
|
// block table discovers the same problem at once.
|
|
let lock = {
|
|
let mut locks = self.content_repairs.lock().await;
|
|
locks.entry(uid.clone()).or_default().clone()
|
|
};
|
|
let guard = lock.clone().lock_owned().await;
|
|
let downloaded = match self.cache.read_range(uid, mtime, fsize, bstart, blen) {
|
|
// The first arrival did the work; the rest just read its blob.
|
|
Some(bytes) if bytes.len() as u64 == blen => Ok(Some(bytes)),
|
|
_ => self
|
|
.download_whole_revision(uid, mtime, fsize)
|
|
.await
|
|
.map(|()| None),
|
|
};
|
|
// Before releasing, so the map entry is only dropped once no waiter is
|
|
// still holding it.
|
|
drop(guard);
|
|
self.release_repair_lock(uid, lock).await;
|
|
if let Some(bytes) = downloaded? {
|
|
return Ok(bytes);
|
|
}
|
|
match self.cache.read_range(uid, mtime, fsize, bstart, blen) {
|
|
Some(bytes) if bytes.len() as u64 == blen => {
|
|
info!(%uid, bidx = span.idx, "served a short block from a repaired whole-file download");
|
|
Ok(bytes)
|
|
}
|
|
_ => {
|
|
error!(%uid, bidx = span.idx, fsize, "repairing a short block did not produce the block");
|
|
Err(Errno::EIO)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Download the whole of `uid`'s active revision to disk and adopt it as the
|
|
/// cached blob, so the range path is bypassed entirely for this file.
|
|
///
|
|
/// Streamed into a scratch file rather than memory: this runs for files up
|
|
/// to [`REPAIR_MAX`], and buffering that per concurrent reader is not a cost
|
|
/// a read should carry. A download that does not produce exactly `fsize`
|
|
/// bytes is discarded — a short blob would fail the cache's own validity
|
|
/// check anyway, and storing it would only hide the disagreement one layer
|
|
/// further down.
|
|
async fn download_whole_revision(
|
|
&self,
|
|
uid: &NodeUid,
|
|
mtime: i64,
|
|
fsize: u64,
|
|
) -> Result<(), Errno> {
|
|
let (mut file, path) = self.cache.create_scratch().map_err(|error| {
|
|
error!(%uid, %error, "creating a scratch file for a repair download failed");
|
|
Errno::EIO
|
|
})?;
|
|
let outcome = self.drive.download_file_to(uid, &mut file).await;
|
|
let stored = (|| {
|
|
outcome.map_err(|error| {
|
|
warn!(%uid, %error, "repair download failed");
|
|
Errno::EIO
|
|
})?;
|
|
file.sync_all().map_err(|error| {
|
|
warn!(%uid, %error, "syncing a repair download failed");
|
|
Errno::EIO
|
|
})?;
|
|
let got = file.metadata().map(|m| m.len()).unwrap_or(0);
|
|
if got != fsize {
|
|
warn!(%uid, fsize, got, "repair download did not match the node size");
|
|
return Err(Errno::EIO);
|
|
}
|
|
self.cache
|
|
.store_file(uid, mtime, fsize, &path)
|
|
.map_err(|error| {
|
|
error!(%uid, %error, "adopting a repair download as cached content failed");
|
|
Errno::EIO
|
|
})
|
|
})();
|
|
// The blob is a hardlink to the same inode, so unlinking the scratch
|
|
// name leaves it intact.
|
|
let _ = std::fs::remove_file(&path);
|
|
stored
|
|
}
|
|
|
|
/// Drop a repair lock's map entry once nobody else holds it, so a file
|
|
/// repaired long ago does not keep a mutex alive forever.
|
|
async fn release_repair_lock(&self, uid: &NodeUid, lock: Arc<tokio::sync::Mutex<()>>) {
|
|
let mut locks = self.content_repairs.lock().await;
|
|
// Two: the map's and ours. Anything more means a waiter still needs it.
|
|
if Arc::strong_count(&lock) <= 2 {
|
|
locks.remove(uid);
|
|
}
|
|
}
|
|
|
|
/// 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 geometry = self.block_geometry(uid, mtime, fsize);
|
|
let last_block = geometry.blocks().saturating_sub(1);
|
|
let wanted: Vec<BlockSpan> = ((last + 1)..=(last + depth).min(last_block))
|
|
.filter_map(|bidx| geometry.span(bidx))
|
|
.filter(|&span| {
|
|
// 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, span)
|
|
.is_none()
|
|
&& !(cache_blocks && self.cache.has_block(uid, mtime, fsize, span))
|
|
})
|
|
.collect();
|
|
if wanted.is_empty() {
|
|
return;
|
|
}
|
|
let core = self.clone();
|
|
let uid = uid.clone();
|
|
self.rt.spawn(async move {
|
|
let Ok(reader) = core.revision_reader(&uid, mtime, fsize).await else {
|
|
return;
|
|
};
|
|
// Concurrently, for the same reason the demand path fetches its
|
|
// misses concurrently: serially awaited blocks cost the sum of their
|
|
// round-trips, and the reader catches up with the window before it
|
|
// finishes.
|
|
let mut set = tokio::task::JoinSet::new();
|
|
for span in wanted {
|
|
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, span, cache_blocks)
|
|
.await
|
|
{
|
|
debug!(%uid, bidx = span.idx, ?error, "prefetch failed");
|
|
}
|
|
});
|
|
}
|
|
while set.join_next().await.is_some() {}
|
|
});
|
|
}
|
|
|
|
/// Read from the content cache, else the remote. The base-content path, with
|
|
/// no awareness of queued writes — callers wanting the file's *current*
|
|
/// content want [`Core::read_range`]. Gap-filling a staged blob is the one
|
|
/// caller that genuinely means "the base", since that is what its zeroed
|
|
/// ranges have to be filled from.
|
|
fn read_range_remote(
|
|
&self,
|
|
uid: &NodeUid,
|
|
mtime: i64,
|
|
fsize: u64,
|
|
offset: u64,
|
|
len: u64,
|
|
cache_blocks: bool,
|
|
) -> Result<Vec<u8>, Errno> {
|
|
if let Some(bytes) = self.cache.read_range(uid, mtime, fsize, offset, len) {
|
|
return Ok(bytes);
|
|
}
|
|
if offset >= fsize || len == 0 {
|
|
return Ok(Vec::new());
|
|
}
|
|
let end = offset.saturating_add(len).min(fsize);
|
|
let mut out = Vec::with_capacity((end - offset) as usize);
|
|
// On the revision's own boundaries where they are known, and on the
|
|
// 4 MiB fallback where they are not (`docs/BUGS.md` B85).
|
|
let wanted = self.block_geometry(uid, mtime, fsize).spans(offset, end);
|
|
let (Some(&head), Some(&tail)) = (wanted.first(), wanted.last()) else {
|
|
return Ok(out);
|
|
};
|
|
let (first, last) = (head.idx, tail.idx);
|
|
|
|
// Collect the blocks overlapping the request, serving any already cached
|
|
// and fetching the rest concurrently. A multi-block read (e.g. a media
|
|
// player buffering, or a large sequential read split into one FUSE call)
|
|
// would otherwise stall on each block round-trip in turn; downloading the
|
|
// misses in parallel saturates the connection and bounds latency at the
|
|
// slowest single block instead of their sum.
|
|
let mut blocks: Vec<Option<Arc<Vec<u8>>>> = Vec::with_capacity(wanted.len());
|
|
let mut misses: Vec<BlockSpan> = Vec::new();
|
|
for &span in &wanted {
|
|
// 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, span);
|
|
let hit = hit.or_else(|| {
|
|
if !cache_blocks {
|
|
return None;
|
|
}
|
|
let bytes = Arc::new(self.cache.cached_block(uid, mtime, fsize, span)?);
|
|
// Promote it, so the other ~31 reads of this block are answered
|
|
// from memory.
|
|
self.block_ring
|
|
.lock()
|
|
.insert(uid, mtime, fsize, span, bytes.clone());
|
|
Some(bytes)
|
|
});
|
|
match hit {
|
|
Some(b) => blocks.push(Some(b)),
|
|
None => {
|
|
blocks.push(None);
|
|
misses.push(span);
|
|
}
|
|
}
|
|
}
|
|
|
|
if !misses.is_empty() {
|
|
// One deadline for the whole read, including any wait for a lost
|
|
// link. A fetch that failed because the link went down is tried
|
|
// once more when it is back, so a blip is a slow read, not an `EIO`.
|
|
let deadline = Instant::now() + READ_FETCH_TIMEOUT;
|
|
let mut retried = false;
|
|
let fetched = loop {
|
|
if !self.is_online() {
|
|
let wait =
|
|
READ_OFFLINE_WAIT.min(deadline.saturating_duration_since(Instant::now()));
|
|
if !self.wait_online(wait) {
|
|
debug!(%uid, offset, len, "read of uncached blocks while offline; failing it");
|
|
return Err(Errno::EIO);
|
|
}
|
|
}
|
|
let fetch = async {
|
|
// Resolve the file's keys and block table once, then read every
|
|
// missing block through the shared reader. Previously each block
|
|
// called `download_range`, which redid that resolution per block.
|
|
let reader = self.revision_reader(uid, mtime, fsize).await?;
|
|
|
|
let mut set = tokio::task::JoinSet::new();
|
|
for &span in &misses {
|
|
let (core, reader, uid) = (self.clone(), reader.clone(), uid.clone());
|
|
set.spawn(async move {
|
|
core.fetch_block(&reader, &uid, mtime, fsize, span, cache_blocks)
|
|
.await
|
|
.map(|bytes| (span.idx, bytes))
|
|
});
|
|
}
|
|
let mut out = Vec::with_capacity(misses.len());
|
|
while let Some(joined) = set.join_next().await {
|
|
// A join error means the task panicked; surface it as EIO.
|
|
out.push(joined.map_err(|_| Errno::EIO)??);
|
|
}
|
|
Ok::<_, Errno>(out)
|
|
};
|
|
// Dropping the fetch on the deadline aborts the block tasks with
|
|
// it; blocks that already landed stay cached for the retry. The
|
|
// timer is built inside the runtime: a FUSE worker has no reactor
|
|
// of its own, and building it out here panicked every remote read.
|
|
let budget = deadline.saturating_duration_since(Instant::now());
|
|
match self
|
|
.rt
|
|
.block_on(async { tokio::time::timeout(budget, fetch).await })
|
|
{
|
|
Ok(Ok(fetched)) => break fetched,
|
|
// The fetch saw the link go down; the top of the loop waits
|
|
// for it within what is left of the deadline.
|
|
Ok(Err(_)) if !retried && !self.is_online() => retried = true,
|
|
Ok(Err(e)) => return Err(e),
|
|
Err(_) => {
|
|
warn!(
|
|
%uid, offset, len, secs = READ_FETCH_TIMEOUT.as_secs(),
|
|
"read got no answer from the network in time; failing it"
|
|
);
|
|
self.evict_reader(uid);
|
|
self.mark_offline("read", &crate::link::no_answer(READ_FETCH_TIMEOUT));
|
|
return Err(Errno::EIO);
|
|
}
|
|
}
|
|
};
|
|
for (bidx, bytes) in fetched {
|
|
blocks[(bidx - first) as usize] = Some(bytes);
|
|
}
|
|
}
|
|
// 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 (span, block) in wanted.into_iter().zip(blocks) {
|
|
let bstart = span.start;
|
|
// Every slot is populated: cache hits up front, misses by the fetch above.
|
|
let block = block.expect("block fetched or cached");
|
|
// Both sources guarantee the length the geometry says the block has
|
|
// — the cache validates it on read, the network path in `read_block`
|
|
// — so this refuses rather than quietly contributing fewer bytes
|
|
// than the range covers, which is what turned a bad block into a
|
|
// truncated file (`docs/BUGS.md` B84).
|
|
if block.len() as u64 != span.len {
|
|
error!(
|
|
%uid, bidx = span.idx, len = block.len(), expected = span.len, fsize,
|
|
"assembling a read hit a short block"
|
|
);
|
|
return Err(Errno::EIO);
|
|
}
|
|
let s = (offset.max(bstart) - bstart) as usize;
|
|
let e = (end.min(bstart + block.len() as u64) - bstart) as usize;
|
|
if s < e {
|
|
out.extend_from_slice(&block[s..e]);
|
|
}
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
/// Serve a read against an open write handle: stitch authored ranges (from
|
|
/// the scratch file) and untouched ranges (from the remote base via the
|
|
/// block cache) in order, zero-filling any region past the base.
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub(super) fn serve_open_read(
|
|
&self,
|
|
file: &Arc<File>,
|
|
len: u64,
|
|
uid: &NodeUid,
|
|
base_mtime: i64,
|
|
base_size: u64,
|
|
written: &Intervals,
|
|
offset: u64,
|
|
size: u64,
|
|
) -> Result<Vec<u8>, Errno> {
|
|
let read = assemble_open_read(file, len, uid, base_size, written, offset, size, |s, n| {
|
|
self.read_range(uid, base_mtime, base_size, s, n, true)
|
|
.map(Some)
|
|
})?;
|
|
Ok(read.unwrap_or_default())
|
|
}
|
|
|
|
/// [`Core::serve_open_read`] from bytes already on this machine only:
|
|
/// `Ok(None)` the moment a range of the base would have to be downloaded.
|
|
///
|
|
/// The dispatch loop answers a read on a write handle with this, and hands
|
|
/// it to a worker only when it says `None`. Fetching the base there held
|
|
/// every request of the mount for as long as the download took, up to the
|
|
/// read deadline (`docs/BUGS.md` B227).
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub(super) fn serve_open_read_local(
|
|
&self,
|
|
file: &Arc<File>,
|
|
len: u64,
|
|
uid: &NodeUid,
|
|
base_mtime: i64,
|
|
base_size: u64,
|
|
written: &Intervals,
|
|
offset: u64,
|
|
size: u64,
|
|
) -> Result<Option<Vec<u8>>, Errno> {
|
|
assemble_open_read(file, len, uid, base_size, written, offset, size, |s, n| {
|
|
Ok(self.cached_range(uid, base_mtime, base_size, s, n))
|
|
})
|
|
}
|
|
|
|
/// Bytes `[offset, offset + size)` of file `ino` as `read` would serve
|
|
/// them, but only from this machine: an open write handle, a queued write,
|
|
/// or the caches. `Ok(None)` the moment any of it would have to be
|
|
/// downloaded, and for a size still provisional (B100), which only the
|
|
/// network settles. Short or empty past the end of the file.
|
|
///
|
|
/// `copy_file_range` answers with this and leaves anything else to the
|
|
/// kernel's read/write copy.
|
|
pub(super) fn read_local(
|
|
&self,
|
|
ino: u64,
|
|
offset: u64,
|
|
size: u64,
|
|
) -> Result<Option<Vec<u8>>, Errno> {
|
|
let handle = self.state().active_writes.get(&ino).map(|h| {
|
|
(
|
|
h.file.clone(),
|
|
h.len,
|
|
h.uid.clone(),
|
|
h.base_mtime,
|
|
h.base_size,
|
|
h.written.clone(),
|
|
)
|
|
});
|
|
if let Some((file, len, uid, base_mtime, base_size, written)) = handle {
|
|
return self.serve_open_read_local(
|
|
&file, len, &uid, base_mtime, base_size, &written, offset, size,
|
|
);
|
|
}
|
|
let (uid, mtime, fsize, provisional) = {
|
|
let st = self.state();
|
|
match st.entries.get(&ino) {
|
|
Some(e) if e.node.is_file() => (
|
|
e.uid.clone(),
|
|
e.node.modification_time,
|
|
node_size(&e.node),
|
|
matches!(
|
|
&e.node.kind,
|
|
NodeKind::File {
|
|
claimed_size: None,
|
|
..
|
|
}
|
|
),
|
|
),
|
|
Some(_) => return Err(Errno::EISDIR),
|
|
None => return Err(Errno::ENOENT),
|
|
}
|
|
};
|
|
let uid = match is_local_uid(&uid) {
|
|
true => self.landed_stand_in_uid(&uid).unwrap_or(uid),
|
|
false => uid,
|
|
};
|
|
if let Some(pending) = self.pending.lock().get(&uid).cloned() {
|
|
let m = &pending.meta;
|
|
let file = File::open(&pending.path).map_err(|e| {
|
|
error!(%uid, path = %pending.path.display(), error = %e, "open staged blob failed");
|
|
Errno::EIO
|
|
})?;
|
|
let mut written = Intervals::default();
|
|
for &(s, e) in &m.authored {
|
|
written.add(s, e);
|
|
}
|
|
return self.serve_open_read_local(
|
|
&Arc::new(file),
|
|
m.len,
|
|
&uid,
|
|
m.base_mtime,
|
|
m.base_size,
|
|
&written,
|
|
offset,
|
|
size,
|
|
);
|
|
}
|
|
// Created here and never written: an empty file.
|
|
if is_local_uid(&uid) {
|
|
return Ok(Some(Vec::new()));
|
|
}
|
|
if provisional {
|
|
return Ok(None);
|
|
}
|
|
if offset >= fsize || size == 0 {
|
|
return Ok(Some(Vec::new()));
|
|
}
|
|
Ok(self.cached_range(&uid, mtime, fsize, offset, size.min(fsize - offset)))
|
|
}
|
|
|
|
/// 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<Vec<u8>> {
|
|
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 span in self.block_geometry(uid, mtime, fsize).spans(offset, end) {
|
|
let ring = self.block_ring.lock().get(uid, mtime, fsize, span);
|
|
let block = match ring {
|
|
Some(bytes) => bytes,
|
|
None => Arc::new(self.cache.cached_block(uid, mtime, fsize, span)?),
|
|
};
|
|
let bstart = span.start;
|
|
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.
|
|
///
|
|
/// This is the step a partial overwrite cannot skip: only the authored bytes
|
|
/// were ever written to disk, and a revision upload sends the whole file.
|
|
/// The gaps come from the *remote base* (through the block cache, so a small
|
|
/// edit of a large file does not pull all of it), which is exactly why this
|
|
/// can fail with no network — see `StagedWrite`.
|
|
pub(super) fn fill_gaps(
|
|
&self,
|
|
uid: &NodeUid,
|
|
file: &File,
|
|
len: u64,
|
|
base_mtime: i64,
|
|
base_size: u64,
|
|
written: &Intervals,
|
|
) -> Result<(), Errno> {
|
|
file.set_len(len).map_err(|e| {
|
|
error!(%uid, error = %e, "resize scratch file failed");
|
|
Errno::EIO
|
|
})?;
|
|
for (s, e, authored) in written.segments(0, len) {
|
|
if authored {
|
|
continue;
|
|
}
|
|
let bend = e.min(base_size);
|
|
if s >= bend {
|
|
continue; // wholly past the base: already zero-filled on disk
|
|
}
|
|
let bytes = self.read_range_remote(uid, base_mtime, base_size, s, bend - s, true)?;
|
|
file.write_all_at(&bytes, s).map_err(|err| {
|
|
error!(%uid, error = %err, "scratch gap-fill write failed");
|
|
Errno::EIO
|
|
})?;
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// The bytes `[offset, offset + size)` of a file open for writing: authored
|
|
/// ranges from its scratch `file`, the rest from its base through `base`,
|
|
/// and zeros past the base. `Ok(None)` when `base` says it cannot answer.
|
|
#[allow(clippy::too_many_arguments)]
|
|
fn assemble_open_read(
|
|
file: &File,
|
|
len: u64,
|
|
uid: &NodeUid,
|
|
base_size: u64,
|
|
written: &Intervals,
|
|
offset: u64,
|
|
size: u64,
|
|
mut base: impl FnMut(u64, u64) -> Result<Option<Vec<u8>>, Errno>,
|
|
) -> Result<Option<Vec<u8>>, Errno> {
|
|
if offset >= len || size == 0 {
|
|
return Ok(Some(Vec::new()));
|
|
}
|
|
let end = offset.saturating_add(size).min(len);
|
|
let mut out = Vec::with_capacity((end - offset) as usize);
|
|
for (s, e, authored) in written.segments(offset, end) {
|
|
if authored {
|
|
let mut buf = vec![0u8; (e - s) as usize];
|
|
file.read_exact_at(&mut buf, s).map_err(|err| {
|
|
warn!(%uid, error = %err, "scratch read failed");
|
|
Errno::EIO
|
|
})?;
|
|
out.extend_from_slice(&buf);
|
|
} else {
|
|
let bend = e.min(base_size);
|
|
if s < bend {
|
|
match base(s, bend - s)? {
|
|
Some(bytes) => out.extend_from_slice(&bytes),
|
|
None => return Ok(None),
|
|
}
|
|
}
|
|
// Anything past the base is a hole: zero-fill.
|
|
let zeros = e.saturating_sub(s.max(base_size));
|
|
out.resize(out.len() + zeros as usize, 0);
|
|
}
|
|
}
|
|
Ok(Some(out))
|
|
}
|
|
|
|
#[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<Vec<u8>> {
|
|
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
|
|
}
|
|
|
|
/// Block `idx` under the uniform geometry — the shape these tests are
|
|
/// written against, since what they exercise is tagging and eviction rather
|
|
/// than block layout.
|
|
fn span(idx: u64) -> BlockSpan {
|
|
BlockSpan {
|
|
idx,
|
|
start: idx * BLOCK_SIZE,
|
|
len: BLOCK_SIZE,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_stored_block_is_served_back() {
|
|
let mut ring = BlockRing::default();
|
|
let node = uid("a");
|
|
ring.insert(&node, 7, 100, span(3), block(0xab));
|
|
assert_eq!(ring.get(&node, 7, 100, span(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, span(0), block(1));
|
|
// Same node, newer revision: the tag moved, so the old bytes go.
|
|
assert!(ring.get(&node, 8, 100, span(0)).is_none());
|
|
assert!(ring.get(&node, 7, 101, span(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, span(0), block(1));
|
|
ring.insert(&node, 8, 100, span(0), block(2));
|
|
assert_eq!(ring.get(&node, 8, 100, span(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, span(idx), block(idx as u8));
|
|
}
|
|
// Re-reading block 0 makes block 1 the oldest instead.
|
|
assert!(ring.get(&node, 1, u64::MAX, span(0)).is_some());
|
|
ring.insert(&node, 1, u64::MAX, span(capacity()), block(0xff));
|
|
assert!(ring.get(&node, 1, u64::MAX, span(0)).is_some());
|
|
assert!(ring.get(&node, 1, u64::MAX, span(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, span(0), block(1));
|
|
for idx in 0..capacity() {
|
|
ring.insert(&new, 1, u64::MAX, span(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, span(0), block(1));
|
|
ring.insert(&b, 1, 100, span(0), block(2));
|
|
ring.drop_node(&a);
|
|
assert!(ring.get(&a, 1, 100, span(0)).is_none());
|
|
assert_eq!(ring.get(&b, 1, 100, span(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<u64> = (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);
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod open_read_tests {
|
|
use super::*;
|
|
use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId};
|
|
|
|
fn scratch(bytes: &[u8]) -> (File, PathBuf) {
|
|
let path = std::env::temp_dir().join(format!(
|
|
"pdfs-open-read-{}-{}",
|
|
std::process::id(),
|
|
std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.unwrap()
|
|
.as_nanos()
|
|
));
|
|
let mut file = File::options()
|
|
.create_new(true)
|
|
.read(true)
|
|
.write(true)
|
|
.open(&path)
|
|
.unwrap();
|
|
file.write_all(bytes).unwrap();
|
|
(file, path)
|
|
}
|
|
|
|
/// The dispatch loop answers from what is on this machine and says so
|
|
/// when the base would have to be downloaded, for a worker to do it
|
|
/// (`docs/BUGS.md` B227).
|
|
#[test]
|
|
fn a_read_needing_the_base_says_so_instead_of_fetching() {
|
|
let uid = NodeUid::new(VolumeId::from("v"), LinkId::from("f"));
|
|
let (file, path) = scratch(b"ABCDEFGH");
|
|
let mut written = Intervals::default();
|
|
written.add(0, 4);
|
|
|
|
let authored = assemble_open_read(&file, 8, &uid, 8, &written, 0, 4, |_, _| {
|
|
panic!("an authored range never asks the base")
|
|
});
|
|
assert_eq!(authored.unwrap().as_deref(), Some(&b"ABCD"[..]));
|
|
|
|
let missing = assemble_open_read(&file, 8, &uid, 8, &written, 2, 6, |_, _| Ok(None));
|
|
assert_eq!(missing.unwrap(), None);
|
|
|
|
let base = assemble_open_read(&file, 10, &uid, 8, &written, 2, 8, |s, n| {
|
|
assert_eq!((s, n), (4, 4));
|
|
Ok(Some(b"efgh".to_vec()))
|
|
});
|
|
assert_eq!(base.unwrap().as_deref(), Some(&b"CDefgh\0\0"[..]));
|
|
let _ = std::fs::remove_file(path);
|
|
}
|
|
}
|