From 22d3fcb1a2ee2bb2d051036a0ac447e9c0ad8b61 Mon Sep 17 00:00:00 2001 From: John Children Date: Wed, 24 Jun 2026 16:30:37 +0100 Subject: [PATCH 1/3] feat: Add back process semaphore Reverts the change that removed the process semaphore for async subprocesses as having too many processes in large test suites was causing resource exhaustion on Apple devices. Adds a dependency on num_cpus to figure out an appropriate number of permits rather than just guessing 8 as before. A similar change may be necessary for sync process management. --- Cargo.toml | 3 ++- src/asynchronous.rs | 12 +++++++++--- src/errors.rs | 4 ++++ src/lib.rs | 12 +++++++++++- 4 files changed, 26 insertions(+), 5 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 27a3e2a..a9acfd3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tmp-postgrust" -version = "0.11.0" +version = "0.11.1" authors = ["John Children "] license = "MIT" edition = "2018" @@ -23,6 +23,7 @@ tokio = { version = "1.8", features = ["parking_lot", "rt", "sync", "io-util", " tracing = "0.1" which = "4.0" futures-util = { version = "0.3", optional = true } +num_cpus = "1.17.0" [dev-dependencies] test-log = { version = "0.2", default-features = false, features = ["trace"] } diff --git a/src/asynchronous.rs b/src/asynchronous.rs index b93c813..f52913e 100644 --- a/src/asynchronous.rs +++ b/src/asynchronous.rs @@ -1,14 +1,15 @@ -use nix::sys::signal::{self, Signal}; -use nix::unistd::{Pid, Uid, User}; use std::convert::TryInto; use std::path::Path; use std::process::Stdio; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; +use nix::sys::signal::{self, Signal}; +use nix::unistd::{Pid, Uid, User}; 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}, @@ -19,6 +20,9 @@ use crate::errors::{ProcessCapture, TmpPostgrustError, TmpPostgrustResult}; use crate::search::{all_dir_entries, build_copy_dst_path}; use crate::POSTGRES_UID_GID; +/// Limit the total processes that can be running at any one time. +pub(crate) static MAX_CONCURRENT_PROCESSES: OnceLock = OnceLock::new(); + #[instrument(skip(command, fail))] async fn exec_process( command: &mut Command, @@ -196,6 +200,8 @@ 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/errors.rs b/src/errors.rs index 9d61e80..fed75cd 100644 --- a/src/errors.rs +++ b/src/errors.rs @@ -75,6 +75,10 @@ pub enum TmpPostgrustError { /// The explicit directory that was searched, if one was provided. searched_dir: Option, }, + /// Error when a process permit cannot be acquired for a new postgres subprocess. + #[cfg(feature = "tokio-process")] + #[error("acquiring process permit failed")] + AsyncProcessPermitAcquireError(#[source] tokio::sync::AcquireError), } /// Result type for `TmpPostgrustError`, used by functions in this crate. diff --git a/src/lib.rs b/src/lib.rs index 5e2ef6c..837b2a8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -30,8 +30,10 @@ use std::{fs::File, io::Write}; use ctor::dtor; use nix::unistd::{Gid, Uid}; use tempfile::{Builder, TempDir}; +use tokio::sync::Semaphore; use tracing::{debug, info, instrument}; +use crate::asynchronous::MAX_CONCURRENT_PROCESSES; use crate::errors::{TmpPostgrustError, TmpPostgrustResult}; use crate::search::PostgresBinaries; @@ -498,7 +500,7 @@ impl TmpPostgrustFactory { self.start_postgresql_async(&Arc::new(data_directory)).await } - #[cfg(feature = "tokio-process")] + // #[cfg(feature = "tokio-process")] #[instrument(skip(self))] async fn start_postgresql_async( &self, @@ -516,6 +518,13 @@ impl TmpPostgrustFactory { .fetch_add(1, std::sync::atomic::Ordering::SeqCst); asynchronous::chown_to_non_root(dir.path()).await?; + + let permit = MAX_CONCURRENT_PROCESSES + .get_or_init(|| Semaphore::new(num_cpus::get())) + .acquire() + .await + .map_err(TmpPostgrustError::AsyncProcessPermitAcquireError)?; + let mut postgres_process_handle = asynchronous::start_postgres_subprocess(&self.binaries.postgres, dir.path(), port)?; let stdout = postgres_process_handle.stdout.take().unwrap(); @@ -542,6 +551,7 @@ impl TmpPostgrustFactory { _cache_directory: Arc::clone(&self.cache_dir), socket_dir: Arc::clone(&self.socket_dir), postgres_process: postgres_process_handle, + _process_permit: permit, }) } } From 717ba9909fe14cda52918cb6c8b50ac65687892d Mon Sep 17 00:00:00 2001 From: John Children Date: Wed, 24 Jun 2026 16:33:33 +0100 Subject: [PATCH 2/3] chore: Update email address --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index a9acfd3..4e7c487 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "tmp-postgrust" version = "0.11.1" -authors = ["John Children "] +authors = ["John Children "] license = "MIT" edition = "2018" description = "Temporary postgresql instances for testing" From 820b187d4315cdf9e54cd5ff8ecadc0d9b9eaf69 Mon Sep 17 00:00:00 2001 From: John Children Date: Wed, 24 Jun 2026 16:36:44 +0100 Subject: [PATCH 3/3] chore: Fix feature gating error --- src/lib.rs | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 837b2a8..5d0cb8b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -30,10 +30,8 @@ use std::{fs::File, io::Write}; use ctor::dtor; use nix::unistd::{Gid, Uid}; use tempfile::{Builder, TempDir}; -use tokio::sync::Semaphore; use tracing::{debug, info, instrument}; -use crate::asynchronous::MAX_CONCURRENT_PROCESSES; use crate::errors::{TmpPostgrustError, TmpPostgrustResult}; use crate::search::PostgresBinaries; @@ -500,7 +498,7 @@ impl TmpPostgrustFactory { self.start_postgresql_async(&Arc::new(data_directory)).await } - // #[cfg(feature = "tokio-process")] + #[cfg(feature = "tokio-process")] #[instrument(skip(self))] async fn start_postgresql_async( &self, @@ -519,8 +517,8 @@ impl TmpPostgrustFactory { asynchronous::chown_to_non_root(dir.path()).await?; - let permit = MAX_CONCURRENT_PROCESSES - .get_or_init(|| Semaphore::new(num_cpus::get())) + let permit = crate::asynchronous::MAX_CONCURRENT_PROCESSES + .get_or_init(|| tokio::sync::Semaphore::new(num_cpus::get())) .acquire() .await .map_err(TmpPostgrustError::AsyncProcessPermitAcquireError)?;