added block cache

This commit is contained in:
nvrl committed 2026-06-30 06:22:40 +02:00
1 parent cc2f370670
commit 81a0ecce54
3 files changed
+614 -95

No files matched your search

+203
View File
@@ -27,6 +27,11 @@ use sha2::{Digest, Sha256};
use crate::error::Result; use crate::error::Result;
/// Block size for the on-demand block cache. Matches the SDK content block size
/// (`DEFAULT_BLOCK_SIZE`, 4 MiB) so each cached block maps to exactly one
/// `download_range` fetch with no straddling.
pub const BLOCK_SIZE: u64 = 1 << 22;
/// Validity tag stored alongside a cached blob. A blob is fresh only if both /// Validity tag stored alongside a cached blob. A blob is fresh only if both
/// fields still match the node's current metadata. /// fields still match the node's current metadata.
#[derive(Serialize, Deserialize, Clone, Copy, PartialEq, Eq)] #[derive(Serialize, Deserialize, Clone, Copy, PartialEq, Eq)]
@@ -61,6 +66,13 @@ pub struct ContentCache {
/// Subdirectory holding cached thumbnails (`<key>.t<n>` + `.meta`). Kept out /// Subdirectory holding cached thumbnails (`<key>.t<n>` + `.meta`). Kept out
/// of `content_dir` so the budget scan never sees thumbnail files. /// of `content_dir` so the budget scan never sees thumbnail files.
thumb_dir: PathBuf, thumb_dir: PathBuf,
/// Subdirectory holding on-demand block-cache files (`<key>.b<idx>` +
/// `.meta`). Kept out of `content_dir` so the whole-file budget scan never
/// sees them; blocks carry their own LRU budget.
block_dir: PathBuf,
/// Subdirectory for write-handle scratch files (disk-backed write buffers).
/// Emptied on open so a crashed run leaves no orphans.
scratch_dir: PathBuf,
/// JSON pin registry path. /// JSON pin registry path.
pins_path: PathBuf, pins_path: PathBuf,
/// Soft cap on total blob bytes. Exceeded only transiently: a `store` /// Soft cap on total blob bytes. Exceeded only transiently: a `store`
@@ -78,12 +90,21 @@ impl ContentCache {
std::fs::create_dir_all(&content_dir)?; std::fs::create_dir_all(&content_dir)?;
let thumb_dir = content_dir.join("thumbs"); let thumb_dir = content_dir.join("thumbs");
std::fs::create_dir_all(&thumb_dir)?; std::fs::create_dir_all(&thumb_dir)?;
let block_dir = content_dir.join("blocks");
std::fs::create_dir_all(&block_dir)?;
// Scratch holds disk-backed write buffers; a previous run's leftovers are
// worthless, so start clean.
let scratch_dir = content_dir.join("scratch");
let _ = std::fs::remove_dir_all(&scratch_dir);
std::fs::create_dir_all(&scratch_dir)?;
if let Some(parent) = pins_path.parent() { if let Some(parent) = pins_path.parent() {
std::fs::create_dir_all(parent)?; std::fs::create_dir_all(parent)?;
} }
Ok(Self { Ok(Self {
content_dir, content_dir,
thumb_dir, thumb_dir,
block_dir,
scratch_dir,
pins_path, pins_path,
max_bytes, max_bytes,
}) })
@@ -180,6 +201,119 @@ impl ContentCache {
Ok(()) Ok(())
} }
fn block_blob(&self, uid: &NodeUid, idx: u64) -> PathBuf {
self.block_dir.join(format!("{}.b{idx}", Self::key(uid)))
}
fn block_meta(&self, uid: &NodeUid, idx: u64) -> PathBuf {
self.block_dir.join(format!("{}.b{idx}.meta", Self::key(uid)))
}
/// 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<Vec<u8>> {
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 {
return None;
}
let mut f = std::fs::File::open(self.block_blob(uid, idx)).ok()?;
let _ = f.set_times(std::fs::FileTimes::new().set_modified(SystemTime::now()));
let mut buf = Vec::new();
f.read_to_end(&mut buf).ok()?;
Some(buf)
}
/// Store `bytes` as cached block `idx` of `uid`, tagged `(mtime, size)`.
/// Temp-file-then-rename like [`store`](Self::store); meta written last so a
/// crash mid-store fails validation. Enforces the block-cache LRU budget.
pub fn store_block(
&self,
uid: &NodeUid,
mtime: i64,
size: u64,
idx: u64,
bytes: &[u8],
) -> Result<()> {
let blob = self.block_blob(uid, idx);
let tmp = self
.block_dir
.join(format!("{}.b{idx}.tmp", Self::key(uid)));
std::fs::write(&tmp, bytes)?;
std::fs::rename(&tmp, &blob)?;
std::fs::write(self.block_meta(uid, idx), serde_json::to_vec(&Meta { mtime, size })?)?;
self.enforce_block_budget();
Ok(())
}
/// Evict least-recently-used block-cache files until the block dir fits
/// `max_bytes`. No-op when the cap is disabled (`0`). All blocks are
/// evictable — pinned files are served from whole-file blobs, never blocks.
fn enforce_block_budget(&self) {
if self.max_bytes == 0 {
return;
}
let mut total: u64 = 0;
let mut blobs: Vec<(PathBuf, u64, SystemTime)> = Vec::new();
let Ok(rd) = std::fs::read_dir(&self.block_dir) else {
return;
};
for entry in rd.flatten() {
let path = entry.path();
let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
continue;
};
if name.ends_with(".meta") || name.ends_with(".tmp") {
continue;
}
let Ok(meta) = entry.metadata() else { continue };
if !meta.is_file() {
continue;
}
total += meta.len();
let atime = meta.modified().unwrap_or(SystemTime::UNIX_EPOCH);
blobs.push((path, meta.len(), atime));
}
if total <= self.max_bytes {
return;
}
blobs.sort_by_key(|(_, _, atime)| *atime);
for (path, len, _) in blobs {
if total <= self.max_bytes {
break;
}
let meta = path.with_file_name(format!(
"{}.meta",
path.file_name().unwrap_or_default().to_string_lossy()
));
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(meta);
total = total.saturating_sub(len);
}
}
/// Create a fresh, empty read-write scratch file for a disk-backed write
/// handle. Returns the open file and its path (for cleanup on release).
pub fn create_scratch(&self) -> Result<(std::fs::File, PathBuf)> {
use std::sync::atomic::{AtomicU64, Ordering};
static N: AtomicU64 = AtomicU64::new(0);
let path = self.scratch_dir.join(format!(
"w-{}-{}",
std::process::id(),
N.fetch_add(1, Ordering::Relaxed)
));
let file = std::fs::OpenOptions::new()
.create(true)
.truncate(true)
.read(true)
.write(true)
.open(&path)?;
Ok((file, path))
}
/// Remove any cached blob + meta for `uid`, including its thumbnails (best /// Remove any cached blob + meta for `uid`, including its thumbnails (best
/// effort; absence is fine). Thumbnails are tied to the revision, so a /// effort; absence is fine). Thumbnails are tied to the revision, so a
/// content eviction must drop them too. /// content eviction must drop them too.
@@ -190,6 +324,19 @@ impl ContentCache {
let _ = std::fs::remove_file(self.thumb_blob(uid, ttype)); let _ = std::fs::remove_file(self.thumb_blob(uid, ttype));
let _ = std::fs::remove_file(self.thumb_meta(uid, ttype)); let _ = std::fs::remove_file(self.thumb_meta(uid, ttype));
} }
// Drop every cached block (and its meta/tmp) for this uid.
let prefix = format!("{}.b", Self::key(uid));
if let Ok(rd) = std::fs::read_dir(&self.block_dir) {
for entry in rd.flatten() {
if entry
.file_name()
.to_str()
.is_some_and(|n| n.starts_with(&prefix))
{
let _ = std::fs::remove_file(entry.path());
}
}
}
} }
fn thumb_blob(&self, uid: &NodeUid, ttype: i32) -> PathBuf { fn thumb_blob(&self, uid: &NodeUid, ttype: i32) -> PathBuf {
@@ -559,4 +706,60 @@ mod tests {
c.evict(&u); c.evict(&u);
assert!(!c.is_cached(&u, 1, 3)); assert!(!c.is_cached(&u, 1, 3));
} }
#[test]
fn block_store_then_read() {
let (c, _d) = cache();
let u = uid("a");
c.store_block(&u, 100, 4096, 0, b"block-zero").unwrap();
assert_eq!(c.cached_block(&u, 100, 4096, 0).unwrap(), b"block-zero");
// A different index is a separate cache entry, absent here.
assert!(c.cached_block(&u, 100, 4096, 1).is_none());
// A new revision (mtime/size bump) invalidates the block.
assert!(c.cached_block(&u, 101, 4096, 0).is_none());
assert!(c.cached_block(&u, 100, 5000, 0).is_none());
}
#[test]
fn evict_drops_blocks() {
let (c, _d) = cache();
let u = uid("a");
c.store_block(&u, 1, 8, 0, b"aaaa").unwrap();
c.store_block(&u, 1, 8, 1, b"bbbb").unwrap();
c.evict(&u);
assert!(c.cached_block(&u, 1, 8, 0).is_none());
assert!(c.cached_block(&u, 1, 8, 1).is_none());
}
#[test]
fn block_budget_evicts_lru() {
// Cap fits two 4-byte blocks but not three.
let (c, _d) = cache_capped(8);
let u = uid("a");
c.store_block(&u, 1, 12, 0, b"aaaa").unwrap();
std::thread::sleep(std::time::Duration::from_millis(10));
c.store_block(&u, 1, 12, 1, b"bbbb").unwrap();
// Touch block 0 so block 1 is the least-recently-used.
std::thread::sleep(std::time::Duration::from_millis(10));
assert!(c.cached_block(&u, 1, 12, 0).is_some());
std::thread::sleep(std::time::Duration::from_millis(10));
c.store_block(&u, 1, 12, 2, b"cccc").unwrap();
assert!(c.cached_block(&u, 1, 12, 0).is_some(), "recently-read survives");
assert!(c.cached_block(&u, 1, 12, 1).is_none(), "LRU block evicted");
assert!(c.cached_block(&u, 1, 12, 2).is_some(), "newest survives");
}
#[test]
fn scratch_file_is_writable_and_isolated() {
use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
let (c, _d) = cache();
let (mut f1, p1) = c.create_scratch().unwrap();
let (_f2, p2) = c.create_scratch().unwrap();
assert_ne!(p1, p2, "each scratch file is unique");
f1.write_all(b"scratch").unwrap();
f1.seek(SeekFrom::Start(0)).unwrap();
let mut s = String::new();
f1.read_to_string(&mut s).unwrap();
assert_eq!(s, "scratch");
}
} }
+410 -94
View File
@@ -23,10 +23,19 @@
//! into the on-disk [`ContentCache`] and every later `read` is served from disk //! into the on-disk [`ContentCache`] and every later `read` is served from disk
//! instead of the network. Writes and remote events evict the cache so it never //! instead of the network. Writes and remote events evict the cache so it never
//! goes stale. //! goes stale.
//!
//! Reads of unpinned files no longer hit the network per call: [`Core::read_range`]
//! fetches and caches [`BLOCK_SIZE`]-aligned blocks, so sequential/sparse reads
//! reuse the on-disk block cache. Writes are disk-backed: each [`WriteHandle`]
//! stages authored bytes in a scratch file and tracks them with an [`Intervals`]
//! set, so a multi-GiB write never buffers in RAM and only the untouched
//! remainder of the file is pulled from the remote — lazily, at commit.
use std::collections::HashMap; use std::collections::HashMap;
use std::ffi::OsStr; use std::ffi::OsStr;
use std::fs::File;
use std::io::{BufRead, BufReader, Write as _}; use std::io::{BufRead, BufReader, Write as _};
use std::os::unix::fs::FileExt;
use std::os::unix::net::{UnixListener, UnixStream}; use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Component, Path, PathBuf}; use std::path::{Component, Path, PathBuf};
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
@@ -39,7 +48,7 @@ use fuser::{
ReplyAttr, ReplyCreate, ReplyData, ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyOpen, ReplyAttr, ReplyCreate, ReplyData, ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyOpen,
ReplyWrite, Request, Session, TimeOrNow, WriteFlags, ReplyWrite, Request, Session, TimeOrNow, WriteFlags,
}; };
use pdfs_core::cache::ContentCache; use pdfs_core::cache::{BLOCK_SIZE, ContentCache};
use pdfs_core::control::{DirEntry, PhotoItem, Request as CtlRequest, Response as CtlResponse}; use pdfs_core::control::{DirEntry, PhotoItem, Request as CtlRequest, Response as CtlResponse};
use proton_drive_rs::proton_sdk::ids::{DriveEventId, LinkId, NodeUid, VolumeId}; use proton_drive_rs::proton_sdk::ids::{DriveEventId, LinkId, NodeUid, VolumeId};
use proton_drive_rs::{ use proton_drive_rs::{
@@ -75,18 +84,90 @@ struct Entry {
node: Node, node: Node,
} }
/// Buffered state for a file opened for writing. The full plaintext is /// A set of non-overlapping `[start, end)` byte ranges, kept sorted and merged.
/// assembled here and uploaded as one new revision on flush/release, because /// Tracks which bytes of a [`WriteHandle`]'s scratch file were authored locally
/// the SDK seals whole revisions rather than byte ranges. /// (vs. still living only in the remote base), so reads and the commit gap-fill
/// know which regions to pull from the network.
#[derive(Clone, Default)]
struct Intervals(Vec<(u64, u64)>);
impl Intervals {
/// Mark `[start, end)` as authored, coalescing with any touching ranges.
fn add(&mut self, start: u64, end: u64) {
if start >= end {
return;
}
self.0.push((start, end));
self.0.sort_by_key(|&(s, _)| s);
let mut merged: Vec<(u64, u64)> = Vec::with_capacity(self.0.len());
for &(s, e) in &self.0 {
match merged.last_mut() {
Some(last) if s <= last.1 => last.1 = last.1.max(e),
_ => merged.push((s, e)),
}
}
self.0 = merged;
}
/// Drop everything at or beyond `len` (a shrink/truncate).
fn clip(&mut self, len: u64) {
self.0.retain(|&(s, _)| s < len);
for iv in &mut self.0 {
iv.1 = iv.1.min(len);
}
}
/// Split `[start, end)` into contiguous `(s, e, authored)` segments, in
/// order. `authored == true` means the bytes live in the scratch file;
/// `false` means they must come from the remote base (or are a hole).
fn segments(&self, start: u64, end: u64) -> Vec<(u64, u64, bool)> {
let mut out = Vec::new();
let mut pos = start;
for &(s, e) in &self.0 {
if e <= start {
continue;
}
if s >= end {
break;
}
let ws = s.max(start);
let we = e.min(end);
if pos < ws {
out.push((pos, ws, false));
}
out.push((ws, we, true));
pos = we;
}
if pos < end {
out.push((pos, end, false));
}
out
}
}
/// State for a file opened for writing. Authored bytes are staged in an on-disk
/// scratch file (positional reads/writes) rather than RAM, so a multi-GiB write
/// never balloons the daemon. On flush/release the scratch file — gap-filled
/// from the remote base where untouched — is streamed up as one new revision,
/// since the SDK seals whole revisions rather than byte ranges.
struct WriteHandle { struct WriteHandle {
ino: u64, ino: u64,
uid: NodeUid, uid: NodeUid,
/// The plaintext being assembled; valid only once `seeded`. /// Disk-backed staging buffer. Shared (`Arc`) so reads can use it without
buf: Vec<u8>, /// holding the state lock across I/O. Accessed positionally (`read_at`/
/// Whether `buf` holds the file's current content (or was emptied by a /// `write_at`), so a clone never disturbs another's file offset.
/// truncate). Until then a partial write must hydrate the base first. file: Arc<File>,
seeded: bool, /// Scratch file path, removed on release.
/// Whether `buf` diverged from the remote and needs an upload. path: PathBuf,
/// Byte ranges authored into `file`. Everything else in `[0, len)` is base.
written: Intervals,
/// Logical file size (may exceed authored bytes after a truncate-extend).
len: u64,
/// Size of the remote base at open, for serving untouched ranges.
base_size: u64,
/// Modification time of the remote base at open, validating its block cache.
base_mtime: i64,
/// Whether anything diverged from the remote and needs an upload.
dirty: bool, dirty: bool,
} }
@@ -289,35 +370,174 @@ impl Core {
nodes.into_iter().next().ok_or(Errno::ENOENT) nodes.into_iter().next().ok_or(Errno::ENOENT)
} }
/// Upload a write handle's buffer as a new revision if it is dirty, clearing /// Serve bytes `[offset, offset + len)` of `uid`'s active revision, hitting
/// the dirty flag and refreshing the entry's size/mtime on success. No-op for /// the on-disk caches before the network: a whole-file blob (pinned files)
/// a clean (or unknown) handle. Network I/O happens without the lock held. /// first, then the block cache — fetching only the [`BLOCK_SIZE`]-aligned
/// blocks that overlap the request and caching each. `mtime`/`fsize` validate
/// both caches. Network I/O runs without any lock held.
fn read_range(
&self,
uid: &NodeUid,
mtime: i64,
fsize: u64,
offset: u64,
len: u64,
) -> 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);
let first = offset / BLOCK_SIZE;
let last = (end - 1) / BLOCK_SIZE;
for bidx in first..=last {
let bstart = bidx * BLOCK_SIZE;
let block = match self.cache.cached_block(uid, mtime, fsize, bidx) {
Some(b) => b,
None => {
let blen = BLOCK_SIZE.min(fsize - bstart);
let bytes = self
.rt
.block_on(self.client.download_range(uid, bstart, blen))
.map_err(|e| {
warn!(%uid, bstart, blen, error = %e, "download_range failed");
Errno::EIO
})?;
let _ = self.cache.store_block(uid, mtime, fsize, bidx, &bytes);
bytes
}
};
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)]
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> {
if offset >= len || size == 0 {
return Ok(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 {
out.extend_from_slice(&self.read_range(uid, base_mtime, base_size, s, bend - s)?);
}
// 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(out)
}
/// Upload a write handle's scratch file as a new revision if it is dirty.
/// Untouched bytes within the base are filled from the remote first (reusing
/// the block cache), so a partial overwrite never had to pre-download the
/// whole file. On success the handle's base is advanced to the just-sealed
/// revision and `written` cleared, so later reads of untouched regions see
/// the new content. No-op for a clean (or unknown) handle. Network I/O runs
/// without the lock held.
fn commit(&self, fh: u64) -> Result<(), Errno> { fn commit(&self, fh: u64) -> Result<(), Errno> {
let (uid, buf, ino) = { let (uid, file, path, len, base_mtime, base_size, written, ino) = {
let st = self.state.lock().unwrap(); let st = self.state.lock().unwrap();
match st.handles.get(&fh) { match st.handles.get(&fh) {
Some(h) if h.dirty => (h.uid.clone(), h.buf.clone(), h.ino), Some(h) if h.dirty => (
h.uid.clone(),
h.file.clone(),
h.path.clone(),
h.len,
h.base_mtime,
h.base_size,
h.written.clone(),
h.ino,
),
_ => return Ok(()), _ => return Ok(()),
} }
}; };
// Materialize the full new content in the scratch file: ensure its length,
// then fill every untouched range that overlaps the base with base bytes.
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(&uid, base_mtime, base_size, s, bend - s)?;
file.write_all_at(&bytes, s).map_err(|err| {
error!(%uid, error = %err, "scratch gap-fill write failed");
Errno::EIO
})?;
}
// Stream the scratch file up as one revision (fresh handle reads from 0).
let reader = File::open(&path).map_err(|e| {
error!(%uid, error = %e, "reopen scratch file failed");
Errno::EIO
})?;
self.rt self.rt
.block_on(self.client.upload_new_revision(&uid, &buf)) .block_on(
self.client
.upload_new_revision_from(&uid, reader, len as i64, Vec::new(), None),
)
.map_err(|e| { .map_err(|e| {
error!(%uid, error = %e, "upload new revision failed"); error!(%uid, error = %e, "upload new revision failed");
Errno::EIO Errno::EIO
})?; })?;
let now = now_secs(); let now = now_secs();
{
let mut st = self.state.lock().unwrap(); let mut st = self.state.lock().unwrap();
if let Some(h) = st.handles.get_mut(&fh) { if let Some(h) = st.handles.get_mut(&fh) {
h.dirty = false; h.dirty = false;
// The scratch file now equals the sealed revision; treat it all as
// base so further reads of untouched bytes hit the new content.
h.written = Intervals::default();
h.base_mtime = now;
h.base_size = len;
} }
st.set_size(ino, buf.len() as u64); st.set_size(ino, len);
st.touch_mtime(ino, now); st.touch_mtime(ino, now);
drop(st); }
// The sealed content differs from any cached blob; if the file is // The sealed content differs from any cached blob/blocks; refresh a pinned
// pinned, refresh the cache so reads stay served from disk. // file's whole-file blob, otherwise evict so reads re-fetch fresh.
if self.cache.is_pinned(&uid) { if self.cache.is_pinned(&uid) {
let _ = self.cache.store(&uid, now, buf.len() as u64, &buf); if let Ok(bytes) = std::fs::read(&path) {
let _ = self.cache.store(&uid, now, len, &bytes);
}
} else { } else {
self.cache.evict(&uid); self.cache.evict(&uid);
} }
@@ -821,10 +1041,12 @@ impl Filesystem for ProtonFs {
} }
fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) { fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) {
let uid = { let (uid, base_mtime, base_size) = {
let st = self.core.state.lock().unwrap(); let st = self.core.state.lock().unwrap();
match st.entries.get(&ino.0) { match st.entries.get(&ino.0) {
Some(e) if e.node.is_file() => e.uid.clone(), Some(e) if e.node.is_file() => {
(e.uid.clone(), e.node.modification_time, node_size(&e.node))
}
Some(_) => { Some(_) => {
reply.error(Errno::EISDIR); reply.error(Errno::EISDIR);
return; return;
@@ -835,12 +1057,21 @@ impl Filesystem for ProtonFs {
} }
} }
}; };
// Read-only opens stay stateless (fh 0). A write open allocates a buffer // Read-only opens stay stateless (fh 0). A write open allocates a
// handle, hydrated lazily on the first write that needs the base content. // disk-backed handle; bytes are authored into a scratch file and the
// untouched remainder is pulled from the base lazily (on read / commit).
if flags.acc_mode() == OpenAccMode::O_RDONLY { if flags.acc_mode() == OpenAccMode::O_RDONLY {
reply.opened(FileHandle(0), FopenFlags::empty()); reply.opened(FileHandle(0), FopenFlags::empty());
return; return;
} }
let (file, path) = match self.core.cache.create_scratch() {
Ok(x) => x,
Err(e) => {
error!(%uid, error = %e, "create scratch file failed");
reply.error(Errno::EIO);
return;
}
};
let mut st = self.core.state.lock().unwrap(); let mut st = self.core.state.lock().unwrap();
let fh = st.next_fh; let fh = st.next_fh;
st.next_fh += 1; st.next_fh += 1;
@@ -849,8 +1080,14 @@ impl Filesystem for ProtonFs {
WriteHandle { WriteHandle {
ino: ino.0, ino: ino.0,
uid, uid,
buf: Vec::new(), file: Arc::new(file),
seeded: false, path,
written: Intervals::default(),
// Starts at the current size; reads in [0, base_size) come from
// the base until overwritten.
len: base_size,
base_size,
base_mtime,
dirty: false, dirty: false,
}, },
); );
@@ -868,16 +1105,33 @@ impl Filesystem for ProtonFs {
_lock_owner: Option<LockOwner>, _lock_owner: Option<LockOwner>,
reply: ReplyData, reply: ReplyData,
) { ) {
let (uid, mtime, fsize) = { // A file open for writing is served from its handle so reads see the
// in-flight (possibly unsaved) content: authored bytes come from the
// scratch file, untouched bytes from the base.
let handle = {
let st = self.core.state.lock().unwrap(); let st = self.core.state.lock().unwrap();
// If the file is open for writing and its buffer is populated, serve st.handles.values().find(|h| h.ino == ino.0).map(|h| {
// from there so reads see the in-flight (possibly unsaved) content. (
if let Some(h) = st.handles.values().find(|h| h.ino == ino.0 && h.seeded) { h.file.clone(),
let off = (offset as usize).min(h.buf.len()); h.len,
let end = (off + size as usize).min(h.buf.len()); h.uid.clone(),
reply.data(&h.buf[off..end]); h.base_mtime,
h.base_size,
h.written.clone(),
)
})
};
if let Some((file, len, uid, base_mtime, base_size, written)) = handle {
match self.core.serve_open_read(
&file, len, &uid, base_mtime, base_size, &written, offset, size as u64,
) {
Ok(bytes) => reply.data(&bytes),
Err(e) => reply.error(e),
}
return; return;
} }
let (uid, mtime, fsize) = {
let st = self.core.state.lock().unwrap();
match st.entries.get(&ino.0) { match st.entries.get(&ino.0) {
Some(e) if e.node.is_file() => { Some(e) if e.node.is_file() => {
(e.uid.clone(), e.node.modification_time, node_size(&e.node)) (e.uid.clone(), e.node.modification_time, node_size(&e.node))
@@ -892,25 +1146,9 @@ impl Filesystem for ProtonFs {
} }
} }
}; };
// Pinned (or otherwise cached) files are served straight from disk. match self.core.read_range(&uid, mtime, fsize, offset, size as u64) {
if let Some(bytes) = self
.core
.cache
.read_range(&uid, mtime, fsize, offset, size as u64)
{
reply.data(&bytes);
return;
}
match self
.core
.rt
.block_on(self.core.client.download_range(&uid, offset, size as u64))
{
Ok(bytes) => reply.data(&bytes), Ok(bytes) => reply.data(&bytes),
Err(e) => { Err(e) => reply.error(e),
warn!(%uid, offset, size, error = %e, "download_range failed");
reply.error(Errno::EIO);
}
} }
} }
@@ -962,6 +1200,14 @@ impl Filesystem for ProtonFs {
return; return;
} }
}; };
let (file, path) = match self.core.cache.create_scratch() {
Ok(x) => x,
Err(e) => {
error!(%new_uid, error = %e, "create scratch file failed");
reply.error(Errno::EIO);
return;
}
};
let mut st = self.core.state.lock().unwrap(); let mut st = self.core.state.lock().unwrap();
let ino = st.intern(parent, node); let ino = st.intern(parent, node);
if let Some(kids) = st.children.get_mut(&parent) if let Some(kids) = st.children.get_mut(&parent)
@@ -973,11 +1219,16 @@ impl Filesystem for ProtonFs {
st.next_fh += 1; st.next_fh += 1;
st.handles.insert( st.handles.insert(
fh, fh,
// A brand-new file: empty base, everything written is authored.
WriteHandle { WriteHandle {
ino, ino,
uid: new_uid, uid: new_uid,
buf: Vec::new(), file: Arc::new(file),
seeded: true, path,
written: Intervals::default(),
len: 0,
base_size: 0,
base_mtime: now_secs(),
dirty: false, dirty: false,
}, },
); );
@@ -1004,60 +1255,34 @@ impl Filesystem for ProtonFs {
reply: ReplyWrite, reply: ReplyWrite,
) { ) {
let fh = fh.0; let fh = fh.0;
// Partial writes need the file's current content; hydrate it once. // Stage the bytes straight into the scratch file (no base download): only
let need_seed = { // the untouched remainder is pulled from the remote, and only at commit.
let file = {
let st = self.core.state.lock().unwrap(); let st = self.core.state.lock().unwrap();
match st.handles.get(&fh) { match st.handles.get(&fh) {
Some(h) => !h.seeded, Some(h) => h.file.clone(),
None => { None => {
reply.error(Errno::EBADF); reply.error(Errno::EBADF);
return; return;
} }
} }
}; };
if need_seed { if let Err(e) = file.write_all_at(data, offset) {
let uid = self error!(ino = ino.0, fh, error = %e, "scratch write failed");
.core
.state
.lock()
.unwrap()
.handles
.get(&fh)
.map(|h| h.uid.clone());
let Some(uid) = uid else {
reply.error(Errno::EBADF);
return;
};
let base = match self.core.rt.block_on(self.core.client.download_file(&uid)) {
Ok(b) => b,
Err(e) => {
warn!(%uid, error = %e, "seed write buffer failed");
reply.error(Errno::EIO); reply.error(Errno::EIO);
return; return;
} }
};
let mut st = self.core.state.lock().unwrap();
if let Some(h) = st.handles.get_mut(&fh)
&& !h.seeded
{
h.buf = base;
h.seeded = true;
}
}
let off = offset as usize;
let new_len = { let new_len = {
let mut st = self.core.state.lock().unwrap(); let mut st = self.core.state.lock().unwrap();
let Some(h) = st.handles.get_mut(&fh) else { let Some(h) = st.handles.get_mut(&fh) else {
reply.error(Errno::EBADF); reply.error(Errno::EBADF);
return; return;
}; };
let end = off + data.len(); let end = offset + data.len() as u64;
if h.buf.len() < end { h.written.add(offset, end);
h.buf.resize(end, 0); h.len = h.len.max(end);
}
h.buf[off..end].copy_from_slice(data);
h.dirty = true; h.dirty = true;
let len = h.buf.len() as u64; let len = h.len;
st.set_size(ino.0, len); st.set_size(ino.0, len);
len len
}; };
@@ -1067,7 +1292,7 @@ impl Filesystem for ProtonFs {
offset, offset,
len = data.len(), len = data.len(),
new_len, new_len,
"buffered write" "staged write"
); );
reply.written(data.len() as u32); reply.written(data.len() as u32);
} }
@@ -1100,8 +1325,16 @@ impl Filesystem for ProtonFs {
let mut st = self.core.state.lock().unwrap(); let mut st = self.core.state.lock().unwrap();
match st.handles.get_mut(&fh) { match st.handles.get_mut(&fh) {
Some(h) => { Some(h) => {
h.buf.resize(size as usize, 0); if size < h.len {
h.seeded = true; // Shrink: drop authored ranges past the new end.
h.written.clip(size);
} else if size > h.len {
// Grow: the new tail is defined as zeros, so claim
// it as authored rather than base content.
h.written.add(h.len, size);
}
let _ = h.file.set_len(size);
h.len = size;
h.dirty = true; h.dirty = true;
true true
} }
@@ -1207,7 +1440,10 @@ impl Filesystem for ProtonFs {
reply: ReplyEmpty, reply: ReplyEmpty,
) { ) {
let res = self.core.commit(fh.0); let res = self.core.commit(fh.0);
self.core.state.lock().unwrap().handles.remove(&fh.0); let handle = self.core.state.lock().unwrap().handles.remove(&fh.0);
if let Some(h) = handle {
let _ = std::fs::remove_file(&h.path);
}
match res { match res {
Ok(()) => reply.ok(), Ok(()) => reply.ok(),
Err(e) => reply.error(e), Err(e) => reply.error(e),
@@ -1846,3 +2082,83 @@ pub fn mount(
let _ = std::fs::remove_file(control_socket); let _ = std::fs::remove_file(control_socket);
Ok(outcome) Ok(outcome)
} }
#[cfg(test)]
mod tests {
use super::Intervals;
/// Flatten `segments` into a readable form for assertions.
fn segs(iv: &Intervals, start: u64, end: u64) -> Vec<(u64, u64, bool)> {
iv.segments(start, end)
}
#[test]
fn append_merges_into_one_run() {
// Sequential writes (an append) coalesce into a single authored range.
let mut iv = Intervals::default();
iv.add(0, 10);
iv.add(10, 20);
iv.add(20, 25);
assert_eq!(segs(&iv, 0, 25), vec![(0, 25, true)]);
}
#[test]
fn partial_overwrite_leaves_base_gap() {
// Author [0,4) and [8,12); [4,8) stays base. A read of [0,12) must stitch
// authored / base / authored in order.
let mut iv = Intervals::default();
iv.add(0, 4);
iv.add(8, 12);
assert_eq!(
segs(&iv, 0, 12),
vec![(0, 4, true), (4, 8, false), (8, 12, true)]
);
}
#[test]
fn overlapping_writes_coalesce() {
let mut iv = Intervals::default();
iv.add(0, 10);
iv.add(5, 15);
iv.add(14, 20);
assert_eq!(segs(&iv, 0, 20), vec![(0, 20, true)]);
}
#[test]
fn segments_clamp_to_request_window() {
let mut iv = Intervals::default();
iv.add(0, 100);
// A sub-window of one big authored range is a single authored segment.
assert_eq!(segs(&iv, 20, 50), vec![(20, 50, true)]);
// A window entirely outside any authored range is all base.
let empty = Intervals::default();
assert_eq!(segs(&empty, 0, 8), vec![(0, 8, false)]);
}
#[test]
fn truncate_shrink_drops_tail() {
// Grow-then-shrink: clip removes/truncates authored ranges past the end.
let mut iv = Intervals::default();
iv.add(0, 100);
iv.clip(40);
assert_eq!(segs(&iv, 0, 40), vec![(0, 40, true)]);
// Authored ranges wholly past the new end disappear.
let mut iv2 = Intervals::default();
iv2.add(0, 10);
iv2.add(50, 60);
iv2.clip(40);
assert_eq!(segs(&iv2, 0, 40), vec![(0, 10, true), (10, 40, false)]);
}
#[test]
fn truncate_extend_authors_zero_tail() {
// setattr grow claims the new tail as authored (defined zeros), so commit
// never pulls it from the base.
let mut iv = Intervals::default();
iv.add(0, 10); // base content authored over
let old_len = 10u64;
let new_len = 30u64;
iv.add(old_len, new_len);
assert_eq!(segs(&iv, 0, 30), vec![(0, 30, true)]);
}
}
+1 -1
View File
@@ -8,7 +8,7 @@
# Build + install: cd packaging && makepkg -fi # Build + install: cd packaging && makepkg -fi
pkgname=proton-drive-linux pkgname=proton-drive-linux
pkgver=0.1.0 pkgver=0.1.2
pkgrel=1 pkgrel=1
pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray" pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray"
arch=('x86_64') arch=('x86_64')