frozen-core 0.0.20

//! A buffer pool implementation with dynamic memory allocations
//!
//! ## Memory Allocation
//!
//! Allocations are directly allocated using the global memory allocator rather then maintaining a pre-allocated
//! memory pool.
//!
//! Every allocation reserves a memory budget and only after is allowed to allocate memory, otherwise faced with
//! backpressure to limit the total size of memory used. So the memory usage stays bounded through memory budgeting
//! rather then a fixed-capacity memory pool.
//!
//! ## Benchmarks
//!
//! Allocation latency for `N` buffers,
//!
//! ```md
//! | Buffers  | Latency  |
//! |:---------|:---------|
//! | 1        | 246 ns   |
//! | 0x10     | 251 ns   |
//! | 0x400    | 300 ns   |
//! ```
//!
//! Allocation throughput,
//!
//! ```md
//! | Metric              | Value         |
//! |:--------------------|:--------------|
//! | Allocations / sec   | ~3.94 Million |
//! | Avg latency / alloc | ~254 ns       |
//! ```
//!
//! *NOTE:* Measurements include allocation and deallocation.
//!
//! Environment used for benching,
//!
//! - OS: NixOS (WSL2)
//! - Architecture: x86_64
//! - Memory: 8 GiB RAM (DDR4)
//! - Rust: rustc 1.86.0 w/ cargo 1.86.0
//! - Kernel: Linux 6.6.87.2-microsoft-standard-WSL2
//! - CPU: Intel® Core™ i5-10300H @ 2.50GHz (4C / 8T)
//!
//! ## Example
//!
//! ```
//! use frozen_core::{
//!     bufpool::{BufPool, BufPoolCfg, BufferPointer},
//!     utils::BufferSize,
//! };
//!
//! let pool = BufPool::new(BufPoolCfg {
//!     module_id: 0,
//!     buffer_size: BufferSize::S16,
//!     max_memory: 0x0A * 0x100,
//! });
//!
//! let alloc = pool.allocate(0x2A);
//!
//! assert_eq!(alloc.length(), 0x2A);
//! assert!(!alloc.first().is_null());
//! assert_eq!(alloc.allocated_bytes(), BufferSize::S16.bytes() * 0x2A);
//!
//! let ptrs: Vec<BufferPointer> = alloc.iter().collect();
//! assert_eq!(ptrs.len(), 0x2A);
//! ```

use crate::utils::BufferSize;
use std::{alloc, ptr, sync, sync::atomic};

/// Raw pointer to a single buffer in memory allocated using [`BufPool::allocate`]
///
/// ## Safety
///
/// As the pointer is untyped and uninitialized, caller are responsible for:
///
/// - Writes stay within the buffer boundry
/// - Reads only occur after initilization (write)
///
/// *NOTE:* The pointer must never outlive the lifetime of [`BufPoolAllocation`] object.
///
/// ## Example
///
/// ```
/// use frozen_core::{
///     bufpool::{BufPool, BufPoolCfg},
///     utils::BufferSize,
/// };
///
/// let pool = BufPool::new(BufPoolCfg {
///     module_id: 0,
///     buffer_size: BufferSize::S16,
///     max_memory: 0x0A * 0x10,
/// });
///
/// let alloc = pool.allocate(1);
/// unsafe {
///     alloc.first().write(0xAA);
/// }
/// ```
pub type BufferPointer = *mut u8;

/// All the available configrations for [`BufPool`]
///
/// ## Example
///
/// ```
/// use frozen_core::{bufpool::BufPoolCfg, utils::BufferSize};
///
/// let cfg = BufPoolCfg {
///     module_id: 0,
///     max_memory: 0x10000,
///     buffer_size: BufferSize::S64,
/// };
///
/// assert_ne!(cfg.max_memory, 0);
/// assert!(cfg.max_memory > cfg.buffer_size.bytes());
/// ```
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct BufPoolCfg {
    /// Identifier used for error propagation by [`frozen_core::error::FrozenErr`]
    pub module_id: u8,

    /// Size (in bytes) of a single buffer unit returned via [`BufPoolAllocation`] upon successful allocation
    pub buffer_size: BufferSize,

    /// Maximum allowed memory (in bytes) to be allocated by [`BufPool`]
    ///
    /// *IMPORTANT:* When trying to allocate more memory then [`BufPoolCfg::max_memory`] via [`BufPool::allocate`],
    /// a deadlock will take place due to memory budgeting in place. Caller must make sure the value is high enough
    /// to avoid this scenerio.
    ///
    /// After the limit is exhausted, incoming allocation request are polled using backpressure to limit the size
    /// of memory being used. The polling ends when existing allocations are dropped from memory decrementing current
    /// memory usage.
    ///
    /// Backpressure can be handy while running in resource constrained environments to help limit the amount of
    /// resources being used by the system.
    ///
    /// *NOTE:* To avoid backpressure, set the `max_memory` to an arbitrary large value. This would not have any direct
    /// impact on performance or resource usage, just that it'll try not to constrain the caller.
    pub max_memory: usize,
}

