one_collect 0.1.34811

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT license.

use std::marker::PhantomData;
use std::arch::asm;
#[cfg(target_os = "linux")]
use std::os::fd::BorrowedFd;
use std::rc::Rc;

#[cfg(target_os = "linux")]
use libc::*;

use tracing::{debug, info, trace, warn, error};
use super::abi;
use super::*;

pub mod source;

/* Arch: X64 */
#[cfg(target_arch = "x86_64")]
unsafe fn rmb() {
    asm!("lfence");
}

#[cfg(target_arch = "x86_64")]
unsafe fn mb() {
    asm!("mfence");
}

/* Arch: ARM64 */
#[cfg(target_arch = "aarch64")]
unsafe fn rmb() {
    asm!("dsb ld");
}

#[cfg(target_arch = "aarch64")]
unsafe fn mb() {
    asm!("dsb sy");
}

pub trait RingBufOptions {
    fn clone_options(&self) -> Self;

    fn attributes_mut(&mut self) -> &mut perf_event_attr;

    fn with_callchain_data(&self) -> Self where Self: Sized {
        let mut clone = self.clone_options();
        let attributes = clone.attributes_mut();

        attributes.sample_type |= abi::PERF_SAMPLE_CALLCHAIN;

        clone
    }

    fn without_user_callchain_data(&self) -> Self where Self: Sized {
        let mut clone = self.clone_options();
        let attributes = clone.attributes_mut();

        attributes.flags |= FLAG_EXCLUDE_CALLCHAIN_USER;

        clone
    }

    fn without_kernel_callchain_data(&self) -> Self where Self: Sized {
        let mut clone = self.clone_options();
        let attributes = clone.attributes_mut();

        attributes.flags |= FLAG_EXCLUDE_CALLCHAIN_KERNEL;

        clone
    }

    fn with_ip(&self) -> Self where Self: Sized {
        let mut clone = self.clone_options();
        let attributes = clone.attributes_mut();

        attributes.sample_type |= abi::PERF_SAMPLE_IP;
        attributes.flags |= FLAG_PRECISE_IP;

        clone
    }

    fn with_user_regs_data(
        &self,
        regs: u64) -> Self where Self: Sized {
        let mut clone = self.clone_options();
        let attributes = clone.attributes_mut();

        attributes.sample_type |= abi::PERF_SAMPLE_REGS_USER;
        attributes.sample_regs_user = regs;

        clone
    }

    fn with_user_stack_data(
        &self,
        stack_bytes: u32) -> Self where Self: Sized {
        let mut clone = self.clone_options();
        let attributes = clone.attributes_mut();

        attributes.sample_type |= abi::PERF_SAMPLE_STACK_USER;
        attributes.sample_stack_user = stack_bytes;

        clone
    }

    fn with_cgroup_data(&self) -> Self where Self: Sized {
        let mut clone = self.clone_options();
        let attributes = clone.attributes_mut();

        attributes.sample_type |= abi::PERF_SAMPLE_CGROUP;

        clone
    }
}

/// Determine whether cgroup ids can be recorded in samples.
///
/// This is done by checking whether the running Linux kernel supports
/// `PERF_SAMPLE_CGROUP`, which was added in kernel 5.7+. We check
/// this separately, otherwise `perf_event_open` would fail fatally.
pub(crate) fn cgroup_sample_supported() -> bool {
    let attr = perf_event_attr {
        size: PERF_ATTR_SIZE_VER4,
        event_type: PERF_TYPE_SOFTWARE,
        config: PERF_COUNT_SW_DUMMY,
        sample_type: abi::PERF_SAMPLE_CGROUP,
        flags: FLAG_DISABLED,
        .. Default::default()
    };

    // SAFETY: `attr` is the fully initialized perf_event_attr that
    // outlives the call. The syscall only reads through the pointer
    // and returns an fd or -1.
    let result = unsafe {
        syscall(
            SYS_perf_event_open,
            &attr as *const perf_event_attr as usize,
            0,                /* pid: calling process */
            (-1i32) as usize, /* cpu: any */
            (-1i32) as usize, /* group_fd: none */
            0)
    };

    if result >= 0 {
        // SAFETY: `result` is an fd we just opened and solely own.
        // Closed once now.
        unsafe { close(result as i32); }
        true
    } else {
        let err = std::io::Error::last_os_error();

        if err.raw_os_error() == Some(EINVAL) {
            warn!("PERF_SAMPLE_CGROUP unsupported; cgroup ids will not be recorded");
            false
        } else {
            /* Not a kernel support failure (but, e.g., permissions) */
            true
        }
    }
}

pub fn cpu_count() -> u32 {
    unsafe {
        const SC_NPROCESSORS_ONLN: i32 = 84;

        sysconf(SC_NPROCESSORS_ONLN) as u32
    }
}

/// Size in bytes of the per-CPU ring's data area for a user-supplied page
/// count.
///
/// The kernel requires the data area to be a power-of-two number of
/// pages, so the user's `page_count` is rounded up via
/// `next_power_of_two()`. The actual `mmap` is one additional metadata
/// page on top of this (`ring_data_bytes(n) + PAGE_SIZE`).
///
/// This is the bound the `wakeup_events_watermark` must be strictly
/// smaller than for the kernel to ever wake the perf fd.
pub(super) fn ring_data_bytes(page_count: usize) -> usize {
    let page_size = unsafe { sysconf(_SC_PAGE_SIZE) as usize };
    page_count.next_power_of_two() * page_size
}

pub fn perf_timestamp(
    attr: &perf_event_attr) -> u64 {
    unsafe {
        let mut tp = timespec {
            tv_sec: 0,
            tv_nsec: 0,
        };

        match clock_gettime(
            attr.clockid,
            &mut tp) {
            0 => {
                ((tp.tv_sec * 1000000000) + tp.tv_nsec) as u64
            }
            _ => {
                0
            }
        }
    }
}

fn perf_event_open(
    attr: &perf_event_attr,
    pid: i32,
    cpu: i32,
    group_fd: i32,
    flags: usize) -> IOResult<usize> {
    unsafe {
        match syscall(
            SYS_perf_event_open,
            attr as *const perf_event_attr as usize,
            pid as usize,
            cpu as usize,
            group_fd as usize,
            flags) {
            -1 => {
                let err = std::io::Error::last_os_error();
                error!("perf_event_open failed: pid={}, cpu={}, error={}", pid, cpu, err);
                Err(err)
            },
            result => {
                debug!("perf_event_open succeeded: pid={}, cpu={}, fd={}", pid, cpu, result);
                Ok(result as usize)
            },
        }
    }
}

pub struct Profiling;
pub struct ContextSwitches;
pub struct PageFaults;
pub struct Tracepoint;
pub struct Kernel;
pub struct Bpf;

pub struct RingBufBuilder<T = Profiling> {
    attributes: perf_event_attr,
    _type: PhantomData<T>,
}

