use std::{
cell::RefCell,
collections::HashMap,
ffi::OsStr,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
time::{Duration, Instant, SystemTime},
};
use parking_lot::{Mutex, RwLock};
use chrono::{DateTime, Utc};
use fuser::{
AccessFlags, Config, Errno, FileAttr, FileHandle, FileType, Filesystem, FopenFlags, Generation,
INodeNo, MountOption, OpenFlags, ReplyAttr, ReplyData, ReplyDirectory, ReplyDirectoryPlus,
ReplyEmpty, ReplyEntry, ReplyOpen, ReplyStatfs, ReplyXattr, Request,
};
use crate::{
common::{CvmfsError, CvmfsResult, FileLike},
directory_entry::DirectoryEntry,
repository::Repository,
};
thread_local! {
static READ_BUF: RefCell<Vec<u8>> = RefCell::new(Vec::with_capacity(128 * 1024));
}
const TTL: Duration = Duration::from_secs(3600);
const FUSE_ROOT_ID: u64 = 1;
#[cfg(target_os = "linux")]
fn num_cpus() -> usize {
std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4)
}
type DirListing = Vec<(u64, FileType, String, FileAttr)>;
static NEXT_FH: AtomicU64 = AtomicU64::new(1);
#[allow(clippy::unnecessary_cast)]
fn map_dirent_type_to_fs_kind(dirent: &DirectoryEntry) -> FileType {
if dirent.is_directory() {
FileType::Directory
} else if dirent.is_symlink() {
FileType::Symlink
} else {
let mode = dirent.mode as u32;
let ifmt = libc::S_IFMT as u32;
match mode & ifmt {
m if m == libc::S_IFSOCK as u32 => FileType::Socket,
m if m == libc::S_IFIFO as u32 => FileType::NamedPipe,
m if m == libc::S_IFBLK as u32 => FileType::BlockDevice,
m if m == libc::S_IFCHR as u32 => FileType::CharDevice,
_ => FileType::RegularFile,
}
}
}
fn make_file_attr(ino: u64, entry: &DirectoryEntry) -> Option<FileAttr> {
let time = SystemTime::from(DateTime::<Utc>::from_timestamp(entry.mtime, 0)?);
let size = entry.size as u64;
Some(FileAttr {
ino: INodeNo(ino),
size,
blocks: 1 + size / 512,
atime: time,
mtime: time,
ctime: time,
crtime: time,
kind: map_dirent_type_to_fs_kind(entry),
perm: entry.mode & 0o7777,
nlink: entry.nlink(),
uid: entry.uid,
gid: entry.gid,
rdev: 0,
blksize: 512,
flags: 0,
})
}
#[derive(Debug)]
struct InodeTable {
next_ino: AtomicU64,
path_to_ino: RwLock<HashMap<String, u64>>,
ino_to_path: RwLock<HashMap<u64, String>>,
}
impl InodeTable {
fn new() -> Self {
let mut path_to_ino = HashMap::new();
let mut ino_to_path = HashMap::new();
path_to_ino.insert(String::new(), FUSE_ROOT_ID);
ino_to_path.insert(FUSE_ROOT_ID, String::new());
Self {
next_ino: AtomicU64::new(2),
path_to_ino: RwLock::new(path_to_ino),
ino_to_path: RwLock::new(ino_to_path),
}
}
fn get_or_insert(&self, path: &str) -> u64 {
if let Some(&ino) = self.path_to_ino.read().get(path) {
return ino;
}
let ino = self.next_ino.fetch_add(1, Ordering::Relaxed);
let mut p2i = self.path_to_ino.write();
if let Some(&existing) = p2i.get(path) {
return existing;
}
p2i.insert(path.into(), ino);
self.ino_to_path.write().insert(ino, path.into());
ino
}
fn get_path(&self, ino: u64) -> Option<String> {
self.ino_to_path.read().get(&ino).cloned()
}
}
#[derive(Debug, Clone, Copy)]
struct CachedStatfs {
blocks: u64,
files: u64,
}
#[derive(Debug)]
pub struct CernvmFileSystem {
repository: Repository,
inodes: InodeTable,
opened_files: RwLock<HashMap<u64, Box<dyn FileLike>>>,
opened_dirs: RwLock<HashMap<u64, DirListing>>,
lookup_cache: RwLock<HashMap<String, Arc<DirectoryEntry>>>,
cached_statfs: Mutex<Option<(Instant, CachedStatfs)>>,
}
impl Filesystem for CernvmFileSystem {
fn lookup(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) {
let name = match name.to_str() {
Some(n) => n,
None => return reply.error(Errno::ENOENT),
};
let parent_path = match self.inodes.get_path(parent.0) {
Some(p) => p,
None => return reply.error(Errno::ENOENT),
};
let child_path = if parent_path.is_empty() {
format!("/{name}")
} else {
format!("{parent_path}/{name}")
};
let entry = match self.cached_lookup(&child_path) {
Ok(e) => e,
Err(_) => return reply.error(Errno::ENOENT),
};
let ino = self.inodes.get_or_insert(&child_path);
match make_file_attr(ino, &entry) {
Some(attr) => reply.entry(&TTL, &attr, Generation(0)),
None => reply.error(Errno::EIO),
}
}
fn getattr(&self, _req: &Request, ino: INodeNo, _fh: Option<FileHandle>, reply: ReplyAttr) {
let path = match self.inodes.get_path(ino.0) {
Some(p) => p,
None => return reply.error(Errno::ENOENT),
};
let cvmfs_path = if path.is_empty() { "" } else { &path };
let entry = match self.cached_lookup(cvmfs_path) {
Ok(e) => e,
Err(_) => return reply.error(Errno::ENOENT),
};
match make_file_attr(ino.0, &entry) {
Some(attr) => reply.attr(&TTL, &attr),
None => reply.error(Errno::EIO),
}
}
fn readlink(&self, _req: &Request, ino: INodeNo, reply: ReplyData) {
let path = match self.inodes.get_path(ino.0) {
Some(p) => p,
None => return reply.error(Errno::ENOENT),
};
let entry = match self.cached_lookup(&path) {
Ok(e) => e,
Err(_) => return reply.error(Errno::ENOENT),
};
if !entry.is_symlink() {
return reply.error(Errno::EINVAL);
}
match &entry.symlink {
Some(target) => reply.data(target.as_bytes()),
None => reply.error(Errno::EIO),
}
}
fn open(&self, _req: &Request, ino: INodeNo, _flags: OpenFlags, reply: ReplyOpen) {
let path = match self.inodes.get_path(ino.0) {
Some(p) => p,
None => return reply.error(Errno::ENOENT),
};
let entry = match self.cached_lookup_with_chunks(&path) {
Ok(e) => e,
Err(_) => return reply.error(Errno::ENOENT),
};
if !entry.is_file() {
return reply.error(Errno::EISDIR);
}
let file = match self.repository.retrieve_object(&entry, &path) {
Ok(f) => f,
Err(_) => return reply.error(Errno::EIO),
};
let fh = NEXT_FH.fetch_add(1, Ordering::Relaxed);
self.opened_files.write().insert(fh, file);
reply.opened(FileHandle(fh), FopenFlags::FOPEN_KEEP_CACHE);
}
fn read(
&self,
_req: &Request,
_ino: INodeNo,
fh: FileHandle,
offset: u64,
size: u32,
_flags: OpenFlags,
_lock_owner: Option<fuser::LockOwner>,
reply: ReplyData,
) {
let opened_files = self.opened_files.read();
let file = match opened_files.get(&fh.0) {
Some(f) => f,
None => return reply.error(Errno::EBADF),
};
READ_BUF.with(|buf| {
let mut data = buf.borrow_mut();
data.resize(size as usize, 0);
match file.read_at(offset, &mut data) {
Ok(n) => reply.data(&data[..n]),
Err(_) => reply.error(Errno::EIO),
}
});
}
fn release(
&self,
_req: &Request,
_ino: INodeNo,
fh: FileHandle,
_flags: OpenFlags,
_lock_owner: Option<fuser::LockOwner>,
_flush: bool,
reply: ReplyEmpty,
) {
self.opened_files.write().remove(&fh.0);
reply.ok();
}
fn opendir(&self, _req: &Request, ino: INodeNo, _flags: OpenFlags, reply: ReplyOpen) {
let path = match self.inodes.get_path(ino.0) {
Some(p) => p,
None => return reply.error(Errno::ENOENT),
};
let cvmfs_path = if path.is_empty() { "" } else { &path };
let entries = match self.repository.list_directory(cvmfs_path) {
Ok(e) => e,
Err(_) => return reply.error(Errno::ENOTDIR),
};
let mut all_entries: DirListing = Vec::with_capacity(entries.len() + 2);
let dot_attr = FileAttr {
ino,
size: 0,
blocks: 0,
atime: SystemTime::UNIX_EPOCH,
mtime: SystemTime::UNIX_EPOCH,
ctime: SystemTime::UNIX_EPOCH,
crtime: SystemTime::UNIX_EPOCH,
kind: FileType::Directory,
perm: 0o755,
nlink: 2,
uid: 0,
gid: 0,
rdev: 0,
blksize: 512,
flags: 0,
};
all_entries.push((ino.0, FileType::Directory, ".".into(), dot_attr));
let parent_ino = if ino.0 == FUSE_ROOT_ID {
FUSE_ROOT_ID
} else {
let parent_path = path.rsplit_once('/').map_or(String::new(), |(p, _)| p.into());
self.inodes.get_or_insert(&parent_path)
};
let dotdot_attr = FileAttr { ino: INodeNo(parent_ino), ..dot_attr };
all_entries.push((parent_ino, FileType::Directory, "..".into(), dotdot_attr));
{
let mut cache = self.lookup_cache.write();
self.evict_if_full(&mut cache, entries.len());
for entry in &entries {
let child_path = if path.is_empty() {
format!("/{}", entry.name)
} else {
format!("{}/{}", path, entry.name)
};
let child_ino = self.inodes.get_or_insert(&child_path);
let attr = make_file_attr(child_ino, entry).unwrap_or(dot_attr);
all_entries.push((
child_ino,
map_dirent_type_to_fs_kind(entry),
entry.name.clone(),
attr,
));
cache.insert(child_path, Arc::new(entry.clone()));
}
}
let fh = NEXT_FH.fetch_add(1, Ordering::Relaxed);
self.opened_dirs.write().insert(fh, all_entries);
reply.opened(FileHandle(fh), FopenFlags::empty());
}
fn readdir(
&self,
_req: &Request,
_ino: INodeNo,
fh: FileHandle,
offset: u64,
mut reply: ReplyDirectory,
) {
let dirs = self.opened_dirs.read();
let all_entries = match dirs.get(&fh.0) {
Some(e) => e,
None => return reply.error(Errno::EBADF),
};
for (i, (ino, kind, name, _)) in all_entries.iter().enumerate().skip(offset as usize) {
if reply.add(INodeNo(*ino), (i + 1) as u64, *kind, name) {
break;
}
}
reply.ok();
}
fn readdirplus(
&self,
_req: &Request,
_ino: INodeNo,
fh: FileHandle,
offset: u64,
mut reply: ReplyDirectoryPlus,
) {
let dirs = self.opened_dirs.read();
let all_entries = match dirs.get(&fh.0) {
Some(e) => e,
None => return reply.error(Errno::EBADF),
};
for (i, (ino, _, name, attr)) in all_entries.iter().enumerate().skip(offset as usize) {
if reply.add(INodeNo(*ino), (i + 1) as u64, name, &TTL, attr, Generation(0)) {
break;
}
}
reply.ok();
}
fn releasedir(
&self,
_req: &Request,
_ino: INodeNo,
fh: FileHandle,
_flags: OpenFlags,
reply: ReplyEmpty,
) {
self.opened_dirs.write().remove(&fh.0);
reply.ok();
}
fn statfs(&self, _req: &Request, _ino: INodeNo, reply: ReplyStatfs) {
if let Some((ts, cached)) = *self.cached_statfs.lock()
&& ts.elapsed() < Duration::from_secs(5)
{
return reply.statfs(cached.blocks, 0, 0, cached.files, 0, 512, 255, 0);
}
let (blocks, files) = match self.repository.get_statistics() {
Ok(stats) => (1 + stats.file_size as u64 / 512, stats.regular as u64),
Err(_) => (0, 0),
};
*self.cached_statfs.lock() = Some((Instant::now(), CachedStatfs { blocks, files }));
reply.statfs(blocks, 0, 0, files, 0, 512, 255, 0);
}
fn getxattr(&self, _req: &Request, _ino: INodeNo, name: &OsStr, size: u32, reply: ReplyXattr) {
let name = match name.to_str() {
Some(n) => n,
None => return reply.error(Errno::ENOENT),
};
let value = match name {
"user.fqrn" => self.repository.fqrn.clone(),
"user.revision" => self.repository.manifest.revision.to_string(),
"user.hash" => self.repository.manifest.root_catalog.clone(),
"user.host" => self.repository.fetcher_source(),
"user.expires" => self.repository.manifest.last_modified.to_rfc3339(),
"user.nclg" => self.repository.opened_catalogs.read().len().to_string(),
_ => return reply.error(Errno::ENOENT),
};
let bytes = value.into_bytes();
if size == 0 {
reply.size(bytes.len() as u32);
} else {
reply.data(&bytes);
}
}
fn access(&self, _req: &Request, ino: INodeNo, _mask: AccessFlags, reply: ReplyEmpty) {
let path = match self.inodes.get_path(ino.0) {
Some(p) => p,
None => return reply.error(Errno::ENOENT),
};
let cvmfs_path = if path.is_empty() { "" } else { &path };
match self.cached_lookup(cvmfs_path) {
Ok(_) => reply.ok(),
Err(_) => reply.error(Errno::ENOENT),
}
}
}
const LOOKUP_CACHE_CAP: usize = 65_536;
impl CernvmFileSystem {
fn cached_lookup(&self, path: &str) -> CvmfsResult<Arc<DirectoryEntry>> {
let lookup_path = if path.is_empty() { "/" } else { path };
if let Some(entry) = self.lookup_cache.read().get(lookup_path).cloned() {
return Ok(entry);
}
let cvmfs_path = if path == "/" { "" } else { path };
let entry = Arc::new(self.repository.lookup(cvmfs_path)?);
let mut cache = self.lookup_cache.write();
self.evict_if_full(&mut cache, 1);
cache.insert(lookup_path.into(), Arc::clone(&entry));
Ok(entry)
}
fn cached_lookup_with_chunks(&self, path: &str) -> CvmfsResult<Arc<DirectoryEntry>> {
let lookup_path = if path.is_empty() { "/" } else { path };
if let Some(entry) = self.lookup_cache.read().get(lookup_path).cloned()
&& (!entry.chunks.is_empty() || !entry.is_file())
{
return Ok(entry);
}
let cvmfs_path = if path == "/" { "" } else { path };
let entry = Arc::new(self.repository.lookup_with_chunks(cvmfs_path)?);
let mut cache = self.lookup_cache.write();
self.evict_if_full(&mut cache, 1);
cache.insert(lookup_path.into(), Arc::clone(&entry));
Ok(entry)
}
fn evict_if_full(&self, cache: &mut HashMap<String, Arc<DirectoryEntry>>, incoming: usize) {
if cache.len() + incoming > LOOKUP_CACHE_CAP {
cache.clear();
}
}
pub fn new(repository: Repository) -> CvmfsResult<Self> {
Ok(Self {
repository,
inodes: InodeTable::new(),
opened_files: Default::default(),
opened_dirs: Default::default(),
lookup_cache: Default::default(),
cached_statfs: Mutex::new(None),
})
}
pub fn mount(self, mountpoint: &str) -> std::io::Result<()> {
let mut config = Config::default();
config.mount_options =
vec![MountOption::RO, MountOption::FSName("cernvmfs".into()), MountOption::NoAtime];
#[cfg(target_os = "linux")]
{
config.n_threads = Some(num_cpus());
config.clone_fd = true;
}
fuser::mount2(self, mountpoint, &config)
}
pub fn do_lookup(&self, path: &str) -> CvmfsResult<(u64, FileAttr)> {
let entry = self.cached_lookup(path)?;
let lookup_path = if path.is_empty() || path == "/" { "/" } else { path };
let ino = self.inodes.get_or_insert(lookup_path);
let attr = make_file_attr(ino, &entry).ok_or(CvmfsError::InvalidTimestamp)?;
Ok((ino, attr))
}
pub fn do_readlink(&self, path: &str) -> CvmfsResult<String> {
let entry = self.cached_lookup(path)?;
if !entry.is_symlink() {
return Err(CvmfsError::NotASymlink);
}
entry.symlink.clone().ok_or(CvmfsError::FileNotFound)
}
pub fn do_open(&self, path: &str) -> CvmfsResult<u64> {
let entry = self.cached_lookup_with_chunks(path)?;
if !entry.is_file() {
return Err(CvmfsError::NotAFile);
}
let file = self.repository.retrieve_object(&entry, path)?;
let fh = NEXT_FH.fetch_add(1, Ordering::Relaxed);
self.opened_files.write().insert(fh, file);
Ok(fh)
}
pub fn do_release(&self, fh: u64) -> CvmfsResult<()> {
self.opened_files.write().remove(&fh).ok_or(CvmfsError::FileNotFound)?;
Ok(())
}
pub fn do_read(&self, fh: u64, offset: u64, size: u32) -> CvmfsResult<Vec<u8>> {
let files = self.opened_files.read();
let file = files.get(&fh).ok_or(CvmfsError::FileNotFound)?;
let mut buf = vec![0u8; size as usize];
let n = file.read_at(offset, &mut buf).map_err(|_| CvmfsError::FileNotFound)?;
buf.truncate(n);
Ok(buf)
}
pub fn do_readdir(&self, path: &str) -> CvmfsResult<Vec<(FileType, String)>> {
let cvmfs_path = if path == "/" { "" } else { path };
let entry = self.cached_lookup(cvmfs_path)?;
if !entry.is_directory() {
return Err(CvmfsError::FileNotFound);
}
let entries = self.repository.list_directory(cvmfs_path)?;
Ok(entries.into_iter().map(|e| (map_dirent_type_to_fs_kind(&e), e.name)).collect())
}
pub fn do_statfs(&self) -> CvmfsResult<(u64, u64)> {
let stats = self.repository.get_statistics()?;
Ok((1 + stats.file_size as u64 / 512, stats.regular as u64))
}
pub fn do_getxattr(&self, name: &str) -> CvmfsResult<String> {
match name {
"user.fqrn" => Ok(self.repository.fqrn.clone()),
"user.revision" => Ok(self.repository.manifest.revision.to_string()),
"user.hash" => Ok(self.repository.manifest.root_catalog.clone()),
"user.host" => Ok(self.repository.fetcher_source()),
"user.expires" => Ok(self.repository.manifest.last_modified.to_rfc3339()),
"user.nclg" => Ok(self.repository.opened_catalogs.read().len().to_string()),
_ => Err(CvmfsError::FileNotFound),
}
}
pub fn do_access(&self, path: &str) -> CvmfsResult<()> {
let cvmfs_path = if path == "/" { "" } else { path };
self.cached_lookup(cvmfs_path).map(|_| ())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::directory_entry::Flags;
fn make_entry(flags: u32, mode: u16) -> DirectoryEntry {
DirectoryEntry {
md5_path_1: 0,
md5_path_2: 0,
parent_1: 0,
parent_2: 0,
content_hash: None,
flags,
size: 0,
mode,
mtime: 0,
name: String::new(),
symlink: None,
uid: 0,
gid: 0,
xattr: None,
content_hash_type: crate::directory_entry::ContentHashTypes::Sha1,
chunks: Vec::new(),
hardlinks: 0,
}
}
#[test]
fn map_type_directory() {
let entry = make_entry(Flags::Directory as u32, 0o40755);
assert_eq!(map_dirent_type_to_fs_kind(&entry), FileType::Directory);
}
#[test]
fn map_type_symlink() {
let entry = make_entry(Flags::Link as u32, 0o120777);
assert_eq!(map_dirent_type_to_fs_kind(&entry), FileType::Symlink);
}
#[test]
fn map_type_regular_file() {
let entry = make_entry(Flags::File as u32, 0o100644);
assert_eq!(map_dirent_type_to_fs_kind(&entry), FileType::RegularFile);
}
#[test]
fn map_type_socket() {
let entry = make_entry(Flags::File as u32, 0o140755);
assert_eq!(map_dirent_type_to_fs_kind(&entry), FileType::Socket);
}
#[test]
fn map_type_named_pipe() {
let entry = make_entry(Flags::File as u32, 0o010644);
assert_eq!(map_dirent_type_to_fs_kind(&entry), FileType::NamedPipe);
}
#[test]
fn map_type_block_device() {
let entry = make_entry(Flags::File as u32, 0o060660);
assert_eq!(map_dirent_type_to_fs_kind(&entry), FileType::BlockDevice);
}
#[test]
fn map_type_char_device() {
let entry = make_entry(Flags::File as u32, 0o020666);
assert_eq!(map_dirent_type_to_fs_kind(&entry), FileType::CharDevice);
}
#[test]
fn map_type_zero_mode_defaults_to_regular() {
let entry = make_entry(Flags::File as u32, 0);
assert_eq!(map_dirent_type_to_fs_kind(&entry), FileType::RegularFile);
}
}