/// A buffer pool implementation with dynamic memory allocations
///
/// ## Memory Budgeting
///
/// The total amout of memory simultenously allocated by the pool is limited by [`BufPoolCfg::max_memory`]. The caller
/// is blocked until previous allocations are dropped and previously allocated memory is deallocated.
///
/// ## Graceful Shutdown
///
/// Dropping the [`BufPool`] blocks until all the allocated [`BufPoolAllocation`] and there reference are dropped from
/// memory. This is done to prevent memory leaks, as well as to enable transfer of [`BufPoolAllocation`] across
/// threads.
///
/// ## Example
///
/// ```
/// use frozen_core::{
///     bufpool::{BufPool, BufPoolCfg, BufferPointer},
///     utils::BufferSize,
/// };
///
/// let pool = BufPool::new(BufPoolCfg {
///     module_id: 0,
///     buffer_size: BufferSize::S16,
///     max_memory: 0x0A * 0x100,
/// });
///
/// let alloc = pool.allocate(0x2A);
///
/// assert_eq!(alloc.length(), 0x2A);
/// assert!(!alloc.first().is_null());
/// assert_eq!(alloc.allocated_bytes(), BufferSize::S16.bytes() * 0x2A);
///
/// let ptrs: Vec<BufferPointer> = alloc.iter().collect();
/// assert_eq!(ptrs.len(), 0x2A);
/// ```
#[derive(Debug)]
pub struct BufPool {
    active_allocations: atomic::AtomicUsize,
    allocation_cv: sync::Condvar,
    allocation_lock: sync::Mutex<()>,
    allocated_memory: atomic::AtomicUsize,
    cfg: BufPoolCfg,
    shutdown_cv: sync::Condvar,
    shutdown_lock: sync::Mutex<()>,
}

unsafe impl Send for BufPool {}
unsafe impl Sync for BufPool {}

impl BufPool {
    /// Create a new instance of [`BufPool`]
    ///
    /// *NOTE:* There are no pre-allocated buffers, and all the allocations are allocated using gloabl allocator
    /// on-the-fly
    ///
    /// ## Debug Assertions
    ///
    /// In debug builds, this function uses `debug_assertion` to prevent invalid configurations. Caller must refer
    /// to [`BufPoolCfg`] for detailed info on the config params.
    ///
    /// ## Example
    ///
    /// ```
    /// use frozen_core::{
    ///     bufpool::{BufPool, BufPoolCfg},
    ///     utils::BufferSize,
    /// };
    ///
    /// let pool = BufPool::new(BufPoolCfg {
    ///     module_id: 0,
    ///     buffer_size: BufferSize::S16,
    ///     max_memory: 0x10 * 0x0A,
    /// });
    ///
    /// let alloc = pool.allocate(1);
    /// assert_eq!(alloc.length(), 1);
    /// ```
    #[inline]
    pub fn new(cfg: BufPoolCfg) -> Self {
        // sanity check
        debug_assert!(
            cfg.buffer_size.bytes() < cfg.max_memory,
            "MAX_MEMORY should always be larger then the BUFFER_SIZE"
        );

        Self {
            cfg,
            active_allocations: atomic::AtomicUsize::new(0),
            allocated_memory: atomic::AtomicUsize::new(0),
            allocation_cv: sync::Condvar::new(),
            allocation_lock: sync::Mutex::new(()),
            shutdown_cv: sync::Condvar::new(),
            shutdown_lock: sync::Mutex::new(()),
        }
    }

