joularcore 0.2.0

/*
 * Copyright (c) 2025-2026, Adel Noureddine.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the
 * GNU Lesser General Public License v3.0 only (LGPL-3.0-only)
 * which accompanies this distribution, and is available at
 * https://www.gnu.org/licenses/lgpl-3.0.en.html
 *
 * Author : Adel Noureddine
 */

//! Shared-memory ring buffer for low-latency IPC with other processes.
//!
//! # Layout
//!
//! The shared region is a native-endian `u64` head counter followed by
//! [`BUFFER_SIZE`] slots of [`PowerRecord`]:
//!
//! ```text
//! offset 0                8              8 + 48         8 + 2*48   ...
//!        +----------------+--------------+--------------+----------+
//!        | head: u64      | slot 0       | slot 1       | ...      |
//!        +----------------+--------------+--------------+----------+
//! ```
//!
//! # Reader protocol
//!
//! The writer stores the incremented head *after* the slot it describes, with
//! release ordering. A reader should therefore:
//!
//! 1. Load `head` with acquire ordering. `head == 0` means nothing was written.
//! 2. Read slot `(head - 1) % BUFFER_SIZE`.
//! 3. Load `head` again. If it advanced by [`BUFFER_SIZE`] or more, the writer
//!    lapped the reader mid-copy; discard the value and retry.
//!
//! At the usual 1 Hz sample rate a lap takes seconds, so step 3 effectively
//! never fails, but skipping it makes torn reads possible.

use crate::monitor::PowerRecord;
use crate::{Error, Result};

#[cfg(unix)]
use {
    memmap2::MmapMut,
    std::fs::{File, OpenOptions},
    std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt},
    std::path::Path,
    std::sync::Mutex,
};

#[cfg(windows)]
use {
    std::os::windows::ffi::OsStrExt,
    std::path::Path,
    std::sync::Mutex,
    windows::Win32::Foundation::{ERROR_ALREADY_EXISTS, GetLastError, INVALID_HANDLE_VALUE},
    windows::Win32::System::Memory::{
        CreateFileMappingW, FILE_MAP_ALL_ACCESS, MapViewOfFile, PAGE_READWRITE,
    },
    windows::core::PCWSTR,
};

use std::mem::size_of;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};

/// How many samples the ring buffer holds before overwriting the oldest.
pub const BUFFER_SIZE: usize = 5;

/// Bytes reserved for the head counter at the start of the region.
const HEADER_SIZE: usize = size_of::<u64>();

/// One sample as laid out in shared memory: a [`PowerRecord`], which is
/// `#[repr(C)]` and free of padding so it can be copied byte for byte by readers
/// written in any language.
const ENTRY_SIZE: usize = size_of::<PowerRecord>();
const _: () = assert!(ENTRY_SIZE == 48, "PowerRecord layout changed");

/// Total size of the shared region.
const REGION_SIZE: usize = HEADER_SIZE + BUFFER_SIZE * ENTRY_SIZE;

/// Writes samples into the shared-memory ring buffer.
///
/// Drop the writer to release the mapping.
pub struct RingBufferWriter {
    #[cfg(unix)]
    mmap: Mutex<MmapMut>,
    #[cfg(windows)]
    mapping: Mutex<WindowsMapping>,
}

impl RingBufferWriter {
    /// The default shared-memory path (Unix) or object name (Windows).
    #[must_use]
    pub fn default_path() -> PathBuf {
        PathBuf::from(if cfg!(target_os = "macos") {
            "/tmp/joularcorering"
        } else if cfg!(windows) {
            r"Local\JoularCoreRing"
        } else {
            "/dev/shm/joularcorering"
        })
    }

    /// Open the ring buffer at the default location.
    ///
    /// # Errors
    ///
    /// See [`RingBufferWriter::with_path`].
    pub fn new() -> Result<Self> {
        Self::with_path(Self::default_path())
    }