impl RingBufBuilder {
    pub(crate) fn common_attributes() -> perf_event_attr {
        perf_event_attr {
            size: PERF_ATTR_SIZE_VER4,
            flags: FLAG_USE_CLOCKID |
                FLAG_SAMPLE_ID_ALL |
                FLAG_DISABLED |
                FLAG_EXCLUDE_HV |
                FLAG_EXCLUDE_IDLE |
                FLAG_INHERIT,
            clockid: CLOCK_MONOTONIC_RAW,
            read_format: abi::PERF_FORMAT_ID,
            sample_type: abi::PERF_SAMPLE_IDENTIFIER |
                abi::PERF_SAMPLE_TIME |
                abi::PERF_SAMPLE_TID,
            /* Leave rest default */
            .. Default::default()
        }
    }

    pub fn for_kernel() -> RingBufBuilder<Kernel> {
        let mut attributes = Self::common_attributes();

        attributes.event_type = PERF_TYPE_SOFTWARE;
        attributes.config = PERF_COUNT_SW_DUMMY;

        RingBufBuilder::<Kernel> {
            attributes,
            _type: PhantomData::<Kernel>,
        }
    }

    pub fn for_cswitches() -> RingBufBuilder<ContextSwitches> {
        let mut attributes = Self::common_attributes();

        attributes.event_type = PERF_TYPE_SOFTWARE;
        attributes.config = PERF_COUNT_SW_CONTEXT_SWITCHES;
        attributes.sample_period_freq = 1;

        RingBufBuilder::<ContextSwitches> {
            attributes,
            _type: PhantomData::<ContextSwitches>,
        }
    }

    pub fn for_soft_page_faults() -> RingBufBuilder<PageFaults> {
        let mut attributes = Self::common_attributes();

        attributes.event_type = PERF_TYPE_SOFTWARE;
        attributes.config = PERF_COUNT_SW_PAGE_FAULTS_MIN;
        attributes.sample_period_freq = 1;

        RingBufBuilder::<PageFaults> {
            attributes,
            _type: PhantomData::<PageFaults>,
        }
    }

    pub fn for_hard_page_faults() -> RingBufBuilder<PageFaults> {
        let mut attributes = Self::common_attributes();

        attributes.event_type = PERF_TYPE_SOFTWARE;
        attributes.config = PERF_COUNT_SW_PAGE_FAULTS_MAJ;
        attributes.sample_period_freq = 1;

        RingBufBuilder::<PageFaults> {
            attributes,
            _type: PhantomData::<PageFaults>,
        }
    }

    pub fn for_profiling(
        sampling_frequency: u64) -> RingBufBuilder<Profiling> {
        let mut attributes = Self::common_attributes();

        attributes.event_type = PERF_TYPE_SOFTWARE;
        attributes.config = PERF_COUNT_SW_CPU_CLOCK;
        attributes.sample_period_freq = sampling_frequency;
        attributes.flags |= FLAG_FREQ;

        RingBufBuilder::<Profiling> {
            attributes,
            _type: PhantomData::<Profiling>,
        }
    }

    pub fn for_tracepoint() -> RingBufBuilder<Tracepoint> {
        let mut attributes = Self::common_attributes();

        attributes.event_type = PERF_TYPE_TRACEPOINT;
        attributes.sample_period_freq = 1;

        attributes.sample_type |= abi::PERF_SAMPLE_RAW;

        RingBufBuilder::<Tracepoint> {
            attributes,
            _type: PhantomData::<Tracepoint>,
        }
    }

    pub fn for_bpf() -> RingBufBuilder<Bpf> {
        let mut attributes = Self::common_attributes();

        attributes.event_type = PERF_TYPE_SOFTWARE;
        attributes.config = PERF_COUNT_SW_BPF_OUTPUT;
        attributes.sample_period_freq = 1;

        attributes.sample_type |= abi::PERF_SAMPLE_RAW;

        RingBufBuilder::<Bpf> {
            attributes,
            _type: PhantomData::<Bpf>,
        }
    }
}

impl<T> RingBufBuilder<T> {
    /// Configure the kernel to wake up the perf fd only after the ring
    /// buffer has accumulated at least `bytes` of data.
    ///
    /// Sets `FLAG_WATERMARK` and `wakeup_events_watermark` on the
    /// underlying `perf_event_attr`. A value of `0` keeps the kernel
    /// default of waking on every event.
    pub(crate) const fn set_wakeup_watermark(&mut self, bytes: u32) {
        self.attributes.flags |= FLAG_WATERMARK;
        self.attributes.wakeup_events_watermark = bytes;
    }
}

impl RingBufOptions for RingBufBuilder<Profiling> {
    fn clone_options(&self) -> Self {
        Self {
            attributes: self.attributes,
            _type: self._type,
        }
    }

    fn attributes_mut(&mut self) -> &mut perf_event_attr {
        &mut self.attributes
    }
}

impl RingBufBuilder<Profiling> {
    pub(crate) fn build(&self) -> CommonRingBuf {
        CommonRingBuf::new(self.attributes)
    }
}

impl RingBufOptions for RingBufBuilder<ContextSwitches> {
    fn clone_options(&self) -> Self {
        Self {
            attributes: self.attributes,
            _type: self._type,
        }
    }

    fn attributes_mut(&mut self) -> &mut perf_event_attr {
        &mut self.attributes
    }
}

impl RingBufBuilder<ContextSwitches> {
    pub(crate) fn build(&self) -> CommonRingBuf {
        CommonRingBuf::new(self.attributes)
    }
}

impl RingBufOptions for RingBufBuilder<PageFaults> {
    fn clone_options(&self) -> Self {
        Self {
            attributes: self.attributes,
            _type: self._type,
        }
    }

    fn attributes_mut(&mut self) -> &mut perf_event_attr {
        &mut self.attributes
    }
}

impl RingBufBuilder<PageFaults> {
    pub(crate) fn build(&self) -> CommonRingBuf {
        CommonRingBuf::new(self.attributes)
    }
}

impl RingBufOptions for RingBufBuilder<Tracepoint> {
    fn clone_options(&self) -> Self {
        Self {
            attributes: self.attributes,
            _type: self._type,
        }
    }

    fn attributes_mut(&mut self) -> &mut perf_event_attr {
        &mut self.attributes
    }
}

impl RingBufBuilder<Tracepoint> {
    pub(crate) fn build(
        &self,
        tracepoint_id: u64) -> CommonRingBuf {
        let mut attributes = self.attributes;

        /*
         * We need to support live adding more events so we
         * copy then update the tracepoint id live here.
         */
        attributes.config = tracepoint_id;

        CommonRingBuf::new(attributes)
    }
}

impl RingBufOptions for RingBufBuilder<Bpf> {
    fn clone_options(&self) -> Self {
        Self {
            attributes: self.attributes,
            _type: self._type,
        }
    }

    fn attributes_mut(&mut self) -> &mut perf_event_attr {
        &mut self.attributes
    }
}

impl RingBufBuilder<Bpf> {
    pub(crate) fn build(
        &self) -> CommonRingBuf {
        CommonRingBuf::new(self.attributes)
    }
}