    /// Allocate `required` number of buffers each of [`BufPoolCfg::buffer_size`] size
    ///
    /// ## Backpressure
    ///
    /// If allocating requested memory (i.e. `required * BufPoolCfg::buffer_size`) would exceed the configured memory
    /// budget, i.e. [`BufPoolCfg::max_memory`], the caller is blocked until previous allocations are dropped and
    /// previously allocated memory is deallocated.
    ///
    /// ## Memory Layout
    ///
    /// All the buffers in the [`BufPoolAlocation`] object, are stored contiguously in memory as a single block of
    /// memory,
    ///
    /// ```text
    /// let allocated_buffer = vec![u8; required * BUFFER_SIZE];
    /// ```
    ///
    /// ## Memory Allocation
    ///
    /// Allocations are directly allocated using the global memory allocator rather then maintaining a pre-allocated
    /// memory pool.
    ///
    /// Every allocation reserves a memory budget and only after is allowed to allocate memory, otherwise faced with
    /// backpressure to limit the total size of memory used. So the memory usage stays bounded through memory budgeting
    /// rather then a fixed-capacity memory pool.
    ///
    /// ## RAII Safety
    ///
    /// The allocated object, i.e. [`BufPoolAllocation`], is RAII safe, as the the underlying memory is deallocated as
    /// soon as the last referece to the object is dropped, while also relaxing the memory budget to eliminate the
    /// backpressure (if any).
    ///
    /// ## Important Considerations
    ///
    /// The number of buffers required should never exceed `u16::MAX`. This is an abstract soft limit and should be
    /// enforced via public interface to avoid any weird exhaustion issues or bugs across the storage system.
    ///
    /// As `u16::MAX` is large enough value to almost never cause any problems for a single write operation, this soft
    /// limit acts as a guidline to safely operate arithmatic operations across storage engine's, including but not
    /// limited to [`frozen_core`].
    ///
    /// ## Example
    ///
    /// ```
    /// use frozen_core::{
    ///     bufpool::{BufPool, BufPoolCfg, BufferPointer},
    ///     utils::BufferSize,
    /// };
    ///
    /// let pool = BufPool::new(BufPoolCfg {
    ///     module_id: 0,
    ///     buffer_size: BufferSize::S16,
    ///     max_memory: 0x0A * 0x14,
    /// });
    ///
    /// let alloc = pool.allocate(0x0A);
    ///
    /// assert_eq!(alloc.length(), 0x0A);
    /// assert!(!alloc.first().is_null());
    /// assert_eq!(alloc.allocated_bytes(), BufferSize::S16.bytes() * 0x0A);
    ///
    /// let ptrs: Vec<BufferPointer> = alloc.iter().collect();
    /// assert_eq!(ptrs.len(), 0x0A);
    /// ```
    #[inline(always)]
    pub fn allocate(&self, required: usize) -> BufPoolAllocation {
        // sanity checks
        debug_assert!(required > 0, "required buffers must never be 0");
        debug_assert!(
            required * self.cfg.buffer_size.bytes() <= self.cfg.max_memory,
            "Total required bytes must be smaller then the MAX_MEMORY allowed to avoid deadlock"
        );
        debug_assert!(
            required * self.cfg.buffer_size.bytes() <= self.cfg.max_memory,
            "Total required bytes must never exceed `u16::MAX` to avoid arithmatic overflows"
        );

        let required_bytes = self.cfg.buffer_size.bytes() * required;
        loop {
            let current_bytes = self.allocated_memory.load(atomic::Ordering::Acquire);
            if current_bytes + required_bytes > self.cfg.max_memory {
                self.backpressure(required_bytes);
                continue;
            }

            match self.allocated_memory.compare_exchange(
                current_bytes,
                current_bytes + required_bytes,
                atomic::Ordering::AcqRel,
                atomic::Ordering::Acquire,
            ) {
                Ok(_) => break,
                Err(_) => continue,
            }
        }

        let layout = create_layout(required_bytes);
        let pointer = allocate_layout(layout);
        self.active_allocations.fetch_add(1, atomic::Ordering::Relaxed);

        BufPoolAllocation {
            layout,
            pointer,
            required_bytes,
            buffers: required,
            pool: ptr::NonNull::from(self),
        }
    }

