tuwunel_database/engine/
env.rs

1use std::{
2	ptr::eq as ptr_eq,
3	sync::{Arc, LockResult, Mutex, MutexGuard, PoisonError, Weak},
4};
5
6use tuwunel_core::{Result, Server, debug, implement};
7
8use crate::or_else;
9
10/// The shared rocksdb environment.
11///
12/// The inner mutex guards the handle itself, which the engine needs while
13/// opening a database or a backup. Releasing the last reference shuts down and
14/// joins the environment's background threads, so an engine must hold one for
15/// as long as it is open.
16pub(super) struct Env(Mutex<rocksdb::Env>);
17
18/// The process-global rocksdb environment, held weakly so it lives exactly as
19/// long as some context needs it.
20///
21/// `Env::new` returns rocksdb's default environment singleton, so every handle
22/// addresses the same object and its thread pools are shared by every engine
23/// open in this process. Locking this slot across both acquisition and
24/// teardown is what keeps the two mutually exclusive.
25static ENV: Mutex<Weak<Env>> = Mutex::new(Weak::new());
26
27/// Take a reference to the shared environment, creating one when no context
28/// currently holds it.
29///
30/// The slot is held for the whole body so an acquisition cannot interleave
31/// with the teardown in [`Drop for Env`]. The priority knobs apply to
32/// the environment rather than to any one context, so they are read from the
33/// config of whichever server first needs it.
34#[implement(Env)]
35pub(super) fn acquire(server: &Server) -> Result<Arc<Self>> {
36	let mut slot = ENV.lock().expect("environment slot locked");
37
38	if let Some(env) = slot.upgrade() {
39		return Ok(env);
40	}
41
42	let config = &server.config;
43	let mut env = rocksdb::Env::new().or_else(or_else)?;
44
45	if config.rocksdb_compaction_prio_idle {
46		env.lower_thread_pool_cpu_priority();
47	}
48
49	if config.rocksdb_compaction_ioprio_idle {
50		env.lower_thread_pool_io_priority();
51	}
52
53	let env = Arc::new(Self(env.into()));
54	*slot = Arc::downgrade(&env);
55
56	Ok(env)
57}
58
59#[implement(Env)]
60#[inline]
61pub(super) fn lock(&self) -> LockResult<MutexGuard<'_, rocksdb::Env>> { self.0.lock() }
62
63impl Drop for Env {
64	#[cold]
65	fn drop(&mut self) {
66		let mut slot = ENV.lock().expect("environment slot locked");
67
68		// A context which acquired after our last strong reference went away
69		// owns the same environment now, so the shutdown is its job.
70		if !ptr_eq(slot.as_ptr(), self) {
71			return;
72		}
73
74		*slot = Weak::new();
75
76		let env = self
77			.0
78			.get_mut()
79			.unwrap_or_else(PoisonError::into_inner);
80
81		debug!("Shutting down background threads");
82		env.set_high_priority_background_threads(0);
83		env.set_low_priority_background_threads(0);
84		env.set_bottom_priority_background_threads(0);
85		env.set_background_threads(0);
86
87		debug!("Joining background threads...");
88		env.join_all_threads();
89	}
90}
tuwunel_database/engine/env.rs

tuwunel_database/engine/
env.rs