impl RingBufBuilder<Kernel> {
    pub fn with_executable_mmap_records(&self) -> Self {
        let mut attributes = self.attributes;

        attributes.flags |= FLAG_MMAP | FLAG_MMAP2;

        Self {
            attributes,
            _type: self._type,
        }
    }

    pub fn with_all_mmap_records(&self) -> Self {
        let mut attributes = self.attributes;

        attributes.flags |= FLAG_MMAP | FLAG_MMAP2 | FLAG_MMAP_DATA;

        Self {
            attributes,
            _type: self._type,
        }
    }

    pub fn with_comm_records(&self) -> Self {
        let mut attributes = self.attributes;

        attributes.flags |= FLAG_COMM | FLAG_COMM_EXEC;

        Self {
            attributes,
            _type: self._type,
        }
    }

    pub fn with_task_records(&self) -> Self {
        let mut attributes = self.attributes;

        attributes.flags |= FLAG_TASK;

        Self {
            attributes,
            _type: self._type,
        }
    }

    pub fn with_cswitch_records(&self) -> Self {
        let mut attributes = self.attributes;

        attributes.flags |= FLAG_CONTEXT_SWITCH;

        Self {
            attributes,
            _type: self._type,
        }
    }

    pub(crate) fn build(&self) -> CommonRingBuf {
        CommonRingBuf::new(self.attributes)
    }
}

#[repr(C)]
#[derive(Default)]
struct read_format {
    value: u64,
    id: u64,
}

pub(crate) struct CommonRingBuf {
    attributes: Rc<perf_event_attr>,
}

impl CommonRingBuf {
    pub(crate) fn new(
        attributes: perf_event_attr) -> Self {
        Self {
            attributes: Rc::new(attributes),
        }
    }

    pub(crate) fn without_callstack(
        self) -> Self {
        /* If no callchain/stack, then don't do anything */
        if !self.attributes.has_format(PERF_SAMPLE_CALLCHAIN) &&
           !self.attributes.has_format(PERF_SAMPLE_STACK_USER) {
            return self;
        }

        let mut clone = self;
        let mut attributes = *clone.attributes;

        /* Clear callchain/stack samples */
        attributes.sample_type &= !PERF_SAMPLE_CALLCHAIN;
        attributes.sample_type &= !PERF_SAMPLE_STACK_USER;
        attributes.sample_type &= !PERF_SAMPLE_REGS_USER;

        /* Enable IP only sample */
        attributes.sample_type |= PERF_SAMPLE_IP;

        clone.attributes = Rc::new(attributes);
        clone
    }

    pub(crate) fn for_cpu(
        &self,
        cpu: u32) -> CpuRingBuf {
        CpuRingBuf::new(
            cpu,
            self.attributes.clone())
    }
}

#[derive(Default)]
pub(crate) struct CpuRingCursor {
    start: u64,
    end: u64,
}

impl CpuRingCursor {
    pub(crate) fn set(
        &mut self,
        start: u64,
        end: u64) {
        self.start = start;
        self.end = end;
    }

    pub(crate) fn advance(
        &mut self,
        len: u16) {
        self.start += len as u64;
    }

    pub(crate) fn more(&self) -> bool {
        self.start < self.end
    }

    pub(crate) fn start(&self) -> u64 {
        self.start
    }
}

pub(crate) struct CpuRingReader {
    pages: *mut u8,
    pages_len: usize,
    data_offset: u64,
    data_size: u64,
    data_mask: u64,
    owned: bool,
}

impl<'a> CpuRingReader {
    pub(crate) fn new(
        pages: *mut u8,
        pages_len: usize) -> Self {
        let slice = unsafe {
            std::slice::from_raw_parts(
                pages,
                pages_len)
        };

        let data_offset = u64::from_ne_bytes(
            slice[1040..1048].try_into().unwrap());

        let data_size = u64::from_ne_bytes(
            slice[1048..1056].try_into().unwrap());

        debug!(
            "CpuRingReader initialized: data_offset={:#x}, data_size={:#x}, pages_len={}",
            data_offset, data_size, pages_len
        );

        Self {
            pages,
            pages_len,
            data_offset,
            data_size,
            data_mask: data_size - 1,
            owned: true,
        }
    }

    pub(crate) fn new_unowned(
        pages: *mut u8,
        pages_len: usize) -> Self {
        let mut reader = Self::new(pages, pages_len);
        reader.owned = false;
        reader
    }

    const fn slice(&self) -> &[u8] {
        unsafe {
            std::slice::from_raw_parts(
                self.pages,
                self.pages_len)
        }
    }

    pub(crate) fn begin_reading(
        &self,
        cursor: &mut CpuRingCursor) {
        let head = self.head();

        unsafe {
            rmb();
        }

        let tail = self.tail();

        cursor.set(tail, head);

        trace!(
            "begin_reading: head={:#x}, tail={:#x}, data_available={}",
            head, tail, head.saturating_sub(tail)
        );
    }

    pub(crate) fn data_slice(
        &'a self) -> &'a [u8] {
        let slice = self.slice();
        let data_start = self.data_offset as usize;
        let data_end = data_start + self.data_size as usize;
        &slice[data_start..data_end]
    }

    pub(crate) fn peek_header(
        &'a self,
        cursor: &CpuRingCursor,
        data_slice: &'a [u8],
        start: &mut usize) -> IOResult<abi::Header<'a>> {
        *start = (cursor.start() & self.data_mask) as usize;
        let end = *start + abi::Header::data_offset();
        let header_slice = &data_slice[*start .. end];

        match abi::Header::from_slice(header_slice) {
            Ok(header) => Ok(header),
            Err(_) => {
                trace!(
                    "peek_header failed: header slice too small, start={:#x}, end={:#x}",
                    *start, end
                );
                Err(io_error("Header slice was not large enough."))
            }
        }
    }

    pub(crate) fn peek_u64(
        &self,
        cursor: &CpuRingCursor,
        offset: u64) -> u64 {
        let start = ((cursor.start() + offset) & self.data_mask) as usize;
        let end = start + 8;

        let data_slice = self.data_slice();
        u64::from_ne_bytes(data_slice[start..end].try_into().unwrap())
    }

    pub(crate) fn read(
        &'a self,
        cursor: &mut CpuRingCursor,
        temp: &'a mut Vec<u8>) -> IOResult<&'a [u8]> {
        let data_slice = self.data_slice();
        let mut header_start = 0;

        let header = self.peek_header(
            cursor,
            data_slice,
            &mut header_start)?;

        let data_size = header.size as usize;
        let data_end = header_start + data_size;

        cursor.advance(header.size);

        if header_start + data_size <= self.data_size as usize {
            /* Fits within slice, no copy */
            trace!(
                "read: entry_type={}, size={}, header_start={:#x}, no_wrap",
                header.entry_type, header.size, header_start
            );
            Ok(&data_slice[header_start .. data_end])
        } else {
            /* Data wrapped, requires copy */
            trace!(
                "read: entry_type={}, size={}, header_start={:#x}, wrapped",
                header.entry_type, header.size, header_start
            );
            temp.clear();
            temp.extend_from_slice(&data_slice[header_start..]);
            let remaining = data_size - temp.len();
            temp.extend_from_slice(&data_slice[0..remaining]);

            Ok(&temp[0..])
        }
    }

    pub(crate) fn end_reading(
        &mut self,
        cursor: &CpuRingCursor) {
        trace!("end_reading: updating tail to {:#x}", cursor.start());
        unsafe {
            mb();
            let tail: *mut u64 = self.pages.offset(1032) as *mut u64;
            *tail = cursor.start();
        }
    }

    fn head(&self) -> u64 {
        let slice = self.slice();
        u64::from_ne_bytes(
            slice[1024..1032].try_into().unwrap())
    }

    fn tail(&self) -> u64 {
        let slice = self.slice();
        u64::from_ne_bytes(
            slice[1032..1040].try_into().unwrap())
    }
}