    /// Applies backpressure until enough memory budget is available for the allocation
    ///
    /// ## Why we ignore [`std::sync::PoisonError`]?
    ///
    /// The mutex used for lock, is solely used as a parking primitive for [`Condvar`] and does not protect any mutable
    /// state. All the pool invariants and accounting are maintained via atomics and are completely seperated from
    /// the mutex.
    ///
    /// A poisoned mutex only indicates that another tx panicked while holding the lock, and indicates an inconsistent
    /// state of the protected value. Since no state can be left partially modified under this lock, there is no
    /// possible consistency risk to recover from and propagating the poison error would only introduce unnecessary
    /// failures into the allocation path.
    ///
    /// Therefore, as best effort, we consume the [`std::sync::PoisonError`] and continue operating with the recovered
    /// guard.
    #[inline]
    fn backpressure(&self, required_bytes: usize) {
        let mut guard = self.allocation_lock.lock().unwrap_or_else(|e| e.into_inner());
        while self.allocated_memory.load(atomic::Ordering::Acquire) + required_bytes > self.cfg.max_memory {
            guard = self.allocation_cv.wait(guard).unwrap_or_else(|e| e.into_inner());
        }
    }
}

impl Drop for BufPool {
    /// *NOTE:* See [`BufPool::backpressure`] for rationale behind poison recovery
    fn drop(&mut self) {
        let mut guard = self.shutdown_lock.lock().unwrap_or_else(|e| e.into_inner());
        while self.active_allocations.load(atomic::Ordering::Acquire) != 0 {
            guard = self.shutdown_cv.wait(guard).unwrap_or_else(|e| e.into_inner());
        }
    }
}

/// A RAII safe allocation object returned by [`BufPool::allocate`]
///
/// ## Lifetime
///
/// The object may/can outlive the scope that created it, while also being able to transfer across threads.
///
/// Internally, the [`BufPool`] tracks all the active allocations and delays the shutdown/drop until every allocation
/// and all there references are dropped from memory.
///
/// ## RAII Safety
///
/// As soon as the object and all its references are dropped, the underlying memory allocation is deallocated to prevent
/// any memory leaks. This also pins the allocation as long as the allocation object stays in the memory.
///
/// **NOTE:** The memory allocation must never outlive the scope that created them, i.e. [`BufPoolAllocation`]
///
/// ## Example
///
/// ```
/// use frozen_core::{
///     bufpool::{BufPool, BufPoolCfg, BufferPointer},
///     utils::BufferSize,
/// };
///
/// let pool = BufPool::new(BufPoolCfg {
///     module_id: 0,
///     buffer_size: BufferSize::S16,
///     max_memory: 0x0A * 0x14,
/// });
///
/// let alloc = pool.allocate(0x0A);
///
/// assert_eq!(alloc.length(), 0x0A);
/// assert!(!alloc.first().is_null());
/// assert_eq!(alloc.allocated_bytes(), BufferSize::S16.bytes() * 0x0A);
///
/// let ptrs: Vec<BufferPointer> = alloc.iter().collect();
/// assert_eq!(ptrs.len(), 0x0A);
/// ```
#[derive(Debug)]
pub struct BufPoolAllocation {
    buffers: usize,
    layout: alloc::Layout,
    pointer: ptr::NonNull<u8>,
    pool: ptr::NonNull<BufPool>,
    required_bytes: usize,
}

unsafe impl Send for BufPoolAllocation {}

impl BufPoolAllocation {
    /// Returns a [`BufferPointer`] to the first buffer in the allocation
    ///
    /// *NOTE:* Returned [`BufferPointer`] can also be used as a _base_pointer_ for the entire allocated memory slice.
    ///
    /// ## Example
    ///
    /// ```
    /// use frozen_core::{
    ///     bufpool::{BufPool, BufPoolCfg},
    ///     utils::BufferSize,
    /// };
    ///
    /// let pool = BufPool::new(BufPoolCfg {
    ///     module_id: 0,
    ///     buffer_size: BufferSize::S16,
    ///     max_memory: 0x10 * 0x0A,
    /// });
    ///
    /// let alloc = pool.allocate(0x0A);
    /// assert!(!alloc.first().is_null());
    /// ```
    #[inline]
    pub const fn first(&self) -> BufferPointer {
        self.pointer.as_ptr()
    }

