some_global_executor 0.1.6

// SPDX-License-Identifier: MIT OR Apache-2.0

//! A bounded record of thread pools, exposed through exfiltrate.
//!
//! "Why did my task not run" usually has one of two answers, and neither is
//! visible from outside the process: there is no executor installed at all, or
//! there is one and its queue is not draining. The first is
//! `snapshot --subsystem executor` in `some_executor`; this is the second.
//!
//! # What this reports, and what it deliberately does not
//!
//! Pool-level state: the executor's name, how many worker threads it was built
//! with, how many tasks it currently has in flight, and whether it is still
//! alive. `running_tasks` is read live at query time rather than mirrored, so
//! it cannot go stale.
//!
//! **Per-worker state is not reported.** Whether an individual thread is idle,
//! running, or blocked means different things on a native thread pool and on a
//! pool of browser workers, and the issue this implements is gated on
//! coordinating that with `wasm_lite_std`'s worker registry. Pool-level status
//! needs none of that, so it ships now and the harder question stays open
//! rather than being pre-empted by a native-shaped answer.
//!
//! # Bounded, and honest about it
//!
//! The most recent `N`, from `SOME_GLOBAL_EXECUTOR_REGISTRY_CAPACITY`,
//! defaulting to [`DEFAULT_CAPACITY`]. Executors are few and long-lived, so in
//! practice this never evicts — but it is bounded anyway, and drops are
//! counted and reported like every sibling registry.
//!
//! Behind the `exfiltrate` feature.

use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

use wasm_lite_std::time::Instant;

use crate::DrainNotify;

/// Pools kept when `SOME_GLOBAL_EXECUTOR_REGISTRY_CAPACITY` is unset.
pub const DEFAULT_CAPACITY: usize = 64;

#[derive(Clone)]
/// One thread pool the registry is holding.
pub struct Entry {
    /// A locally minted id; what `--id` selects.
    pub id: u64,
    /// The name the pool was built with.
    pub name: String,
    /// Worker threads currently configured for the pool.
    pub threads: usize,
    /// When the pool was created.
    pub created: Instant,
    /// False once the pool has been dropped.
    pub alive: bool,
    /// Read live at query time, so it cannot be stale.
    drain_notify: Arc<DrainNotify>,
    current_threads: Arc<AtomicUsize>,
    current_alive: Arc<AtomicBool>,
}

impl Entry {
    /// Tasks the pool currently has in flight, read live.
    pub fn running_tasks(&self) -> u64 {
        self.drain_notify.running_tasks.load(Ordering::Relaxed) as u64
    }
}

impl std::fmt::Debug for Entry {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("Entry")
            .field("id", &self.id)
            .field("name", &self.name)
            .field("threads", &self.current_threads.load(Ordering::Relaxed))
            .field("alive", &self.current_alive.load(Ordering::Acquire))
            .field("running_tasks", &self.running_tasks())
            .finish()
    }
}

struct Registry {
    capacity: usize,
    entries: VecDeque<Entry>,
}

impl Registry {
    /// Inserts, evicting if full. Returns how many entries it forgot.
    ///
    /// A method so the policy is testable on a local registry rather than the
    /// process-global one.
    fn push(&mut self, entry: Entry) -> u64 {
        let mut forgotten = 0;
        while self.entries.len() >= self.capacity {
            let victim = self
                .entries
                .iter()
                .position(|entry| !entry.current_alive.load(Ordering::Acquire))
                .unwrap_or(0);
            self.entries.remove(victim);
            forgotten += 1;
        }
        self.entries.push_back(entry);
        forgotten
    }
}

static REGISTRY: Mutex<Option<Registry>> = Mutex::new(None);
static DROPPED: AtomicU64 = AtomicU64::new(0);
static NEXT_ID: AtomicU64 = AtomicU64::new(1);

fn configured_capacity() -> usize {
    std::env::var("SOME_GLOBAL_EXECUTOR_REGISTRY_CAPACITY")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|capacity| *capacity > 0)
        .unwrap_or(DEFAULT_CAPACITY)
}

fn with<R>(f: impl FnOnce(&mut Registry) -> R) -> Option<R> {
    let mut guard = REGISTRY.try_lock().ok()?;
    let registry = guard.get_or_insert_with(|| Registry {
        capacity: configured_capacity(),
        entries: VecDeque::new(),
    });
    Some(f(registry))
}

