diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 6c096837..836c365b 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -12,13 +12,24 @@ jobs: steps: - uses: actions/checkout@v3 + + - name: Install FUSE dependencies + run: | + sudo apt update + sudo apt install -y fuse3 libfuse3-dev + echo "user_allow_other" | sudo tee -a /etc/fuse.conf + - name: Formatting run: cargo fmt -- --check + - name: Clippy run: cargo clippy --all-targets --tests -- -D warnings + - name: Build run: cargo build --all-features --all-targets --verbose + - name: Run tests run: cargo test --verbose + - name: Run binary run: cargo build -p chunkfscli --verbose diff --git a/Cargo.lock b/Cargo.lock index a02e7695..6ef0bf46 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -156,6 +156,12 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" +[[package]] +name = "cfg_aliases" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" + [[package]] name = "chrono" version = "0.4.40" @@ -182,7 +188,11 @@ dependencies = [ "criterion", "csv", "fastcdc", + "filetime", + "fuser", "itertools 0.14.0", + "libc", + "nix 0.30.1", "rand 0.9.0", "serde", "serde_json", @@ -481,12 +491,40 @@ version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" +[[package]] +name = "filetime" +version = "0.2.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35c0522e981e68cbfa8c3f978441a5f34b30b96e146b33cd3359176b50fe8586" +dependencies = [ + "cfg-if", + "libc", + "libredox", + "windows-sys", +] + [[package]] name = "fnv" version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "fuser" +version = "0.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53274f494609e77794b627b1a3cddfe45d675a6b2e9ba9c0fdc8d8eee2184369" +dependencies = [ + "libc", + "log", + "memchr", + "nix 0.29.0", + "page_size", + "pkg-config", + "smallvec", + "zerocopy", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -684,6 +722,17 @@ version = "0.2.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8355be11b20d696c8f18f6cc018c4e372165b1fa8126cef092399c9951984ffa" +[[package]] +name = "libredox" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0ff37bd590ca25063e35af745c343cb7a0271906fb7b37e4813e8f79f00268d" +dependencies = [ + "bitflags", + "libc", + "redox_syscall", +] + [[package]] name = "linux-raw-sys" version = "0.9.3" @@ -702,6 +751,30 @@ version = "2.7.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "78ca9ab1a0babb1e7d5695e3530886289c18cf2f87ec19a575a0abdce112e3a3" +[[package]] +name = "nix" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71e2746dc3a24dd78b3cfcb7be93368c6de9963d30f43a6a73998a9cf4b17b46" +dependencies = [ + "bitflags", + "cfg-if", + "cfg_aliases", + "libc", +] + +[[package]] +name = "nix" +version = "0.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "74523f3a35e05aba87a1d978330aef40f67b0304ac79c1c00b294c9830543db6" +dependencies = [ + "bitflags", + "cfg-if", + "cfg_aliases", + "libc", +] + [[package]] name = "num-conv" version = "0.1.0" @@ -730,6 +803,22 @@ version = "11.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" +[[package]] +name = "page_size" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30d5b2194ed13191c1999ae0704b7839fb18384fa22e49b57eeaa97d79ce40da" +dependencies = [ + "libc", + "winapi", +] + +[[package]] +name = "pkg-config" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" + [[package]] name = "plotters" version = "0.3.7" @@ -887,6 +976,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "redox_syscall" +version = "0.5.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f103c6d277498fbceb16e84d317e2a400f160f46904d5f5410848c829511a3" +dependencies = [ + "bitflags", +] + [[package]] name = "regex" version = "1.11.1" @@ -1048,6 +1146,12 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +[[package]] +name = "smallvec" +version = "1.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7fcf8323ef1faaee30a44a340193b1ac6814fd9b7b4e88e9d4519a3e4abe1cfd" + [[package]] name = "strsim" version = "0.11.1" @@ -1280,6 +1384,22 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + [[package]] name = "winapi-util" version = "0.1.9" @@ -1289,6 +1409,12 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows-core" version = "0.52.0" diff --git a/Cargo.toml b/Cargo.toml index 00350d2e..4510b650 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,6 +28,9 @@ csv = { version = "1", optional = true } serde = { version = "1", optional = true, features = ["derive"] } serde_with = { version = "3", optional = true } chrono = { version = "0.4", optional = true, features = ["serde"] } +libc = "0.2.171" +fuser = { version = "0.15.1", features = ["abi-7-28"] } +nix = { version = "0.30.1", features = ["ioctl"] } [features] chunkers = ["cdc-chunkers", "fastcdc"] @@ -40,6 +43,7 @@ criterion = "0.5" sha3 = "0.10" approx = "0.5" tempfile = "3.14" +filetime = "0.2.25" serde_json = "1" @@ -50,3 +54,7 @@ harness = false [[bench]] name = "chunker_sizes" harness = false + +[[bench]] +name = "fuse_write_read" +harness = false diff --git a/README.md b/README.md index 477775c9..b9f21e7d 100644 --- a/README.md +++ b/README.md @@ -82,6 +82,21 @@ To use provided chunkers and hashers, use the corresponding features: chunkfs = { version = "0.1", features = ["chunkers", "hashers"] } ``` +## FUSE native dependepcies + +To use FuseFS structure, mountable with FUSE, you need to install the following native dependencies: + +```bash +sudo apt update +sudo apt install -y fuse3 libfuse3-dev +``` + +To use the file system correctly, you need to allow other users to access the mounted file system. + +```bash +echo "user_allow_other" | sudo tee -a /etc/fuse.conf +``` + ## Examples Examples for chunkfs usage and benching are provided in [examples](examples) folder. diff --git a/benches/chunker_sizes.rs b/benches/chunker_sizes.rs index 0289f09a..4ee90588 100644 --- a/benches/chunker_sizes.rs +++ b/benches/chunker_sizes.rs @@ -54,8 +54,8 @@ fn get_chunker(algorithm: Algorithms, params: SizeParams) -> ChunkerRef { match algorithm { Algorithms::Rabin => RabinChunker::new(params).into(), Algorithms::Leap => LeapChunker::new(params).into(), - Algorithms::Super => UltraChunker::new(params).into(), - Algorithms::Ultra => SuperChunker::new(params).into(), + Algorithms::Super => SuperChunker::new(params).into(), + Algorithms::Ultra => UltraChunker::new(params).into(), } } diff --git a/benches/fuse_write_read.rs b/benches/fuse_write_read.rs new file mode 100644 index 00000000..5633e193 --- /dev/null +++ b/benches/fuse_write_read.rs @@ -0,0 +1,256 @@ +use cdc_chunkers::SizeParams; +use chunkfs::bench::Dataset; +use chunkfs::chunkers::{LeapChunker, RabinChunker, SuperChunker, UltraChunker}; +use chunkfs::hashers::Sha256Hasher; +use chunkfs::{ChunkerRef, FuseFS, IOC_GET_AVG_CHUNK_SIZE, IOC_GET_DEDUP_RATIO, MB}; +use criterion::measurement::WallTime; +use criterion::{BatchSize, BenchmarkGroup, BenchmarkId, Criterion, Throughput}; +use fuser::MountOption::AutoUnmount; +use libc::O_DIRECT; +use std::collections::HashMap; +use std::fs; +use std::fs::{File, OpenOptions}; +use std::io::{Read, Write}; +use std::os::fd::AsRawFd; +use std::os::unix::fs::OpenOptionsExt; +use std::path::Path; + +const SAMPLE_SIZE: usize = 30; + +#[allow(dead_code)] +#[derive(Copy, Clone, Debug)] +enum Algorithms { + Rabin, + Leap, + Super, + Ultra, +} + +#[allow(dead_code)] +fn chunkers() -> Vec { + vec![ + Algorithms::Rabin, + Algorithms::Leap, + Algorithms::Super, + Algorithms::Ultra, + ] +} + +#[allow(dead_code)] +fn get_chunker(algorithm: Algorithms, params: SizeParams) -> ChunkerRef { + match algorithm { + Algorithms::Rabin => RabinChunker::new(params).into(), + Algorithms::Leap => LeapChunker::new(params).into(), + Algorithms::Super => SuperChunker::new(params).into(), + Algorithms::Ultra => UltraChunker::new(params).into(), + } +} + +#[allow(dead_code)] +fn get_default_sizes(algorithm: Algorithms) -> SizeParams { + match algorithm { + Algorithms::Rabin => SizeParams::rabin_default(), + Algorithms::Leap => SizeParams::leap_default(), + Algorithms::Super => SizeParams::super_default(), + Algorithms::Ultra => SizeParams::ultra_default(), + } +} + +pub fn bench(c: &mut Criterion) { + let datasets = vec![Dataset::new("archX4.tar", "archX4").unwrap()]; + + for dataset in datasets { + let mut group = c.benchmark_group("FuseChunkers"); + group.sample_size(SAMPLE_SIZE); + group.throughput(Throughput::Bytes(dataset.size as u64)); + + for chunker in chunkers() { + let params = get_default_sizes(chunker); + dedup_ratio_and_avg_chunk(&dataset, chunker, params); + } + + for chunker in chunkers() { + let params = get_default_sizes(chunker); + bench_write(&dataset, &mut group, chunker, params); + } + + for chunker in chunkers() { + let params = get_default_sizes(chunker); + bench_read(&dataset, &mut group, chunker, params); + } + } +} + +fn dedup_ratio_and_avg_chunk(dataset: &Dataset, algorithm: Algorithms, params: SizeParams) { + let mount_point = Path::new("mount_dir/mount_point"); + let db = HashMap::default(); + let chunker = get_chunker(algorithm, params); + let fuse_fs = FuseFS::new(db, Sha256Hasher::default(), chunker); + + fs::create_dir_all(mount_point).unwrap(); + let session = fuser::spawn_mount2(fuse_fs, mount_point, &[AutoUnmount]).unwrap(); + + let fuse_path = mount_point.join("file"); + let mut fuse_file = OpenOptions::new() + .write(true) + .read(true) + .create(true) + .custom_flags(O_DIRECT) + .truncate(true) + .open(&fuse_path) + .unwrap(); + + let mut source = File::open(&dataset.path).unwrap(); + let mut buf = vec![0; 50 * MB]; + loop { + let bytes_read = source.read(&mut buf).unwrap(); + if bytes_read == 0 { + break; + } + fuse_file.write_all(&buf[..bytes_read]).unwrap(); + } + fuse_file.flush().unwrap(); + + let mut dedup_ratio = [0u8; size_of::()]; + let mut avg_chunk_size = [0u8; size_of::()]; + unsafe { libc::ioctl(fuse_file.as_raw_fd(), IOC_GET_DEDUP_RATIO, &mut dedup_ratio) }; + unsafe { + libc::ioctl( + fuse_file.as_raw_fd(), + IOC_GET_AVG_CHUNK_SIZE, + &mut avg_chunk_size, + ) + }; + + let dedup_ratio = f64::from_ne_bytes(dedup_ratio); + let avg_chunk_size = usize::from_ne_bytes(avg_chunk_size); + println!("{:?}-{} dedup ratio: {:.2}", algorithm, params, dedup_ratio); + println!( + "{:?}-{} average chunk size: {}", + algorithm, params, avg_chunk_size + ); + + drop(session) +} + +fn bench_write( + dataset: &Dataset, + group: &mut BenchmarkGroup, + algorithm: Algorithms, + params: SizeParams, +) { + let bench_name = dataset.path.clone(); + let parameter = format!("write_fuse-{:?}-{}", algorithm, params); + group.bench_function(BenchmarkId::new(bench_name, parameter), |b| { + b.iter_batched( + || { + let mount_point = Path::new("mount_dir/mount_point"); + let db = HashMap::default(); + let chunker = get_chunker(algorithm, params); + let fuse_fs = FuseFS::new(db, Sha256Hasher::default(), chunker); + + fs::create_dir_all(mount_point).unwrap(); + let session = fuser::spawn_mount2(fuse_fs, mount_point, &[AutoUnmount]).unwrap(); + + let fuse_path = mount_point.join("file"); + let fuse_file = OpenOptions::new() + .write(true) + .read(true) + .create(true) + .custom_flags(O_DIRECT) + .truncate(true) + .open(&fuse_path) + .unwrap(); + + let source = File::open(&dataset.path).unwrap(); + + (session, source, fuse_file) + }, + |(_session, mut source, mut fuse_file)| { + let mut buf = vec![0; 50 * MB]; + loop { + let bytes_read = source.read(&mut buf).unwrap(); + if bytes_read == 0 { + break; + } + fuse_file.write_all(&buf[..bytes_read]).unwrap(); + } + fuse_file.flush().unwrap(); + drop(fuse_file); + }, + BatchSize::PerIteration, + ) + }); +} + +fn bench_read( + dataset: &Dataset, + group: &mut BenchmarkGroup, + algorithm: Algorithms, + params: SizeParams, +) { + let mount_point = Path::new("mount_dir/mount_point"); + let db = HashMap::default(); + let chunker = get_chunker(algorithm, params); + let fuse_fs = FuseFS::new(db, Sha256Hasher::default(), chunker); + + fs::create_dir_all(mount_point).unwrap(); + let _session = fuser::spawn_mount2(fuse_fs, mount_point, &[AutoUnmount]).unwrap(); + + let fuse_path = mount_point.join("file"); + let mut fuse_file = OpenOptions::new() + .write(true) + .read(true) + .create(true) + .truncate(true) + .open(&fuse_path) + .unwrap(); + + let mut source = File::open(&dataset.path).unwrap(); + + let mut buf = vec![0; 50 * MB]; + loop { + let bytes_read = source.read(&mut buf).unwrap(); + if bytes_read == 0 { + break; + } + fuse_file.write_all(&buf[..bytes_read]).unwrap(); + } + fuse_file.flush().unwrap(); + drop(fuse_file); + + let bench_name = dataset.path.clone(); + let parameter = format!("read_fuse-{:?}-{}", algorithm, params); + group.bench_function(BenchmarkId::new(bench_name, parameter), |b| { + b.iter_batched( + || { + OpenOptions::new() + .read(true) + .custom_flags(O_DIRECT) + .open(&fuse_path) + .unwrap() + }, + |mut fuse_file| { + let mut buf = vec![0; 50 * MB]; + loop { + let bytes_read = fuse_file.read(&mut buf).unwrap(); + if bytes_read == 0 { + break; + } + } + }, + BatchSize::PerIteration, + ) + }); +} + +pub fn benches() { + let mut criterion: Criterion<_> = Criterion::default().configure_from_args(); + bench(&mut criterion); +} + +fn main() { + benches(); + + Criterion::default().configure_from_args().final_summary(); +} diff --git a/benches/write_read.rs b/benches/write_read.rs index 2abab07f..48ef0799 100644 --- a/benches/write_read.rs +++ b/benches/write_read.rs @@ -33,8 +33,8 @@ fn get_chunker(algorithm: Algorithms) -> ChunkerRef { match algorithm { Algorithms::Rabin => RabinChunker::default().into(), Algorithms::Leap => LeapChunker::default().into(), - Algorithms::Super => UltraChunker::default().into(), - Algorithms::Ultra => SuperChunker::default().into(), + Algorithms::Super => SuperChunker::default().into(), + Algorithms::Ultra => UltraChunker::default().into(), } } diff --git a/examples/mount.rs b/examples/mount.rs new file mode 100644 index 00000000..2d9d8d30 --- /dev/null +++ b/examples/mount.rs @@ -0,0 +1,48 @@ +use chunkfs::chunkers::SuperChunker; +use chunkfs::hashers::SimpleHasher; +use chunkfs::{FuseFS, MB}; +use fuser::MountOption::AutoUnmount; +use std::collections::HashMap; +use std::fs; +use std::fs::OpenOptions; +use std::io::{Read, Seek, SeekFrom, Write}; +use std::os::unix::fs::FileExt; +use std::path::Path; + +fn main() { + let mount_point = Path::new("./mount_point"); + let db = HashMap::default(); + let fuse_fs = FuseFS::new(db, SimpleHasher, SuperChunker::default()); + + fs::create_dir_all(mount_point).unwrap(); + + let session = fuser::spawn_mount2(fuse_fs, mount_point, &[AutoUnmount]).unwrap(); + + let file_path = mount_point.join("file"); + // careful: writing is sequential only + let mut file = OpenOptions::new() + .write(true) + .read(true) + .create(true) + .truncate(true) + .open(&file_path) + .unwrap(); + + let data1 = vec![1u8; 2 * MB]; + let data2 = vec![2u8; 5 * MB]; + file.write_all(&data1).unwrap(); + file.write_all(&data2).unwrap(); + + let expected: Vec = vec![1, 1, 1, 1, 2, 2, 2]; + let mut actual = vec![0u8; expected.len()]; + file.read_exact_at(&mut actual, 2 * MB as u64 - 4).unwrap(); + assert_eq!(expected, actual); + + file.seek(SeekFrom::Start(0)).unwrap(); + let mut file_data = vec![0u8; 7 * MB]; + let read_size = file.read_to_end(&mut file_data).unwrap(); + assert_eq!(read_size, 7 * MB); + + drop(session); + fs::remove_dir_all(mount_point).unwrap(); +} diff --git a/src/lib.rs b/src/lib.rs index 16acc3a8..12019292 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,6 +5,7 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; pub use system::database::{ContainerDatabase, Database, IterableDatabase}; +pub use system::fuse_filesystem::{FuseFS, IOC_GET_AVG_CHUNK_SIZE, IOC_GET_DEDUP_RATIO}; pub use system::scrub::{ ClusteringMeasurements, CopyScrubber, DumbScrubber, Scrub, ScrubMeasurements, }; @@ -73,15 +74,15 @@ impl Chunk { /// Chunks that are found are returned by [`chunk_data`][Chunker::chunk_data] method. /// If some contents were cut because the end of `data` and not the end of the chunk was reached, /// it must be returned with [`rest`][Chunker::rest] instead of storing it in the [`chunk_data`][Chunker::chunk_data]'s output. -pub trait Chunker: Debug { - /// Goes through whole `data` and finds chunks. If last chunk is not actually a chunk but a leftover, +pub trait Chunker: Debug + Send { + /// Goes through whole `data` and finds chunks. If the last chunk is not actually a chunk but a leftover, /// it is returned via [`rest`][Chunker::rest] method and is not contained in the vector. /// /// `empty` is an empty vector whose capacity is determined by [`estimate_chunk_count`][Chunker::estimate_chunk_count]. - /// Resulting chunks should be written right to it, and it should be returned as result. + /// Resulting chunks should be written right to it, and it should be returned as a result. fn chunk_data(&mut self, data: &[u8], empty: Vec) -> Vec; - /// Returns an estimate amount of chunks that will be created once the algorithm runs through the whole + /// Returns an estimate number of chunks that will be created once the algorithm runs through the whole /// data buffer. Used to pre-allocate the buffer with the required size so that allocation times are not counted /// towards total chunking time. fn estimate_chunk_count(&self, data: &[u8]) -> usize; @@ -127,7 +128,7 @@ impl Debug for ChunkerRef { } /// Functionality for an object that hashes the input. -pub trait Hasher { +pub trait Hasher: Send { /// Hash type that would be returned by the hasher. type Hash: ChunkHash; diff --git a/src/system/database.rs b/src/system/database.rs index e434754b..aafea30a 100644 --- a/src/system/database.rs +++ b/src/system/database.rs @@ -14,7 +14,7 @@ pub trait Database { /// Retrieves a value by a given key. Note that it returns a value, not a reference. /// /// # Errors - /// Should return [ErrorKind::NotFound], if the key-value pair + /// Should return [ErrorKind::NotFound] if the key-value pair /// was not found in the storage. fn get(&self, key: &K) -> io::Result; @@ -27,7 +27,7 @@ pub trait Database { } /// Retrieves a multitude of values, corresponding to the keys, in the correct order. - fn get_multi(&self, keys: &[K]) -> io::Result> { + fn get_multi(&self, keys: &[&K]) -> io::Result> { keys.iter().map(|key| self.get(key)).collect() } diff --git a/src/system/file_layer.rs b/src/system/file_layer.rs index 2fc0462d..3a7189f9 100644 --- a/src/system/file_layer.rs +++ b/src/system/file_layer.rs @@ -1,10 +1,12 @@ +use std::cmp::min; use std::collections::HashMap; +use std::hash::Hash; use std::io; use std::io::ErrorKind; use crate::system::storage::SpansInfo; +use crate::WriteMeasurements; use crate::{ChunkHash, ChunkerRef}; -use crate::{WriteMeasurements, SEG_SIZE}; /// Hashed span, starting at `offset`. #[derive(Debug, PartialEq, Eq, Default, Clone, Hash)] @@ -14,6 +16,20 @@ pub struct FileSpan { len: usize, } +impl FileSpan { + pub fn hash(&self) -> &Hash { + &self.hash + } + + pub fn offset(&self) -> usize { + self.offset + } + + pub fn len(&self) -> usize { + self.len + } +} + /// A named file, doesn't store actual contents, /// but rather hashes for them. #[derive(Clone)] @@ -68,12 +84,21 @@ impl FileHandle { } } + pub fn offset(&self) -> usize { + self.offset + } + + /// Sets the offset for the file handle. + pub fn set_offset(&mut self, offset: usize) { + self.offset = offset; + } + /// Returns name of the file. pub fn name(&self) -> &str { &self.file_name } - /// Closes handle and returns [`WriteMeasurements`] made while file was open. + /// Closes the handle and returns [`WriteMeasurements`] made while a file was open. pub(crate) fn close(self) -> WriteMeasurements { self.measurements } @@ -124,12 +149,9 @@ impl FileLayer { } /// Reads all hashes of the file, from beginning to end. - pub fn read_complete(&self, handle: &FileHandle) -> Vec { + pub fn read_complete(&self, handle: &FileHandle) -> Vec<&Hash> { let file = self.find_file(handle); - file.spans - .iter() - .map(|span| span.hash.clone()) // cloning hashes, takes a lot of time - .collect() + file.spans.iter().map(|span| span.hash()).collect() } /// Writes spans to the end of the file. @@ -147,31 +169,29 @@ impl FileLayer { handle.measurements += info.measurements; } - /// Reads 1 MB of data from the open file and returns received hashes, - /// starting point is based on the `FileHandle`'s offset. - pub fn read(&self, handle: &mut FileHandle) -> Vec { + /// Reads the specified amount of data from the open file and + /// returns FileSpans, in which the necessary data is stored (the side FileSpans may contain it partially). + /// The Starting point is based on the `FileHandle`'s offset. + /// + /// If `size` + file handle offset is greater than file size, then returns FileSpans up to the end of the file. + pub fn read(&self, handle: &mut FileHandle, size: usize) -> Vec<&FileSpan> { let file = self.find_file(handle); - let mut bytes_read = 0; - let hashes = file + let spans: Vec<_> = file .spans .iter() - .skip_while(|span| span.offset < handle.offset) // find current span in the file - .take_while(|span| { - bytes_read += span.len; - if bytes_read > SEG_SIZE { - bytes_read -= span.len; - false - } else { - true - } - }) // take 1 MB of spans after current one - .map(|span| span.hash.clone()) // take their hashes + .skip_while(|span| span.offset + span.len < handle.offset) // find the first span that contains required data + .take_while(|span| span.offset < handle.offset + size) .collect(); - handle.offset += bytes_read; + if spans.is_empty() { + return spans; + } - hashes + let last_span = spans.last().unwrap(); + let read_size_possible = last_span.offset + last_span.len - handle.offset; + handle.offset += min(read_size_possible, size); + spans } /// Checks if the file with the given name exists. @@ -206,7 +226,8 @@ impl FileLayer { } #[cfg(feature = "bench")] - /// Generate a new dataset with set deduplication ratio from the existing one. + /// Generate a new dataset with + /// a set deduplication ratio from the existing one. /// /// Returns the name of the new file. pub fn get_to_dedup_ratio(&mut self, name: &str, dedup_ratio: f64) -> io::Result { diff --git a/src/system/fuse_filesystem.rs b/src/system/fuse_filesystem.rs new file mode 100644 index 00000000..581e619b --- /dev/null +++ b/src/system/fuse_filesystem.rs @@ -0,0 +1,761 @@ +use crate::system::file_layer::FileHandle; +use crate::{ + create_cdc_filesystem, ChunkHash, ChunkerRef, DataContainer, FileSystem, Hasher, + IterableDatabase, MB, +}; +use fuser::consts::FUSE_BIG_WRITES; +use fuser::FileType::RegularFile; +use fuser::TimeOrNow::Now; +use fuser::{ + FileAttr, FileType, Filesystem, KernelConfig, ReplyAttr, ReplyCreate, ReplyData, + ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyIoctl, ReplyOpen, ReplyWrite, Request, TimeOrNow, +}; +use libc::c_int; +use std::cmp::min; +use std::collections::HashMap; +use std::ffi::OsStr; +use std::io; +use std::time::{Duration, SystemTime}; + +/// The File is opened for execution. +const FMODE_EXEC: i32 = 0x20; +/// Total cache size of all files, after which all caches will be [`dropped and shrank`][FuseFS::drop_and_shrink_cache] to the [`underlying filesystem`][FileSystem]. +const FILESYSTEM_CACHE_MAX_SIZE: usize = 250 * MB; +/// Maximum cache size, after which it will drop [`dropped and shrank`][FuseFS::drop_and_shrink_cache] to the [`underlying filesystem`][FileSystem]. +const FILE_CACHE_MAX_SIZE: usize = 100 * MB; + +/// Command number for filesystem dedup ratio request via ioctl. +pub const IOC_GET_DEDUP_RATIO: u64 = nix::request_code_read!('S', 0, size_of::()); +/// Command number for filesystem average chunk size request via ioctl. +pub const IOC_GET_AVG_CHUNK_SIZE: u64 = nix::request_code_read!('S', 1, size_of::()); + +type Inode = u64; +type Fh = u64; + +/// [`FuseFS`] file entity. +#[derive(Clone)] +struct FuseFile { + /// File cache. Cache fills up with every [`write`][FuseFS::write] on a file's opened [`file handles`][FuseFileHandle], + /// until it reaches its [`maximum size`][FILE_CACHE_MAX_SIZE], + /// or the total size of the file system caches reaches the [`maximum size`][FILESYSTEM_CACHE_MAX_SIZE]. + cache: Vec, + /// File attributes. + attr: FileAttr, + /// File name. + name: String, + /// Number of opened file handles. + handles: u64, +} + +/// [`FuseFS`] file handle entity. +struct FuseFileHandle { + /// Underlying file handle. + underlying_file_handle: FileHandle, + /// Read permission. + read: bool, + /// Write permission. + write: bool, + /// Inode number with which the file handle is associated. + inode: Inode, +} + +/// Wrap around [`FileSystem`] for implementing [`Filesystem`] trait. +/// +/// After creation, it should be passed to [`mount2`][fuser::mount2] or [`spawn_mount2`][fuser::spawn_mount2]. +pub struct FuseFS +where + B: IterableDatabase>, + Hash: ChunkHash, +{ + /// Underlying [chunkfs][`FileSystem`]. + underlying_fs: FileSystem>>, + /// Files map <[`Inode`], [`FuseFile`]>. + files: HashMap, + /// File handles map <[`Fh`], [`FuseFileHandle`]>. + file_handles: HashMap, + /// Inodes map . + inodes: HashMap, + /// Number for the next file handle to be created. Increments on every [`FuseFS::open`]. + next_fh: u64, + /// Chunker set for the filesystem. + chunker: ChunkerRef, + /// Total cache size of opened file handles. + total_cache: usize, +} + +impl FuseFS +where + B: IterableDatabase>, + Hash: ChunkHash, +{ + /// Creates a file system with the given [`hasher`][Hasher], [`database`][Database] and [`chunker`][ChunkerRef]. + /// + /// After creation, it should be passed to [`mount2`][fuser::mount2] or [`spawn_mount2`][fuser::spawn_mount2]. + pub fn new(base: B, hasher: H, chunker: C) -> Self + where + H: Into + 'static>>, + C: Into, + { + let underlying_fs = create_cdc_filesystem(base, hasher); + + let uid = unsafe { libc::getuid() }; + let gid = unsafe { libc::getgid() }; + let now = SystemTime::now(); + let root_attr = FileAttr { + ino: 1, + size: 0, + blocks: 0, + atime: now, + mtime: now, + ctime: now, + crtime: now, + kind: FileType::Directory, + perm: 0o755, + nlink: 2, + uid, + gid, + rdev: 0, + flags: 0, + blksize: 512, + }; + let root_dir = FuseFile { + cache: Vec::new(), + attr: root_attr, + name: ".".to_string(), + handles: 0, + }; + let mut parent_dir = root_dir.clone(); + parent_dir.name = "..".to_string(); + parent_dir.attr.ino = 0; + let files = HashMap::from([(0, parent_dir), (1, root_dir)]); + + let inodes = HashMap::from([("..".to_string(), 0), (".".to_string(), 1)]); + Self { + underlying_fs, + files, + inodes, + file_handles: HashMap::default(), + next_fh: 0, + chunker: chunker.into(), + total_cache: 0, + } + } + + /// Get new inode number for a new file. After creation, the new inode is inserted into the inodes map, + /// incrementing the next new inode number. The filesystem does not provide a file removing, so the inodes do not overlap. + fn get_new_inode(&self) -> Inode { + self.inodes.len() as Inode + } + + /// Get new file handle number for a new file handle. Increments next fh counter on every call. + fn get_new_fh(&mut self) -> Fh { + let next_fh = self.next_fh; + self.next_fh += 1; + next_fh + } + + /// Writes cache content to the [`underlying filesystem`][FileSystem] and shrinks its size. + fn drop_and_shrink_cache(&mut self, file: Inode, handle: Fh) -> io::Result<()> { + self.drop_cache(file, handle)?; + + let file = self.files.get_mut(&file).ok_or(io::ErrorKind::NotFound)?; + self.total_cache -= file.cache.len(); + file.cache = vec![]; + Ok(()) + } + + /// Writes cache content to the [`underlying filesystem`][FileSystem]. + fn drop_cache(&mut self, file: Inode, handle: Fh) -> io::Result<()> { + let file = self.files.get_mut(&file).ok_or(io::ErrorKind::NotFound)?; + let handle = self + .file_handles + .get_mut(&handle) + .ok_or(io::ErrorKind::NotFound)?; + + handle + .underlying_file_handle + .set_offset(file.attr.size as usize - file.cache.len()); + self.underlying_fs + .write_to_file(&mut handle.underlying_file_handle, &file.cache)?; + + file.cache.clear(); + Ok(()) + } + + /// Writes all file caches content to the [`underlying filesystem`][FileSystem] and shrinks its size. + /// + /// Called after the [`maximum filesystem cache size`][FILESYSTEM_CACHE_MAX_SIZE] has been reached. + fn drop_and_shrink_caches(&mut self) -> io::Result<()> { + for handle in self.file_handles.values_mut() { + let file = self + .files + .get_mut(&handle.inode) + .ok_or(io::ErrorKind::NotFound)?; + handle + .underlying_file_handle + .set_offset(file.attr.size as usize - file.cache.len()); + self.underlying_fs + .write_to_file(&mut handle.underlying_file_handle, &file.cache)?; + + self.total_cache -= file.cache.len(); + file.cache = vec![]; + } + Ok(()) + } +} + +/// Checks the request rights for the file with the specified access mask (flags). +fn check_access(file_attr: &FileAttr, req: &Request, access_mask: i32) -> bool { + let file_uid = file_attr.uid; + let file_gid = file_attr.gid; + let file_mode = file_attr.perm; + let uid = req.uid(); + let gid = req.gid(); + + let mut access_mask = access_mask; + // F_OK tests for existence of a file + if access_mask == libc::F_OK { + return true; + } + let file_mode = i32::from(file_mode); + + // root is allowed to read and write anything + if uid == 0 { + // root only allowed to exec if one of the Exec bits is set + access_mask &= libc::X_OK; + access_mask -= access_mask & (file_mode >> 6); + access_mask -= access_mask & (file_mode >> 3); + access_mask -= access_mask & file_mode; + return access_mask == 0; + } + + if uid == file_uid { + access_mask -= access_mask & (file_mode >> 6); + } else if gid == file_gid { + access_mask -= access_mask & (file_mode >> 3); + } else { + access_mask -= access_mask & file_mode; + } + + access_mask == 0 +} + +impl Filesystem for FuseFS +where + B: IterableDatabase>, + Hash: ChunkHash, +{ + fn init(&mut self, _req: &Request<'_>, config: &mut KernelConfig) -> Result<(), c_int> { + let _ = config.add_capabilities(FUSE_BIG_WRITES); + if let Err(nearest) = config.set_max_write(128 * MB as u32) { + let _ = config.set_max_write(nearest); + }; + Ok(()) + } + + fn lookup(&mut self, req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEntry) { + let name = name.to_str().unwrap().to_owned(); + if parent != 1 { + reply.error(libc::EINVAL); + return; + } + + let Some(inode) = self.inodes.get::(&name) else { + reply.error(libc::ENOENT); + return; + }; + let parent_attr = self.files.get(&parent).unwrap().attr; + if !check_access(&parent_attr, req, libc::X_OK) { + reply.error(libc::EACCES); + return; + } + + let file = self.files.get(inode).unwrap(); + reply.entry(&Duration::new(0, 0), &file.attr, 0) + } + + fn getattr(&mut self, _req: &Request<'_>, ino: u64, _fh: Option, reply: ReplyAttr) { + match self.files.get(&ino) { + Some(file) => reply.attr(&Duration::new(0, 0), &file.attr), + None => reply.error(libc::ENOENT), + } + } + + fn setattr( + &mut self, + req: &Request<'_>, + ino: u64, + mode: Option, + _uid: Option, + _gid: Option, + _size: Option, + atime: Option, + mtime: Option, + ctime: Option, + _fh: Option, + _crtime: Option, + _chgtime: Option, + _bkuptime: Option, + _flags: Option, + reply: ReplyAttr, + ) { + let Some(file) = self.files.get_mut(&ino) else { + reply.error(libc::ENOENT); + return; + }; + + let now = SystemTime::now(); + let attr = &mut file.attr; + if let Some(mode) = mode { + if req.uid() != 0 && req.uid() != attr.uid { + reply.error(libc::EPERM); + return; + } else { + attr.perm = mode as u16; + } + attr.ctime = now; + reply.attr(&Duration::new(0, 0), &file.attr); + return; + } + + let set_time_with_check = |time: TimeOrNow| { + if attr.uid != req.uid() && req.uid() != 0 && time != Now { + return None; + } + + match time { + TimeOrNow::SpecificTime(time) => Some(time), + Now => Some(now), + } + }; + + if let Some(atime) = atime { + let Some(atime) = set_time_with_check(atime) else { + reply.error(libc::EPERM); + return; + }; + attr.atime = atime; + attr.ctime = now; + } + + if let Some(mtime) = mtime { + let Some(mtime) = set_time_with_check(mtime) else { + reply.error(libc::EPERM); + return; + }; + attr.mtime = mtime; + attr.ctime = now; + } + + if let Some(ctime) = ctime { + let Some(ctime) = set_time_with_check(TimeOrNow::SpecificTime(ctime)) else { + reply.error(libc::EPERM); + return; + }; + attr.ctime = ctime; + } + + reply.attr(&Duration::new(0, 0), attr); + } + + fn open(&mut self, req: &Request<'_>, ino: u64, flags: i32, reply: ReplyOpen) { + let Some(file) = self.files.get_mut(&ino) else { + reply.error(libc::ENOENT); + return; + }; + + let (access_mask, read, write) = match flags & libc::O_ACCMODE { + libc::O_RDONLY => { + if flags & libc::O_TRUNC != 0 { + reply.error(libc::EACCES); + return; + } + if flags & FMODE_EXEC != 0 { + // Open is from internal exec syscall + (libc::X_OK, true, false) + } else { + (libc::R_OK, true, false) + } + } + libc::O_WRONLY => (libc::W_OK, false, true), + libc::O_RDWR => (libc::R_OK | libc::W_OK, true, true), + // Exactly one access mode flag must be specified + _ => { + reply.error(libc::EINVAL); + return; + } + }; + + if !check_access(&file.attr, req, access_mask) { + reply.error(libc::EACCES); + return; + } + + let Ok(underlying_file_handle) = self + .underlying_fs + .open_file(&file.name, self.chunker.clone()) + else { + reply.error(libc::EBADF); + return; + }; + + let file_handle = FuseFileHandle { + underlying_file_handle, + inode: ino, + read, + write, + }; + file.handles += 1; + let fh = self.get_new_fh(); + self.file_handles.insert(fh, file_handle); + + reply.opened(fh, flags as u32) + } + + fn read( + &mut self, + req: &Request<'_>, + ino: u64, + fh: u64, + offset: i64, + size: u32, + _flags: i32, + _lock_owner: Option, + reply: ReplyData, + ) { + let Some(file_handle) = self.file_handles.get_mut(&fh) else { + reply.error(libc::EBADF); + return; + }; + if file_handle.inode != ino { + reply.error(libc::ESTALE); + return; + } + let Some(file) = self.files.get_mut(&ino) else { + reply.error(libc::ENOENT); + return; + }; + if offset < 0 { + reply.error(libc::EINVAL); + return; + } + let offset = offset as usize; + let size = size as usize; + + if !check_access(&file.attr, req, libc::R_OK) || !file_handle.read { + reply.error(libc::EACCES); + return; + } + let underlying_fh = &mut file_handle.underlying_file_handle; + underlying_fh.set_offset(offset); + + let now = SystemTime::now(); + file.attr.atime = now; + file.attr.ctime = now; + + let Ok(mut data) = self.underlying_fs.read(underlying_fh, size) else { + reply.error(libc::EIO); + return; + }; + + let read_size = data.len(); + let new_offset = offset + read_size; + underlying_fh.set_offset(new_offset); + + if read_size > size || file.cache.len() > file.attr.size as usize { + reply.error(libc::EIO); + return; + } + if read_size == size || new_offset >= file.attr.size as usize { + reply.data(&data); + return; + } + + let missing_size = size - read_size; + let disk_data_size = file.attr.size as usize - file.cache.len(); + if new_offset < disk_data_size { + reply.error(libc::EIO); + return; + } + + let cache_start_offset = new_offset - disk_data_size; + let cache_end_offset = min(file.cache.len(), cache_start_offset + missing_size); + data.extend_from_slice(&file.cache[cache_start_offset..cache_end_offset]); + let new_offset = offset + data.len(); + + underlying_fh.set_offset(new_offset); + reply.data(&data); + } + + fn write( + &mut self, + req: &Request<'_>, + ino: u64, + fh: u64, + offset: i64, + data: &[u8], + _write_flags: u32, + _flags: i32, + _lock_owner: Option, + reply: ReplyWrite, + ) { + let Some(file_handle) = self.file_handles.get_mut(&fh) else { + reply.error(libc::EBADF); + return; + }; + if file_handle.inode != ino { + reply.error(libc::ESTALE); + return; + } + let Some(file) = self.files.get_mut(&ino) else { + reply.error(libc::ENOENT); + return; + }; + if offset < 0 || offset as u64 != file.attr.size { + reply.error(libc::EINVAL); + return; + } + + if !check_access(&file.attr, req, libc::W_OK) || !file_handle.write { + reply.error(libc::EACCES); + return; + } + + let cache_cap_before = file.cache.capacity(); + file.cache.extend_from_slice(data); + let cache_cap_after = file.cache.capacity(); + self.total_cache += cache_cap_after - cache_cap_before; + file.attr.size += data.len() as u64; + + if file.cache.len() > FILE_CACHE_MAX_SIZE && self.drop_cache(ino, fh).is_err() { + reply.error(libc::EIO); + return; + } + if self.total_cache > FILESYSTEM_CACHE_MAX_SIZE && self.drop_and_shrink_caches().is_err() { + reply.error(libc::EIO); + return; + } + + let now = SystemTime::now(); + let file = self.files.get_mut(&ino).unwrap(); + file.attr.ctime = now; + file.attr.mtime = now; + + reply.written(data.len() as u32); + } + + fn flush( + &mut self, + _req: &Request<'_>, + ino: u64, + fh: u64, + _lock_owner: u64, + reply: ReplyEmpty, + ) { + let Some(file_handle) = self.file_handles.get_mut(&fh) else { + reply.error(libc::EBADF); + return; + }; + if file_handle.inode != ino { + reply.error(libc::ESTALE); + return; + } + let Some(_) = self.files.get_mut(&ino) else { + reply.error(libc::ENOENT); + return; + }; + if !file_handle.write && !file_handle.read { + reply.error(libc::EACCES); + return; + } + + if self.drop_cache(ino, fh).is_err() { + reply.error(libc::EIO); + return; + } + reply.ok() + } + + fn release( + &mut self, + _req: &Request<'_>, + ino: u64, + fh: u64, + _flags: i32, + _lock_owner: Option, + _flush: bool, + reply: ReplyEmpty, + ) { + let Some(file) = self.files.get(&ino) else { + reply.error(libc::EINVAL); + return; + }; + if file.handles == 0 { + reply.error(libc::EINVAL); + return; + } + if !self.file_handles.contains_key(&fh) { + reply.error(libc::EINVAL); + return; + } + + if self.drop_and_shrink_cache(ino, fh).is_err() { + reply.error(libc::EIO); + return; + } + + let Some(file_handle) = self.file_handles.remove(&fh) else { + reply.error(libc::EINVAL); + return; + }; + file_handle.underlying_file_handle.close(); + let file = self.files.get_mut(&ino).unwrap(); + file.handles -= 1; + reply.ok() + } + + fn readdir( + &mut self, + req: &Request<'_>, + ino: u64, + _fh: u64, + offset: i64, + mut reply: ReplyDirectory, + ) { + if ino != 1 { + reply.error(libc::EINVAL); + return; + } + let dir = self.files.get(&ino).unwrap(); + if !check_access(&dir.attr, req, libc::R_OK) { + reply.error(libc::EACCES); + return; + } + + let entries = self + .files + .iter() + .map(|(inode, file)| (inode, file.attr.kind, &file.name)); + for (i, entry) in entries.enumerate().skip(offset as usize) { + let (inode, kind, name) = entry; + if reply.add(*inode, offset + i as i64 + 1, kind, name) { + break; + } + } + + reply.ok() + } + + fn create( + &mut self, + req: &Request<'_>, + parent: u64, + name: &OsStr, + mode: u32, + umask: u32, + flags: i32, + reply: ReplyCreate, + ) { + let name = name.to_str().unwrap().to_owned(); + if parent != 1 { + reply.error(libc::EINVAL); + return; + } + let Ok(underlying_file_handle) = self + .underlying_fs + .create_file(name.clone(), self.chunker.clone()) + else { + reply.error(libc::EEXIST); + return; + }; + + let ino = self.get_new_inode(); + let now = SystemTime::now(); + let attr = FileAttr { + ino, + size: 0, + blocks: 0, + atime: now, + mtime: now, + ctime: now, + crtime: now, + kind: RegularFile, + perm: (mode & !umask) as u16, + nlink: 1, + uid: req.uid(), + gid: req.gid(), + rdev: 000, + blksize: 000, + flags: flags as u32, + }; + + let (read, write) = match flags & libc::O_ACCMODE { + libc::O_RDONLY => (true, false), + libc::O_WRONLY => (false, true), + libc::O_RDWR => (true, true), + // Exactly one access mode flag must be specified + _ => { + reply.error(libc::EINVAL); + return; + } + }; + + let file_handle = FuseFileHandle { + underlying_file_handle, + inode: ino, + read, + write, + }; + let file = FuseFile { + cache: Vec::new(), + attr, + name: name.clone(), + handles: 1, + }; + + let fh = self.get_new_fh(); + self.files.insert(ino, file.clone()); + self.inodes.insert(name, ino); + self.file_handles.insert(fh, file_handle); + + reply.created(&Duration::new(0, 0), &file.attr, 0, fh, flags as u32); + } + + fn ioctl( + &mut self, + _req: &Request<'_>, + ino: u64, + fh: u64, + _flags: u32, + cmd: u32, + _in_data: &[u8], + _out_size: u32, + reply: ReplyIoctl, + ) { + let Some(file_handle) = self.file_handles.get_mut(&fh) else { + reply.error(libc::EBADF); + return; + }; + if file_handle.inode != ino { + reply.error(libc::ESTALE); + return; + } + if self.drop_and_shrink_caches().is_err() { + reply.error(libc::EIO); + return; + } + match cmd.into() { + IOC_GET_AVG_CHUNK_SIZE => { + let avg_chunk_size = self.underlying_fs.average_chunk_size(); + reply.ioctl(0, &avg_chunk_size.to_ne_bytes()); + } + IOC_GET_DEDUP_RATIO => { + let dedup_ratio = self.underlying_fs.cdc_dedup_ratio(); + reply.ioctl(0, &dedup_ratio.to_ne_bytes()); + } + _ => { + reply.error(libc::EINVAL); + } + } + } +} diff --git a/src/system/mod.rs b/src/system/mod.rs index c37f09d3..086254ad 100644 --- a/src/system/mod.rs +++ b/src/system/mod.rs @@ -1,3 +1,4 @@ +use std::cmp::min; use std::collections::HashMap; use std::io; use std::io::Write; @@ -8,10 +9,11 @@ use file_layer::{FileHandle, FileLayer}; use scrub::{Scrub, ScrubMeasurements}; use storage::{ChunkStorage, DataContainer}; -use super::{ChunkHash, ChunkerRef, Hasher, WriteMeasurements}; +use super::{ChunkHash, ChunkerRef, Hasher, WriteMeasurements, SEG_SIZE}; pub mod database; pub mod file_layer; +pub mod fuse_filesystem; pub mod scrub; pub mod storage; @@ -151,12 +153,43 @@ where Ok(self.storage.retrieve(&hashes)?.concat()) // it assumes that all retrieved data segments are in correct order } - /// Reads at most 1 MB of data from a file and returns it. + /// Reads 1 MB of data from a file and returns it. + /// If file handle offset + 1 MB is greater than file size, then returns data starting from the offset to the end of the file /// /// **Careful:** it modifies internal `FileHandle` data. After using this `write_to_file` should not be used on the same FileHandle. pub fn read_from_file(&self, handle: &mut FileHandle) -> io::Result> { - let hashes = self.file_layer.read(handle); - Ok(self.storage.retrieve(&hashes)?.concat()) + self.read(handle, SEG_SIZE) + } + + /// Reads the specified amount of data from the open file. + /// Starting point is based on the `FileHandle`'s offset. + /// + /// If `size` + file handle offset is greater than file size, then returns data starting from the offset to the end of the file + /// **Careful:** it modifies internal `FileHandle` data. After using this `write_to_file` should not be used on the same FileHandle. + pub fn read(&self, handle: &mut FileHandle, size: usize) -> io::Result> { + let original_offset = handle.offset(); + let spans = self.file_layer.read(handle, size); + if spans.is_empty() { + return Ok(vec![]); + } + let hashes: Vec<_> = spans.iter().map(|span| span.hash()).collect(); + let mut data_vectors = self.storage.retrieve(&hashes)?; + + // Since we read by offset, which may be somewhere in the middle of the FileSpan offset, + // we need to remove the extra at the beginning of the first span + let first_span = spans.first().unwrap(); + let data_extra_start = original_offset - first_span.offset(); // amount of data from start to be removed + data_vectors.first_mut().unwrap().drain(0..data_extra_start); + + // Same, we need to remove the extra at the end of the last span. + let last_span = spans.last().unwrap(); + let read_size_possible = last_span.offset() + last_span.len() - original_offset; + let read_size_actual = min(size, read_size_possible); + let data_extra_end = read_size_possible - read_size_actual; // amount of data from end to be removed + let last_vector = data_vectors.last_mut().unwrap(); + last_vector.truncate(last_vector.len() - data_extra_end); + + Ok(data_vectors.concat()) } /// Gives out a distribution of the chunks with the same hash for the given file. diff --git a/src/system/scrub.rs b/src/system/scrub.rs index 19d4e143..9c53dfcc 100644 --- a/src/system/scrub.rs +++ b/src/system/scrub.rs @@ -19,7 +19,7 @@ use super::storage::DataContainer; /// /// After moving the data from `database` to `target_map`, we should be able to have access to it via the `database`. /// Therefore, after moving, we should leave a `Vec` in place of the source chunk. It is done via [DataContainer::make_target] method. -/// Not using it will lead to either not getting any benefits from the algorithm, or to being unable to access the initial chunk anymore, if it was deleted. +/// Not using it will lead to either not getting any benefits from the algorithm or to being unable to access the initial chunk anymore if it is deleted. /// /// # Arguments /// The only method [scrub][Scrub::scrub] takes two arguments: @@ -29,7 +29,7 @@ use super::storage::DataContainer; /// 2. A target map, which contains `Key`-`Vec` pairs, where `Key` is a generic value determined by the implementation. /// The way data is stored is determined by the target map implementation, the only information known to the scrubber is that /// the target map implements [Database] trait. It should only be used for storage purposes and not contain any algorithm logic. -pub trait Scrub +pub trait Scrub: Send where Hash: ChunkHash, B: IterableDatabase>, @@ -41,7 +41,7 @@ where /// [DataContainer::extract] or [DataContainer::extract_mut] should be used. /// /// If the chunk is suitable for being transferred to the `target_map`, it should NOT be deleted, but instead be replaced by the `target_map`'s keys, - /// using which the original chunk can be restored. This is accomplished by the [DataContainer::make_target] method. + /// using which the original chunk can be restored. This is achieved by the [DataContainer::make_target] method. /// /// It should also gather information to return the [measurements][ScrubMeasurements]. /// @@ -56,7 +56,7 @@ where /// /// # CDC Database /// We should be able to iterate over the `database` to process all chunks we had stored before. - /// The [IntoIterator] trait should be implemented for `database`, but it should not be a big concern, because the only structure that should be implemented + /// The [IntoIterator] trait should be implemented for `database`, but it should not be a big concern because the only structure that should be implemented /// for the algorithm is the scrubber itself. `database` should be considered a given entity, along with the `target_map`. fn scrub<'a>(&mut self, database: &mut B, target_map: &mut T) -> io::Result where diff --git a/src/system/storage.rs b/src/system/storage.rs index 1010557a..ffe242af 100644 --- a/src/system/storage.rs +++ b/src/system/storage.rs @@ -73,8 +73,8 @@ where /// Writes 1 MB of data to the [`base`][crate::base::Base] storage after deduplication. /// - /// Returns resulting lengths of [chunks][crate::chunker::Chunk] with corresponding hash, - /// along with amount of time spent on chunking and hashing. + /// Returns the resulting lengths of [chunks][crate::chunker::Chunk] with the corresponding hash, + /// along with the amount of time spent on chunking and hashing. pub fn write(&mut self, data: &[u8], chunker: &ChunkerRef) -> io::Result>> { let mut writer = StorageWriter::new(chunker, &mut self.hasher); @@ -137,8 +137,8 @@ where } /// Retrieves the data from the storage based on hashes of the data [`segments`][Segment], - /// or Error(NotFound) if some of the hashes were not present in the base. - pub fn retrieve(&self, request: &[Hash]) -> io::Result>> { + /// or Error(NotFound) if some hashes were not present in the base. + pub fn retrieve(&self, request: &[&Hash]) -> io::Result>> { let retrieved = self.database.get_multi(request)?; retrieved @@ -147,7 +147,7 @@ where Data::Chunk(chunk) => Ok(chunk.clone()), Data::TargetChunk(keys) => Ok(self .target_map - .get_multi(keys)? + .get_multi(&keys.iter().collect::>())? .into_iter() .flatten() .collect()), @@ -199,7 +199,7 @@ where }) } - /// Calculates deduplication ratio of the storage, not accounting for chunks processed with scrubber. + /// Calculates a deduplication ratio of the storage, not accounting for chunks processed with scrubber. pub fn cdc_dedup_ratio(&self) -> f64 { (self.size_written as f64) / (self.total_cdc_size() as f64) } @@ -297,8 +297,8 @@ where /// Writes 1 MB of data to the [`base`][crate::base::Base] storage after deduplication. /// - /// Returns resulting lengths of [chunks][crate::chunker::Chunk] with corresponding hash, - /// along with amount of time spent on chunking and hashing. + /// Returns the resulting lengths of [chunks][crate::chunker::Chunk] with the corresponding hash, + /// along with the amount of time spent on chunking and hashing. fn write>>( &mut self, data: &[u8], diff --git a/tests/filesystem.rs b/tests/filesystem.rs index 383dda18..e8c37922 100644 --- a/tests/filesystem.rs +++ b/tests/filesystem.rs @@ -11,6 +11,37 @@ use chunkfs::hashers::{Sha256Hasher, SimpleHasher}; use chunkfs::{create_cdc_filesystem, ChunkerRef, DataContainer, Database, WriteMeasurements}; const MB: usize = 1024 * 1024; +const KB: usize = 1024; + +#[test] +fn write_read_with_strange_size() { + let mut fs = create_cdc_filesystem(HashMap::default(), SimpleHasher); + + let mut handle = fs.create_file("file", FSChunker::new(4096)).unwrap(); + + let mut data1 = vec![0; 433 * KB]; + let data2 = vec![1; 2 * MB + 3]; + let data3 = vec![2; 2 * MB]; + fs.write_to_file(&mut handle, &data1).unwrap(); + fs.write_to_file(&mut handle, &data2).unwrap(); + fs.write_to_file(&mut handle, &data3).unwrap(); + fs.close_file(handle).unwrap(); + + let mut handle = fs.open_file_readonly("file").unwrap(); + + data1.append(&mut vec![1; MB]); + let actual = fs.read(&mut handle, 433 * KB + MB).unwrap(); + assert_eq!(actual, data1); + + let mut ones_w_twos = vec![1; MB + 3]; + ones_w_twos.append(&mut vec![2; MB]); + let actual = fs.read(&mut handle, 2 * MB + 3).unwrap(); + assert_eq!(actual, ones_w_twos); + + handle.set_offset(433 * KB + 2 * MB + 3 + MB / 2); + let actual = fs.read(&mut handle, 10 * MB).unwrap(); + assert_eq!(actual, vec![2; MB + MB / 2]); +} #[test] fn write_read_complete_test() { @@ -60,7 +91,7 @@ fn write_read_blocks_test() { buffer.extend_from_slice(&buf); } assert_eq!(buffer.len(), MB * 3 + 50); - assert!(complete == buffer); + assert_eq!(complete, buffer); assert_eq!(fs.read_from_file(&mut handle).unwrap(), []); } @@ -116,11 +147,11 @@ fn non_iterable_database_can_be_used_with_fs() { struct DummyDatabase; impl Database, DataContainer<()>> for DummyDatabase { - fn insert(&mut self, _key: Vec, _value: DataContainer<()>) -> std::io::Result<()> { + fn insert(&mut self, _key: Vec, _value: DataContainer<()>) -> io::Result<()> { unimplemented!() } - fn get(&self, _key: &Vec) -> std::io::Result> { + fn get(&self, _key: &Vec) -> io::Result> { unimplemented!() } diff --git a/tests/fuse_filesystem.rs b/tests/fuse_filesystem.rs new file mode 100644 index 00000000..7a46d819 --- /dev/null +++ b/tests/fuse_filesystem.rs @@ -0,0 +1,883 @@ +use cdc_chunkers::SizeParams; +use chunkfs::chunkers::{LeapChunker, SuperChunker}; +use chunkfs::hashers::Sha256Hasher; +use chunkfs::{ChunkerRef, FuseFS, IOC_GET_AVG_CHUNK_SIZE, IOC_GET_DEDUP_RATIO, MB}; +use filetime::FileTime; +use fuser::BackgroundSession; +use fuser::MountOption::AutoUnmount; +use libc::O_DIRECT; +use std::collections::HashMap; +use std::ffi::OsString; +use std::fs; +use std::fs::{File, OpenOptions, Permissions}; +use std::io::{Read, Write}; +use std::os::fd::AsRawFd; +use std::os::unix::fs::{FileExt, MetadataExt, OpenOptionsExt, PermissionsExt}; +use std::path::{Path, PathBuf}; +use std::time::SystemTime; +use uuid::Uuid; + +fn generate_unique_mount_point() -> String { + Uuid::new_v4().to_string() +} + +struct FuseFixture { + mount_point: PathBuf, + fuse_session: Option, +} + +impl FuseFixture { + fn default() -> Self { + Self::with_chunker(SuperChunker::default()) + } + + fn with_chunker(chunker: C) -> Self + where + C: Into, + { + let mount_dir = Path::new("mount_dir"); + let mount_point = mount_dir.join(generate_unique_mount_point()); + let db = HashMap::default(); + let fuse_fs = FuseFS::new(db, Sha256Hasher::default(), chunker); + fs::create_dir_all(&mount_point).unwrap(); + + let fuse_session = fuser::spawn_mount2(fuse_fs, &mount_point, &[AutoUnmount]).unwrap(); + + Self { + mount_point, + fuse_session: Some(fuse_session), + } + } +} + +impl Drop for FuseFixture { + fn drop(&mut self) { + if let Some(session) = self.fuse_session.take() { + drop(session) + } + fs::remove_dir(&self.mount_point).unwrap(); + } +} + +fn file_size(file: &File) -> u64 { + file.metadata().unwrap().len() +} + +fn to_unix_secs(time: &SystemTime) -> u64 { + time.duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_secs() +} + +fn get_metadata_times(file: &File) -> (u64, u64, u64) { + let metadata = file.metadata().unwrap(); + ( + metadata.atime() as u64, + metadata.mtime() as u64, + metadata.ctime() as u64, + ) +} + +#[test] +fn metadata_times() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + let file_path = mount_point.join("file"); + + let mut file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(true) + .custom_flags(O_DIRECT) + .open(&file_path) + .unwrap(); + let (atime_init, mtime_init, ctime_init) = get_metadata_times(&file); + + std::thread::sleep(std::time::Duration::from_secs(1)); + + file.write_all(&vec![0; 512]).unwrap(); + let (atime1, mtime1, ctime1) = get_metadata_times(&file); + assert!(mtime1 > mtime_init); + assert!(ctime1 > ctime_init); + assert_eq!(atime1, atime_init); + + std::thread::sleep(std::time::Duration::from_secs(1)); + + file.read_at(&mut vec![0; 512], 0).unwrap(); + let (atime2, mtime2, ctime2) = get_metadata_times(&file); + assert!(atime2 > atime1); + assert_eq!(mtime2, mtime1); + assert!(ctime2 > ctime1); +} + +#[test] +fn manual_setattr() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + let file_path = mount_point.join("file"); + + let before_creation = SystemTime::now(); + let file = OpenOptions::new() + .create(true) + .truncate(true) + .write(true) + .custom_flags(O_DIRECT) + .open(&file_path) + .unwrap(); + + let (atime1, mtime1, ctime1) = get_metadata_times(&file); + assert_eq!(atime1, mtime1); + assert_eq!(mtime1, ctime1); + let before_creation_in_unix_secs = to_unix_secs(&before_creation); + assert!(ctime1 >= before_creation_in_unix_secs); + + std::thread::sleep(std::time::Duration::from_secs(1)); + + let now = SystemTime::now(); + let now_minus10s = now - std::time::Duration::from_secs(10); + let now_minus100s = now - std::time::Duration::from_secs(100); + + let new_atime = FileTime::from_system_time(now_minus10s); + let new_mtime = FileTime::from_system_time(now_minus100s); + + filetime::set_file_atime(&file_path, new_atime).unwrap(); + filetime::set_file_mtime(&file_path, new_mtime).unwrap(); + + let (atime2, mtime2, ctime2) = get_metadata_times(&file); + assert_eq!(atime2, to_unix_secs(&now_minus10s)); + assert_eq!(mtime2, to_unix_secs(&now_minus100s)); + assert!(ctime2 > ctime1); +} + +#[test] +fn readdir() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + + File::create(mount_point.join("file1")).unwrap(); + File::create(mount_point.join("file2")).unwrap(); + + let mut files = vec![]; + for entry in fs::read_dir(mount_point).unwrap() { + let entry = entry.unwrap(); + let path = entry.path(); + assert!(path.is_file()); + files.push(path.file_name().unwrap().to_owned()); + } + assert!(files.contains(&OsString::from("file1"))); + assert!(files.contains(&OsString::from("file2"))); + assert_eq!(files.len(), 2) +} + +#[test] +fn permissions() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + File::create(&file_path).unwrap(); + + let read_ok = || { + let mut file = OpenOptions::new().read(true).open(&file_path).unwrap(); + let mut buf = vec![]; + file.read_to_end(&mut buf).unwrap(); + assert_eq!(file_size(&file), buf.len() as u64); + }; + let read_denied = || { + let res = OpenOptions::new().read(true).open(&file_path); + assert_eq!( + res.unwrap_err().kind(), + std::io::ErrorKind::PermissionDenied + ); + }; + let write_ok = || { + let file = OpenOptions::new().write(true).open(&file_path).unwrap(); + let write_len = file.write_at(&vec![0; 512], file_size(&file)).unwrap(); + assert_eq!(write_len, 512); + }; + let write_denied = || { + let res = OpenOptions::new().write(true).open(&file_path); + assert_eq!( + res.unwrap_err().kind(), + std::io::ErrorKind::PermissionDenied + ); + }; + + let perms: Vec<_> = (0o000..=0o777).map(Permissions::from_mode).collect(); + for perm in perms { + fs::set_permissions(&file_path, perm.clone()).unwrap(); + if perm.mode() & 0o400 != 0 { + read_ok(); + } else { + read_denied(); + } + + if perm.mode() & 0o200 != 0 { + write_ok(); + } else { + write_denied(); + } + } +} + +#[test] +fn create_dir_fails() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let dir_path = mount_point.join("directory"); + let res = fs::create_dir(&dir_path); + assert_eq!(res.unwrap_err().raw_os_error(), Some(libc::ENOSYS)); +} + +#[test] +fn write_not_to_end_fails() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + let mut file = File::create(&file_path).unwrap(); + + file.write_all(b"Hello, Chunkfs!").unwrap(); + file.write_all(&vec![0; MB]).unwrap(); + + let res1 = file.write_at(&[1, 2, 3], 10); + let res2 = file.write_at(&[1, 2, 3], file_size(&file) + 1); + assert!(res1.is_err()); + assert!(res2.is_err()); +} + +#[test] +fn filehandles_mods() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + File::create(&file_path).unwrap(); + + let mut file = OpenOptions::new().write(true).open(&file_path).unwrap(); + let res = file.read(&mut vec![0; 512]); + assert!(res.is_err()); + + let file = OpenOptions::new().read(true).open(&file_path).unwrap(); + let res = file.write_at(&vec![0; 512], file_size(&file)); + assert!(res.is_err()); + + let res = OpenOptions::new().open(&file_path); + assert!(res.is_err()); +} + +#[test] +fn write_fuse_fs() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .write(true) + .custom_flags(O_DIRECT) + .open(&file_path) + .unwrap(); + + let mut data1 = vec![1u8; 2000]; + let mut data2 = vec![2u8; 5000]; + file.write_all(&data1).unwrap(); + file.write_at(&data2, data1.len() as u64).unwrap(); + + let mut file = OpenOptions::new() + .custom_flags(O_DIRECT) + .read(true) + .open(&file_path) + .unwrap(); + data1.append(&mut data2); + let mut actual = Vec::new(); + file.read_to_end(&mut actual).unwrap(); + assert_eq!(actual, data1); +} + +#[test] +fn different_data_writes() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .write(true) + .custom_flags(O_DIRECT) + .open(&file_path) + .unwrap(); + + let mut data1 = vec![1u8; 500]; + let mut data2 = vec![2u8; 700]; + let data3 = vec![3u8; 3 * MB]; + let mut data4 = vec![4u8; 10 * MB]; + file.write_all(&data1).unwrap(); + file.write_all(&data2).unwrap(); + file.write_all(&data3).unwrap(); + file.write_all(&data4).unwrap(); + + let mut file = OpenOptions::new() + .custom_flags(O_DIRECT) + .read(true) + .open(&file_path) + .unwrap(); + data1.append(&mut data2); + data1.append(&mut vec![3u8; MB + 11]); + let mut actual = vec![0u8; 500 + 700 + MB + 11]; + file.read_exact(&mut actual).unwrap(); + assert_eq!(actual, data1); + let first_read_len = actual.len(); + + let mut expected = vec![3u8; MB - 11]; + expected.append(&mut data4); + let mut actual = vec![0u8; expected.len()]; + file.read_exact_at(&mut actual, (first_read_len + MB) as u64) + .unwrap(); + assert_eq!(actual, expected); +} + +#[test] +fn read_dropped_cache() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .custom_flags(O_DIRECT) + .write(true) + .read(true) + .open(&file_path) + .unwrap(); + file.write_all(&[0; 10 * MB]).unwrap(); + file.flush().unwrap(); + file.write_all(&[1; 3 * MB]).unwrap(); + + let mut actual = vec![10; 14 * MB]; + assert_eq!(file.read_at(&mut actual, 0).unwrap(), 13 * MB); + let expected = [vec![0; 10 * MB], vec![1; 3 * MB], vec![10; MB]].concat(); + assert_eq!( + actual, expected, + "read entire file with dropped and dirty cache is correct" + ); + + actual = vec![10; 7 * MB]; + assert_eq!(file.read_at(&mut actual, 0).unwrap(), 7 * MB); + assert_eq!( + actual, + [0; 7 * MB], + "read dropped cache from start to end - epsilon is correct" + ); + + actual = vec![10; 12 * MB]; + assert_eq!(file.read_at(&mut actual, 0).unwrap(), 12 * MB); + let expected = [vec![0; 10 * MB], vec![1; 2 * MB]].concat(); + assert_eq!( + actual, expected, + "read dropped cache from start to end + epsilon is correct" + ); + + actual = vec![10; 7 * MB]; + assert_eq!(file.read_at(&mut actual, MB as u64).unwrap(), 7 * MB); + assert_eq!( + actual, + [0; 7 * MB], + "read dropped cache from start + epsilon to end - epsilon is correct" + ); + + actual = vec![10; 10 * MB]; + assert_eq!(file.read_at(&mut actual, 7 * MB as u64).unwrap(), 6 * MB); + let expected = [vec![0; 3 * MB], vec![1; 3 * MB], vec![10; 4 * MB]].concat(); + assert_eq!( + actual, expected, + "read dropped cache from start + epsilon to end + epsilon is correct" + ); +} + +#[test] +fn read_cache() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .write(true) + .custom_flags(O_DIRECT) + .read(true) + .open(&file_path) + .unwrap(); + file.write_all(&[0; 10 * MB]).unwrap(); + file.flush().unwrap(); + file.write_all(&[1; 3 * MB]).unwrap(); + + let mut actual = vec![10; 15 * MB]; + assert_eq!(file.read_at(&mut actual, 0).unwrap(), 13 * MB); + let expected = [vec![0; 10 * MB], vec![1; 3 * MB], vec![10; 2 * MB]].concat(); + assert_eq!( + actual, expected, + "read entire file with dropped and dirty cache is correct" + ); + + actual = vec![10; 10 * MB]; + assert_eq!(file.read_at(&mut actual, 5 * MB as u64).unwrap(), 8 * MB); + let expected = [vec![0; 5 * MB], vec![1; 3 * MB], vec![10; 2 * MB]].concat(); + assert_eq!( + actual, expected, + "read cache from start - epsilon to end + epsilon is correct" + ); + + actual = vec![10; 2 * MB]; + assert_eq!(file.read_at(&mut actual, 10 * MB as u64).unwrap(), 2 * MB); + assert_eq!( + actual, + [1; 2 * MB], + "read cache from start to end - epsilon is correct" + ); + + actual = vec![10; 5 * MB]; + assert_eq!(file.read_at(&mut actual, 11 * MB as u64).unwrap(), 2 * MB); + let expected = [vec![1; 2 * MB], vec![10; 3 * MB]].concat(); + assert_eq!( + actual, expected, + "read cache from start + epsilon to end + epsilon is correct" + ); + + actual = vec![10; MB]; + assert_eq!(file.read_at(&mut actual, 11 * MB as u64).unwrap(), MB); + assert_eq!( + actual, [1; MB], + "read cache from start + epsilon to end - epsilon is correct" + ); +} + +#[test] +fn concurrent_file_handles() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + File::create(&file_path).unwrap(); + + let handle1 = OpenOptions::new() + .append(true) + .custom_flags(O_DIRECT) + .open(&file_path) + .unwrap(); + let handle2 = OpenOptions::new() + .append(true) + .custom_flags(O_DIRECT) + .open(&file_path) + .unwrap(); + let handle3 = OpenOptions::new() + .append(true) + .custom_flags(O_DIRECT) + .open(&file_path) + .unwrap(); + for _ in 0..12 { + handle1 + .write_all_at(&vec![1; MB], file_size(&handle1)) + .unwrap(); + handle2 + .write_all_at(&vec![2; MB], file_size(&handle1)) + .unwrap(); + handle3 + .write_all_at(&vec![3; MB], file_size(&handle1)) + .unwrap(); + } + drop(handle2); + drop(handle3); + drop(handle1); + + let expected = [[1; MB], [2; MB], [3; MB]].concat().repeat(12); + + let mut file = File::open(&file_path).unwrap(); + let mut actual = vec![0; 12 * 3 * MB]; + assert_eq!(file.read(&mut actual).unwrap(), 12 * 3 * MB); + assert_eq!(actual, expected); + assert_eq!(file.metadata().unwrap().len(), 12 * 3 * MB as u64); +} + +#[test] +fn offset_change_does_not_affect_cache_drop() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .custom_flags(O_DIRECT) + .write(true) + .read(true) + .open(&file_path) + .unwrap(); + file.write_all(&[0; 3 * MB]).unwrap(); + assert_eq!(MB, file.read_at(&mut [0; MB], 2 * MB as u64).unwrap()); + drop(file); + + let mut file = OpenOptions::new() + .custom_flags(O_DIRECT) + .read(true) + .append(true) + .open(&file_path) + .unwrap(); + file.write_all(&[0; 3 * MB]).unwrap(); + file.flush().unwrap(); + + let mut actual = vec![10; 6 * MB]; + assert_eq!(file.read_at(&mut actual, 0).unwrap(), 6 * MB); + assert_eq!(actual, [0; 6 * MB]); + assert_eq!(file.metadata().unwrap().len(), 6 * MB as u64); +} + +#[test] +fn single_chunk_read() { + let fuse_fixture = FuseFixture::with_chunker(LeapChunker::new(SizeParams { + min: 1000, + avg: 1000, + max: 1000, + })); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .custom_flags(O_DIRECT) + .write(true) + .read(true) + .open(&file_path) + .unwrap(); + + file.write_all(&[0; 300]).unwrap(); + + let mut actual = vec![10; 100]; + assert_eq!(100, file.read_at(&mut actual, 100).unwrap()); + assert_eq!( + actual, [0; 100], + "read from start + epsilon to end - epsilon of a single chunk is correct" + ); + + actual = vec![10; 200]; + assert_eq!(200, file.read_at(&mut actual, 100).unwrap()); + assert_eq!( + actual, [0; 200], + "read from start + epsilon to end of a single chunk is correct" + ); + + actual = vec![10; 1000]; + assert_eq!(200, file.read_at(&mut actual, 100).unwrap()); + assert_eq!( + actual, + [vec![0; 200], vec![10; 800]].concat(), + "read from start + epsilon to end + epsilon of a single chunk is correct" + ); + + actual = vec![10; 1000]; + assert_eq!(0, file.read_at(&mut actual, 1000).unwrap()); + assert_eq!( + actual, [10; 1000], + "read from end to end + epsilon of a single chunk is correct" + ); +} + +#[test] +fn read_first_chunk_piece() { + let fuse_fixture = FuseFixture::with_chunker(LeapChunker::new(SizeParams { + min: 1000, + avg: 1000, + max: 1000, + })); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .custom_flags(O_DIRECT) + .write(true) + .read(true) + .open(&file_path) + .unwrap(); + + file.write_all(&[0; 1000]).unwrap(); + file.write_all(&[1; 4500]).unwrap(); + + let mut actual = vec![10; 40]; + assert_eq!(40, file.read_at(&mut actual, 0).unwrap()); + assert_eq!( + actual, [0; 40], + "read from start to end - epsilon of first chunk is correct" + ); + + actual = vec![10; 30]; + assert_eq!(30, file.read_at(&mut actual, 40).unwrap()); + assert_eq!( + actual, [0; 30], + "read from start + epsilon to end - epsilon of first chunk is correct" + ); + + actual = vec![10; 960]; + assert_eq!(960, file.read_at(&mut actual, 40).unwrap()); + assert_eq!( + actual, [0; 960], + "read from start + epsilon to end of first chunk is correct" + ); + + actual = vec![10; 60]; + assert_eq!(60, file.read_at(&mut actual, 970).unwrap()); + assert_eq!( + actual, + [[0; 30], [1; 30]].concat(), + "read from start + epsilon to end + epsilon of first chunk is correct" + ); + + actual = vec![10; 0]; + assert_eq!(0, file.read_at(&mut actual, 0).unwrap()); + assert_eq!( + actual, [0; 0], + "read zero bytes from start of first chunk if correct" + ); + + actual = vec![10; 0]; + assert_eq!(0, file.read_at(&mut actual, 1000).unwrap()); + assert_eq!( + actual, [0; 0], + "read zero bytes from end of first chunk if correct" + ); +} + +#[test] +fn read_middle_chunk_piece() { + let fuse_fixture = FuseFixture::with_chunker(LeapChunker::new(SizeParams { + min: 1000, + avg: 1000, + max: 1000, + })); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .custom_flags(O_DIRECT) + .write(true) + .read(true) + .open(&file_path) + .unwrap(); + + file.write_all(&[0; 5500]).unwrap(); + + let mut actual = vec![10; 40]; + assert_eq!(40, file.read_at(&mut actual, 1040).unwrap()); + assert_eq!( + actual, [0; 40], + "read from start + epsilon to end - epsilon of middle chunk is correct" + ); + + actual = vec![10; 0]; + assert_eq!(0, file.read_at(&mut actual, 1040).unwrap()); + assert_eq!( + actual, [0; 0], + "read zero bytes from start + epsilon of middle chunk is correct" + ); +} + +#[test] +fn read_last_chunk_piece() { + let fuse_fixture = FuseFixture::with_chunker(LeapChunker::new(SizeParams { + min: 1000, + avg: 1000, + max: 1000, + })); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .custom_flags(O_DIRECT) + .write(true) + .read(true) + .open(&file_path) + .unwrap(); + + file.write_all(&[0; 5000]).unwrap(); + file.write_all(&[1; 500]).unwrap(); + + let mut actual = vec![10; 40]; + assert_eq!(40, file.read_at(&mut actual, 5000).unwrap()); + assert_eq!( + actual, [1; 40], + "read from start to end - epsilon of last chunk is correct" + ); + + actual = vec![10; 30]; + assert_eq!(30, file.read_at(&mut actual, 5010).unwrap()); + assert_eq!( + actual, [1; 30], + "read from start + epsilon to end - epsilon of last chunk is correct" + ); + + actual = vec![10; 470]; + assert_eq!(470, file.read_at(&mut actual, 5030).unwrap()); + assert_eq!( + actual, [1; 470], + "read from start + epsilon to end of last chunk is correct" + ); + + actual = vec![10; 500]; + assert_eq!(470, file.read_at(&mut actual, 5030).unwrap()); + assert_eq!( + actual, + [vec![1; 470], vec![10; 30]].concat(), + "read from start + epsilon to end + epsilon of last chunk is correct" + ); + + actual = vec![10; 530]; + assert_eq!(500, file.read_at(&mut actual, 5000).unwrap()); + assert_eq!( + actual, + [vec![1; 500], vec![10; 30]].concat(), + "read from start to end + epsilon of last chunk is correct" + ); + + actual = vec![10; 0]; + assert_eq!(0, file.read_at(&mut actual, 5000).unwrap()); + assert_eq!( + actual, [0; 0], + "read zero bytes from start of last chunk if correct" + ); + + actual = vec![10; 0]; + assert_eq!(0, file.read_at(&mut actual, 5500).unwrap()); + assert_eq!( + actual, [0; 0], + "read zero bytes from end of last chunk if correct" + ); + + actual = vec![10; 40]; + assert_eq!(0, file.read_at(&mut actual, 5600).unwrap()); + assert_eq!( + actual, [10; 40], + "read n bytes from end + epsilon of last chunk if correct" + ); +} + +#[test] +fn lookup_permission() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + File::create(mount_point.join("file1")).unwrap(); + File::create(mount_point.join("file2")).unwrap(); + + let lookup_ok = || { + let mut files = vec![]; + + for entry in fs::read_dir(mount_point).unwrap() { + let entry = entry.unwrap(); + let path = entry.path(); + assert!(path.is_file()); + let _ = entry.metadata().unwrap(); + files.push(path.file_name().unwrap().to_str().unwrap().to_string()); + } + assert!(files.contains(&"file1".to_string())); + assert!(files.contains(&"file2".to_string())); + assert_eq!(files.len(), 2); + }; + let lookup_denied = || { + for entry in fs::read_dir(mount_point).unwrap() { + let metadata = entry.unwrap().metadata(); + assert_eq!( + metadata.unwrap_err().kind(), + std::io::ErrorKind::PermissionDenied + ); + } + }; + + let perms: Vec<_> = (0o000..=0o777).map(Permissions::from_mode).collect(); + for perm in perms { + // reading mount directory is handled by the upper filesystem + if perm.mode() & 0o400 == 0 { + continue; + } + + fs::set_permissions(mount_point, perm.clone()).unwrap(); + if perm.mode() & 0o100 != 0 { + lookup_ok(); + } else { + lookup_denied(); + } + } +} + +#[test] +fn concurrent_write_and_lookup() { + let fuse_fixture = FuseFixture::default(); + let mount_point = Path::new(&fuse_fixture.mount_point); + let file_path = mount_point.join("file"); + + let mut file = File::create(&file_path).unwrap(); + let handle = std::thread::spawn(move || { + for _ in 0..10 { + std::thread::sleep(std::time::Duration::from_millis(10)); + let _ = File::open(&file_path).unwrap().metadata().unwrap(); + } + }); + let buf = vec![0; 5 * MB]; + for _ in 0..10 { + file.write_all(&buf).unwrap() + } + + handle.join().unwrap(); + assert_eq!(file.metadata().unwrap().len(), 50 * MB as u64); +} + +#[test] +fn ioctl() { + let fuse_fixture = FuseFixture::with_chunker(SuperChunker::new(SizeParams { + min: 1000, + avg: 1000, + max: 1000, + })); + let mount_point = Path::new(&fuse_fixture.mount_point); + + let file_path = mount_point.join("file"); + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .custom_flags(O_DIRECT) + .write(true) + .read(true) + .open(&file_path) + .unwrap(); + + file.write_all(&[0; 5000]).unwrap(); + + let mut dedup_ratio = [0u8; size_of::()]; + let ret = unsafe { libc::ioctl(file.as_raw_fd(), IOC_GET_DEDUP_RATIO, &mut dedup_ratio) }; + assert_eq!(ret, 0); + + let mut avg_chunk_size = [0u8; size_of::()]; + let ret = unsafe { + libc::ioctl( + file.as_raw_fd(), + IOC_GET_AVG_CHUNK_SIZE, + &mut avg_chunk_size, + ) + }; + assert_eq!(ret, 0); + + let dedup_ratio = f64::from_ne_bytes(dedup_ratio); + let avg_chunk_size = usize::from_ne_bytes(avg_chunk_size); + assert_eq!(dedup_ratio, 5.); + assert_eq!(avg_chunk_size, 1000); +}