    /// Open the ring buffer at a specific location.
    ///
    /// On Unix this is a filesystem path; on Windows it is the name of a
    /// section object, conventionally prefixed with `Local\`.
    ///
    /// # Errors
    ///
    /// Fails if the target already exists and is not a plain file owned by the
    /// current user (Unix), or if a section of that name already exists
    /// (Windows). Both cases mean another program controls the object, and
    /// adopting it would let that program decide where this process writes.
    pub fn with_path(path: impl Into<PathBuf>) -> Result<Self> {
        let path = path.into();

        #[cfg(unix)]
        {
            let file = open_exclusive(&path)?;
            file.set_len(REGION_SIZE as u64)?;

            // SAFETY: the file is a regular file of exactly REGION_SIZE bytes
            // that we just validated as owned by this user with no other hard
            // links. Concurrent readers may map it too, all cross-process
            // accesses to the head counter go through atomics, and slots are
            // plain data whose tearing the reader protocol detects.
            let mmap = unsafe { MmapMut::map_mut(&file) }
                .map_err(|e| Error::config(format!("failed to map {}: {e}", path.display())))?;

            if mmap.len() < REGION_SIZE {
                return Err(Error::config(format!(
                    "mapped {} bytes but {REGION_SIZE} are required",
                    mmap.len()
                )));
            }

            Ok(Self {
                mmap: Mutex::new(mmap),
            })
        }

        #[cfg(windows)]
        {
            Ok(Self {
                mapping: Mutex::new(WindowsMapping::create(&path)?),
            })
        }
    }

    /// Append a sample, overwriting the oldest slot once the buffer is full.
    pub fn write(&self, entry: PowerRecord) {
        #[cfg(unix)]
        {
            let mut mmap = self.mmap.lock().unwrap_or_else(|e| e.into_inner());
            // SAFETY: `with_path` verified the mapping is at least REGION_SIZE
            // bytes, and the mapping is kept alive by `self`.
            unsafe { write_entry(mmap.as_mut_ptr(), entry) };
        }

        #[cfg(windows)]
        {
            let mapping = self.mapping.lock().unwrap_or_else(|e| e.into_inner());
            // SAFETY: `WindowsMapping::create` verified the view is at least
            // REGION_SIZE bytes, and the view is kept alive by `self`.
            unsafe { write_entry(mapping.ptr, entry) };
        }
    }
}

/// Mirrors every sample into the shared-memory ring buffer.
impl crate::output::OutputSink for RingBufferWriter {
    fn send(&mut self, sample: &crate::monitor::MonitorSample) -> crate::Result<()> {
        self.write(sample.into());
        Ok(())
    }
}

/// Store `entry` in the next slot and publish it by advancing the head.
///
/// # Safety
///
/// `base` must point to at least [`REGION_SIZE`] writable bytes that stay valid
/// for the duration of the call, and must be `u64`-aligned.
unsafe fn write_entry(base: *mut u8, entry: PowerRecord) {
    // SAFETY: the caller guarantees `base` is aligned and large enough, so the
    // first 8 bytes are a valid `AtomicU64`. Readers in other processes use the
    // same atomic accesses.
    let head = unsafe { AtomicU64::from_ptr(base.cast::<u64>()) };
    let index = (head.load(Ordering::Relaxed) as usize) % BUFFER_SIZE;

    // SAFETY: `index < BUFFER_SIZE`, so the slot lies inside the region.
    unsafe {
        base.add(HEADER_SIZE + index * ENTRY_SIZE)
            .cast::<PowerRecord>()
            .write_unaligned(entry);
    }

    // Release ordering publishes the slot write above: a reader that observes
    // the new head with acquire ordering also observes the slot contents.
    head.fetch_add(1, Ordering::Release);
}

