hotpath 0.17.0

//! RwLock instrumentation module - tracks read/write lock acquisitions and hold durations.

use crossbeam_channel::{bounded, select, unbounded, Receiver as CbReceiver, Sender as CbSender};
use hdrhistogram::Histogram;
use std::collections::HashMap;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, Mutex, OnceLock, RwLock as StdRwLock};

use crate::batch::{register_thread_batch, BatchRegistry, BatchedMeasurement, MeasurementBatch};
use crate::instant::Instant;
use crate::lib_on::hotpath_guard::{
    WORKER_BATCH_SIZE, WORKER_FLUSH_INTERVAL_MS, WORKER_SHUTDOWN_DRAIN_LIMIT,
};
use crate::lib_on::START_TIME;
use crate::metrics_server::METRICS_SERVER_PORT;

pub(crate) mod wrapper;

// Re-exported to keep the std wrapper reachable at `hotpath::rw_locks::*` for downstream code.
pub use wrapper::std::{RwLock, RwLockReadGuard, RwLockWriteGuard};

static RW_LOCK_ID_COUNTER: AtomicU32 = AtomicU32::new(1);

fn next_rw_lock_id() -> u32 {
    RW_LOCK_ID_COUNTER.fetch_add(1, Ordering::Relaxed)
}

/// Whether an acquisition was a shared (read) or exclusive (write) lock.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum RwLockKind {
    Read,
    Write,
}

/// Events sent to the background lock statistics collection thread.
#[derive(Debug)]
pub(crate) enum RwLockEvent {
    Created {
        id: u32,
        source: &'static str,
        label: Option<String>,
        type_name: &'static str,
    },
    /// Emitted when a guard is dropped. `wait_nanos` is the time blocked
    /// before the lock was granted; `acquire_nanos` is the held duration
    /// (granted -> released).
    Released {
        id: u32,
        kind: RwLockKind,
        wait_nanos: u64,
        acquire_nanos: u64,
        elapsed_ns: u64,
    },
}

/// Statistics for a single instrumented RwLock.
#[derive(Debug, Clone)]
pub(crate) struct RwLockEntry {
    pub(crate) id: u32,
    pub(crate) source: &'static str,
    pub(crate) label: Option<String>,
    pub(crate) type_name: &'static str,
    pub(crate) read_count: u64,
    pub(crate) write_count: u64,
    pub(crate) read_wait_total_nanos: u64,
    pub(crate) write_wait_total_nanos: u64,
    pub(crate) read_acquire_total_nanos: u64,
    pub(crate) write_acquire_total_nanos: u64,
    read_wait_hist: Option<Histogram<u64>>,
    write_wait_hist: Option<Histogram<u64>>,
    read_acquire_hist: Option<Histogram<u64>>,
    write_acquire_hist: Option<Histogram<u64>>,
    pub(crate) iter: u32,
}

impl RwLockEntry {
    const LOW_NS: u64 = 1;
    const HIGH_NS: u64 = 1_000_000_000_000; // 1000s
    const SIGFIGS: u8 = 3;

    fn new_histogram() -> Histogram<u64> {
        Histogram::<u64>::new_with_bounds(Self::LOW_NS, Self::HIGH_NS, Self::SIGFIGS)
            .expect("hdrhistogram init")
    }

    #[inline]
    fn record(hist: &mut Option<Histogram<u64>>, nanos: u64) {
        if let Some(ref mut hist) = hist {
            hist.record(nanos.clamp(Self::LOW_NS, Self::HIGH_NS))
                .unwrap();
        }
    }

    pub(crate) fn count(&self, kind: RwLockKind) -> u64 {
        match kind {
            RwLockKind::Read => self.read_count,
            RwLockKind::Write => self.write_count,
        }
    }

    pub(crate) fn wait_avg_nanos(&self, kind: RwLockKind) -> u64 {
        let (total, count) = match kind {
            RwLockKind::Read => (self.read_wait_total_nanos, self.read_count),
            RwLockKind::Write => (self.write_wait_total_nanos, self.write_count),
        };
        total.checked_div(count).unwrap_or(0)
    }