    /// Returns the number of buffers allocated
    ///
    /// *NOTE:* Returned value is always equal to the `required` value used while [`BufPool::allocate`].
    ///
    /// ## Example
    ///
    /// ```
    /// use frozen_core::{
    ///     bufpool::{BufPool, BufPoolCfg},
    ///     utils::BufferSize,
    /// };
    ///
    /// let pool = BufPool::new(BufPoolCfg {
    ///     module_id: 0,
    ///     buffer_size: BufferSize::S16,
    ///     max_memory: 0x10 * 0x0A,
    /// });
    ///
    /// let alloc = pool.allocate(0x0A);
    /// assert_eq!(alloc.length(), 0x0A);
    /// ```
    #[inline]
    pub const fn length(&self) -> usize {
        self.buffers
    }

    /// Returns the total number of bytes of memory allocated
    ///
    /// ## Example
    ///
    /// ```
    /// use frozen_core::{
    ///     bufpool::{BufPool, BufPoolCfg},
    ///     utils::BufferSize,
    /// };
    ///
    /// let pool = BufPool::new(BufPoolCfg {
    ///     module_id: 0,
    ///     buffer_size: BufferSize::S16,
    ///     max_memory: 0x10 * 0x0A,
    /// });
    ///
    /// let alloc = pool.allocate(0x0A);
    /// assert_eq!(alloc.allocated_bytes(), BufferSize::S16.bytes() * 0x0A);
    /// ```
    #[inline]
    pub const fn allocated_bytes(&self) -> usize {
        self.required_bytes
    }

    /// A custom [`Iterator`] implementation to enable iteration over every allocated buffer in [`BufPoolAllocation`]
    ///
    /// *NOTE:* Each yielded pointer references a distinct buffer separated by [`BufPoolCfg::buffer_size`] bytes.
    ///
    /// ## Example
    ///
    /// ```
    /// use frozen_core::{
    ///     bufpool::{BufPool, BufPoolCfg, BufferPointer},
    ///     utils::BufferSize,
    /// };
    ///
    /// let pool = BufPool::new(BufPoolCfg {
    ///     module_id: 0,
    ///     buffer_size: BufferSize::S16,
    ///     max_memory: 0x10 * 0x14,
    /// });
    ///
    /// let alloc = pool.allocate(0x0A);
    /// let ptrs: Vec<BufferPointer> = alloc.iter().collect();
    /// assert_eq!(ptrs.len(), 0x0A);
    /// ```
    #[inline]
    pub fn iter(&self) -> BufPoolAllocationIter {
        let pool = unsafe { self.pool.as_ref() };
        BufPoolAllocationIter {
            pointer: self.pointer,
            buffer_size: pool.cfg.buffer_size.bytes(),
            remaining: self.buffers,
        }
    }
}

impl Drop for BufPoolAllocation {
    fn drop(&mut self) {
        let pool = unsafe { self.pool.as_ref() };
        deallocate_memory(self.pointer, self.layout);

        pool.allocated_memory
            .fetch_sub(self.required_bytes, atomic::Ordering::Release);
        pool.allocation_cv.notify_one();

        if pool.active_allocations.fetch_sub(1, atomic::Ordering::Release) == 1 {
            pool.shutdown_cv.notify_one();
        }
    }
}

/// Iterator over all buffers belonging to a [`BufPoolAllocation`]
///
/// Buffers are yielded in allocation order and are backed by a single contiguous memory region.
///
/// ## Example
///
/// ```rs
/// use frozen_core::{
///     bufpool::{BufPool, BufPoolCfg, BufferPointer},
///     utils::BufferSize,
/// };
///
/// let pool = BufPool::new(BufPoolCfg {
///     module_id: 0,
///     buffer_size: BufferSize::S16,
///     max_memory: 0x10 * 0x14,
/// });
///
/// let alloc = pool.allocate(0x0A);
/// let ptrs: Vec<BufferPointer> = alloc.iter().collect();
/// assert_eq!(ptrs.len(), 0x0A);
/// ```
#[derive(Debug)]
pub struct BufPoolAllocationIter {
    pointer: ptr::NonNull<u8>,
    buffer_size: usize,
    remaining: usize,
}

