[runtime/tokio] use system thread stack size for all threads (#3475)

Co-authored-by: Patrick O'Grady <me@patrickogrady.xyz>

Author: andresilva

runtime/src/network/iouring.rs

@@ -25,7 +25,7 @@
 
 use crate::{
     iouring::{self, should_retry, OpBuffer, OpFd, OpIovecs},
-    Buf, BufferPool, Error, IoBuf, IoBufMut, IoBufs,
+    utils, Buf, BufferPool, Error, IoBuf, IoBufMut, IoBufs,
 };
 use commonware_utils::channel::oneshot;
 use io_uring::{opcode, types::Fd};
@@ -71,6 +71,8 @@ pub struct Config {
     pub read_buffer_size: usize,
     /// Configuration for the iouring instance.
     pub iouring_config: iouring::Config,
+    /// Stack size for the dedicated send and receive io_uring threads.
+    pub thread_stack_size: usize,
 }
 
 impl Default for Config {
@@ -82,6 +84,7 @@ impl Default for Config {
             read_write_timeout: iouring_config.max_op_timeout,
             iouring_config,
             read_buffer_size: DEFAULT_READ_BUFFER_SIZE,
+            thread_stack_size: utils::thread::system_thread_stack_size(),
         }
     }
 }
@@ -134,13 +137,13 @@ impl Network {
         let sender_registry = registry.sub_registry_with_prefix("iouring_sender");
         let (send_submitter, send_loop) =
             iouring::IoUringLoop::new(cfg.iouring_config.clone(), sender_registry);
-        std::thread::spawn(move || send_loop.run());
+        utils::thread::spawn(cfg.thread_stack_size, move || send_loop.run());
 
         // Create an io_uring instance to handle receive operations.
         let receiver_registry = registry.sub_registry_with_prefix("iouring_receiver");
         let (recv_submitter, recv_loop) =
             iouring::IoUringLoop::new(cfg.iouring_config, receiver_registry);
-        std::thread::spawn(move || recv_loop.run());
+        utils::thread::spawn(cfg.thread_stack_size, move || recv_loop.run());
 
         Ok(Self {
             tcp_nodelay: cfg.tcp_nodelay,
@@ -664,7 +667,8 @@ mod tests {
             iouring::{Config, Network},
             tests,
         },
-        BufferPool, BufferPoolConfig, Error, Listener as _, Network as _, Sink as _, Stream as _,
+        thread, BufferPool, BufferPoolConfig, Error, Listener as _, Network as _, Sink as _,
+        Stream as _,
     };
     use commonware_macros::{select, test_group};
     use prometheus_client::registry::Registry;
@@ -677,6 +681,14 @@ mod tests {
         BufferPool::new(BufferPoolConfig::for_network(), &mut Registry::default())
     }
 
+    #[test]
+    fn test_default_thread_stack_size_uses_system_default() {
+        assert_eq!(
+            Config::default().thread_stack_size,
+            thread::system_thread_stack_size()
+        );
+    }
+
     #[tokio::test]
     async fn test_trait() {
         tests::test_network_trait(|| {

runtime/src/storage/iouring.rs

@@ -24,7 +24,7 @@
 use super::Header;
 use crate::{
     iouring::{self, should_retry, OpBuffer, OpFd, OpIovecs},
-    Buf, BufferPool, Error, IoBuf, IoBufs, IoBufsMut,
+    utils, Buf, BufferPool, Error, IoBuf, IoBufs, IoBufsMut,
 };
 use commonware_codec::Encode;
 use commonware_utils::{channel::oneshot, from_hex, hex};
@@ -69,6 +69,8 @@ pub struct Config {
     pub storage_directory: PathBuf,
     /// Configuration for the iouring instance.
     pub iouring_config: iouring::Config,
+    /// Stack size for the dedicated io_uring worker thread.
+    pub thread_stack_size: usize,
 }
 
 #[derive(Clone)]
@@ -80,22 +82,28 @@ pub struct Storage {
 
 impl Storage {
     /// Returns a new `Storage` instance.
-    pub fn start(mut cfg: Config, registry: &mut Registry, pool: BufferPool) -> Self {
+    pub fn start(cfg: Config, registry: &mut Registry, pool: BufferPool) -> Self {
+        let Config {
+            storage_directory,
+            mut iouring_config,
+            thread_stack_size,
+        } = cfg;
+
         // Optimize performance by hinting the kernel that a single task will
         // submit requests. This is safe because each iouring instance runs in a
         // dedicated thread, which guarantees that the same thread that creates
         // the ring is the only thread submitting work to it.
-        cfg.iouring_config.single_issuer = true;
+        iouring_config.single_issuer = true;
 
-        let (io_submitter, iouring_loop) = iouring::IoUringLoop::new(cfg.iouring_config, registry);
+        let (io_submitter, iouring_loop) = iouring::IoUringLoop::new(iouring_config, registry);
 
         let storage = Self {
-            storage_directory: cfg.storage_directory,
+            storage_directory,
             io_submitter,
             pool,
         };
 
-        std::thread::spawn(move || iouring_loop.run());
+        utils::thread::spawn(thread_stack_size, move || iouring_loop.run());
         storage
     }
 }
@@ -579,7 +587,8 @@ impl crate::Blob for Blob {
 mod tests {
     use super::{Header, *};
     use crate::{
-        storage::tests::run_storage_tests, Blob, BufferPool, BufferPoolConfig, Storage as _,
+        storage::tests::run_storage_tests, utils::thread, Blob, BufferPool, BufferPoolConfig,
+        Storage as _,
     };
     use rand::{Rng as _, SeedableRng as _};
     use std::env;
@@ -595,6 +604,7 @@ mod tests {
             Config {
                 storage_directory: storage_directory.clone(),
                 iouring_config: Default::default(),
+                thread_stack_size: thread::system_thread_stack_size(),
             },
             &mut Registry::default(),
             pool,

runtime/src/tokio/runtime.rs

@@ -17,7 +17,9 @@ use crate::{
     signal::Signal,
     storage::metered::Storage as MeteredStorage,
     telemetry::metrics::task::Label,
-    utils::{add_attribute, signal::Stopper, supervision::Tree, Panicker, Registry, ScopeGuard},
+    utils::{
+        self, add_attribute, signal::Stopper, supervision::Tree, Panicker, Registry, ScopeGuard,
+    },
     BufferPool, BufferPoolConfig, Clock, Error, Execution, Handle, Metrics as _, SinkOf,
     Spawner as _, StreamOf, METRICS_PREFIX,
 };
@@ -42,7 +44,6 @@ use std::{
     num::NonZeroUsize,
     path::PathBuf,
     sync::Arc,
-    thread,
     time::{Duration, SystemTime},
 };
 use tokio::runtime::{Builder, Runtime};
@@ -141,6 +142,14 @@ pub struct Config {
     /// operations that require blocking (like `fs` and writing to `Stdout`).
     max_blocking_threads: usize,
 
+    /// Stack size to use for runtime-owned threads.
+    ///
+    /// Defaults to the system stack size when the current platform exposes it,
+    /// and otherwise falls back to Rust's default spawned-thread stack size.
+    ///
+    /// See [utils::thread::system_thread_stack_size].
+    thread_stack_size: usize,
+
     /// Whether or not to catch panics.
     catch_panics: bool,
 
@@ -170,6 +179,7 @@ impl Config {
         Self {
             worker_threads: 2,
             max_blocking_threads: 512,
+            thread_stack_size: utils::thread::system_thread_stack_size(),
             catch_panics: false,
             storage_directory,
             maximum_buffer_size: 2 * 1024 * 1024, // 2 MB
@@ -191,6 +201,11 @@ impl Config {
         self
     }
     /// See [Config]
+    pub const fn with_thread_stack_size(mut self, n: usize) -> Self {
+        self.thread_stack_size = n;
+        self
+    }
+    /// See [Config]
     pub const fn with_catch_panics(mut self, b: bool) -> Self {
         self.catch_panics = b;
         self
@@ -241,6 +256,10 @@ impl Config {
         self.max_blocking_threads
     }
     /// See [Config]
+    pub const fn thread_stack_size(&self) -> usize {
+        self.thread_stack_size
+    }
+    /// See [Config]
     pub const fn catch_panics(&self) -> bool {
         self.catch_panics
     }
@@ -297,6 +316,7 @@ pub struct Executor {
     runtime: Runtime,
     shutdown: Mutex<Stopper>,
     panicker: Panicker,
+    thread_stack_size: usize,
 }
 
 /// Implementation of [crate::Runner] for the `tokio` runtime.
@@ -334,6 +354,7 @@ impl crate::Runner for Runner {
         let runtime = Builder::new_multi_thread()
             .worker_threads(self.cfg.worker_threads)
             .max_blocking_threads(self.cfg.max_blocking_threads)
+            .thread_stack_size(self.cfg.thread_stack_size)
             .enable_all()
             .build()
             .expect("failed to create Tokio runtime");
@@ -368,6 +389,7 @@ impl crate::Runner for Runner {
                         IoUringConfig {
                             storage_directory: self.cfg.storage_directory.clone(),
                             iouring_config: Default::default(),
+                            thread_stack_size: self.cfg.thread_stack_size,
                         },
                         iouring_registry,
                         storage_buffer_pool.clone(),
@@ -404,6 +426,7 @@ impl crate::Runner for Runner {
                         shutdown_timeout: Some(self.cfg.network_cfg.read_write_timeout),
                         ..Default::default()
                     },
+                    thread_stack_size: self.cfg.thread_stack_size,
                     ..Default::default()
                 };
                 let network = MeteredNetwork::new(
@@ -435,6 +458,7 @@ impl crate::Runner for Runner {
             runtime,
             shutdown: Mutex::new(Stopper::default()),
             panicker,
+            thread_stack_size: self.cfg.thread_stack_size,
         });
 
         // Get metrics
@@ -578,7 +602,7 @@ impl crate::Spawner for Context {
         );
 
         if matches!(past, Execution::Dedicated) {
-            thread::spawn({
+            utils::thread::spawn(executor.thread_stack_size, {
                 // Ensure the task can access the tokio runtime
                 let handle = executor.runtime.handle().clone();
                 move || {
@@ -871,6 +895,21 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_default_thread_stack_size_uses_system_default() {
+        let cfg = Config::new();
+        assert_eq!(
+            cfg.thread_stack_size(),
+            utils::thread::system_thread_stack_size()
+        );
+    }
+
+    #[test]
+    fn test_thread_stack_size_override() {
+        let cfg = Config::new().with_thread_stack_size(4 * 1024 * 1024);
+        assert_eq!(cfg.thread_stack_size(), 4 * 1024 * 1024);
+    }
+
     #[test]
     fn test_explicit_buffer_pool_configs_override_worker_threads() {
         // Order does not matter -- explicit configs always win.

runtime/src/utils/mod.rs

@@ -14,6 +14,8 @@ use std::{
 
 commonware_macros::stability_mod!(BETA, pub mod buffer);
 pub mod signal;
+#[cfg(not(target_arch = "wasm32"))]
+pub(crate) mod thread;
 
 mod handle;
 pub use handle::Handle;

runtime/src/utils/thread.rs

@@ -0,0 +1,113 @@
+//! Helpers for resolving the configured thread stack size.
+
+use std::{env, sync::OnceLock, thread};
+
+/// Cached configured thread stack size.
+static SYSTEM_THREAD_STACK_SIZE: OnceLock<usize> = OnceLock::new();
+
+/// Rust's default thread stack size.
+///
+/// See <https://doc.rust-lang.org/std/thread/#stack-size>.
+const RUST_DEFAULT_THREAD_STACK_SIZE: usize = 2 * 1024 * 1024;
+
+/// Returns the value of the `RUST_MIN_STACK` environment variable, if set.
+fn rust_min_stack() -> Option<usize> {
+    env::var_os("RUST_MIN_STACK").and_then(|s| s.to_str().and_then(|s| s.parse().ok()))
+}
+
+/// Resolves the stack size to use for runtime-owned threads.
+///
+/// If `RUST_MIN_STACK` is set, this uses that value so runtime-owned threads
+/// preserve Rust's process-wide spawned-thread override.
+///
+/// Otherwise, on Unix platforms other than macOS, this queries the default
+/// stack size for newly created pthreads via `pthread_attr_init` and
+/// `pthread_attr_getstacksize`.
+///
+/// On macOS, this instead uses `RLIMIT_STACK`. macOS distinguishes between the
+/// process stack limit and the smaller default stack size for secondary
+/// pthreads, so `pthread_attr_getstacksize` would otherwise resolve the wrong
+/// default for this use case. In practice, that means preferring the larger
+/// 8 MB process default over the 512 KB secondary-thread pthread default.
+///
+/// On other platforms, or if the platform-specific query fails, this falls back
+/// to [RUST_DEFAULT_THREAD_STACK_SIZE].
+pub(crate) fn system_thread_stack_size() -> usize {
+    *SYSTEM_THREAD_STACK_SIZE.get_or_init(|| {
+        rust_min_stack()
+            .or(system_thread_stack_size_impl())
+            .unwrap_or(RUST_DEFAULT_THREAD_STACK_SIZE)
+    })
+}
+
+#[cfg(all(unix, not(target_os = "macos")))]
+fn system_thread_stack_size_impl() -> Option<usize> {
+    let mut attr = std::mem::MaybeUninit::<libc::pthread_attr_t>::uninit();
+
+    // SAFETY: `attr` points to uninitialized storage reserved for
+    // `pthread_attr_t`, exactly as required by `pthread_attr_init`.
+    if unsafe { libc::pthread_attr_init(attr.as_mut_ptr()) } != 0 {
+        return None;
+    }
+
+    // SAFETY: `pthread_attr_init` succeeded, so `attr` is now initialized.
+    let mut attr = unsafe { attr.assume_init() };
+    let mut stack_size = 0;
+    // SAFETY: `attr` is a valid initialized pthread attribute object and
+    // `stack_size` points to writable storage for the result.
+    let get_result = unsafe { libc::pthread_attr_getstacksize(&attr, &mut stack_size) };
+    // SAFETY: `attr` remains initialized until it is destroyed here.
+    let destroy_result = unsafe { libc::pthread_attr_destroy(&mut attr) };
+
+    if get_result != 0 || destroy_result != 0 || stack_size == 0 {
+        return None;
+    }
+
+    Some(stack_size)
+}
+
+#[cfg(target_os = "macos")]
+fn system_thread_stack_size_impl() -> Option<usize> {
+    // macOS uses different defaults for the main thread and spawned threads:
+    // the main thread stack is 8 MB, while secondary threads default to
+    // 512 KB. We use `RLIMIT_STACK` here to avoid inheriting the smaller
+    // secondary-thread default through `pthread_attr_getstacksize`.
+    let mut stack_limit = std::mem::MaybeUninit::<libc::rlimit>::uninit();
+
+    // SAFETY: `stack_limit` points to uninitialized storage reserved for
+    // `rlimit`, exactly as required by `getrlimit`.
+    let limit_result = unsafe { libc::getrlimit(libc::RLIMIT_STACK, stack_limit.as_mut_ptr()) };
+    if limit_result != 0 {
+        return None;
+    }
+
+    // SAFETY: `getrlimit` succeeded, so `stack_limit` is initialized.
+    let stack_limit = unsafe { stack_limit.assume_init() };
+    if stack_limit.rlim_cur == libc::RLIM_INFINITY {
+        return None;
+    }
+
+    let limit = usize::try_from(stack_limit.rlim_cur).ok()?;
+    if limit == 0 {
+        return None;
+    }
+
+    Some(limit)
+}
+
+#[cfg(not(unix))]
+const fn system_thread_stack_size_impl() -> Option<usize> {
+    None
+}
+
+/// Spawns a thread with an explicit stack size.
+pub(crate) fn spawn<F, T>(stack_size: usize, f: F) -> thread::JoinHandle<T>
+where
+    F: FnOnce() -> T + Send + 'static,
+    T: Send + 'static,
+{
+    thread::Builder::new()
+        .stack_size(stack_size)
+        .spawn(f)
+        .expect("failed to spawn thread")
+}