    pub(crate) fn acquire_avg_nanos(&self, kind: RwLockKind) -> u64 {
        let (total, count) = match kind {
            RwLockKind::Read => (self.read_acquire_total_nanos, self.read_count),
            RwLockKind::Write => (self.write_acquire_total_nanos, self.write_count),
        };
        total.checked_div(count).unwrap_or(0)
    }

    fn percentile(hist: &Option<Histogram<u64>>, count: u64, p: f64) -> u64 {
        match hist {
            Some(hist) if count > 0 => hist.value_at_percentile(p.clamp(0.0, 100.0)),
            _ => 0,
        }
    }

    pub(crate) fn wait_percentile_nanos(&self, kind: RwLockKind, p: f64) -> u64 {
        let (hist, count) = match kind {
            RwLockKind::Read => (&self.read_wait_hist, self.read_count),
            RwLockKind::Write => (&self.write_wait_hist, self.write_count),
        };
        Self::percentile(hist, count, p)
    }

    pub(crate) fn acquire_percentile_nanos(&self, kind: RwLockKind, p: f64) -> u64 {
        let (hist, count) = match kind {
            RwLockKind::Read => (&self.read_acquire_hist, self.read_count),
            RwLockKind::Write => (&self.write_acquire_hist, self.write_count),
        };
        Self::percentile(hist, count, p)
    }
}

pub(crate) struct RwLocksInternalState {
    pub(crate) stats: HashMap<u32, RwLockEntry>,
}

pub(crate) struct RwLocksState {
    pub(crate) event_tx: CbSender<Vec<RwLockEvent>>,
    pub(crate) inner: Arc<StdRwLock<RwLocksInternalState>>,
    pub(crate) shutdown_tx: Mutex<Option<CbSender<()>>>,
    pub(crate) completion_rx: Mutex<Option<CbReceiver<()>>>,
}

pub(crate) static RW_LOCKS_STATE: OnceLock<RwLocksState> = OnceLock::new();

pub(crate) fn get_sorted_rw_lock_entries() -> Vec<RwLockEntry> {
    let Some(state) = RW_LOCKS_STATE.get() else {
        return Vec::new();
    };
    let guard = state.inner.read().unwrap();
    let mut stats: Vec<RwLockEntry> = guard.stats.values().cloned().collect();
    stats.sort_by(compare_rw_lock_entries);
    stats
}

pub(crate) fn get_rw_locks_json() -> crate::json::JsonRwLocksList {
    let entries = get_sorted_rw_lock_entries();
    let elapsed = std::time::Duration::from_nanos(crate::lib_on::current_elapsed_ns());
    crate::lib_on::report::collect_rw_locks_json(
        &entries,
        elapsed,
        &crate::lib_on::hotpath_guard::configured_percentiles(),
    )
}

#[inline]
pub(crate) fn elapsed_nanos(start: Instant) -> u64 {
    start.elapsed().as_nanos() as u64
}

static EVENT_REGISTRY: BatchRegistry<RwLockEvent> = BatchRegistry::new();

thread_local! {
    static EVENT_BATCH: std::sync::Arc<std::sync::Mutex<MeasurementBatch<RwLockEvent>>> =
        register_thread_batch(&EVENT_REGISTRY);
}

#[inline]
pub(crate) fn send_rw_lock_event(event: RwLockEvent) {
    let _suspend = crate::lib_on::SuspendAllocTracking::new();
    EVENT_BATCH.with(|b| {
        if let Ok(mut b) = b.lock() {
            b.add(event);
        }
    });
}

/// Flushes every thread's buffered rw_lock events into the worker channel.
/// Called at shutdown before the worker is signalled to stop.
pub(crate) fn flush_rw_lock_batch() {
    EVENT_REGISTRY.flush_all();
}

impl BatchedMeasurement for RwLockEvent {
    fn elapsed_since_start_ns(&self) -> u64 {
        match self {
            RwLockEvent::Released { elapsed_ns, .. } => *elapsed_ns,
            _ => 0,
        }
    }

    fn fetch_sender() -> Option<CbSender<Vec<Self>>> {
        Some(RW_LOCKS_STATE.get()?.event_tx.clone())
    }