impl Iterator for BufPoolAllocationIter {
    type Item = BufferPointer;

    #[inline(always)]
    fn next(&mut self) -> Option<Self::Item> {
        if self.remaining == 0 {
            return None;
        }

        let curr_ptr = self.pointer;

        self.pointer = unsafe { self.pointer.add(self.buffer_size) };
        self.remaining -= 1;

        Some(curr_ptr.as_ptr())
    }
}

/// Creates a array layout with given `capacity`
///
/// *NOTE:* Use of `unwrap` is totally safe as the panic, if any, would be caught by unit tests and would be the
/// indication of incorrect impl and not any runtime failures
#[inline]
fn create_layout(required_bytes: usize) -> alloc::Layout {
    match alloc::Layout::array::<u8>(required_bytes) {
        Ok(layout) => layout,
        Err(e) => panic!("Invalid Layout: {e}"),
    }
}

/// Allocate a memory slice with given allocation `layout`
///
/// ## Allocation Failure
///
/// If the allocator is unable to satisfy the request (typically due to an OOM condition), [`alloc::alloc`] returns
/// a null pointer. In such cases we delegate to [`alloc::handle_alloc_error`], matching the behavior of std library
/// types such as [`Vec`], [`Box`] and [`String`].
///
/// This path aborts the process and never returns. Allocation failures are therefore treated as fatal runtime conditions
/// rather than recoverable errors.
///
/// Under normal operation this path should never be reached, as memory usage is expected to be bounded by the buffer
/// pools memory budget and backpressure mechanisms.
///
/// ## Why not return `FrozenErr`?
///
/// A null return from [`alloc::alloc`] indicates that the global allocator itself was unable to satisfy the request.
///
/// Delegating to [`alloc::handle_alloc_error`] matches the behavior of standard library containers and avoids continuing
/// execution after a catastrophic allocator failure, where further allocations required for error handling, logging
/// or recovery may also fail.
#[inline]
fn allocate_layout(layout: alloc::Layout) -> ptr::NonNull<u8> {
    let pointer = unsafe { alloc::alloc(layout) };
    match ptr::NonNull::new(pointer) {
        Some(p) => p,
        None => alloc::handle_alloc_error(layout),
    }
}