/// Open the ring buffer file, refusing anything we do not fully control.
///
/// The default paths live in world-writable, sticky directories (`/tmp`,
/// `/dev/shm`), so another local user can create the path first. Every check
/// runs against the **open descriptor** rather than the path, which closes the
/// window between checking and opening, and `O_NOFOLLOW` stops the open from
/// being redirected through a symlink into a file we would then truncate.
#[cfg(unix)]
fn open_exclusive(path: &Path) -> Result<File> {
    let mut options = OpenOptions::new();
    options
        .read(true)
        .write(true)
        // Mode 0o600: same-user consumers (the IPC case) can still map the
        // buffer, other local users cannot read live power telemetry.
        .mode(0o600)
        .custom_flags(libc::O_NOFOLLOW);

    let file = match options.clone().create_new(true).open(path) {
        Ok(file) => return Ok(file),
        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => options
            .open(path)
            .map_err(|e| Error::config(format!("failed to open {}: {e}", path.display())))?,
        Err(e) => {
            return Err(Error::config(format!(
                "failed to create {}: {e}",
                path.display()
            )));
        }
    };

    let meta = file.metadata()?;
    // SAFETY: `geteuid` reads process state and cannot fail.
    let euid = unsafe { libc::geteuid() };

    if !meta.is_file() {
        return Err(Error::config(format!(
            "{} is not a regular file; refusing to use it",
            path.display()
        )));
    }
    if meta.uid() != euid {
        return Err(Error::config(format!(
            "{} is owned by uid {} but this process runs as uid {euid}; \
             remove it or run as the owner",
            path.display(),
            meta.uid()
        )));
    }
    if meta.nlink() != 1 {
        return Err(Error::config(format!(
            "{} has {} hard links; refusing to use it",
            path.display(),
            meta.nlink()
        )));
    }

    // The file may predate this build and carry a laxer mode from the umask.
    file.set_permissions(std::fs::Permissions::from_mode(0o600))?;

    Ok(file)
}

/// A named section object plus its mapped view.
#[cfg(windows)]
struct WindowsMapping {
    handle: windows::Win32::Foundation::HANDLE,
    ptr: *mut u8,
}

// SAFETY: the handle and view are owned exclusively by this struct, which is
// only ever reached behind a `Mutex`.
#[cfg(windows)]
unsafe impl Send for WindowsMapping {}

#[cfg(windows)]
impl WindowsMapping {
    fn create(name: &Path) -> Result<Self> {
        let wide: Vec<u16> = name
            .as_os_str()
            .encode_wide()
            .chain(std::iter::once(0))
            .collect();

        // SAFETY: `wide` is a NUL-terminated UTF-16 string that outlives the
        // call, and the size arguments describe the region we want.
        let handle = unsafe {
            CreateFileMappingW(
                INVALID_HANDLE_VALUE,
                None,
                PAGE_READWRITE,
                0,
                REGION_SIZE as u32,
                PCWSTR::from_raw(wide.as_ptr()),
            )
        }
        .map_err(|e| Error::config(format!("CreateFileMappingW failed: {e}")))?;

        // CreateFileMappingW *opens* an existing section of the same name
        // instead of failing. Adopting it would let whoever created it choose
        // the section's size, and every subsequent write would run past the end
        // of a deliberately undersized view.
        // SAFETY: `GetLastError` only reads this thread's last-error value.
        if unsafe { GetLastError() } == ERROR_ALREADY_EXISTS {
            // SAFETY: `handle` is a valid handle returned above.
            let _ = unsafe { windows::Win32::Foundation::CloseHandle(handle) };
            return Err(Error::config(format!(
                "a shared memory section named {} already exists; \
                 another Joular Core instance may be running",
                name.display()
            )));
        }

        // Map exactly REGION_SIZE bytes rather than "the whole section", so an
        // unexpected section size fails here instead of silently truncating.
        // SAFETY: `handle` is a valid section handle.
        let view = unsafe { MapViewOfFile(handle, FILE_MAP_ALL_ACCESS, 0, 0, REGION_SIZE) };
        if view.Value.is_null() {
            // SAFETY: `handle` is a valid handle returned above.
            let _ = unsafe { windows::Win32::Foundation::CloseHandle(handle) };
            return Err(Error::config("MapViewOfFile failed"));
        }

        Ok(Self {
            handle,
            ptr: view.Value.cast::<u8>(),
        })
    }
}