pub(crate) fn record_created(
    name: &str,
    threads: usize,
    current_threads: Arc<AtomicUsize>,
    current_alive: Arc<AtomicBool>,
    drain_notify: Arc<DrainNotify>,
) -> u64 {
    let id = NEXT_ID.fetch_add(1, Ordering::Relaxed);
    let recorded = with(|registry| {
        let forgotten = registry.push(Entry {
            id,
            name: name.to_string(),
            threads,
            created: Instant::now(),
            alive: true,
            drain_notify,
            current_threads,
            current_alive,
        });
        DROPPED.fetch_add(forgotten, Ordering::Relaxed);
    });
    if recorded.is_none() {
        DROPPED.fetch_add(1, Ordering::Relaxed);
    }
    id
}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
/// Registry accounting.
pub struct RegistryStats {
    /// Entries forgotten to stay bounded, or because the lock was held.
    pub dropped: u64,
    /// Entries currently held.
    pub retained: usize,
    /// The configured bound.
    pub capacity: usize,
}

/// A snapshot of the registry accounting.
pub fn stats() -> RegistryStats {
    let (retained, capacity) = with(|registry| (registry.entries.len(), registry.capacity))
        .unwrap_or((0, configured_capacity()));
    RegistryStats {
        dropped: DROPPED.load(Ordering::Relaxed),
        retained,
        capacity,
    }
}

/// Retained entries, oldest first.
///
/// `None` when the registry was locked; report `busy` rather than an empty
/// list.
pub fn entries() -> Option<Vec<Entry>> {
    with(|registry| {
        registry
            .entries
            .iter()
            .cloned()
            .map(|mut entry| {
                entry.threads = entry.current_threads.load(Ordering::Relaxed);
                entry.alive = entry.current_alive.load(Ordering::Acquire);
                entry
            })
            .collect()
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    fn entry(id: u64, alive: bool) -> Entry {
        Entry {
            id,
            name: format!("pool-{id}"),
            threads: 1,
            created: Instant::now(),
            alive,
            drain_notify: Arc::new(DrainNotify::new()),
            current_threads: Arc::new(AtomicUsize::new(1)),
            current_alive: Arc::new(AtomicBool::new(alive)),
        }
    }

    fn registry(capacity: usize) -> Registry {
        Registry {
            capacity,
            entries: VecDeque::new(),
        }
    }

    /// Overflow forgets a dropped pool before a live one.
    #[test]
    fn overflow_forgets_dropped_before_live() {
        let mut registry = registry(2);
        registry.push(entry(1, true));
        registry.push(entry(2, false));
        assert_eq!(registry.push(entry(3, true)), 1);
        let ids: Vec<u64> = registry.entries.iter().map(|entry| entry.id).collect();
        assert_eq!(ids, vec![1, 3]);
    }

    #[test]
    fn overflow_of_all_live_drops_the_oldest_and_counts_it() {
        let mut registry = registry(2);
        registry.push(entry(1, true));
        registry.push(entry(2, true));
        assert_eq!(registry.push(entry(3, true)), 1);
        let ids: Vec<u64> = registry.entries.iter().map(|entry| entry.id).collect();
        assert_eq!(ids, vec![2, 3]);
    }

    /// `running_tasks` is read through the live counter, not a copy taken at
    /// registration -- a mirrored value would be stale exactly when it matters.
    #[test]
    fn running_tasks_reads_through_to_the_live_counter() {
        let entry = entry(1, true);
        assert_eq!(entry.running_tasks(), 0);
        entry
            .drain_notify
            .running_tasks
            .fetch_add(3, Ordering::Relaxed);
        assert_eq!(entry.running_tasks(), 3);
    }

    #[test]
    fn dropping_while_registry_is_busy_still_marks_entry_dead() {
        let id = NEXT_ID.fetch_add(1, Ordering::Relaxed);
        let alive = Arc::new(AtomicBool::new(true));
        let mut guard = REGISTRY
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        *guard = Some(Registry {
            capacity: 1,
            entries: VecDeque::from([{
                let mut entry = entry(id, true);
                entry.current_alive = alive.clone();
                entry
            }]),
        });

        drop(crate::RegistryRegistration {
            id,
            threads: Arc::new(AtomicUsize::new(1)),
            alive,
        });
        drop(guard);

        let retained = entries().expect("the released registry lock should be available");
        assert!(
            !retained.iter().find(|entry| entry.id == id).unwrap().alive,
            "a contended registry lock must not lose the final liveness update"
        );
    }
}