diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/database/file/mod.rs | 357 |
1 files changed, 188 insertions, 169 deletions
diff --git a/src/database/file/mod.rs b/src/database/file/mod.rs index ee7d30b..5d6ff47 100644 --- a/src/database/file/mod.rs +++ b/src/database/file/mod.rs @@ -1,7 +1,6 @@ use crate::database::{filter_post, ErrorKind, Result, Storage, StorageError}; -use async_std::fs::{File, OpenOptions}; -use async_std::io::prelude::*; -use async_std::io::ErrorKind as IOErrorKind; +use std::fs::{File, OpenOptions}; +use std::io::{ErrorKind as IOErrorKind, Seek, SeekFrom, Read, Write}; use async_std::task::spawn_blocking; use async_trait::async_trait; use fd_lock::RwLock; @@ -26,11 +25,6 @@ impl From<std::io::Error> for StorageError { } } -async fn get_lockable_file(file: File) -> RwLock<File> { - debug!("Trying to create a file lock"); - spawn_blocking(move || RwLock::new(file)).await -} - // Copied from https://stackoverflow.com/questions/39340924 // This routine is adapted from the *old* Path's `path_relative_from` // function, which works differently from the new `relative_from` function. @@ -257,29 +251,31 @@ impl Storage for FileStorage { async fn get_post(&self, url: &str) -> Result<Option<serde_json::Value>> { let path = url_to_path(&self.root_dir, url); debug!("Opening {:?}", path); - // We have to special-case in here because the function should return Ok(None) on 404 - match File::open(path).await { - Ok(f) => { - let lock = get_lockable_file(f).await; - let guard = lock.read()?; - - let mut content = String::new(); - // Apparently this typechecks. Somehow. - // I can take &mut for a &File because File is not a real type - // The operating system guards it using something - // that looks like a mutex to Rust's runtime, allowing me - // to grab a mutable reference from an immutable reference - (&mut &*guard).read_to_string(&mut content).await?; - Ok(Some(serde_json::from_str(&content)?)) - } - Err(err) => { - if err.kind() == IOErrorKind::NotFound { - Ok(None) - } else { - Err(err.into()) + // Use exclusively synchronous operations to never transfer a lock over an await boundary + spawn_blocking(move || { + match File::open(&path) { + Ok(file) => { + let lock = RwLock::new(file); + debug!("Trying to get a lock for file {:?}", &path); + let guard = lock.read()?; + + let mut content = String::new(); + // Typechecks because OS magic acts on references + // to FDs as if they were behind a mutex + (&mut &*guard).read_to_string(&mut content)?; + Ok(Some(serde_json::from_str(&content)?)) + } + Err(err) => { + // We have to special-case in here because + // the function should return Ok(None) on 404 + if err.kind() == IOErrorKind::NotFound { + Ok(None) + } else { + Err(err.into()) + } } } - } + }).await } async fn put_post<'a>(&self, post: &'a serde_json::Value, user: &'a str) -> Result<()> { @@ -289,24 +285,31 @@ impl Storage for FileStorage { let path = url_to_path(&self.root_dir, key); debug!("Creating {:?}", path); + // To move these into the closure, we have to clone values + let post_json = post.to_string(); + let post_path = path.clone(); + // Use exclusively synchronous operations to never transfer a lock over an await boundary + spawn_blocking(move || { + let parent = post_path.parent().unwrap().to_owned(); + if !parent.is_dir() { + std::fs::create_dir_all(post_path.parent().unwrap())?; + } - let parent = path.parent().unwrap().to_owned(); - if !spawn_blocking(move || parent.is_dir()).await { - async_std::fs::create_dir_all(path.parent().unwrap()).await?; - } + let f = OpenOptions::new() + .write(true) + .create_new(true) + .open(&post_path)?; - let f = OpenOptions::new() - .write(true) - .create_new(true) - .open(&path) - .await?; + let mut lock = RwLock::new(f); + debug!("Waiting for lock on {:?}", &post_path); + let mut guard = lock.write()?; - let mut lock = get_lockable_file(f).await; - let mut guard = lock.write()?; + (*guard).write_all(post_json.as_bytes())?; + (*guard).flush()?; + drop(guard); - (*guard).write_all(post.to_string().as_bytes()).await?; - (*guard).flush().await?; - drop(guard); + Result::Ok(()) + }).await?; if post["properties"]["url"].is_array() { for url in post["properties"]["url"] @@ -319,33 +322,33 @@ impl Storage for FileStorage { let link = url_to_path(&self.root_dir, url); debug!("Creating a symlink at {:?}", link); let orig = path.clone(); - spawn_blocking::<_, Result<()>>(move || { - // We're supposed to have a parent here. - let basedir = link.parent().ok_or_else(|| { - StorageError::new( - ErrorKind::Backend, - "Failed to calculate parent directory when creating a symlink", - ) - })?; - let relative = path_relative_from(&orig, basedir).unwrap(); - println!("{:?} - {:?} = {:?}", &orig, &basedir, &relative); + // We're supposed to have a parent here. + let basedir = link.parent().ok_or_else(|| { + StorageError::new( + ErrorKind::Backend, + "Failed to calculate parent directory when creating a symlink", + ) + })?; + let relative = path_relative_from(&orig, basedir).unwrap(); + println!("{:?} - {:?} = {:?}", &orig, &basedir, &relative); + spawn_blocking(move || { println!("Created a symlink at {:?}", &link); let symlink_result; #[cfg(unix)] { symlink_result = std::os::unix::fs::symlink(relative, link); } - // Wow it even supports windows. Not sure if I need it to run on Windows but oh well + // Wow it even supports windows. Windows is weird #[cfg(windows)] { symlink_result = std::os::windows::fs::symlink_file(relative, link); } - match symlink_result { - Ok(()) => Ok(()), - Err(e) => Err(e.into()), + if let Err(e) = symlink_result { + Err(e.into()) + } else { + Result::Ok(()) } - }) - .await?; + }).await?; } } } @@ -363,65 +366,74 @@ impl Storage for FileStorage { path.push("channels"); let path = path.to_path(&self.root_dir); - let file = OpenOptions::new() - .read(true) - .write(true) - .truncate(false) - .create(true) - .open(&path) - .await?; - let mut lock = get_lockable_file(file).await; - let mut guard = lock.write()?; + let channel_name = post["properties"]["name"][0] + .as_str() + .map(|s| s.to_string()) + .unwrap_or_else(String::default); + let key = key.to_string(); + drop(post); + spawn_blocking(move || { + let file = OpenOptions::new() + .read(true) + .write(true) + .truncate(false) + .create(true) + .open(&path)?; + let mut lock = RwLock::new(file); + debug!("Trying to lock feed {:?}", &path); + let mut guard = lock.write()?; - let mut content = String::new(); - guard.read_to_string(&mut content).await?; - let mut channels: Vec<super::MicropubChannel>; - if !content.is_empty() { - channels = serde_json::from_str(&content)?; - } else { - channels = Vec::default(); - } + let mut content = String::new(); - channels.push(super::MicropubChannel { - uid: key.to_string(), - name: post["properties"]["name"][0] - .as_str() - .map(|s| s.to_string()) - .unwrap_or_else(String::default), - }); - guard.seek(std::io::SeekFrom::Start(0)).await?; - guard.set_len(0).await?; - guard - .write_all(serde_json::to_string(&channels)?.as_bytes()) - .await?; - } + (*guard).read_to_string(&mut content)?; + + let mut channels: Vec<super::MicropubChannel>; + if !content.is_empty() { + channels = serde_json::from_str(&content)?; + } else { + channels = Vec::default(); + } + + channels.push(super::MicropubChannel { + uid: key.to_string(), + name: channel_name, + }); + (*guard).seek(SeekFrom::Start(0))?; + (*guard).set_len(0)?; + (*guard).write_all(serde_json::to_string(&channels)?.as_bytes())?; + Result::Ok(()) + }).await?; + } Ok(()) } async fn update_post<'a>(&self, url: &'a str, update: serde_json::Value) -> Result<()> { let path = url_to_path(&self.root_dir, url); - let f = OpenOptions::new() - .write(true) - .read(true) - .truncate(false) - .open(&path) - .await?; - - let mut lock = get_lockable_file(f).await; - let mut guard = lock.write()?; - - let mut content = String::new(); - guard.read_to_string(&mut content).await?; - let json: serde_json::Value = serde_json::from_str(&content)?; - // Apply the editing algorithms - let new_json = modify_post(&json, &update)?; - - (*guard).set_len(0).await?; - (*guard).seek(std::io::SeekFrom::Start(0)).await?; - (*guard).write_all(new_json.to_string().as_bytes()).await?; - (*guard).flush().await?; - drop(guard); + + let (old_json, new_json) = spawn_blocking(move || { + let f = OpenOptions::new() + .write(true) + .read(true) + .truncate(false) + .open(&path)?; + + let mut lock = RwLock::new(f); + let mut guard = lock.write()?; + + let mut content = String::new(); + (*guard).read_to_string(&mut content)?; + let json: serde_json::Value = serde_json::from_str(&content)?; + // Apply the editing algorithms + let new_json = modify_post(&json, &update)?; + + (*guard).set_len(0)?; + (*guard).seek(SeekFrom::Start(0))?; + (*guard).write_all(new_json.to_string().as_bytes())?; + (*guard).flush()?; + + Result::Ok((json, new_json)) + }).await?; // TODO check if URLs changed between old and new JSON Ok(()) } @@ -432,28 +444,30 @@ impl Storage for FileStorage { path.push("channels"); let path = path.to_path(&self.root_dir); - match File::open(&path).await { - Ok(f) => { - let lock = get_lockable_file(f).await; - let guard = lock.read()?; - - let mut content = String::new(); - (&mut &*guard).read_to_string(&mut content).await?; - // This should not happen, but if it does, let's handle it gracefully instead of failing. - if content.is_empty() { - return Ok(vec![]); + spawn_blocking(move || { + match File::open(&path) { + Ok(f) => { + let lock = RwLock::new(f); + let guard = lock.read()?; + + let mut content = String::new(); + (&mut &*guard).read_to_string(&mut content)?; + // This should not happen, but if it does, handle it gracefully + if content.is_empty() { + return Ok(vec![]); + } + let channels: Vec<super::MicropubChannel> = serde_json::from_str(&content)?; + Ok(channels) } - let channels: Vec<super::MicropubChannel> = serde_json::from_str(&content)?; - Ok(channels) - } - Err(e) => { - if e.kind() == IOErrorKind::NotFound { - Ok(vec![]) - } else { - Err(e.into()) + Err(e) => { + if e.kind() == IOErrorKind::NotFound { + Ok(vec![]) + } else { + Err(e.into()) + } } } - } + }).await } async fn read_feed_with_limit<'a>( @@ -520,6 +534,7 @@ impl Storage for FileStorage { if let Err(e) = async_std::fs::remove_file(path).await { Err(e.into()) } else { + // TODO check for dangling references in the channel list Ok(()) } } @@ -532,19 +547,22 @@ impl Storage for FileStorage { path.push("settings"); let path = path.to_path(&self.root_dir); - let lock = get_lockable_file(File::open(path).await?).await; - let guard = lock.read()?; - - let mut content = String::new(); - (&mut &*guard).read_to_string(&mut content).await?; - drop(guard); - let settings: HashMap<String, String> = serde_json::from_str(&content)?; - // XXX consider returning string slices instead of cloning a string every time - // it might come with a performance hit and/or memory usage inflation - settings - .get(setting) - .cloned() - .ok_or_else(|| StorageError::new(ErrorKind::Backend, "Setting not set")) + let setting = setting.to_string(); + spawn_blocking(move || { + let lock = RwLock::new(File::open(path)?); + let guard = lock.read()?; + + let mut content = String::new(); + (&mut &*guard).read_to_string(&mut content)?; + drop(guard); + let settings: HashMap<String, String> = serde_json::from_str(&content)?; + // XXX consider returning string slices instead of cloning a string every time + // it might come with a performance hit and/or memory usage inflation + settings + .get(&setting) + .cloned() + .ok_or_else(|| StorageError::new(ErrorKind::Backend, "Setting not set")) + }).await } async fn set_setting<'a>(&self, setting: &'a str, user: &'a str, value: &'a str) -> Result<()> { @@ -560,32 +578,33 @@ impl Storage for FileStorage { async_std::fs::create_dir_all(path.parent().unwrap()).await?; } - let file = OpenOptions::new() - .write(true) - .read(true) - .truncate(false) - .create(true) - .open(&path) - .await?; - let mut lock = get_lockable_file(file).await; - log::debug!("Created a lock. Locking for writing..."); - let mut guard = lock.write()?; - - log::debug!("Locked. Writing."); - let mut content = String::new(); - guard.read_to_string(&mut content).await?; - let mut settings: HashMap<String, String> = if content.is_empty() { - HashMap::default() - } else { - serde_json::from_str(&content)? - }; - - settings.insert(setting.to_string(), value.to_string()); - guard.seek(std::io::SeekFrom::Start(0)).await?; - guard.set_len(0).await?; - guard - .write_all(serde_json::to_string(&settings)?.as_bytes()) - .await?; - Ok(()) + let (setting, value) = (setting.to_string(), value.to_string()); + + spawn_blocking(move || { + let file = OpenOptions::new() + .write(true) + .read(true) + .truncate(false) + .create(true) + .open(&path)?; + let mut lock = RwLock::new(file); + log::debug!("Created a lock. Locking for writing..."); + let mut guard = lock.write()?; + + log::debug!("Locked. Writing."); + let mut content = String::new(); + (&mut &*guard).read_to_string(&mut content)?; + let mut settings: HashMap<String, String> = if content.is_empty() { + HashMap::default() + } else { + serde_json::from_str(&content)? + }; + + settings.insert(setting.to_string(), value.to_string()); + (&mut *guard).seek(SeekFrom::Start(0))?; + (&mut *guard).set_len(0)?; + (&mut *guard).write_all(serde_json::to_string(&settings)?.as_bytes())?; + Result::Ok(()) + }).await } } |