impl Drop for CpuRingReader {
    fn drop(&mut self) {
        if self.owned {
            unsafe {
                munmap(self.pages as *mut c_void, self.pages_len);
            }
        }
    }
}

pub(crate) struct CpuRingBuf {
    cpu: u32,
    attributes: Rc<perf_event_attr>,
    sample_time_offset: u16,
    fd: Option<i32>,
    id: Option<u64>,
}

impl CpuRingBuf {
    pub(crate) fn new(
        cpu: u32,
        attributes: Rc<perf_event_attr>) -> Self {
        let mut sample_time_offset = abi::Header::data_offset() as u16;

        if attributes.has_format(abi::PERF_SAMPLE_IDENTIFIER) {
            sample_time_offset += 8;
        }

        if attributes.has_format(abi::PERF_SAMPLE_IP) {
            sample_time_offset += 8;
        }

        if attributes.has_format(abi::PERF_SAMPLE_TID) {
            sample_time_offset += 8;
        }

        debug!(
            "CpuRingBuf created: cpu={}, sample_time_offset={}",
            cpu, sample_time_offset
        );

        Self {
            cpu,
            attributes,
            sample_time_offset,
            fd: None,
            id: None,
        }
    }

    pub(crate) fn ancillary(&self) -> AncillaryData {
        AncillaryData {
            cpu: self.cpu,
            attributes: self.attributes.clone(),
        }
    }

    /// Borrow the per-CPU perf fd for the lifetime of `self`.
    ///
    /// Returns `None` if the buffer has not been opened yet.
    #[cfg(target_os = "linux")]
    pub(crate) fn borrowed_fd(&self) -> Option<BorrowedFd<'_>> {
        /* SAFETY: `self.fd` is an open file descriptor for the lifetime
         * of this `CpuRingBuf`. The returned `BorrowedFd` is tied to
         * `&self` so the caller cannot outlive it. */
        self.fd.map(|fd| unsafe { BorrowedFd::borrow_raw(fd) })
    }

    fn read_id(&self) -> IOResult<u64> {
        match &self.fd {
            Some(fd) => {
                let mut id = read_format::default();

                unsafe {
                    let result = read(
                        *fd,
                        &mut id as *mut read_format as *mut c_void,
                        16);

                    if result == -1 {
                        let err = IOError::last_os_error();
                        warn!("read_id failed: cpu={}, fd={}, error={}", self.cpu, fd, err);
                        return Err(err);
                    }
                }

                trace!("read_id succeeded: cpu={}, id={}", self.cpu, id.id);
                Ok(id.id)
            },

            None => {
                warn!("read_id failed: ring buffer not open, cpu={}", self.cpu);
                Err(io_error("Ring buffer is not open."))
            }
        }
    }

    pub(crate) fn sample_time_offset(&self) -> u16 {
        self.sample_time_offset
    }

    pub(crate) fn id(&self) -> Option<u64> {
        self.id
    }

    pub(crate) fn is_open(&self) -> bool {
        self.fd.is_some()
    }

    pub(crate) fn open(
        &mut self,
        target_pid: Option<i32>) -> IOResult<()> {
        let pid = target_pid.unwrap_or(-1);

        let fd = perf_event_open(
            &self.attributes,
            pid,
            self.cpu as i32,
            -1,
            0)?;

        self.fd = Some(fd as i32);
        self.id = Some(self.read_id()?);

        info!(
            "CpuRingBuf opened: cpu={}, pid={}, fd={}, id={}",
            self.cpu, pid, fd, self.id.unwrap()
        );

        Ok(())
    }

    pub(crate) fn create_reader(
        &self,
        page_count: usize) -> IOResult<CpuRingReader> {
        if self.fd.is_none() {
            warn!("create_reader failed: ring buffer not open, cpu={}", self.cpu);
            return Err(io_error(
                "Ring buffer is not open."));
        }

        let data_bytes = ring_data_bytes(page_count);

        unsafe {
            let page_size = sysconf(_SC_PAGE_SIZE) as usize;
            let pages_len = data_bytes + page_size; /* + metadata page */

            let pages = mmap(
                std::ptr::null_mut::<u8>() as *mut c_void,
                pages_len,
                PROT_READ | PROT_WRITE,
                MAP_SHARED,
                self.fd.unwrap(),
                0);

            if pages == MAP_FAILED {
                let err = IOError::last_os_error();
                warn!(
                    "create_reader mmap failed: cpu={}, page_count={}, pages_len={}, error={}",
                    self.cpu, page_count, pages_len, err
                );
                return Err(err);
            }

            debug!(
                "CpuRingReader created: cpu={}, page_count={}, pages_len={}",
                self.cpu, page_count, pages_len
            );

            Ok(CpuRingReader::new(
                pages as *mut u8,
                pages_len))
        }
    }

    pub(crate) fn enable(
        &self) -> IOResult<()> {
        if self.fd.is_none() {
            warn!("enable failed: ring buffer not open, cpu={}", self.cpu);
            return Err(io_error(
                "Ring buffer is not open."));
        }

        unsafe {
            let result = ioctl(
                self.fd.unwrap(),
                PERF_EVENT_IOC_ENABLE as _);

            if result != 0 {
                let err = IOError::last_os_error();
                warn!("enable ioctl failed: cpu={}, fd={}, error={}", self.cpu, self.fd.unwrap(), err);
                return Err(err);
            }
        };

        debug!("CpuRingBuf enabled: cpu={}, id={:?}", self.cpu, self.id);

        Ok(())
    }

    pub(crate) fn disable(
        &self) -> IOResult<()> {
        if self.fd.is_none() {
            warn!("disable failed: ring buffer not open, cpu={}", self.cpu);
            return Err(io_error(
                "Ring buffer is not open."));
        }

        unsafe {
            let result = ioctl(
                self.fd.unwrap(),
                PERF_EVENT_IOC_DISABLE as _);

            if result != 0 {
                let err = IOError::last_os_error();
                warn!("disable ioctl failed: cpu={}, fd={}, error={}", self.cpu, self.fd.unwrap(), err);
                return Err(err);
            }
        };

        debug!("CpuRingBuf disabled: cpu={}, id={:?}", self.cpu, self.id);

        Ok(())
    }

    pub(crate) fn redirect_to(
        &self, 
        target: &Self) -> IOResult<()> {
        if self.fd.is_none() || target.fd.is_none() {
            warn!(
                "redirect_to failed: ring buffer or target not open, cpu={}, target_cpu={}",
                self.cpu, target.cpu
            );
            return Err(io_error(
                "Ring buffer or target is not open."));
        }

        unsafe {
            let result = ioctl(
                self.fd.unwrap(),
                PERF_EVENT_IOC_SET_OUTPUT as _,
                target.fd.unwrap());

            if result == -1 {
                let err = IOError::last_os_error();
                warn!(
                    "redirect_to ioctl failed: cpu={}, target_cpu={}, error={}",
                    self.cpu, target.cpu, err
                );
                return Err(err);
            }

            debug!(
                "CpuRingBuf redirected: cpu={}, id={:?}, target_cpu={}, target_id={:?}",
                self.cpu, self.id, target.cpu, target.id
            );

            Ok(())
        }
    }

    pub(crate) fn set_filter(&self, filter: &std::ffi::CStr) -> IOResult<()> {
        if self.fd.is_none() {
            warn!("set_filter failed: ring buffer not open, cpu={}", self.cpu);
            return Err(io_error("Ring buffer is not open."));
        }

        unsafe {
            let result = ioctl(
                self.fd.unwrap(),
                abi::PERF_EVENT_IOC_SET_FILTER as _,
                filter.as_ptr(),
            );

            if result == -1 {
                let err = IOError::last_os_error();
                warn!(
                    "set_filter ioctl failed: cpu={}, fd={}, error={}",
                    self.cpu,
                    self.fd.unwrap(),
                    err
                );
                return Err(err);
            }
        };

        debug!("CpuRingBuf filter set: cpu={}, id={:?}", self.cpu, self.id);

        Ok(())
    }
}

