//! Process-wide simulator cache ownership and inactivity reaping. use crate::{Parcel, Primitive, Settings, Simulator, SimulatorDataPool}; use libremetaverse_types::Error; use std::collections::HashMap; use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak}; use std::thread::JoinHandle; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; const REAPER_INTERVAL: Duration = Duration::from_secs(180); const ACTIVE_TIMESTAMP_SECONDS: u64 = 253_402_300_799; fn mutex(value: &Mutex) -> std::sync::MutexGuard<'_, T> { value .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) } struct PoolRecord { active_clients: i32, inactive_since: Option<(Instant, SystemTime)>, prim_cache: HashMap, parcels: HashMap, parcel_map: Vec, downloading_parcel_map: bool, } impl PoolRecord { fn new() -> Self { Self { active_clients: 0, inactive_since: None, prim_cache: HashMap::new(), parcels: HashMap::new(), parcel_map: vec![0; 64 * 64], downloading_parcel_map: false, } } fn snapshot(&self, handle: u64) -> SimulatorDataPool { SimulatorDataPool { active_clients: self.active_clients, downloading_parcel_map: self.downloading_parcel_map, handle, inactive_since: self.inactive_since.map_or_else( || UNIX_EPOCH + Duration::from_secs(ACTIVE_TIMESTAMP_SECONDS), |(_, timestamp)| timestamp, ), parcel_map: self.parcel_map.clone(), parcels: std::sync::RwLock::new(self.parcels.clone()), prim_cache: self.prim_cache.clone(), } } } struct Runtime { records: Mutex>, worker: Mutex>>, wake: Condvar, stopped: Mutex, } fn runtime() -> &'static Arc { static RUNTIME: OnceLock> = OnceLock::new(); RUNTIME.get_or_init(|| { Arc::new(Runtime { records: Mutex::new(HashMap::new()), worker: Mutex::new(None), wake: Condvar::new(), stopped: Mutex::new(false), }) }) } fn reap(runtime: &Runtime) { let timeout = Duration::from_millis( u64::try_from(Settings::simulator_pool_timeout().max(1)).unwrap_or(1), ); mutex(&runtime.records).retain(|_, record| { record .inactive_since .is_none_or(|(instant, _)| instant.elapsed() < timeout) }); } fn ensure_reaper() { let runtime = Arc::clone(runtime()); let mut worker = mutex(&runtime.worker); if worker.is_some() { return; } *mutex(&runtime.stopped) = false; let weak: Weak = Arc::downgrade(&runtime); *worker = Some(std::thread::spawn(move || { let Some(runtime) = weak.upgrade() else { return; }; let mut stopped = mutex(&runtime.stopped); loop { let (guard, _) = runtime .wake .wait_timeout(stopped, REAPER_INTERVAL) .unwrap_or_else(std::sync::PoisonError::into_inner); stopped = guard; if *stopped { return; } drop(stopped); reap(&runtime); stopped = mutex(&runtime.stopped); } })); } pub(crate) fn snapshots() -> HashMap { reap(runtime()); mutex(&runtime().records) .iter() .map(|(handle, record)| (*handle, record.snapshot(*handle))) .collect() } pub(crate) fn get(handle: u64) -> Result { let mut records = mutex(&runtime().records); let record = records.entry(handle).or_insert_with(PoolRecord::new); Ok(record.snapshot(handle)) } pub(crate) fn add(simulator: Simulator) -> Result<(), Error> { ensure_reaper(); let mut records = mutex(&runtime().records); let record = records .entry(simulator.handle) .or_insert_with(PoolRecord::new); record.active_clients = record.active_clients.saturating_add(1).max(1); record.inactive_since = None; Ok(()) } pub(crate) fn release(simulator: Simulator) -> Result<(), Error> { let mut records = mutex(&runtime().records); let record = records .entry(simulator.handle) .or_insert_with(PoolRecord::new); record.active_clients = record.active_clients.saturating_sub(1); if record.active_clients <= 0 { record.active_clients = 0; record.inactive_since = Some((Instant::now(), SystemTime::now())); } Ok(()) } pub(crate) fn shutdown() -> Result<(), Error> { let runtime = runtime(); *mutex(&runtime.stopped) = true; runtime.wake.notify_all(); if let Some(worker) = mutex(&runtime.worker).take() && worker.thread().id() != std::thread::current().id() { worker.join().map_err(|_| Error::InvalidOperation)?; } Ok(()) }