MetaCrate/crates/libremetaverse/src/simulator_data_pool.rs

//! 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<T>(value: &Mutex<T>) -> 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<u32, Primitive>,
    parcels: HashMap<i32, Parcel>,
    parcel_map: Vec<i32>,
    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<HashMap<u64, PoolRecord>>,
    worker: Mutex<Option<JoinHandle<()>>>,
    wake: Condvar,
    stopped: Mutex<bool>,
}

fn runtime() -> &'static Arc<Runtime> {
    static RUNTIME: OnceLock<Arc<Runtime>> = 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<Runtime> = 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<u64, SimulatorDataPool> {
    reap(runtime());
    mutex(&runtime().records)
        .iter()
        .map(|(handle, record)| (*handle, record.snapshot(*handle)))
        .collect()
}

pub(crate) fn get(handle: u64) -> Result<SimulatorDataPool, Error> {
    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(())
}