diff --git a/lib/common/common/src/universal_io/conformance.rs b/lib/common/common/src/universal_io/conformance.rs index e8f4f499d6..1a6f6d35a3 100644 --- a/lib/common/common/src/universal_io/conformance.rs +++ b/lib/common/common/src/universal_io/conformance.rs @@ -153,4 +153,43 @@ where first.read_whole::().unwrap().as_ref(), b"aaabbbccc".as_slice(), ); + + run_open_append_conformance(fs, dir); +} + +/// Exercise [`UniversalWriteFileOps::open_append`] against a backend: the +/// filesystem hands out an append handle whose writes the read path observes, +/// whatever file type either side is. +/// +/// Called by [`run_append_conformance`]; run it directly for a filesystem +/// whose own [`UniversalReadFs::File`] does not append. +pub fn run_open_append_conformance(fs: &Fs, dir: &Path) +where + Fs: UniversalReadFs + UniversalWriteFileOps, + Fs::OpenExtra: Default, +{ + let path = dir.join("open_append.dat"); + fs.create(&path, 0).unwrap(); + + // `writeable` is forced on: an append handle opened as read-only is a + // contradiction, not an error to propagate. + let mut file = fs.open_append(&path, open_options(false)).unwrap(); + + file.append(0, b"one".as_slice()).unwrap(); + file.append_batch(3, [b"two".as_slice(), b"three".as_slice()]) + .unwrap(); + (file.flusher())().unwrap(); + + // Same compare-and-swap contract as any other append handle. + let err = file.append(0, b"x".as_slice()).unwrap_err(); + assert!(matches!(err, UniversalIoError::AppendOffsetConflict { .. })); + + // The read path sees the appended bytes. + let reader = fs + .open(&path, open_options(false), Fs::OpenExtra::default()) + .unwrap(); + assert_eq!( + reader.read_whole::().unwrap().as_ref(), + b"onetwothree".as_slice(), + ); } diff --git a/lib/common/common/src/universal_io/disk_cache/mod.rs b/lib/common/common/src/universal_io/disk_cache/mod.rs index e652aa7a73..bc64cbc9ea 100644 --- a/lib/common/common/src/universal_io/disk_cache/mod.rs +++ b/lib/common/common/src/universal_io/disk_cache/mod.rs @@ -8,7 +8,7 @@ use crate::ext::aligned_vec::ACow; use crate::generic_consts::AccessPattern; use crate::universal_io::{ ListedFile, OpenOptions, UioResult, UniversalIoError, UniversalRead, UniversalReadFileOps, - UniversalReadFs, UniversalWriteFileOps, UserData, local_file_ops, + UniversalReadFs, UserData, local_file_ops, }; mod cached_slice; @@ -100,27 +100,10 @@ impl UniversalReadFileOps for BlockCacheFs { } } -impl UniversalWriteFileOps for BlockCacheFs { - fn create(&self, path: &Path, expected_length: usize) -> UioResult<()> { - local_file_ops::local_create(path, expected_length) - } - - fn create_dir(&self, path: &Path) -> UioResult<()> { - local_file_ops::local_create_dir(path) - } - - fn remove(&self, path: &Path) -> UioResult<()> { - local_file_ops::local_remove(path) - } - - fn remove_dir(&self, path: &Path) -> UioResult<()> { - local_file_ops::local_remove_dir(path) - } - - fn atomic_save(&self, path: &Path, bytes: &[u8]) -> UioResult<()> { - local_file_ops::local_atomic_save(path, bytes) - } -} +// Deliberately no `UniversalWriteFileOps` impl: the block cache is strictly +// read-only ([`CachedSlice`] neither writes nor appends, and `open` rejects +// writeable opens). Mutations go straight to the underlying local +// filesystem — `MmapFs`/`IoUringFs` over the very same paths. impl UniversalReadFs for BlockCacheFs { type File = CachedSlice; diff --git a/lib/common/common/src/universal_io/io_uring/mod.rs b/lib/common/common/src/universal_io/io_uring/mod.rs index 3d9d6e7a9f..cc7187cf92 100644 --- a/lib/common/common/src/universal_io/io_uring/mod.rs +++ b/lib/common/common/src/universal_io/io_uring/mod.rs @@ -78,6 +78,8 @@ impl UniversalReadFileOps for IoUringFs { } impl UniversalWriteFileOps for IoUringFs { + type AppendFile = IoUringFile; + fn create(&self, path: &Path, expected_length: usize) -> UioResult<()> { local_file_ops::local_create(path, expected_length) } @@ -97,6 +99,13 @@ impl UniversalWriteFileOps for IoUringFs { fn atomic_save(&self, path: &Path, bytes: &[u8]) -> UioResult<()> { local_file_ops::local_atomic_save(path, bytes) } + + /// The very handle [`UniversalReadFs::open`] hands out, opened with the + /// default extras: an `O_DIRECT` handle cannot append (its block-aligned + /// I/O requirements rule out appends of arbitrary sizes). + fn open_append(&self, path: impl AsRef, options: OpenOptions) -> UioResult { + self.open(path, options.for_append(), IoUringOpenExtra::default()) + } } /// Per-open backend extras for [`IoUringFs::open`]. diff --git a/lib/common/common/src/universal_io/mmap/mod.rs b/lib/common/common/src/universal_io/mmap/mod.rs index 84968b3413..fa87bdc962 100644 --- a/lib/common/common/src/universal_io/mmap/mod.rs +++ b/lib/common/common/src/universal_io/mmap/mod.rs @@ -38,6 +38,8 @@ impl UniversalReadFileOps for MmapFs { } impl UniversalWriteFileOps for MmapFs { + type AppendFile = MmapFile; + fn create(&self, path: &Path, expected_length: usize) -> UioResult<()> { local_file_ops::local_create(path, expected_length) } @@ -57,6 +59,12 @@ impl UniversalWriteFileOps for MmapFs { fn atomic_save(&self, path: &Path, bytes: &[u8]) -> UioResult<()> { local_file_ops::local_atomic_save(path, bytes) } + + /// The very handle [`UniversalReadFs::open`] hands out: an [`MmapFile`] + /// appends through a dedicated `O_APPEND` fd of its own. + fn open_append(&self, path: impl AsRef, options: OpenOptions) -> UioResult { + MmapFile::open_inner(path, options.for_append()) + } } impl UniversalReadFs for MmapFs { diff --git a/lib/common/common/src/universal_io/simple_disk_cache/fs.rs b/lib/common/common/src/universal_io/simple_disk_cache/fs.rs index f574af77a8..48391b59e4 100644 --- a/lib/common/common/src/universal_io/simple_disk_cache/fs.rs +++ b/lib/common/common/src/universal_io/simple_disk_cache/fs.rs @@ -12,7 +12,7 @@ use crate::universal_io::simple_disk_cache::REMOTE_OPEN_OPTIONS; use crate::universal_io::simple_disk_cache::local_state::LocalState; use crate::universal_io::{ ListedFile, OpenExtra, OpenOptions, OwnedPipeline, Populate, UioResult, UniversalIoError, - UniversalRead, UniversalReadFileOps, UniversalReadFs, UniversalWriteFileOps, + UniversalRead, UniversalReadFileOps, UniversalReadFs, }; /// Construction context for [`DiskCacheFs`]: carries the @@ -123,31 +123,11 @@ where } } -impl UniversalWriteFileOps for DiskCacheFs -where - R: UniversalRead, - R::Fs: UniversalWriteFileOps, -{ - fn create(&self, path: &Path, expected_length: usize) -> UioResult<()> { - self.remote_fs.create(path, expected_length) - } - - fn create_dir(&self, path: &Path) -> UioResult<()> { - self.remote_fs.create_dir(path) - } - - fn remove(&self, path: &Path) -> UioResult<()> { - self.remote_fs.remove(path) - } - - fn remove_dir(&self, path: &Path) -> UioResult<()> { - self.remote_fs.remove_dir(path) - } - - fn atomic_save(&self, path: &Path, bytes: &[u8]) -> UioResult<()> { - self.remote_fs.atomic_save(path, bytes) - } -} +// Deliberately no `UniversalWriteFileOps` impl: the disk cache is strictly +// read-only, at the filesystem level as much as at the file level (see the +// `assert_not_impl_any!` on `DiskCache`). Mutations — creating, removing and +// appending to files — go straight to the backing storage, whose handle the +// caller holds anyway to build this one. /// Make the mirror path unique per open, so concurrently-alive [`DiskCache`] /// instances for the same remote path never share (and truncate) each other's diff --git a/lib/common/common/src/universal_io/simple_disk_cache/tests.rs b/lib/common/common/src/universal_io/simple_disk_cache/tests.rs index dd0da80d06..2eb1b5b2d9 100644 --- a/lib/common/common/src/universal_io/simple_disk_cache/tests.rs +++ b/lib/common/common/src/universal_io/simple_disk_cache/tests.rs @@ -16,15 +16,18 @@ use crate::universal_io::cached_fs::FileInfo; use crate::universal_io::{ CachedFs, CachedReadFs, MmapFile, OpenOptions, Populate, ReadPipeline, ReadRange, UniversalAppend, UniversalFlush, UniversalIoError, UniversalRead, UniversalReadFileOps, - UniversalReadFs, UniversalWrite, + UniversalReadFs, UniversalWrite, UniversalWriteFileOps, }; // The disk cache is strictly read-only: mutating it must stay a // compile-time error, on top of writeable opens being rejected at runtime -// (covered per backend variant below). +// (covered per backend variant below). This holds at the filesystem level +// too — creating, removing and appending to files goes to the backing +// storage, never through the cache. static_assertions::assert_not_impl_any!( DiskCache: UniversalAppend, UniversalFlush, UniversalWrite ); +static_assertions::assert_not_impl_any!(DiskCacheFs: UniversalWriteFileOps); fn make_test_data(n_bytes: usize) -> Vec { (0..n_bytes).map(|i| (i % 251) as u8).collect() diff --git a/lib/common/common/src/universal_io/traits/append.rs b/lib/common/common/src/universal_io/traits/append.rs index 90bcd46d5e..e76c479a6e 100644 --- a/lib/common/common/src/universal_io/traits/append.rs +++ b/lib/common/common/src/universal_io/traits/append.rs @@ -39,8 +39,10 @@ use crate::universal_io::{ByteOffset, UioResult}; /// acknowledgement). Retrying at the *same* offset is safe — it conflicts /// rather than duplicating; re-check the length before appending at a new /// offset. -/// - Requires a handle opened with `writeable: true`. Not supported on -/// `prevent_caching` (`O_DIRECT`) handles. +/// - Requires a handle opened with `writeable: true` — either through +/// [`UniversalWriteFileOps::open_append`], or through +/// [`UniversalReadFs::open`] on a backend whose read handle appends. Not +/// supported on `prevent_caching` (`O_DIRECT`) handles. /// - Appending no bytes trivially succeeds, without touching the file or /// validating `offset`. /// - Offsets are plain byte offsets; no `T`-alignment is guaranteed or @@ -50,6 +52,7 @@ use crate::universal_io::{ByteOffset, UioResult}; /// /// [`AppendOffsetConflict`]: crate::universal_io::UniversalIoError::AppendOffsetConflict /// [`MmapFile`]: crate::universal_io::MmapFile +/// [`UniversalReadFs::open`]: super::UniversalReadFs::open /// [`UniversalWrite`]: super::UniversalWrite /// [`UniversalWrite::write`]: super::UniversalWrite::write /// [`len`]: UniversalRead::len diff --git a/lib/common/common/src/universal_io/traits/file_ops.rs b/lib/common/common/src/universal_io/traits/file_ops.rs index ac20c5e6b7..b5bff09d69 100644 --- a/lib/common/common/src/universal_io/traits/file_ops.rs +++ b/lib/common/common/src/universal_io/traits/file_ops.rs @@ -2,6 +2,7 @@ use std::fmt::Debug; use std::path::Path; use crate::universal_io::cached_fs::FileInfo; +use crate::universal_io::traits::append::UniversalAppend; use crate::universal_io::traits::open_extra::OpenExtra; use crate::universal_io::traits::read::UniversalRead; use crate::universal_io::{ListedFile, OpenOptions, UioResult}; @@ -51,11 +52,21 @@ pub trait UniversalReadFileOps: Clone + Debug + Send + Sync + Sized { /// Filesystem-level handle for mutating operations. /// -/// Extends [`UniversalReadFileOps`] with create/remove/save operations. -/// Read-only backends (e.g. `ReadOnlyFs`) implement only the read side, -/// making the absence of write support a compile-time property instead of -/// a runtime error. +/// Extends [`UniversalReadFileOps`] with create/remove/save operations and +/// with opening append handles ([`Self::open_append`]). Read-only backends +/// (e.g. `ReadOnlyFs`, the disk caches) implement only the read side, making +/// the absence of write support a compile-time property instead of a runtime +/// error. pub trait UniversalWriteFileOps: UniversalReadFileOps { + /// File handle type produced by [`Self::open_append`]. + /// + /// Deliberately not tied to [`UniversalReadFs::File`]: the two capabilities + /// live on independent traits, so a backend may serve reads through one + /// handle type and appends through another (and a filesystem that opens no + /// read handles at all still names an append handle here). Backends whose + /// read handle appends — every local one — simply name it twice. + type AppendFile: UniversalAppend; + /// Create or truncate a file at the given path. /// /// Local backends use `expected_length` to pre-size the file. Backends @@ -81,6 +92,20 @@ pub trait UniversalWriteFileOps: UniversalReadFileOps { /// backends may overwrite the full object. fn atomic_save(&self, path: &Path, bytes: &[u8]) -> UioResult<()>; + /// Open an existing file for appending, per the [`UniversalAppend`] + /// contract. The file must exist — [`Self::create`] it first. + /// + /// `options` are honored as by [`UniversalReadFs::open`], except that + /// `writeable` is forced on ([`OpenOptions::for_append`]). There is no + /// [`OpenExtra`] counterpart: the append handle may come from a different + /// backend than [`UniversalReadFs::File`], whose per-open knobs would then + /// not apply. + fn open_append( + &self, + path: impl AsRef, + options: OpenOptions, + ) -> UioResult; + // When adding provided methods, don't forget to update impls in // `crate::universal_io::wrappers::*`. } diff --git a/lib/common/common/src/universal_io/types.rs b/lib/common/common/src/universal_io/types.rs index 96b8dfc34d..552932d08f 100644 --- a/lib/common/common/src/universal_io/types.rs +++ b/lib/common/common/src/universal_io/types.rs @@ -119,6 +119,28 @@ pub struct OpenOptions { } impl OpenOptions { + /// The same options with `writeable` forced on, as + /// [`UniversalWriteFileOps::open_append`] opens them: an append handle + /// mutates the file by definition, so the flag carries no information + /// there. + /// + /// [`UniversalWriteFileOps::open_append`]: super::UniversalWriteFileOps::open_append + pub fn for_append(self) -> Self { + let Self { + writeable: _, + need_sequential, + populate, + advice, + } = self; + + Self { + writeable: true, + need_sequential, + populate, + advice, + } + } + /// Default values for [`OpenOptions`]. #[cfg(any(test, feature = "testing"))] pub fn new_for_test() -> Self { diff --git a/lib/common/io_bridge/src/fs.rs b/lib/common/io_bridge/src/fs.rs index 2c13ead3d7..f2b3bf264a 100644 --- a/lib/common/io_bridge/src/fs.rs +++ b/lib/common/io_bridge/src/fs.rs @@ -6,7 +6,7 @@ use common::universal_io::{ UniversalWriteFileOps, }; -use crate::{AsyncRead, AsyncWrite, BlobFile, BridgeRuntime}; +use crate::{AsyncAppend, AsyncRead, BlobFile, BridgeRuntime}; /// Filesystem handle for an object-store backend: an [`AsyncRead`] handle plus /// the [`BridgeRuntime`] used to drive its async operations. Opens per-object @@ -74,7 +74,14 @@ impl UniversalReadFileOps for BlobFs { } } -impl UniversalWriteFileOps for BlobFs { +/// Requires [`AsyncAppend`], not just [`AsyncWrite`](crate::AsyncWrite): a write-capable +/// filesystem must hand out append handles, and only backends with a native +/// single-request append can produce one (`BlobFile` appends through +/// [`AsyncAppend::append`]). Backends that can only put whole objects stay +/// read-only through universal I/O. +impl UniversalWriteFileOps for BlobFs { + type AppendFile = BlobFile; + fn create(&self, path: &Path, _expected_length: usize) -> UioResult<()> { // Object stores have no fixed-size preallocation; the expected // length is ignored, as the trait allows. @@ -100,6 +107,16 @@ impl UniversalWriteFileOps for BlobFs { self.runtime .block_on(self.inner.save(path, Bytes::copy_from_slice(bytes))) } + + /// The very handle [`UniversalReadFs::open`] hands out, always writeable. + /// Performs no IO — the object is not touched until the first append or + /// read. Blob handles have no open-time knobs, so `options` is unused. + fn open_append(&self, path: impl AsRef, _options: OpenOptions) -> UioResult> { + Ok( + BlobFile::new(self.inner.clone(), self.runtime.clone(), path.as_ref()) + .with_writeable(true), + ) + } } impl UniversalReadFs for BlobFs {