impl Drop for CpuRingBuf {
    fn drop(&mut self) {
        if let Some(fd) = self.fd {
            unsafe {
                close(fd);
            }
        }
    }
}

/// A ring buffer backed by user-space memory instead of kernel memory.
///
/// This struct uses the same memory layout as the kernel's perf ring buffer,
/// allowing it to be read by `CpuRingReader`. It is used during
/// `capture_environment` so that synthetic events (COMM, MMAP2) can be
/// written from a background thread while the main parse loop reads them
/// alongside real kernel events.
pub(crate) struct InProcessRingBuf {
    data: Vec<u8>,
    data_offset: usize,
    data_size: usize,
    data_mask: usize,
    head: usize,
}

/// Writer half of an `InProcessRingBuf`.
///
/// This is `Send` so it can be moved to another thread. The writer
/// appends perf-event records and updates the head pointer with the
/// appropriate memory barriers so a concurrent `CpuRingReader` sees the
/// data.
pub struct InProcessRingBufWriter {
    data: *mut u8,
    data_offset: usize,
    data_size: usize,
    data_mask: usize,
    head: usize,
    lost_count: u64,
}

// SAFETY: InProcessRingBufWriter only writes to memory via raw pointer.
// The ring buffer protocol (single writer, single reader, memory barriers)
// ensures safe concurrent access with a CpuRingReader on another thread.
unsafe impl Send for InProcessRingBufWriter {}

impl InProcessRingBuf {
    /// Create a new in-process ring buffer with the given number of data pages.
    /// The data_pages value will be rounded up to the next power of two.
    pub(crate) fn new(data_pages: usize) -> Self {
        let page_size = unsafe { sysconf(_SC_PAGE_SIZE) as usize };
        let data_pages = data_pages.next_power_of_two();
        let data_size = data_pages * page_size;
        let data_offset = page_size; /* first page is metadata */
        let total_size = data_offset + data_size;

        let mut data = vec![0u8; total_size];

        /* Set data_offset at byte 1040 */
        data[1040..1048].copy_from_slice(
            &(data_offset as u64).to_ne_bytes());

        /* Set data_size at byte 1048 */
        data[1048..1056].copy_from_slice(
            &(data_size as u64).to_ne_bytes());

        debug!(
            "InProcessRingBuf created: data_pages={}, data_size={:#x}, total_size={:#x}",
            data_pages, data_size, total_size
        );

        Self {
            data,
            data_offset,
            data_size,
            data_mask: data_size - 1,
            head: 0,
        }
    }

    /// Create a `CpuRingReader` that reads from this buffer's memory.
    /// The reader does NOT own the memory.
    pub(crate) fn create_reader(&mut self) -> CpuRingReader {
        CpuRingReader::new_unowned(
            self.data.as_mut_ptr(),
            self.data.len())
    }

    /// Split into a writer (movable to another thread) and a reader.
    /// The writer and reader share the underlying memory.
    ///
    /// # Safety
    /// The returned `InProcessRingBufWriter` holds a raw pointer into
    /// `self.data`. The caller must ensure that `self` outlives both the
    /// writer and the reader.
    pub(crate) fn writer(&mut self) -> InProcessRingBufWriter {
        InProcessRingBufWriter {
            data: self.data.as_mut_ptr(),
            data_offset: self.data_offset,
            data_size: self.data_size,
            data_mask: self.data_mask,
            head: self.head,
            lost_count: 0,
        }
    }
}

impl InProcessRingBufWriter {
    /// Maximum time to spin-wait for space to become available (100 ms).
    const WAIT_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(100);

    /// Size of a PERF_RECORD_LOST record with sample_id_all fields:
    /// 8-byte header + (id + lost) + (pid/tid + time + id).
    const LOST_RECORD_SIZE: usize = 48;

    /// Read the current tail value written by the reader.
    fn tail(&self) -> usize {
        unsafe {
            let tail_ptr = self.data.add(1032) as *const u64;
            std::ptr::read_volatile(tail_ptr) as usize
        }
    }

    /// Returns the available space in the data region.
    fn available(&self) -> usize {
        let tail = self.tail();
        self.data_size - (self.head - tail)
    }

    /// Spin-wait until at least `needed` bytes are available or timeout
    /// elapses.  Returns `true` if space is available.
    fn wait_for_space(&self, needed: usize) -> bool {
        let start = std::time::Instant::now();

        loop {
            if self.available() >= needed {
                return true;
            }

            if start.elapsed() >= Self::WAIT_TIMEOUT {
                return false;
            }

            std::thread::sleep(std::time::Duration::from_millis(10));
        }
    }