    fn is_flush_boundary(&self) -> bool {
        matches!(self, RwLockEvent::Created { .. })
    }
}

fn process_rw_lock_event(state: &mut RwLocksInternalState, event: RwLockEvent) {
    match event {
        RwLockEvent::Created {
            id,
            source,
            label,
            type_name,
        } => {
            let iter = state.stats.values().filter(|s| s.source == source).count() as u32;
            state.stats.insert(
                id,
                RwLockEntry {
                    id,
                    source,
                    label,
                    type_name,
                    read_count: 0,
                    write_count: 0,
                    read_wait_total_nanos: 0,
                    write_wait_total_nanos: 0,
                    read_acquire_total_nanos: 0,
                    write_acquire_total_nanos: 0,
                    read_wait_hist: Some(RwLockEntry::new_histogram()),
                    write_wait_hist: Some(RwLockEntry::new_histogram()),
                    read_acquire_hist: Some(RwLockEntry::new_histogram()),
                    write_acquire_hist: Some(RwLockEntry::new_histogram()),
                    iter,
                },
            );
        }
        RwLockEvent::Released {
            id,
            kind,
            wait_nanos,
            acquire_nanos,
            elapsed_ns: _,
        } => {
            if let Some(entry) = state.stats.get_mut(&id) {
                match kind {
                    RwLockKind::Read => {
                        entry.read_count += 1;
                        entry.read_wait_total_nanos += wait_nanos;
                        entry.read_acquire_total_nanos += acquire_nanos;
                        RwLockEntry::record(&mut entry.read_wait_hist, wait_nanos);
                        RwLockEntry::record(&mut entry.read_acquire_hist, acquire_nanos);
                    }
                    RwLockKind::Write => {
                        entry.write_count += 1;
                        entry.write_wait_total_nanos += wait_nanos;
                        entry.write_acquire_total_nanos += acquire_nanos;
                        RwLockEntry::record(&mut entry.write_wait_hist, wait_nanos);
                        RwLockEntry::record(&mut entry.write_acquire_hist, acquire_nanos);
                    }
                }
            }
        }
    }
}

/// Registers a new RwLock with the profiling subsystem.
pub(crate) fn register_rw_lock<T>(source: &'static str, label: Option<String>) -> u32 {
    let type_name = std::any::type_name::<T>();
    init_rw_locks_state();
    let id = next_rw_lock_id();

    send_rw_lock_event(RwLockEvent::Created {
        id,
        source,
        label,
        type_name,
    });

    id
}

/// Initialize the lock statistics collection system (called on first instrumented lock).
pub(crate) fn init_rw_locks_state() -> &'static RwLocksState {
    RW_LOCKS_STATE.get_or_init(|| {
        START_TIME.get_or_init(Instant::now);

        let (event_tx, event_rx) = unbounded::<Vec<RwLockEvent>>();
        let (shutdown_tx, shutdown_rx) = bounded::<()>(1);
        let (completion_tx, completion_rx) = bounded::<()>(1);

        let inner = Arc::new(StdRwLock::new(RwLocksInternalState {
            stats: HashMap::new(),
        }));
        let inner_clone = Arc::clone(&inner);

        std::thread::Builder::new()
            .name("hp-rw-locks".into())
            .spawn(move || {
                let mut local_buffer: Vec<RwLockEvent> = Vec::with_capacity(WORKER_BATCH_SIZE);
                let flush_interval = std::time::Duration::from_millis(WORKER_FLUSH_INTERVAL_MS);

                loop {
                    select! {
                        recv(event_rx) -> result => {
                            match result {
                                Ok(events) => {
                                    local_buffer.extend(events);
                                    if local_buffer.len() >= WORKER_BATCH_SIZE {
                                        if let Ok(mut shared) = inner_clone.write() {
                                            for e in local_buffer.drain(..) {
                                                process_rw_lock_event(&mut shared, e);
                                            }
                                        }
                                    }
                                }
                                Err(_) => {
                                    if !local_buffer.is_empty() {
                                        if let Ok(mut shared) = inner_clone.write() {
                                            for e in local_buffer.drain(..) {
                                                process_rw_lock_event(&mut shared, e);
                                            }
                                        }
                                    }
                                    break;
                                }
                            }
                        }
                        recv(shutdown_rx) -> _ => {
                            let mut drained_events = Vec::with_capacity(WORKER_BATCH_SIZE);
                            for _ in 0..WORKER_SHUTDOWN_DRAIN_LIMIT {
                                match event_rx.try_recv() {
                                    Ok(events) => drained_events.extend(events),
                                    Err(_) => break,
                                }
                            }

                            if let Ok(mut shared) = inner_clone.write() {
                                for e in local_buffer.drain(..) {
                                    process_rw_lock_event(&mut shared, e);
                                }
                                for event in drained_events {
                                    process_rw_lock_event(&mut shared, event);
                                }
                            }
                            break;
                        }
                        default(flush_interval) => {
                            if !local_buffer.is_empty() {
                                if let Ok(mut shared) = inner_clone.write() {
                                    for e in local_buffer.drain(..) {
                                        process_rw_lock_event(&mut shared, e);
                                    }
                                }
                            }
                        }
                    }
                }

                let _ = completion_tx.send(());
            })
            .expect("Failed to spawn rw_lock-stats-collector thread");

        crate::metrics_server::start_metrics_server_once(*METRICS_SERVER_PORT);

        RwLocksState {
            event_tx,
            inner,
            shutdown_tx: Mutex::new(Some(shutdown_tx)),
            completion_rx: Mutex::new(Some(completion_rx)),
        }
    })
}

