diff --git a/.envrc b/.envrc new file mode 100644 index 0000000..4a4726a --- /dev/null +++ b/.envrc @@ -0,0 +1 @@ +use_nix diff --git a/Cargo.toml b/Cargo.toml index c51132c..bcfd643 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tmp-postgrust" -version = "0.9.0" +version = "0.10.0" authors = ["John Children "] license = "MIT" edition = "2018" diff --git a/shell.nix b/shell.nix new file mode 100644 index 0000000..309de8e --- /dev/null +++ b/shell.nix @@ -0,0 +1,18 @@ +{ pkgs ? import {} }: + +pkgs.mkShell { + name = "diesel-tracing-dev-env"; + buildInputs = with pkgs; [ + cargo + rustc + rustfmt + clippy + + pkg-config + postgresql + libmysqlclient + sqlite + ]; + + LD_LIBRARY_PATH = "${pkgs.postgresql.lib}/lib"; +} diff --git a/src/asynchronous.rs b/src/asynchronous.rs index 868efa7..0c72aee 100644 --- a/src/asynchronous.rs +++ b/src/asynchronous.rs @@ -9,7 +9,6 @@ use tempfile::TempDir; use tokio::fs::create_dir_all; use tokio::io::Lines; use tokio::process::{ChildStderr, ChildStdout}; -use tokio::sync::{Semaphore, SemaphorePermit}; use tokio::{ io::BufReader, process::{Child, Command}, @@ -20,9 +19,6 @@ use crate::errors::{ProcessCapture, TmpPostgrustError, TmpPostgrustResult}; use crate::search::{all_dir_entries, build_copy_dst_path, find_postgresql_command}; use crate::POSTGRES_UID_GID; -/// Limit the total processes that can be running at any one time. -pub(crate) static MAX_CONCURRENT_PROCESSES: Semaphore = Semaphore::const_new(8); - #[instrument(skip(command, fail))] async fn exec_process( command: &mut Command, @@ -94,12 +90,18 @@ pub(crate) async fn exec_copy_dir(src_dir: &Path, dst_dir: &Path) -> TmpPostgrus .await .map_err(TmpPostgrustError::CopyCachedInitDBFailedFileNotFound)?; } + let src_dir = src_dir.to_owned(); + let dst_dir = dst_dir.to_owned(); - for entry in others { - reflink_copy::reflink_or_copy(&entry, build_copy_dst_path(&entry, src_dir, dst_dir)?) - .map_err(TmpPostgrustError::CopyCachedInitDBFailedFileNotFound)?; - } - + tokio::task::spawn_blocking(move || { + for entry in others { + reflink_copy::reflink_or_copy(&entry, build_copy_dst_path(&entry, &src_dir, &dst_dir)?) + .map_err(TmpPostgrustError::CopyCachedInitDBFailedFileNotFound)?; + } + Ok(()) + }) + .await + .unwrap()?; Ok(()) } @@ -193,8 +195,6 @@ pub struct ProcessGuard { /// Socket directory for connection to the running process. pub(crate) socket_dir: Arc, pub(crate) postgres_process: Child, - // Limit the total concurrent processes. - pub(crate) _process_permit: SemaphorePermit<'static>, } impl Drop for ProcessGuard { diff --git a/src/lib.rs b/src/lib.rs index 595c69b..db523ea 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -19,11 +19,12 @@ mod search; /// Methods for Synchronous API pub mod synchronous; +use std::fmt::Write as FmtWrite; use std::fs::{metadata, set_permissions}; use std::io::{BufRead, BufReader}; use std::path::Path; use std::sync::atomic::AtomicU32; -use std::sync::{Arc, Mutex, OnceLock}; +use std::sync::{Arc, OnceLock, RwLock}; use std::{fs::File, io::Write}; use ctor::dtor; @@ -45,23 +46,25 @@ pub(crate) static POSTGRES_UID_GID: OnceLock<(Uid, Gid)> = OnceLock::new(); fn cleanup_static() { #[cfg(feature = "tokio-process")] if let Some(factory_mutex) = TOKIO_POSTGRES_FACTORY.get() { - let mut guard = factory_mutex.blocking_lock(); + let mut guard = factory_mutex.blocking_write(); drop(guard.take()); } if let Some(factory_mutex) = DEFAULT_POSTGRES_FACTORY.get() { let mut guard = factory_mutex - .lock() + .write() .expect("Failed to lock default factory mutex."); drop(guard.take()); } } -static DEFAULT_POSTGRES_FACTORY: OnceLock>> = OnceLock::new(); +static DEFAULT_POSTGRES_FACTORY: OnceLock>> = OnceLock::new(); /// Create a new default instance, initializing the `DEFAULT_POSTGRES_FACTORY` if it /// does not already exist. /// +/// fsync is disabled in the default process. +/// /// # Errors /// /// Will return `Err` if postgres is not installed on system @@ -72,12 +75,15 @@ static DEFAULT_POSTGRES_FACTORY: OnceLock>> = /// is called. pub fn new_default_process() -> TmpPostgrustResult { let factory_mutex = DEFAULT_POSTGRES_FACTORY.get_or_init(|| { - Mutex::new(Some( - TmpPostgrustFactory::try_new().expect("Failed to initialize default postgres factory."), + RwLock::new(Some( + TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig { + disable_fsync: true, + }) + .expect("Failed to initialize default postgres factory."), )) }); let guard = factory_mutex - .lock() + .read() .expect("Failed to lock default factory mutex."); let factory = guard .as_ref() @@ -89,6 +95,8 @@ pub fn new_default_process() -> TmpPostgrustResult { /// does not already exist. The function passed as the `migrate` parameters /// will be run the first time the factory is initialised. /// +/// fsync is disabled in the default process. +/// /// # Errors /// /// Will return `Err` if postgres is not installed on system @@ -101,16 +109,18 @@ pub fn new_default_process_with_migrations( migrate: impl Fn(&str) -> Result<(), Box>, ) -> TmpPostgrustResult { let factory_mutex = DEFAULT_POSTGRES_FACTORY.get_or_init(|| { - let factory = - TmpPostgrustFactory::try_new().expect("Failed to initialize default postgres factory."); + let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig { + disable_fsync: true, + }) + .expect("Failed to initialize default postgres factory."); factory .run_migrations(migrate) .expect("Failed to run migrations"); - Mutex::new(Some(factory)) + RwLock::new(Some(factory)) }); let guard = factory_mutex - .lock() + .read() .expect("Failed to lock default factory mutex."); let factory = guard .as_ref() @@ -121,12 +131,14 @@ pub fn new_default_process_with_migrations( /// Static factory that can be re-used between tests. #[cfg(feature = "tokio-process")] static TOKIO_POSTGRES_FACTORY: tokio::sync::OnceCell< - tokio::sync::Mutex>, + tokio::sync::RwLock>, > = tokio::sync::OnceCell::const_new(); /// Create a new default instance, initializing the `TOKIO_POSTGRES_FACTORY` if it /// does not already exist. /// +/// fsync is disabled in the default process. +/// /// # Errors /// /// Will return `Err` if postgres is not installed on system @@ -139,12 +151,14 @@ static TOKIO_POSTGRES_FACTORY: tokio::sync::OnceCell< pub async fn new_default_process_async() -> TmpPostgrustResult { let factory_mutex = TOKIO_POSTGRES_FACTORY .get_or_try_init(|| async { - TmpPostgrustFactory::try_new_async() - .await - .map(|factory| tokio::sync::Mutex::new(Some(factory))) + TmpPostgrustFactory::try_new_async(&TmpPostgrustFactoryConfig { + disable_fsync: true, + }) + .await + .map(|factory| tokio::sync::RwLock::new(Some(factory))) }) .await?; - let guard = factory_mutex.lock().await; + let guard = factory_mutex.read().await; let factory = guard .as_ref() .expect("Default tokio factory is uninitialized or has been dropped."); @@ -155,6 +169,8 @@ pub async fn new_default_process_async() -> TmpPostgrustResult( migrate: F, ) -> TmpPostgrustResult where - F: for<'r> Fn(&'r str) -> futures_util::future::BoxFuture<'r, Result<(), Box>>, + F: for<'r> Fn( + &'r str, + ) -> futures_util::future::BoxFuture< + 'r, + Result<(), Box>, + >, { let factory_mutex = TOKIO_POSTGRES_FACTORY .get_or_try_init(|| async { - let factory = TmpPostgrustFactory::try_new_async().await?; - factory - .run_migrations_async(migrate) - .await?; - Ok(tokio::sync::Mutex::new(Some(factory))) + let factory = TmpPostgrustFactory::try_new_async(&TmpPostgrustFactoryConfig { + disable_fsync: true, + }) + .await?; + factory.run_migrations_async(migrate).await?; + Ok(tokio::sync::RwLock::new(Some(factory))) }) .await?; - let guard = factory_mutex.lock().await; + let guard = factory_mutex.read().await; let factory = guard .as_ref() .expect("Default tokio factory is uninitialized or has been dropped."); @@ -195,26 +217,44 @@ pub struct TmpPostgrustFactory { next_port: AtomicU32, } +/// Configuration for the `TmpPostgrustFactory`. +#[derive(Default, Debug)] +pub struct TmpPostgrustFactoryConfig { + /// Disable fsync this will speed up unit tests in exchange for + /// not guaranteeing that files will be written if postgresql + /// crashes. + disable_fsync: bool, +} + impl TmpPostgrustFactory { /// Build a Postgresql configuration for temporary databases as a String. - fn build_config(socket_dir: &Path) -> String { + fn build_config(factory_config: &TmpPostgrustFactoryConfig, socket_dir: &Path) -> String { let mut config = String::new(); // Minimize chance of running out of shared memory config.push_str("shared_buffers = '12MB'\n"); // Disable TCP connections. config.push_str("listen_addresses = ''\n"); // Listen on UNIX socket. - config.push_str(&format!( - "unix_socket_directories = \'{}\'\n", + writeln!( + &mut config, + "unix_socket_directories = \'{}\'", socket_dir.to_str().unwrap() - )); + ) + .expect("Failed to append unix socket to config"); + // Disable fsync this will speed up unit tests in exchange for + // not guaranteeing that files will be written if postgresql + // crashes. + writeln!(&mut config, "fsync = \'{}\'", !factory_config.disable_fsync) + .expect("Failed to append fsync option to config"); config } /// Try to create a new factory by creating temporary directories and the necessary config. #[instrument] - pub fn try_new() -> TmpPostgrustResult { + pub fn try_new( + factory_config: &TmpPostgrustFactoryConfig, + ) -> TmpPostgrustResult { let socket_dir = Builder::new() .prefix("tmp-postgrust-socket") .tempdir() @@ -228,7 +268,7 @@ impl TmpPostgrustFactory { synchronous::chown_to_non_root(socket_dir.path())?; synchronous::exec_init_db(cache_dir.path())?; - let config = TmpPostgrustFactory::build_config(socket_dir.path()); + let config = TmpPostgrustFactory::build_config(factory_config, socket_dir.path()); let factory = TmpPostgrustFactory { socket_dir: Arc::new(socket_dir), @@ -251,7 +291,9 @@ impl TmpPostgrustFactory { /// Try to create a new factory by creating temporary directories and the necessary config. #[cfg(feature = "tokio-process")] #[instrument] - pub async fn try_new_async() -> TmpPostgrustResult { + pub async fn try_new_async( + factory_config: &TmpPostgrustFactoryConfig, + ) -> TmpPostgrustResult { let socket_dir = Builder::new() .prefix("tmp-postgrust-socket") .tempdir() @@ -265,7 +307,7 @@ impl TmpPostgrustFactory { asynchronous::chown_to_non_root(socket_dir.path()).await?; asynchronous::exec_init_db(cache_dir.path()).await?; - let config = TmpPostgrustFactory::build_config(socket_dir.path()); + let config = TmpPostgrustFactory::build_config(factory_config, socket_dir.path()); let factory = TmpPostgrustFactory { socket_dir: Arc::new(socket_dir), @@ -313,16 +355,20 @@ impl TmpPostgrustFactory { /// Will error if Postgresql is unable to start or if the migrate function returns /// an error. #[cfg(feature = "tokio-process")] - pub async fn run_migrations_async( - &self, - migrate: F, - ) -> TmpPostgrustResult<()> + pub async fn run_migrations_async(&self, migrate: F) -> TmpPostgrustResult<()> where - F: for<'r> Fn(&'r str) -> futures_util::future::BoxFuture<'r, Result<(), Box>>, + F: for<'r> Fn( + &'r str, + ) -> futures_util::future::BoxFuture< + 'r, + Result<(), Box>, + >, { let process = self.start_postgresql(&self.cache_dir)?; - migrate(&process.connection_string()).await.map_err(TmpPostgrustError::MigrationsFailed)?; + migrate(&process.connection_string()) + .await + .map_err(TmpPostgrustError::MigrationsFailed)?; Ok(()) } @@ -346,7 +392,7 @@ impl TmpPostgrustFactory { if !data_directory_path.join("PG_VERSION").exists() { return Err(TmpPostgrustError::EmptyDataDirectory); - }; + } self.start_postgresql(&Arc::new(data_directory)) } @@ -417,7 +463,7 @@ impl TmpPostgrustFactory { if !data_directory_path.join("PG_VERSION").exists() { return Err(TmpPostgrustError::EmptyDataDirectory); - }; + } self.start_postgresql_async(&Arc::new(data_directory)).await } @@ -430,11 +476,6 @@ impl TmpPostgrustFactory { ) -> TmpPostgrustResult { use tokio::io::AsyncBufReadExt; - let process_permit = asynchronous::MAX_CONCURRENT_PROCESSES - .acquire() - .await - .unwrap(); - File::create(dir.path().join("postgresql.conf")) .map_err(TmpPostgrustError::CreateConfigFailed)? .write_all(self.config.as_bytes()) @@ -471,7 +512,6 @@ impl TmpPostgrustFactory { _cache_directory: Arc::clone(&self.cache_dir), socket_dir: Arc::clone(&self.socket_dir), postgres_process: postgres_process_handle, - _process_permit: process_permit, }) } } @@ -486,7 +526,34 @@ mod tests { #[test(tokio::test)] async fn it_works() { - let factory = TmpPostgrustFactory::try_new().expect("failed to create factory"); + let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig { + disable_fsync: false, + }) + .expect("failed to create factory"); + + let postgresql_proc = factory + .new_instance() + .expect("failed to create a new instance"); + + let (client, conn) = tokio_postgres::connect(&postgresql_proc.connection_string(), NoTls) + .await + .expect("failed to connect to postgresql"); + + tokio::spawn(async move { + if let Err(e) = conn.await { + error!("connection error: {}", e); + } + }); + + client.query("SELECT 1;", &[]).await.unwrap(); + } + + #[test(tokio::test)] + async fn it_works_fsync_disabled() { + let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig { + disable_fsync: true, + }) + .expect("failed to create factory"); let postgresql_proc = factory .new_instance() @@ -508,9 +575,11 @@ mod tests { #[cfg(feature = "tokio-process")] #[test(tokio::test)] async fn it_works_async() { - let factory = TmpPostgrustFactory::try_new_async() - .await - .expect("failed to create factory"); + let factory = TmpPostgrustFactory::try_new_async(&TmpPostgrustFactoryConfig { + disable_fsync: false, + }) + .await + .expect("failed to create factory"); let postgresql_proc = factory .new_instance_async() @@ -532,7 +601,10 @@ mod tests { #[test(tokio::test)] async fn two_simulatenous_processes() { - let factory = TmpPostgrustFactory::try_new().expect("failed to create factory"); + let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig { + disable_fsync: false, + }) + .expect("failed to create factory"); let proc1 = factory .new_instance() @@ -569,9 +641,11 @@ mod tests { #[cfg(feature = "tokio-process")] #[test(tokio::test)] async fn two_simulatenous_processes_async() { - let factory = TmpPostgrustFactory::try_new_async() - .await - .expect("failed to create factory"); + let factory = TmpPostgrustFactory::try_new_async(&TmpPostgrustFactoryConfig { + disable_fsync: false, + }) + .await + .expect("failed to create factory"); let proc1 = factory .new_instance_async() diff --git a/src/search.rs b/src/search.rs index 55f3a34..f2f034b 100644 --- a/src/search.rs +++ b/src/search.rs @@ -19,7 +19,7 @@ pub(crate) fn find_postgresql_command(dir: &str, name: &str) -> Result