/// Deallocate the manually allocated slice of memory with help of given `pointer` and memory `layout`
#[inline]
fn deallocate_memory(pointer: ptr::NonNull<u8>, layout: alloc::Layout) {
    unsafe { alloc::dealloc(pointer.as_ptr(), layout) };
}

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

    const MOD_ID: u8 = 0;
    const BUF_SIZE: BufferSize = BufferSize::S32;

    #[inline]
    fn create_bufpool(max_mem: usize) -> BufPool {
        BufPool::new(BufPoolCfg {
            buffer_size: BUF_SIZE,
            max_memory: max_mem,
            module_id: MOD_ID,
        })
    }

    #[test]
    #[should_panic]
    #[cfg(debug_assertions)]
    fn err_new_with_invalid_cfg() {
        create_bufpool(BUF_SIZE.bytes() >> 1);
    }

    #[test]
    #[should_panic]
    #[cfg(debug_assertions)]
    fn err_alloc_zero() {
        let bpool = create_bufpool(BUF_SIZE.bytes());
        let _ = bpool.allocate(0);
    }

    #[test]
    #[should_panic]
    #[cfg(debug_assertions)]
    fn err_alloc_more_then_max_memory() {
        let bpool = create_bufpool(BUF_SIZE.bytes());
        let _ = bpool.allocate(2);
    }

    #[test]
    fn ok_alloc_single() {
        let bpool = create_bufpool(BUF_SIZE.bytes() * 2);
        let alloc = bpool.allocate(1);

        assert_eq!(alloc.buffers, 1);
        assert_eq!(alloc.required_bytes, BUF_SIZE.bytes());
    }

    #[test]
    fn ok_alloc_multiple() {
        let bpool = create_bufpool(BUF_SIZE.bytes() * 0x14);
        let alloc = bpool.allocate(0x10);

        assert_eq!(alloc.buffers, 0x10);
        assert_eq!(alloc.required_bytes, BUF_SIZE.bytes() * 0x10);
    }

    #[test]
    fn ok_alloc_max_memory() {
        let bpool = create_bufpool(BUF_SIZE.bytes() * 0x0A);
        let alloc = bpool.allocate(0x0A);

        assert_eq!(alloc.buffers, 0x0A);
        assert_eq!(alloc.required_bytes, BUF_SIZE.bytes() * 0x0A);
    }

    #[test]
    fn ok_alloc_updates_memory_accounting() {
        let bpool = create_bufpool(BUF_SIZE.bytes() * 0x14);
        let alloc = bpool.allocate(0x10);

        assert_eq!(
            bpool.allocated_memory.load(atomic::Ordering::Acquire),
            BUF_SIZE.bytes() * 0x10
        );
        drop(alloc);
        assert_eq!(bpool.allocated_memory.load(atomic::Ordering::Acquire), 0);
    }

    #[test]
    fn ok_alloc_updates_active_allocation_tracking() {
        let bpool = create_bufpool(BUF_SIZE.bytes() * 0x2A);

        let alloc1 = bpool.allocate(0x10);
        let alloc2 = bpool.allocate(0x10);

        assert_eq!(bpool.active_allocations.load(atomic::Ordering::Acquire), 2);
        let _ = (drop(alloc1), drop(alloc2));
        assert_eq!(bpool.active_allocations.load(atomic::Ordering::Acquire), 0);
    }

    #[test]
    fn ok_alloc_decrments_allocated_memory_after_deallocations() {
        let bpool = create_bufpool(BUF_SIZE.bytes() * 0x80);
        let allocations: Vec<_> = (0..0x20).map(|_| bpool.allocate(2)).collect();

        assert_eq!(bpool.allocated_memory.load(atomic::Ordering::Acquire), 0x20 * 0x40);
        drop(allocations);
        assert_eq!(bpool.allocated_memory.load(atomic::Ordering::Acquire), 0);
    }

    #[test]
    fn ok_backpressure_blocks_till_memory_is_deallocated() {
        let bpool = sync::Arc::new(create_bufpool(BUF_SIZE.bytes() * 2));
        let alloc = bpool.allocate(1);

        let pool2 = bpool.clone();
        let barrier = sync::Arc::new(sync::Barrier::new(2));
        let barrier2 = barrier.clone();

        let handle = std::thread::spawn(move || {
            barrier2.wait();

            let start = std::time::Instant::now();
            let _alloc = pool2.allocate(2);

            start.elapsed()
        });

        barrier.wait();

        std::thread::sleep(std::time::Duration::from_millis(100));
        drop(alloc);

        let elapsed = handle.join().expect("allocation thread should not panic");
        assert!(elapsed >= std::time::Duration::from_millis(100));
    }

    #[test]
    fn ok_concurrent_allocations() {
        let pool = sync::Arc::new(create_bufpool(BUF_SIZE.bytes() * 0x1000));

        let mut handles = Vec::new();
        for _ in 0..0x0A {
            let pool = pool.clone();

            handles.push(std::thread::spawn(move || {
                for _ in 0..0x64 {
                    drop(pool.allocate(1));
                }
            }));
        }

        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(pool.allocated_memory.load(atomic::Ordering::Acquire), 0);
        assert_eq!(pool.active_allocations.load(atomic::Ordering::Acquire), 0);
    }

    mod drop {
        use super::*;

        #[test]
        fn ok_partial_drop_updates_accounting() {
            let bpool = create_bufpool(BUF_SIZE.bytes() * 0x0A);

            let alloc1 = bpool.allocate(2);
            let alloc2 = bpool.allocate(2);

            assert_eq!(
                bpool.allocated_memory.load(atomic::Ordering::Acquire),
                BUF_SIZE.bytes() * 4
            );
            drop(alloc1);

            assert_eq!(
                bpool.allocated_memory.load(atomic::Ordering::Acquire),
                BUF_SIZE.bytes() * 2
            );
            drop(alloc2);

            assert_eq!(bpool.allocated_memory.load(atomic::Ordering::Acquire), 0);
        }

        #[test]
        fn ok_drop_waits_for_active_allocations() {
            let bpool = sync::Arc::new(create_bufpool(BUF_SIZE.bytes() * 0x1A));
            let alloc = bpool.allocate(0x10);

            let handle = std::thread::spawn(move || {
                drop(bpool);
            });

            std::thread::sleep(std::time::Duration::from_millis(0x64));
            assert!(!handle.is_finished());
            drop(alloc);

            handle.join().unwrap();
        }
    }

    mod memory_tests {
        use super::*;

        #[test]
        fn ok_create_layout() {
            let layout = create_layout(0x1000);

            assert_eq!(layout.align(), 1);
            assert_eq!(layout.size(), 0x1000);
        }

        #[test]
        #[should_panic(expected = "Invalid Layout")]
        fn err_create_layout() {
            create_layout(usize::MAX);
        }

        #[test]
        fn ok_allocate_layout() {
            let layout = create_layout(0x10);
            let pointer = allocate_layout(layout);
            let raw_ptr = pointer.as_ptr();

            assert!(!raw_ptr.is_null());
            deallocate_memory(pointer, layout);
        }

        #[test]
        fn ok_allocate_layout_allows_write() {
            let layout = create_layout(0x80);
            let pointer = allocate_layout(layout);

            unsafe {
                pointer.as_ptr().write(0x40);
                assert_eq!(pointer.as_ptr().read(), 0x40);
            }

            deallocate_memory(pointer, layout);
        }

        #[test]
        fn ok_allocate_layout_allows_write_to_entire_slice() {
            let layout = create_layout(0x200);
            let pointer = allocate_layout(layout);

            unsafe {
                for i in 0..0x200 {
                    pointer.as_ptr().add(i).write((i % 0xFF) as u8);
                }

                for i in 0..0x200 {
                    assert_eq!(pointer.as_ptr().add(i).read(), (i % 0xFF) as u8);
                }
            }

            deallocate_memory(pointer, layout);
        }
    }

    mod alloc_struct {
        use super::*;

        #[test]
        fn ok_first_returns_ptr_to_first_buf_from_alloc() {
            let bpool = create_bufpool(BUF_SIZE.bytes() * 0x20);
            let alloc = bpool.allocate(0x10);

            assert_eq!(alloc.first(), alloc.pointer.as_ptr());
        }

        #[test]
        fn ok_length_returns_length_of_alloc() {
            let bpool = create_bufpool(BUF_SIZE.bytes() * 0x20);
            let alloc = bpool.allocate(0x10);

            assert_eq!(alloc.length(), alloc.buffers);
        }

        #[test]
        fn ok_allocated_bytes_return_total_allocated_bytes() {
            let bpool = create_bufpool(BUF_SIZE.bytes() * 0x20);
            let alloc = bpool.allocate(0x10);

            assert_eq!(alloc.allocated_bytes(), alloc.buffers * BUF_SIZE.bytes());
        }

        #[test]
        fn ok_alloc_can_be_shared_across_threads() {
            let pool = sync::Arc::new(create_bufpool(BUF_SIZE.bytes() * 2));
            let alloc = pool.allocate(1);

            std::thread::spawn(move || {
                drop(alloc);
            })
            .join()
            .unwrap();
        }
    }

    mod iterator {
        use super::*;

        #[test]
        fn ok_iter_yeilds_all_buffers() {
            let bpool = create_bufpool(BUF_SIZE.bytes() * 0x0A);
            let alloc = bpool.allocate(4);

            let ptrs: Vec<_> = alloc.iter().collect();
            assert_eq!(ptrs.len(), 4);

            assert_eq!(ptrs[1] as usize - ptrs[0] as usize, 0x20);
            assert_eq!(ptrs[2] as usize - ptrs[1] as usize, 0x20);
            assert_eq!(ptrs[3] as usize - ptrs[2] as usize, 0x20);
        }

        #[test]
        fn ok_iter_yeilds_none_when_exhausted() {
            let bpool = create_bufpool(BUF_SIZE.bytes() * 0x0A);
            let alloc = bpool.allocate(2);
            let mut iter = alloc.iter();

            assert!(iter.next().is_some());
            assert!(iter.next().is_some());
            assert!(iter.next().is_none());
            assert!(iter.next().is_none());
        }
    }
}