/// Compare two lock stats for sorting. Custom labels first, then by source and iter.
pub(crate) fn compare_rw_lock_entries(a: &RwLockEntry, b: &RwLockEntry) -> std::cmp::Ordering {
    match (a.label.is_some(), b.label.is_some()) {
        (true, false) => std::cmp::Ordering::Less,
        (false, true) => std::cmp::Ordering::Greater,
        (true, true) => a
            .label
            .as_ref()
            .unwrap()
            .cmp(b.label.as_ref().unwrap())
            .then_with(|| a.iter.cmp(&b.iter)),
        (false, false) => a.source.cmp(b.source).then_with(|| a.iter.cmp(&b.iter)),
    }
}

/// Trait for instrumenting RwLocks. Dispatches on the type of the wrapped lock
/// (e.g. [`std::sync::RwLock`] or [`parking_lot::RwLock`]).
///
/// This trait is not intended for direct use. Use the `rw_lock!` macro instead.
#[doc(hidden)]
pub trait InstrumentRwLock {
    type Output;
    fn instrument(self, source: &'static str, label: Option<String>) -> Self::Output;
}

/// Instrument an [`std::sync::RwLock`], [`parking_lot::RwLock`], `async_lock::RwLock`, or
/// `tokio::sync::RwLock` for read/write profiling.
///
/// Returns an instrumented drop-in replacement that proxies to the wrapped lock and records
/// how long read and write locks are held. The wrapper type matches the API of the underlying
/// lock (`std::sync::RwLock` returns `LockResult`s; `parking_lot::RwLock` returns guards directly;
/// `async_lock::RwLock` and `tokio::sync::RwLock` expose async `read`/`write` returning guards).
///
/// `parking_lot::RwLock` support requires the `parking_lot` feature; `async_lock::RwLock`
/// support requires the `async-lock` feature; `tokio::sync::RwLock` support requires the
/// `tokio` feature.
///
/// # Examples
///
/// ```rust,no_run
/// let lock = hotpath::rw_lock!(std::sync::RwLock::new(0u32));
/// *lock.write().unwrap() += 1;
/// let _ = *lock.read().unwrap();
/// ```
#[macro_export]
macro_rules! rw_lock {
    ($expr:expr) => {{
        const RW_LOCK_ID: &'static str = concat!(file!(), ":", line!());
        $crate::InstrumentRwLock::instrument($expr, RW_LOCK_ID, None)
    }};

    ($expr:expr, label = $label:expr) => {{
        const RW_LOCK_ID: &'static str = concat!(file!(), ":", line!());
        $crate::InstrumentRwLock::instrument($expr, RW_LOCK_ID, Some($label.to_string()))
    }};
}