Reorganize main crate for testability.

Signed-off-by: Jason Volk <jason@zemos.net>
This commit is contained in:
Jason Volk
2025-09-27 12:05:38 +00:00
parent ba12773a5a
commit 1313eb0b64
11 changed files with 248 additions and 214 deletions

View File

@@ -17,17 +17,11 @@ crate-type = [
# "dylib",
]
[[bench]]
name = "state_res"
harness = false
[features]
brotli_compression = [
"reqwest/brotli",
]
tuwunel_mods = [
"dep:libloading"
]
console = []
gzip_compression = [
"reqwest/gzip",
]
@@ -53,6 +47,9 @@ release_max_log_level = [
"log/release_max_level_info",
]
sentry_telemetry = []
tuwunel_mods = [
"dep:libloading"
]
zstd_compression = [
"reqwest/zstd",
]
@@ -124,3 +121,7 @@ criterion.workspace = true
[lints]
workspace = true
[[bench]]
name = "state_res"
harness = false

View File

@@ -3,7 +3,8 @@
use std::path::PathBuf;
use clap::{ArgAction, Parser};
use tuwunel_core::{
use crate::{
Err, Result,
config::{Figment, FigmentValue},
err, toml,
@@ -16,37 +17,37 @@ use tuwunel_core::{
about,
long_about = None,
name = "tuwunel",
version = tuwunel_core::version(),
version = crate::version(),
)]
pub(crate) struct Args {
pub struct Args {
#[arg(short, long)]
/// Path to the config TOML file (optional)
pub(crate) config: Option<Vec<PathBuf>>,
pub config: Option<Vec<PathBuf>>,
/// Override a configuration variable using TOML 'key=value' syntax
#[arg(long, short('O'))]
pub(crate) option: Vec<String>,
pub option: Vec<String>,
/// Run in a stricter read-only --maintenance mode.
#[arg(long)]
pub(crate) read_only: bool,
pub read_only: bool,
/// Run in maintenance mode while refusing connections.
#[arg(long)]
pub(crate) maintenance: bool,
pub maintenance: bool,
#[cfg(feature = "console")]
/// Activate admin command console automatically after startup.
#[arg(long, num_args(0))]
pub(crate) console: bool,
pub console: bool,
/// Execute console command automatically after startup.
#[arg(long)]
pub(crate) execute: Vec<String>,
pub execute: Vec<String>,
/// Set functional testing modes if available. Ex '--test=smoke'
#[arg(long, hide(true))]
pub(crate) test: Vec<String>,
pub test: Vec<String>,
/// Override the tokio worker_thread count.
#[arg(
@@ -55,7 +56,7 @@ pub(crate) struct Args {
env = "TOKIO_WORKER_THREADS",
default_value = available_parallelism().to_string(),
)]
pub(crate) worker_threads: usize,
pub worker_threads: usize,
/// Override the tokio global_queue_interval.
#[arg(
@@ -64,7 +65,7 @@ pub(crate) struct Args {
env = "TOKIO_GLOBAL_QUEUE_INTERVAL",
default_value = "192"
)]
pub(crate) global_event_interval: u32,
pub global_event_interval: u32,
/// Override the tokio event_interval.
#[arg(
@@ -73,7 +74,7 @@ pub(crate) struct Args {
env = "TOKIO_EVENT_INTERVAL",
default_value = "512"
)]
pub(crate) kernel_event_interval: u32,
pub kernel_event_interval: u32,
/// Override the tokio max_io_events_per_tick.
#[arg(
@@ -82,7 +83,7 @@ pub(crate) struct Args {
env = "TOKIO_MAX_IO_EVENTS_PER_TICK",
default_value = "512"
)]
pub(crate) kernel_events_per_tick: usize,
pub kernel_events_per_tick: usize,
/// Set the histogram bucket size, in microseconds (tokio_unstable). Default
/// is 25 microseconds. If the values of the histogram don't approach zero
@@ -95,7 +96,7 @@ pub(crate) struct Args {
env = "TUWUNEL_RUNTIME_HISTOGRAM_INTERVAL",
default_value = "25"
)]
pub(crate) worker_histogram_interval: u64,
pub worker_histogram_interval: u64,
/// Set the histogram bucket count (tokio_unstable). Default is 20.
#[arg(
@@ -104,7 +105,7 @@ pub(crate) struct Args {
env = "TUWUNEL_RUNTIME_HISTOGRAM_BUCKETS",
default_value = "20"
)]
pub(crate) worker_histogram_buckets: usize,
pub worker_histogram_buckets: usize,
/// Toggles worker affinity feature.
#[arg(
@@ -117,7 +118,7 @@ pub(crate) struct Args {
default_value = "true",
default_missing_value = "true",
)]
pub(crate) worker_affinity: bool,
pub worker_affinity: bool,
/// Toggles feature to promote memory reclamation by the operating system
/// when tokio worker runs out of work.
@@ -129,7 +130,7 @@ pub(crate) struct Args {
num_args = 0..=1,
require_equals(false),
)]
pub(crate) gc_on_park: Option<bool>,
pub gc_on_park: Option<bool>,
/// Toggles muzzy decay for jemalloc arenas associated with a tokio
/// worker (when worker-affinity is enabled). Setting to false releases
@@ -145,15 +146,19 @@ pub(crate) struct Args {
num_args = 0..=1,
require_equals(false),
)]
pub(crate) gc_muzzy: Option<bool>,
pub gc_muzzy: Option<bool>,
}
impl Default for Args {
fn default() -> Self { Self::parse() }
}
/// Parse commandline arguments into structured data
#[must_use]
pub(super) fn parse() -> Args { Args::parse() }
pub fn parse() -> Args { Args::parse() }
/// Synthesize any command line options with configuration file options.
pub(crate) fn update(mut config: Figment, args: &Args) -> Result<Figment> {
pub fn update(mut config: Figment, args: &Args) -> Result<Figment> {
if args.read_only {
config = config.join(("rocksdb_read_only", true));
}

View File

@@ -1,6 +1,7 @@
#![type_length_limit = "12288"]
pub mod alloc;
pub mod args;
pub mod config;
pub mod debug;
pub mod error;
@@ -9,6 +10,7 @@ pub mod log;
pub mod matrix;
pub mod metrics;
pub mod mods;
pub mod runtime;
pub mod server;
pub mod utils;
@@ -20,12 +22,14 @@ pub use ::smallstr;
pub use ::smallvec;
pub use ::toml;
pub use ::tracing;
pub use args::Args;
pub use config::Config;
pub use error::Error;
pub use info::{rustc_flags_capture, version, version::version};
pub use matrix::{
Event, EventTypeExt, Pdu, PduCount, PduEvent, PduId, RoomVersion, pdu, state_res,
};
pub use runtime::Runtime;
pub use server::Server;
pub use utils::{ctor, dtor, implement, result, result::Result};

View File

@@ -9,15 +9,15 @@ use std::{
};
use tokio::runtime::Builder;
pub use tokio::runtime::{Handle, Runtime};
#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
use tuwunel_core::result::LogDebugErr;
use tuwunel_core::{
Result, debug, is_true,
use crate::result::LogDebugErr;
use crate::{
Args, Result, Server, debug, is_true,
utils::sys::compute::{nth_core_available, set_affinity},
};
use crate::{clap::Args, server::Server};
const WORKER_NAME: &str = "tuwunel:worker";
const WORKER_MIN: usize = 2;
const WORKER_KEEPALIVE: u64 = 36;
@@ -30,7 +30,10 @@ static WORKER_AFFINITY: OnceLock<bool> = OnceLock::new();
static GC_ON_PARK: OnceLock<Option<bool>> = OnceLock::new();
static GC_MUZZY: OnceLock<Option<bool>> = OnceLock::new();
pub(super) fn new(args: &Args) -> Result<tokio::runtime::Runtime> {
pub fn new(args: Option<&Args>) -> Result<Runtime> {
let args_default = args.is_none().then(Args::default);
let args = args.unwrap_or_else(|| args_default.as_ref().expect("default arguments"));
WORKER_AFFINITY
.set(args.worker_affinity)
.expect("set WORKER_AFFINITY from program argument");
@@ -86,9 +89,8 @@ fn enable_histogram(builder: &mut Builder, args: &Args) {
#[cfg(tokio_unstable)]
#[tracing::instrument(name = "stop", level = "info", skip_all)]
pub(super) fn shutdown(server: &Arc<Server>, runtime: tokio::runtime::Runtime) {
pub fn shutdown(server: &Arc<Server>, runtime: Runtime) -> Result {
use tracing::Level;
use tuwunel_core::event;
// The final metrics output is promoted to INFO when tokio_unstable is active in
// a release/bench mode and DEBUG is likely optimized out
@@ -100,21 +102,22 @@ pub(super) fn shutdown(server: &Arc<Server>, runtime: tokio::runtime::Runtime) {
wait_shutdown(server, runtime);
let runtime_metrics = server
.server
.metrics
.runtime_interval()
.unwrap_or_default();
event!(LEVEL, ?runtime_metrics, "Final runtime metrics");
crate::event!(LEVEL, ?runtime_metrics, "Final runtime metrics");
Ok(())
}
#[cfg(not(tokio_unstable))]
#[tracing::instrument(name = "stop", level = "info", skip_all)]
pub(super) fn shutdown(server: &Arc<Server>, runtime: tokio::runtime::Runtime) {
pub fn shutdown(server: &Arc<Server>, runtime: Runtime) -> Result {
wait_shutdown(server, runtime);
Ok(())
}
fn wait_shutdown(_server: &Arc<Server>, runtime: tokio::runtime::Runtime) {
fn wait_shutdown(_server: &Arc<Server>, runtime: Runtime) {
debug!(
timeout = ?SHUTDOWN_TIMEOUT,
"Waiting for runtime..."
@@ -124,7 +127,7 @@ fn wait_shutdown(_server: &Arc<Server>, runtime: tokio::runtime::Runtime) {
// Join any jemalloc threads so they don't appear in use at exit.
#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
tuwunel_core::alloc::je::background_thread_enable(false)
crate::alloc::je::background_thread_enable(false)
.log_debug_err()
.ok();
}
@@ -153,7 +156,7 @@ fn thread_start() {
fn set_worker_affinity() {
static CORES_OCCUPIED: AtomicUsize = AtomicUsize::new(0);
let handle = tokio::runtime::Handle::current();
let handle = Handle::current();
let num_workers = handle.metrics().num_workers();
let i = CORES_OCCUPIED.fetch_add(1, Ordering::Relaxed);
if i >= num_workers {
@@ -170,7 +173,7 @@ fn set_worker_affinity() {
#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
fn set_worker_mallctl(id: usize) {
use tuwunel_core::alloc::je::{
use crate::alloc::je::{
is_affine_arena,
this_thread::{set_arena, set_muzzy_decay},
};
@@ -183,8 +186,7 @@ fn set_worker_mallctl(id: usize) {
.get()
.expect("GC_MUZZY initialized by runtime::new()");
let muzzy_auto_disable =
tuwunel_core::utils::available_parallelism() >= DISABLE_MUZZY_THRESHOLD;
let muzzy_auto_disable = crate::utils::available_parallelism() >= DISABLE_MUZZY_THRESHOLD;
if matches!(muzzy_option, Some(false) | None if muzzy_auto_disable) {
set_muzzy_decay(-1).log_debug_err().ok();
@@ -238,7 +240,7 @@ fn thread_park() {
fn gc_on_park() {
#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
tuwunel_core::alloc::je::this_thread::decay()
crate::alloc::je::this_thread::decay()
.log_debug_err()
.ok();
}

View File

@@ -51,6 +51,11 @@ assets = [
name = "tuwunel"
pkgdesc = "High performance Matrix homeserver written in Rust"
[lib]
path = "lib.rs"
bench = false
crate-type = ["rlib"]
[features]
default = [
"brotli_compression",
@@ -84,6 +89,7 @@ bzip2_compression = [
"tuwunel-service/bzip2_compression",
]
console = [
"tuwunel-core/console",
"tuwunel-service/console",
]
direct_tls = [

116
src/main/lib.rs Normal file
View File

@@ -0,0 +1,116 @@
#![type_length_limit = "4096"] //TODO: reduce me
pub mod logging;
pub mod mods;
pub mod restart;
pub mod sentry;
pub mod server;
pub mod signals;
use std::sync::Arc;
use tuwunel_core::{
Result, Runtime, debug_info, error, mod_ctor, mod_dtor, runtime::shutdown,
rustc_flags_capture,
};
pub use self::server::Server;
mod_ctor! {}
mod_dtor! {}
rustc_flags_capture! {}
pub fn exec(server: &Arc<Server>, runtime: Runtime) -> Result {
run(server, &runtime)?;
shutdown(&server.server, runtime)
}
pub fn run(server: &Arc<Server>, runtime: &Runtime) -> Result {
runtime.spawn(signals::enable(server.clone()));
runtime.block_on(run_async(server))
}
/// Operate the server normally in release-mode static builds. This will start,
/// run and stop the server within the asynchronous runtime.
#[cfg(any(not(tuwunel_mods), not(feature = "tuwunel_mods")))]
#[tracing::instrument(
name = "main",
parent = None,
skip_all
)]
pub async fn run_async(server: &Arc<Server>) -> Result {
extern crate tuwunel_router as router;
match router::start(&server.server).await {
| Ok(services) => server.services.lock().await.insert(services),
| Err(error) => {
error!("Critical error starting server: {error}");
return Err(error);
},
};
if let Err(error) = router::run(
server
.services
.lock()
.await
.as_ref()
.expect("services initialized"),
)
.await
{
error!("Critical error running server: {error}");
return Err(error);
}
if let Err(error) = router::stop(
server
.services
.lock()
.await
.take()
.expect("services initialized"),
)
.await
{
error!("Critical error stopping server: {error}");
return Err(error);
}
debug_info!("Exit runtime");
Ok(())
}
/// Operate the server in developer-mode dynamic builds. This will start, run,
/// and hot-reload portions of the server as-needed before returning for an
/// actual shutdown. This is not available in release-mode or static builds.
#[cfg(all(tuwunel_mods, feature = "tuwunel_mods"))]
pub async fn run_async(server: &Arc<Server>) -> Result {
let mut starts = true;
let mut reloads = true;
while reloads {
if let Err(error) = mods::open(server).await {
error!("Loading router: {error}");
return Err(error);
}
let result = mods::run(server, starts).await;
if let Ok(result) = result {
(starts, reloads) = result;
}
let force = !reloads || result.is_err();
if let Err(error) = mods::close(server, force).await {
error!("Unloading router: {error}");
return Err(error);
}
if let Err(error) = result {
error!("{error}");
return Err(error);
}
}
debug_info!("Exit runtime");
Ok(())
}

View File

@@ -12,6 +12,7 @@ use tuwunel_core::{
#[cfg(feature = "perf_measurements")]
pub(crate) type TracingFlameGuard =
Option<tracing_flame::FlushGuard<std::io::BufWriter<std::fs::File>>>;
#[cfg(not(feature = "perf_measurements"))]
pub(crate) type TracingFlameGuard = ();

View File

@@ -1,30 +1,14 @@
#![type_length_limit = "4096"] //TODO: reduce me
use std::sync::atomic::Ordering;
pub(crate) mod clap;
mod logging;
mod mods;
mod restart;
mod runtime;
mod sentry;
mod server;
mod signal;
use std::sync::{Arc, atomic::Ordering};
use tuwunel_core::{Error, Result, debug_info, error, rustc_flags_capture};
use crate::server::Server;
rustc_flags_capture! {}
use tuwunel::{Server, restart};
use tuwunel_core::{Result, args, debug_info, runtime};
fn main() -> Result {
let args = clap::parse();
let runtime = runtime::new(&args)?;
let server = Server::new(&args, Some(runtime.handle()))?;
let args = args::parse();
let runtime = runtime::new(Some(&args))?;
let server = Server::new(Some(&args), Some(runtime.handle()))?;
runtime.spawn(signal::signal(server.clone()));
runtime.block_on(async_main(&server))?;
runtime::shutdown(&server, runtime);
tuwunel::exec(&server, runtime)?;
#[cfg(unix)]
if server.server.restarting.load(Ordering::Acquire) {
@@ -34,88 +18,3 @@ fn main() -> Result {
debug_info!("Exit");
Ok(())
}
/// Operate the server normally in release-mode static builds. This will start,
/// run and stop the server within the asynchronous runtime.
#[cfg(any(not(tuwunel_mods), not(feature = "tuwunel_mods")))]
#[tracing::instrument(
name = "main",
parent = None,
skip_all
)]
async fn async_main(server: &Arc<Server>) -> Result<(), Error> {
extern crate tuwunel_router as router;
match router::start(&server.server).await {
| Ok(services) => server.services.lock().await.insert(services),
| Err(error) => {
error!("Critical error starting server: {error}");
return Err(error);
},
};
if let Err(error) = router::run(
server
.services
.lock()
.await
.as_ref()
.expect("services initialized"),
)
.await
{
error!("Critical error running server: {error}");
return Err(error);
}
if let Err(error) = router::stop(
server
.services
.lock()
.await
.take()
.expect("services initialized"),
)
.await
{
error!("Critical error stopping server: {error}");
return Err(error);
}
debug_info!("Exit runtime");
Ok(())
}
/// Operate the server in developer-mode dynamic builds. This will start, run,
/// and hot-reload portions of the server as-needed before returning for an
/// actual shutdown. This is not available in release-mode or static builds.
#[cfg(all(tuwunel_mods, feature = "tuwunel_mods"))]
async fn async_main(server: &Arc<Server>) -> Result<(), Error> {
let mut starts = true;
let mut reloads = true;
while reloads {
if let Err(error) = mods::open(server).await {
error!("Loading router: {error}");
return Err(error);
}
let result = mods::run(server, starts).await;
if let Ok(result) = result {
(starts, reloads) = result;
}
let force = !reloads || result.is_err();
if let Err(error) = mods::close(server, force).await {
error!("Unloading router: {error}");
return Err(error);
}
if let Err(error) = result {
error!("{error}");
return Err(error);
}
}
debug_info!("Exit runtime");
Ok(())
}

View File

@@ -5,7 +5,7 @@ use std::{env, os::unix::process::CommandExt, process::Command};
use tuwunel_core::{debug, info, utils};
#[cold]
pub(super) fn restart() -> ! {
pub fn restart() -> ! {
let exe = utils::sys::current_exe().expect("program path must be available");
let envs = env::vars();
let args = env::args().skip(1);

View File

@@ -1,22 +1,23 @@
use std::{path::PathBuf, sync::Arc};
use tokio::{runtime, sync::Mutex};
use tokio::sync::Mutex;
use tuwunel_core::{
Error, Result,
Args, Error, Result, args,
config::Config,
info,
implement, info,
log::Log,
runtime,
utils::{stream, sys},
};
use crate::{clap::Args, logging::TracingFlameGuard};
use crate::logging::TracingFlameGuard;
/// Server runtime state; complete
pub(crate) struct Server {
pub struct Server {
/// Server runtime state; public portion
pub(crate) server: Arc<tuwunel_core::Server>,
pub server: Arc<tuwunel_core::Server>,
pub(crate) services: Mutex<Option<Arc<tuwunel_service::Services>>>,
pub services: Mutex<Option<Arc<tuwunel_service::Services>>>,
_tracing_flame_guard: TracingFlameGuard,
@@ -28,61 +29,58 @@ pub(crate) struct Server {
pub(crate) mods: tokio::sync::RwLock<Vec<tuwunel_core::mods::Module>>,
}
impl Server {
pub(crate) fn new(
args: &Args,
runtime: Option<&runtime::Handle>,
) -> Result<Arc<Self>, Error> {
let _runtime_guard = runtime.map(runtime::Handle::enter);
#[implement(Server)]
pub fn new(args: Option<&Args>, runtime: Option<&runtime::Handle>) -> Result<Arc<Self>, Error> {
let _runtime_guard = runtime.map(runtime::Handle::enter);
let config_paths = args
.config
.as_deref()
.into_iter()
.flat_map(<[_]>::iter)
.map(PathBuf::as_path);
let args_default = args.is_none().then(Args::default);
let args = args.unwrap_or_else(|| args_default.as_ref().expect("default arguments"));
let config_paths = args
.config
.as_deref()
.into_iter()
.flat_map(<[_]>::iter)
.map(PathBuf::as_path);
let config = Config::load(config_paths)
.and_then(|raw| crate::clap::update(raw, args))
.and_then(|raw| Config::new(&raw))?;
let config = Config::load(config_paths)
.and_then(|raw| args::update(raw, args))
.and_then(|raw| Config::new(&raw))?;
let (tracing_reload_handle, tracing_flame_guard, capture) =
crate::logging::init(&config)?;
let (tracing_reload_handle, tracing_flame_guard, capture) = crate::logging::init(&config)?;
config.check()?;
config.check()?;
#[cfg(feature = "sentry_telemetry")]
let sentry_guard = crate::sentry::init(&config);
#[cfg(unix)]
sys::maximize_fd_limit()
.expect("Unable to increase maximum soft and hard file descriptor limit");
let (_old_width, _new_width) = stream::set_width(config.stream_width_default);
let (_old_amp, _new_amp) = stream::set_amplification(config.stream_amplification);
info!(
server_name = %config.server_name,
database_path = ?config.database_path,
log_levels = %config.log,
"{}",
tuwunel_core::version(),
);
let logger = Log { reload: tracing_reload_handle, capture };
Ok(Arc::new(Self {
server: Arc::new(tuwunel_core::Server::new(config, runtime.cloned(), logger)),
services: None.into(),
_tracing_flame_guard: tracing_flame_guard,
#[cfg(feature = "sentry_telemetry")]
let sentry_guard = crate::sentry::init(&config);
_sentry_guard: sentry_guard,
#[cfg(unix)]
sys::maximize_fd_limit()
.expect("Unable to increase maximum soft and hard file descriptor limit");
let (_old_width, _new_width) = stream::set_width(config.stream_width_default);
let (_old_amp, _new_amp) = stream::set_amplification(config.stream_amplification);
info!(
server_name = %config.server_name,
database_path = ?config.database_path,
log_levels = %config.log,
"{}",
tuwunel_core::version(),
);
let logger = Log { reload: tracing_reload_handle, capture };
Ok(Arc::new(Self {
server: Arc::new(tuwunel_core::Server::new(config, runtime.cloned(), logger)),
services: None.into(),
_tracing_flame_guard: tracing_flame_guard,
#[cfg(feature = "sentry_telemetry")]
_sentry_guard: sentry_guard,
#[cfg(all(tuwunel_mods, feature = "tuwunel_mods"))]
mods: tokio::sync::RwLock::new(Vec::new()),
}))
}
#[cfg(all(tuwunel_mods, feature = "tuwunel_mods"))]
mods: tokio::sync::RwLock::new(Vec::new()),
}))
}

View File

@@ -7,7 +7,7 @@ use super::server::Server;
#[cfg(unix)]
#[tracing::instrument(skip_all)]
pub(super) async fn signal(server: Arc<Server>) {
pub async fn enable(server: Arc<Server>) {
use signal::unix;
use unix::SignalKind;
@@ -22,6 +22,7 @@ pub(super) async fn signal(server: Arc<Server>) {
trace!("Installed signal handlers");
let sig: &'static str;
tokio::select! {
() = server.server.until_shutdown() => break,
_ = signal::ctrl_c() => { sig = "SIGINT"; },
_ = quit.recv() => { sig = "SIGQUIT"; },
_ = term.recv() => { sig = "SIGTERM"; },
@@ -46,9 +47,10 @@ pub(super) async fn signal(server: Arc<Server>) {
#[cfg(not(unix))]
#[tracing::instrument(skip_all)]
pub(super) async fn signal(server: Arc<Server>) {
pub async fn enable(server: Arc<Server>) {
loop {
tokio::select! {
() = server.server.until_shutdown() => break,
_ = signal::ctrl_c() => {
warn!("Received Ctrl+C");
if let Err(e) = server.server.signal.send("SIGINT") {