diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs index 5059202..ebbb6a1 100644 --- a/crates/pdfs-fuse/src/filesystem.rs +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -430,58 +430,10 @@ impl Filesystem for ProtonFs { reply.error(error); return; } - let fh = fh.0; - // Stage the bytes straight into the scratch file (no base download): only - // the untouched remainder is pulled from the remote, and only at commit. - let file = { - let st = self.core.state(); - match st.handles.get(&fh).and_then(|&i| st.active_writes.get(&i)) { - Some(aw) => aw.file.clone(), - None => { - reply.error(Errno::EBADF); - return; - } - } - }; - if let Err(e) = file.write_all_at(data, offset) { - error!(ino = ino.0, fh, error = %e, "scratch write failed"); - // A full local disk is not a broken file: answer `ENOSPC` so the - // application can say so (and so `cp`/editors take their own - // out-of-space path), and free what the cache is holding — the - // scratch file the next write goes to shares that filesystem. - if e.raw_os_error() == Some(libc::ENOSPC) { - self.core.cache.emergency_evict(); - reply.error(Errno::ENOSPC); - } else { - reply.error(Errno::EIO); - } - return; + match self.stage_write(ino.0, fh.0, offset, data) { + Ok(()) => reply.written(data.len() as u32), + Err(error) => reply.error(error), } - let new_len = { - let mut st = self.core.state(); - let Some(aw) = st - .handles - .get(&fh) - .copied() - .and_then(|i| st.active_writes.get_mut(&i)) - else { - reply.error(Errno::EBADF); - return; - }; - aw.record_write(offset, offset + data.len() as u64); - let len = aw.len; - st.set_size(ino.0, len); - len - }; - debug!( - ino = ino.0, - fh, - offset, - len = data.len(), - new_len, - "staged write" - ); - reply.written(data.len() as u32); } #[allow(clippy::too_many_arguments)] @@ -1016,20 +968,59 @@ impl Filesystem for ProtonFs { fn copy_file_range( &self, _req: &Request, - _ino_in: INodeNo, + ino_in: INodeNo, _fh_in: FileHandle, - _offset_in: u64, - _ino_out: INodeNo, - _fh_out: FileHandle, - _offset_out: u64, - _len: u64, - _flags: CopyFileRangeFlags, + offset_in: u64, + ino_out: INodeNo, + fh_out: FileHandle, + offset_out: u64, + len: u64, + flags: CopyFileRangeFlags, reply: ReplyWrite, ) { - // Also must stay ENOSYS: there is no server-side copy, and this is the - // errno that sends coreutils (and the kernel) down the read/write path - // instead of failing the copy outright. - reply.error(Errno::ENOSYS); + // Copied here when the source bytes are on this machine, as `write` + // would stage them, minus a FUSE round trip per page: `cp` of a cached + // 256 MiB file ran at about 100 MiB/s through read and write. Anything + // that would have to be downloaded gets `EOPNOTSUPP`, which makes the + // kernel copy that one call through read and write itself; `ENOSYS` + // would turn the op off for the whole mount. There is no server-side + // copy into a file `cp` has already created, so the copy still uploads. + // + // On the dispatch loop, like `write`, so it keeps its order with the + // reads and writes of the same handle. [`COPY_RANGE_MAX`] bounds how + // long one call holds the loop. + let (ino_in, ino_out) = (self.tree(ino_in), self.tree(ino_out)); + if !flags.is_empty() || ino_in == ino_out { + reply.error(Errno::EOPNOTSUPP); + return; + } + if let Err(error) = self.core.require_writable(ino_out.0) { + reply.error(error); + return; + } + let bytes = match self + .core + .read_local(ino_in.0, offset_in, len.min(COPY_RANGE_MAX)) + { + Ok(Some(bytes)) => bytes, + Ok(None) => { + debug!(ino_in = ino_in.0, offset_in, "copy range not local"); + reply.error(Errno::EOPNOTSUPP); + return; + } + Err(error) => { + reply.error(error); + return; + } + }; + if bytes.is_empty() { + reply.written(0); + return; + } + match self.stage_write(ino_out.0, fh_out.0, offset_out, &bytes) { + Ok(()) => reply.written(bytes.len() as u32), + Err(error) => reply.error(error), + } } } @@ -1057,6 +1048,11 @@ pub struct ProtonFs { /// worker busy indefinitely. const OPEN_BASE_ATTEMPTS: u32 = 3; +/// The most one `copy_file_range` call copies; the caller loops for the rest. +/// A local copy of this much takes milliseconds, which is how long it holds +/// the dispatch loop. +const COPY_RANGE_MAX: u64 = 16 * 1024 * 1024; + /// One directory's entries as `readdir` reports them: `.` and `..` first, then /// the interned children in `children` order. /// @@ -1641,6 +1637,48 @@ impl ProtonFs { } /// The body of [`Filesystem::create`], run off the dispatch loop. + /// Stage `data` at `offset` of the write handle `fh` on `ino`, straight into + /// its scratch file (no base download): only the untouched remainder is + /// pulled from the remote, and only at commit. + fn stage_write(&self, ino: u64, fh: u64, offset: u64, data: &[u8]) -> Result<(), Errno> { + let file = { + let st = self.core.state(); + match st.handles.get(&fh).and_then(|&i| st.active_writes.get(&i)) { + Some(aw) => aw.file.clone(), + None => return Err(Errno::EBADF), + } + }; + if let Err(e) = file.write_all_at(data, offset) { + error!(ino, fh, error = %e, "scratch write failed"); + // A full local disk is not a broken file: answer `ENOSPC` so the + // application can say so (and so `cp`/editors take their own + // out-of-space path), and free what the cache is holding — the + // scratch file the next write goes to shares that filesystem. + if e.raw_os_error() == Some(libc::ENOSPC) { + self.core.cache.emergency_evict(); + return Err(Errno::ENOSPC); + } + return Err(Errno::EIO); + } + let new_len = { + let mut st = self.core.state(); + let Some(aw) = st + .handles + .get(&fh) + .copied() + .and_then(|i| st.active_writes.get_mut(&i)) + else { + return Err(Errno::EBADF); + }; + aw.record_write(offset, offset + data.len() as u64); + let len = aw.len; + st.set_size(ino, len); + len + }; + debug!(ino, fh, offset, len = data.len(), new_len, "staged write"); + Ok(()) + } + fn serve_create(&self, parent: u64, name: &str, reply: ReplyCreate) { if let Err(error) = self.core.require_writable(parent) { reply.error(error); diff --git a/crates/pdfs-fuse/src/reads.rs b/crates/pdfs-fuse/src/reads.rs index 5b51a26..a785177 100644 --- a/crates/pdfs-fuse/src/reads.rs +++ b/crates/pdfs-fuse/src/reads.rs @@ -1101,6 +1101,92 @@ impl Core { }) } + /// 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>, 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. /// diff --git a/crates/pdfs-fuse/src/sim/daemon.rs b/crates/pdfs-fuse/src/sim/daemon.rs index 99cbbbc..5a425a3 100644 --- a/crates/pdfs-fuse/src/sim/daemon.rs +++ b/crates/pdfs-fuse/src/sim/daemon.rs @@ -2433,6 +2433,152 @@ mod tests { eprintln!("wifi: syscalls {wifi_syscalls:?}, drain {wifi_drain:?}"); } + /// `copy_file_range` inside the mount is answered from the bytes the daemon + /// holds, and left to the kernel for bytes it would have to download. Every + /// source kind has to come out byte for byte: cached, queued, open with + /// unsaved writes, and on Drive only. + #[test] + #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"] + fn a_copy_inside_the_mount_has_the_source_bytes() { + use std::os::fd::AsRawFd as _; + use std::os::unix::fs::FileExt as _; + + fn copy_range(from: &std::fs::File, at: u64, to: &std::fs::File, to_at: u64, len: usize) { + let (mut at, mut to_at) = (at as i64, to_at as i64); + let mut left = len; + while left > 0 { + // SAFETY: both descriptors are open; the offsets are ours. + let n = unsafe { + libc::copy_file_range( + from.as_raw_fd(), + &mut at, + to.as_raw_fd(), + &mut to_at, + left, + 0, + ) + }; + assert!(n > 0, "copy_file_range: {}", io::Error::last_os_error()); + left -= n as usize; + } + } + + let pattern = |len: usize, seed: u8| -> Vec { + (0..len).map(|i| (i % 253) as u8 ^ seed).collect() + }; + let drive = FakeDrive::new(); + // On Drive before the mount knows it, so nothing of it is cached. + let remote = pattern(9 << 20, 1); + drive.device().write("remote.bin", &remote); + let dir = scratch("copy-range"); + let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap(); + let mount = &daemon.mountpoint; + + // Downloaded through the kernel's own copy, then served from the cache. + std::fs::copy(mount.join("remote.bin"), mount.join("remote-1.bin")).unwrap(); + assert!(std::fs::read(mount.join("remote-1.bin")).unwrap() == remote); + std::fs::copy(mount.join("remote.bin"), mount.join("remote-2.bin")).unwrap(); + assert!(std::fs::read(mount.join("remote-2.bin")).unwrap() == remote); + + // Queued and not on Drive yet: the staged blob is the content. + let paused = |paused| Request::SetSyncPaused { + paused, + until: None, + }; + daemon.request(&paused(true)).unwrap(); + let queued = pattern(20 << 20, 2); + std::fs::write(mount.join("queued.bin"), &queued).unwrap(); + std::fs::copy(mount.join("queued.bin"), mount.join("queued-copy.bin")).unwrap(); + assert!(std::fs::read(mount.join("queued-copy.bin")).unwrap() == queued); + + // Open with writes not saved yet: the copy sees them, as a read would. + let open = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(mount.join("queued.bin")) + .unwrap(); + open.write_all_at(b"UNSAVED", 100).unwrap(); + let target = std::fs::File::create(mount.join("unsaved-copy.bin")).unwrap(); + copy_range(&open, 96, &target, 0, 16); + drop(target); + drop(open); + let mut expected = queued[96..112].to_vec(); + expected[4..11].copy_from_slice(b"UNSAVED"); + assert_eq!( + std::fs::read(mount.join("unsaved-copy.bin")).unwrap(), + expected + ); + + // A range into the middle of a file keeps the bytes around it. + std::fs::write(mount.join("middle.bin"), b"0123456789").unwrap(); + let source = std::fs::File::open(mount.join("remote.bin")).unwrap(); + let target = std::fs::OpenOptions::new() + .write(true) + .open(mount.join("middle.bin")) + .unwrap(); + copy_range(&source, 5, &target, 3, 4); + drop(target); + let mut expected = b"0123456789".to_vec(); + expected[3..7].copy_from_slice(&remote[5..9]); + assert_eq!(std::fs::read(mount.join("middle.bin")).unwrap(), expected); + + daemon.request(&paused(false)).unwrap(); + let mut edited = queued.clone(); + edited[100..107].copy_from_slice(b"UNSAVED"); + let file = |bytes: &[u8]| Some(Entry::File(Arc::new(bytes.to_vec()))); + assert!(wait_until(Duration::from_secs(120), || { + let tree = drive.tree(); + daemon.pending().is_ok_and(|ops| ops.is_empty()) + && tree.get("queued.bin").cloned() == file(&edited) + && tree.get("queued-copy.bin").cloned() == file(&queued) + && tree.get("remote-2.bin").cloned() == file(&remote) + && tree.get("middle.bin").cloned() == file(&expected) + })); + assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown))); + let _ = std::fs::remove_dir_all(&dir); + } + + /// How long `cp` inside the mount takes for a file the daemon already + /// holds, cold and from the page cache. It decides whether the daemon + /// should answer `copy_file_range` itself rather than leave the kernel to + /// copy through read and write. Runs when `PDFS_SIM_MEASURE` is set. + #[test] + #[ignore = "mounts FUSE: run with `PDFS_SIM_MEASURE=1 cargo test -p pdfs-fuse sim:: -- --ignored`"] + fn a_copy_inside_the_mount_measures_its_speed() { + if std::env::var_os("PDFS_SIM_MEASURE").is_none() { + eprintln!("skipped: set PDFS_SIM_MEASURE to measure the copy"); + return; + } + const SIZE: usize = 256 << 20; + let drive = FakeDrive::new(); + let dir = scratch("copy-speed"); + let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap(); + let mount = &daemon.mountpoint; + let bytes: Vec = (0..SIZE).map(|i| (i % 251) as u8).collect(); + std::fs::write(mount.join("big.bin"), &bytes).unwrap(); + assert!(wait_until(Duration::from_secs(120), || { + daemon.pending().is_ok_and(|ops| ops.is_empty()) + })); + for (round, name) in ["copy-1.bin", "copy-2.bin", "copy-3.bin"] + .into_iter() + .enumerate() + { + let start = Instant::now(); + let status = std::process::Command::new("cp") + .arg(mount.join("big.bin")) + .arg(mount.join(name)) + .status() + .unwrap(); + let took = start.elapsed(); + assert!(status.success()); + let mib_per_s = (SIZE >> 20) as f64 / took.as_secs_f64(); + eprintln!("cp round {round}: {took:?}, {mib_per_s:.0} MiB/s"); + assert!(std::fs::read(mount.join(name)).unwrap() == bytes); + } + assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown))); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"] fn half_of_a_wide_folder_unlinked_at_once_leaves_the_other_half() { diff --git a/docs/BUGS.md b/docs/BUGS.md index aad9d31..bf8a4ce 100644 --- a/docs/BUGS.md +++ b/docs/BUGS.md @@ -5877,6 +5877,13 @@ Only the three namespace ops were strictly required (the other four silence themselves once the kernel caches `ENOSYS`), but implementing all seven keeps the refusal explicit and documented at the call site rather than inherited. +`copy_file_range` has since been implemented: the daemon copies a range whose bytes are on this +machine (an open write handle, a queued write, the cache) into the destination's scratch file, +and answers `EOPNOTSUPP` for anything it would have to download. Unlike `ENOSYS`, that errno makes +the kernel copy that one call through read and write and keeps asking the daemon next time. +A warm `cp` of a 256 MiB file inside the mount went from about 100 MiB/s to about 450 MiB/s +(sim, debug build; `a_copy_inside_the_mount_measures_its_speed`). + **Verified:** full live acceptance run after the fix — 19 pass / 1 skip, the capability diff identical to the pre-fix run, and **0** `Not Implemented` lines in the journal across the whole run (was 17, 10 of them `link`). diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index c296761..12d0bd0 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -34,6 +34,10 @@ scratch (user data in `staging/` and `recovery/` is never touched by this). Proton Drive counts it, including items in the trash. From a terminal: `pdfs du PATH`. ### Changed +- **Copying a file inside the mount is faster** when the computer already has the file, for + example with `cp` or a file manager: the copy is made on this computer, about four times + faster for a 256 MiB file, instead of going through the mount page by page. A file that is + only on Proton Drive is copied as before. The copy is then uploaded as usual. - **Reorganised documentation, now also on the [wiki](https://github.com/narrrl/proton-drive-linux/wiki).** A new [Getting started](https://github.com/narrrl/proton-drive-linux/wiki/Getting-Started) page and