    /// Write raw bytes into the ring buffer without any lost-record logic.
    /// Caller must ensure aligned record size is <= `self.data_size` and that
    /// sufficient space is available.
    fn write_raw(&mut self, record: &[u8], aligned_len: usize) {
        let write_pos = self.head & self.data_mask;

        unsafe {
            let dest = self.data.add(self.data_offset);

            if write_pos + record.len() <= self.data_size {
                /* Fits without wrapping */
                std::ptr::copy_nonoverlapping(
                    record.as_ptr(),
                    dest.add(write_pos),
                    record.len());
            } else {
                /* Wraps around */
                let first_part = self.data_size - write_pos;
                std::ptr::copy_nonoverlapping(
                    record.as_ptr(),
                    dest.add(write_pos),
                    first_part);
                std::ptr::copy_nonoverlapping(
                    record.as_ptr().add(first_part),
                    dest,
                    record.len() - first_part);
            }

            let padding = aligned_len - record.len();
            if padding > 0 {
                let padding_start = (write_pos + record.len()) & self.data_mask;

                if padding_start + padding <= self.data_size {
                    std::ptr::write_bytes(dest.add(padding_start), 0, padding);
                } else {
                    let first_part = self.data_size - padding_start;
                    std::ptr::write_bytes(dest.add(padding_start), 0, first_part);
                    std::ptr::write_bytes(dest, 0, padding - first_part);
                }
            }

            self.head += aligned_len;

            /* Memory barrier then update head so the reader sees the data */
            mb();
            let head_ptr = self.data.add(1024) as *mut u64;
            std::ptr::write_volatile(head_ptr, self.head as u64);
        }

        trace!(
            "InProcessRingBufWriter::write_raw: record_len={}, head={:#x}",
            record.len(), self.head
        );
    }

    /// If there are pending lost events and enough space is available,
    /// write a PERF_RECORD_LOST record and reset the counter.
    fn flush_pending_lost(&mut self) {
        if self.lost_count == 0 {
            return;
        }

        if self.available() < Self::LOST_RECORD_SIZE {
            return;
        }

        let mut record = Vec::new();

        /* Payload: id + lost count + sample_id_all(pid/tid,time,id) */
        let mut payload = Vec::new();
        payload.extend_from_slice(&0u64.to_ne_bytes());
        payload.extend_from_slice(&self.lost_count.to_ne_bytes());
        payload.extend_from_slice(&0u64.to_ne_bytes()); /* pid/tid */
        payload.extend_from_slice(
            &perf_timestamp(&RingBufBuilder::common_attributes()).to_ne_bytes()
        );
        payload.extend_from_slice(&0u64.to_ne_bytes()); /* id */

        abi::Header::write(
            abi::PERF_RECORD_LOST,
            0,
            &payload,
            &mut record);

        warn!(
            "InProcessRingBufWriter: writing lost record, lost_count={}",
            self.lost_count
        );

        self.lost_count = 0;
        self.write_raw(&record, abi::align_to_perf_record(record.len()));
    }

    /// Flush any pending lost record. Call this at the end of a session
    /// to ensure lost events are reported even when no more writes follow.
    pub fn flush(&mut self) {
        self.flush_pending_lost();
    }