#[cfg(windows)]
impl Drop for WindowsMapping {
    fn drop(&mut self) {
        // SAFETY: both resources were created in `create` and are owned here.
        unsafe {
            let _ = windows::Win32::System::Memory::UnmapViewOfFile(
                windows::Win32::System::Memory::MEMORY_MAPPED_VIEW_ADDRESS {
                    Value: self.ptr.cast(),
                },
            );
            let _ = windows::Win32::Foundation::CloseHandle(self.handle);
        }
    }
}

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

    fn sample(timestamp: u64) -> PowerRecord {
        PowerRecord {
            timestamp,
            cpu_power: 1.0,
            gpu_power: 2.0,
            total_power: 3.0,
            cpu_usage: 4.0,
            pid_or_app_power: 5.0,
        }
    }

    /// Read a slot the way an external consumer would.
    fn read_slot(region: &[u8], index: usize) -> PowerRecord {
        let start = HEADER_SIZE + index * ENTRY_SIZE;
        let mut bytes = [0u8; ENTRY_SIZE];
        bytes.copy_from_slice(&region[start..start + ENTRY_SIZE]);
        // SAFETY: the bytes came from a `PowerRecord` written by
        // `write_entry`, and the type is plain data.
        unsafe { std::ptr::read_unaligned(bytes.as_ptr().cast()) }
    }

    fn head_of(region: &[u8]) -> u64 {
        u64::from_ne_bytes(region[..HEADER_SIZE].try_into().unwrap())
    }

    #[test]
    fn the_wire_format_is_unchanged() {
        // Readers written in Java, Python and C copy these bytes out by offset.
        // Reordering a field or changing one's type breaks every one of them
        // silently, so the layout is pinned here rather than left to whatever
        // the struct definition happens to say today.
        let record = PowerRecord {
            timestamp: 1_700_000_000,
            cpu_power: 1.0,
            gpu_power: 2.0,
            total_power: 3.0,
            cpu_usage: 4.0,
            pid_or_app_power: 5.0,
        };

        let mut backing: Vec<u64> = vec![0; REGION_SIZE.div_ceil(size_of::<u64>())];
        let base = backing.as_mut_ptr().cast::<u8>();
        // SAFETY: `backing` is u64-aligned and at least REGION_SIZE bytes.
        unsafe { write_entry(base, record) };
        // SAFETY: same allocation, read back as bytes for the duration of the
        // borrow.
        let region = unsafe { std::slice::from_raw_parts(base, REGION_SIZE) };

        // A record is 48 bytes: one u64 then five f64, in this order, with no
        // padding anywhere.
        assert_eq!(ENTRY_SIZE, 48);
        assert_eq!(HEADER_SIZE, 8);

        let slot = &region[HEADER_SIZE..HEADER_SIZE + ENTRY_SIZE];
        let u64_at =
            |offset: usize| u64::from_ne_bytes(slot[offset..offset + 8].try_into().unwrap());
        let f64_at =
            |offset: usize| f64::from_ne_bytes(slot[offset..offset + 8].try_into().unwrap());

        assert_eq!(u64_at(0), 1_700_000_000, "timestamp at offset 0");
        assert_eq!(f64_at(8), 1.0, "cpu_power at offset 8");
        assert_eq!(f64_at(16), 2.0, "gpu_power at offset 16");
        assert_eq!(f64_at(24), 3.0, "total_power at offset 24");
        assert_eq!(f64_at(32), 4.0, "cpu_usage at offset 32");
        assert_eq!(f64_at(40), 5.0, "pid_or_app_power at offset 40");
    }

    #[test]
    fn entries_land_in_successive_slots_and_wrap() {
        // Backed by u64s so the head counter is correctly aligned, as it is in
        // a real page-aligned mapping.
        let mut backing: Vec<u64> = vec![0; REGION_SIZE.div_ceil(size_of::<u64>())];
        let base = backing.as_mut_ptr().cast::<u8>();

        for i in 0..(BUFFER_SIZE as u64 + 2) {
            // SAFETY: `backing` is u64-aligned and at least REGION_SIZE bytes.
            unsafe { write_entry(base, sample(i)) };
        }

        // SAFETY: same allocation, read back as bytes for the duration of the
        // borrow.
        let region = unsafe { std::slice::from_raw_parts(base, REGION_SIZE) };

        assert_eq!(head_of(region), BUFFER_SIZE as u64 + 2);
        // The last two writes lapped and overwrote the two oldest slots.
        assert_eq!(read_slot(region, 0).timestamp, 5);
        assert_eq!(read_slot(region, 1).timestamp, 6);
        assert_eq!(read_slot(region, 2).timestamp, 2);
    }
}