    /// Write a complete perf event record into the ring buffer.
    /// If not enough space is available after a brief spin-wait, the
    /// record is dropped, the lost counter is incremented, and a
    /// warning is logged.  On a successful write, any accumulated
    /// lost events are flushed first.
    pub fn write(&mut self, record: &[u8]) {
        let aligned_len = abi::align_to_perf_record(record.len());

        if aligned_len > self.data_size {
            warn!(
                "InProcessRingBufWriter::write: aligned_record_len={} exceeds data_size={}, dropping",
                aligned_len, self.data_size
            );
            self.lost_count += 1;
            return;
        }

        /* Account for a pending lost record that may need to be written
         * alongside the actual record. */
        let extra = if self.lost_count > 0 {
            Self::LOST_RECORD_SIZE
        } else {
            0
        };

        let needed = aligned_len + extra;

        if self.available() < needed {
            warn!(
                "InProcessRingBufWriter::write: insufficient space for record_len={}, waiting",
                record.len()
            );

            if !self.wait_for_space(needed) {
                warn!(
                    "InProcessRingBufWriter::write: timed out waiting for space, dropping record_len={}",
                    record.len()
                );
                self.lost_count += 1;
                return;
            }
        }

        self.flush_pending_lost();
        self.write_raw(record, aligned_len);
    }
}

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

    fn swap(
        source: &[u8],
        dest: &mut [u8]) {
        let mut i: usize = 0;

        for b in source {
            dest[i] = *b;
            i += 1;
        }
    }

    #[test]
    fn reader() {
        let mut temp = Vec::new();

        let mut data = Vec::new();
        data.resize(2 * 4096, 0);

        let slice = data.as_mut_slice();

        /* Data Offset: 4096 */
        swap(
            &4096u64.to_ne_bytes(),
            &mut slice[1040..1048]);

        /* Data Size: 4096 */
        swap(
            &4096u64.to_ne_bytes(),
            &mut slice[1048..1056]);

        /* Write a few entries */
        let mut entry = Vec::new();

        /* 1 */
        abi::Header::write(1024, 0, &1u64.to_ne_bytes(), &mut entry);

        /* 2 */
        abi::Header::write(1024, 0, &2u64.to_ne_bytes(), &mut entry);

        /* 3 */
        abi::Header::write(1024, 0, &3u64.to_ne_bytes(), &mut entry);

        /* Add entry to ring buffer */
        swap(
            entry.as_slice(),
            &mut slice[4096..]);

        /* Head position */
        swap(
            &(entry.len() as u64).to_ne_bytes(),
            &mut slice[1024..1032]);

        /* Tail position */
        swap(
            &0u64.to_ne_bytes(),
            &mut slice[1032..1040]);

        let mut reader = CpuRingReader::new(
            data.as_mut_ptr(),
            data.len());

        let mut cursor = CpuRingCursor::default();
        reader.begin_reading(&mut cursor);

        assert_eq!(true, cursor.more());

        /* 1 */
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(1024, header.entry_type);
        assert_eq!(16, header.size);
        assert_eq!(16, read.len());
        assert_eq!(1, u64::from_ne_bytes(read[8..16].try_into().unwrap()));

        /* 2 */
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(1024, header.entry_type);
        assert_eq!(16, header.size);
        assert_eq!(16, read.len());
        assert_eq!(2, u64::from_ne_bytes(read[8..16].try_into().unwrap()));

        /* 3 */
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(1024, header.entry_type);
        assert_eq!(16, header.size);
        assert_eq!(16, read.len());
        assert_eq!(3, u64::from_ne_bytes(read[8..16].try_into().unwrap()));

        /* Reading after end results in 0 sized slice */
        assert_eq!(false, cursor.more());
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        assert_eq!(0, read.len());

        reader.end_reading(&cursor);
        drop(reader);

        let slice = data.as_mut_slice();

        /* Add wrapping entry */
        entry.clear();

        /* 4 */
        abi::Header::write(1024, 0, &4u64.to_ne_bytes(), &mut entry);

        /* Add entry to ring buffer */
        swap(
            &entry.as_slice()[0..8],
            &mut slice[8184..8192]);

        swap(
            &entry.as_slice()[8..16],
            &mut slice[4096..4104]);

        /* Head position: 8200 */
        swap(
            &8200u64.to_ne_bytes(),
            &mut slice[1024..1032]);

        /* Tail position: 8184 */
        swap(
            &8184u64.to_ne_bytes(),
            &mut slice[1032..1040]);

        let mut reader = CpuRingReader::new(
            data.as_mut_ptr(),
            data.len());

        reader.begin_reading(&mut cursor);

        assert_eq!(true, cursor.more());

        /* 4 */
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(1024, header.entry_type);
        assert_eq!(16, header.size);
        assert_eq!(16, read.len());
        assert_eq!(4, u64::from_ne_bytes(read[8..16].try_into().unwrap()));

        /* Reading after end results in 0 sized slice */
        assert_eq!(false, cursor.more());
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        assert_eq!(0, read.len());

        reader.end_reading(&cursor);

        /* Ensure update stuck */
        reader.begin_reading(&mut cursor);
        assert_eq!(false, cursor.more());
        reader.end_reading(&cursor);
    }

    #[test]
    #[ignore]
    fn open_close() {
        println!("NOTE: Requires sudo/SYS_CAP_ADMIN/tracefs access.");

        let cpu = 0;
        let pid = Some(0);
        let mut rb_head = RingBufBuilder::for_kernel().build().for_cpu(cpu);

        rb_head.open(pid).unwrap();

        let kernel = RingBufBuilder::for_kernel()
            .with_executable_mmap_records();

        let page_count = 1;
        let _reader = rb_head.create_reader(page_count).unwrap();

        let mut rb = kernel.build().for_cpu(cpu);
        rb.open(pid).unwrap();
        rb.redirect_to(&rb_head).unwrap();
        rb.enable().unwrap();
    }

    #[test]
    fn in_process_ring_buf_write_read() {
        let mut temp = Vec::new();

        /* Create an in-process ring buffer with 1 data page */
        let mut ring_buf = InProcessRingBuf::new(1);

        /* Get writer and reader */
        let mut writer = ring_buf.writer();
        let mut reader = ring_buf.create_reader();

        /* Write three records using abi::Header format */
        let mut record = Vec::new();
        abi::Header::write(1024, 0, &1u64.to_ne_bytes(), &mut record);
        writer.write(&record);

        record.clear();
        abi::Header::write(1024, 0, &2u64.to_ne_bytes(), &mut record);
        writer.write(&record);

        record.clear();
        abi::Header::write(1024, 0, &3u64.to_ne_bytes(), &mut record);
        writer.write(&record);

        /* Read from the ring buffer */
        let mut cursor = CpuRingCursor::default();
        reader.begin_reading(&mut cursor);

        assert!(cursor.more());

        /* Read record 1 */
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(1024, header.entry_type);
        assert_eq!(16, header.size);
        assert_eq!(1, u64::from_ne_bytes(read[8..16].try_into().unwrap()));

        /* Read record 2 */
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(1024, header.entry_type);
        assert_eq!(16, header.size);
        assert_eq!(2, u64::from_ne_bytes(read[8..16].try_into().unwrap()));

        /* Read record 3 */
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(1024, header.entry_type);
        assert_eq!(16, header.size);
        assert_eq!(3, u64::from_ne_bytes(read[8..16].try_into().unwrap()));

        /* No more data */
        assert!(!cursor.more());

        reader.end_reading(&cursor);
    }

    #[test]
    fn in_process_ring_buf_writer_threaded() {
        let mut temp = Vec::new();
        let mut ring_buf = InProcessRingBuf::new(1);

        /* Get the reader before spawning the writer thread */
        let mut reader = ring_buf.create_reader();

        /* Get a writer and move it to another thread */
        let mut writer = ring_buf.writer();

        let handle = std::thread::spawn(move || {
            let mut record = Vec::new();
            for i in 1u64..=5 {
                record.clear();
                abi::Header::write(1024, 0, &i.to_ne_bytes(), &mut record);
                writer.write(&record);
            }
        });

        handle.join().unwrap();

        /* Reader should see all 5 records */
        let mut cursor = CpuRingCursor::default();
        reader.begin_reading(&mut cursor);

        assert!(cursor.more());

        for expected in 1u64..=5 {
            let read = reader.read(&mut cursor, &mut temp).unwrap();
            let header = abi::Header::from_slice(read).unwrap();
            assert_eq!(1024, header.entry_type);
            assert_eq!(16, header.size);
            assert_eq!(
                expected,
                u64::from_ne_bytes(read[8..16].try_into().unwrap()));
        }

        assert!(!cursor.more());
        reader.end_reading(&cursor);
    }

    #[test]
    fn in_process_ring_buf_write_wrap() {
        let mut temp = Vec::new();

        /* Create an in-process ring buffer with 1 data page */
        let mut ring_buf = InProcessRingBuf::new(1);
        let mut writer = ring_buf.writer();
        let mut reader = ring_buf.create_reader();

        /* Each record is 16 bytes (8-byte header + 8-byte payload).
         * Write 255 records to advance head to 4080 bytes, leaving
         * 16 bytes before the end of the 4096-byte data region. */
        let mut record = Vec::new();
        for i in 0u64..255 {
            record.clear();
            abi::Header::write(1024, 0, &i.to_ne_bytes(), &mut record);
            writer.write(&record);
        }

        /* Read and discard those records, advancing the tail to 4080
         * so the writer sees enough free space for the next write. */
        let mut cursor = CpuRingCursor::default();
        reader.begin_reading(&mut cursor);
        while cursor.more() {
            let _ = reader.read(&mut cursor, &mut temp).unwrap();
        }
        reader.end_reading(&cursor);

        /* Write a 24-byte record (8-byte header + 16-byte payload).
         * With head at 4080 and data_size = 4096:
         *   write_pos = 4080, write_pos + 24 = 4104 > 4096
         * so the record must wrap around the end of the buffer. */
        let payload = [0x55u8; 16]; /* arbitrary fill pattern */
        record.clear();
        abi::Header::write(1024, 0, &payload, &mut record);
        assert_eq!(24, record.len());
        writer.write(&record);

        /* Read the wrapped record and verify its contents */
        reader.begin_reading(&mut cursor);
        assert!(cursor.more());

        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(1024, header.entry_type);
        assert_eq!(24, header.size);
        assert_eq!(24, read.len());
        assert_eq!(payload, read[8..24]);

        assert!(!cursor.more());
        reader.end_reading(&cursor);
    }

    #[test]
    fn in_process_ring_buf_write_unaligned_record_aligns_head() {
        let mut temp = Vec::new();
        let mut ring_buf = InProcessRingBuf::new(1);
        let mut writer = ring_buf.writer();
        let mut reader = ring_buf.create_reader();
        let mut cursor = CpuRingCursor::default();

        let mut record = Vec::new();
        let payload = [0xABu8; 5];
        abi::Header::write(1024, 0, &payload, &mut record);
        assert_eq!(16, record.len());
        writer.write(&record);

        reader.begin_reading(&mut cursor);
        assert_eq!(16, cursor.end);
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(16, header.size);
        assert_eq!(payload, read[8..13]);
        assert_eq!([0u8; 3], read[13..16]);
        assert!(!cursor.more());
        reader.end_reading(&cursor);

        record.clear();
        abi::Header::write(1024, 0, &42u64.to_ne_bytes(), &mut record);
        writer.write(&record);

        reader.begin_reading(&mut cursor);
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(16, header.size);
        assert_eq!(42, u64::from_ne_bytes(read[8..16].try_into().unwrap()));
        assert!(!cursor.more());
        reader.end_reading(&cursor);
    }

    #[test]
    fn in_process_ring_buf_lost_record_on_next_write() {
        let mut temp = Vec::new();

        /* 1 data page = 4096 bytes */
        let mut ring_buf = InProcessRingBuf::new(1);
        let mut writer = ring_buf.writer();
        let mut reader = ring_buf.create_reader();

        /* Fill the buffer completely: 256 × 16-byte records = 4096 bytes.
         * After this, available() == 0 so the next write will be dropped. */
        let mut record = Vec::new();
        for i in 0u64..256 {
            record.clear();
            abi::Header::write(1024, 0, &i.to_ne_bytes(), &mut record);
            writer.write(&record);
        }

        /* Attempt a write that will fail due to no space - this should
         * increment lost_count rather than silently dropping. */
        record.clear();
        abi::Header::write(1024, 0, &0xDEADu64.to_ne_bytes(), &mut record);
        writer.write(&record);

        /* Drain all data so the reader advances the tail, freeing space. */
        let mut cursor = CpuRingCursor::default();
        reader.begin_reading(&mut cursor);
        while cursor.more() {
            let _ = reader.read(&mut cursor, &mut temp).unwrap();
        }
        reader.end_reading(&cursor);

        /* Next successful write should first emit a PERF_RECORD_LOST,
         * then the actual record. */
        record.clear();
        abi::Header::write(1024, 0, &42u64.to_ne_bytes(), &mut record);
        writer.write(&record);

        reader.begin_reading(&mut cursor);
        assert!(cursor.more());

        /* First record should be PERF_RECORD_LOST */
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(abi::PERF_RECORD_LOST, header.entry_type);
        assert_eq!(48, header.size);
        let lost_id = u64::from_ne_bytes(read[8..16].try_into().unwrap());
        let lost_count = u64::from_ne_bytes(read[16..24].try_into().unwrap());
        let sample_id = u64::from_ne_bytes(read[40..48].try_into().unwrap());
        assert_eq!(0, lost_id);
        assert_eq!(1, lost_count);
        assert_eq!(0, sample_id);

        /* Second record should be the actual data */
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(1024, header.entry_type);
        assert_eq!(42, u64::from_ne_bytes(read[8..16].try_into().unwrap()));

        assert!(!cursor.more());
        reader.end_reading(&cursor);
    }

    #[test]
    fn in_process_ring_buf_lost_record_accumulates() {
        let mut temp = Vec::new();

        let mut ring_buf = InProcessRingBuf::new(1);
        let mut writer = ring_buf.writer();
        let mut reader = ring_buf.create_reader();

        /* Fill the buffer completely */
        let mut record = Vec::new();
        for i in 0u64..256 {
            record.clear();
            abi::Header::write(1024, 0, &i.to_ne_bytes(), &mut record);
            writer.write(&record);
        }

        /* Attempt 3 writes that will each fail */
        for _ in 0..3 {
            record.clear();
            abi::Header::write(1024, 0, &0u64.to_ne_bytes(), &mut record);
            writer.write(&record);
        }

        /* Drain */
        let mut cursor = CpuRingCursor::default();
        reader.begin_reading(&mut cursor);
        while cursor.more() {
            let _ = reader.read(&mut cursor, &mut temp).unwrap();
        }
        reader.end_reading(&cursor);

        /* Next write should emit a single PERF_RECORD_LOST with count=3 */
        record.clear();
        abi::Header::write(1024, 0, &99u64.to_ne_bytes(), &mut record);
        writer.write(&record);

        reader.begin_reading(&mut cursor);
        assert!(cursor.more());

        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(abi::PERF_RECORD_LOST, header.entry_type);
        assert_eq!(48, header.size);
        let lost_count = u64::from_ne_bytes(read[16..24].try_into().unwrap());
        let sample_id = u64::from_ne_bytes(read[40..48].try_into().unwrap());
        assert_eq!(3, lost_count);
        assert_eq!(0, sample_id);

        /* Followed by the actual data record */
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(1024, header.entry_type);
        assert_eq!(99, u64::from_ne_bytes(read[8..16].try_into().unwrap()));

        assert!(!cursor.more());
        reader.end_reading(&cursor);
    }

    #[test]
    fn in_process_ring_buf_flush_writes_lost_record() {
        let mut temp = Vec::new();

        let mut ring_buf = InProcessRingBuf::new(1);
        let mut writer = ring_buf.writer();
        let mut reader = ring_buf.create_reader();

        /* Fill the buffer */
        let mut record = Vec::new();
        for i in 0u64..256 {
            record.clear();
            abi::Header::write(1024, 0, &i.to_ne_bytes(), &mut record);
            writer.write(&record);
        }

        /* Fail 2 writes */
        for _ in 0..2 {
            record.clear();
            abi::Header::write(1024, 0, &0u64.to_ne_bytes(), &mut record);
            writer.write(&record);
        }

        /* Drain */
        let mut cursor = CpuRingCursor::default();
        reader.begin_reading(&mut cursor);
        while cursor.more() {
            let _ = reader.read(&mut cursor, &mut temp).unwrap();
        }
        reader.end_reading(&cursor);

        /* flush() at end of session should emit the lost record */
        writer.flush();

        reader.begin_reading(&mut cursor);
        assert!(cursor.more());

        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(abi::PERF_RECORD_LOST, header.entry_type);
        assert_eq!(48, header.size);
        let lost_count = u64::from_ne_bytes(read[16..24].try_into().unwrap());
        let sample_id = u64::from_ne_bytes(read[40..48].try_into().unwrap());
        assert_eq!(2, lost_count);
        assert_eq!(0, sample_id);

        assert!(!cursor.more());
        reader.end_reading(&cursor);
    }

    #[test]
    fn in_process_ring_buf_no_lost_record_when_zero() {
        let mut temp = Vec::new();

        let mut ring_buf = InProcessRingBuf::new(1);
        let mut writer = ring_buf.writer();
        let mut reader = ring_buf.create_reader();

        /* Write a normal record */
        let mut record = Vec::new();
        abi::Header::write(1024, 0, &1u64.to_ne_bytes(), &mut record);
        writer.write(&record);

        /* flush() with no lost records should be a no-op */
        writer.flush();

        let mut cursor = CpuRingCursor::default();
        reader.begin_reading(&mut cursor);
        assert!(cursor.more());

        /* Only the normal record, no lost record */
        let read = reader.read(&mut cursor, &mut temp).unwrap();
        let header = abi::Header::from_slice(read).unwrap();
        assert_eq!(1024, header.entry_type);
        assert_eq!(1, u64::from_ne_bytes(read[8..16].try_into().unwrap()));

        assert!(!cursor.more());
        reader.end_reading(&cursor);
    }
}