shared_buffer_rs/

lib.rs

/*-
 * shared-buffer-rs - a buffer managment and sharing crate.
 * 
 * Copyright (C) 2025 Aleksandr Morozov
 * 
 * The scram-rs crate can be redistributed and/or modified
 * under the terms of either of the following licenses:
 *
 *   1. the Mozilla Public License Version 2.0 (the “MPL”) OR
 *                     
 *   2. EUROPEAN UNION PUBLIC LICENCE v. 1.2 EUPL © the European Union 2007, 2016
 */

/*! A small crate which implements a thread safe implementation to allocate
 the buffer of some size, borrow as muable in current context or thread
 and borrow as many read only references as needed which are Send+Sync.

 It acts like Arc but embeds the RefCell functionality without any
 issues with Send and Sync.

 The main purpose it to have a lock free, lightweight buffer I/O for
 writing in one side and broadcast to multiple tasks i.e threads and
 async task making sure that it can not be modifyied.
 */

#![cfg_attr(not(feature = "std"), no_std)]

#[cfg(not(feature = "std"))]
extern crate alloc;


#[cfg(not(feature = "std"))]
use core::
{
    fmt, 
    mem, 
    ops::{Deref, DerefMut}, 
    ptr::{self, NonNull}, 
    sync::atomic::{AtomicU64, Ordering}
};

#[cfg(not(feature = "std"))]
use core::{marker::PhantomData, task::Poll, time::Duration};

#[cfg(not(feature = "std"))]
use alloc::{boxed::Box, collections::vec_deque::VecDeque, vec::Vec};

#[cfg(not(feature = "std"))]
use alloc::vec;
use crossbeam_utils::Backoff;

#[cfg(feature = "std")]
use std::
{
    collections::VecDeque, 
    ops::{Deref, DerefMut}, 
    ptr::{self, NonNull}, 
    sync::atomic::{AtomicU64, Ordering},
    fmt,
    mem
};

#[cfg(feature = "std")]
use std::{marker::PhantomData, task::Poll, time::Duration};

extern crate crossbeam_utils;

/// A local try_clone trait.
pub trait TryClone: Sized 
{
    type Error;

    /// Attempts to clone the instance in reasonable time without completly
    /// blocking the thread.
    fn try_clone(&self) -> Result<Self, Self::Error>;
}

/// An interface of the `async_drop` for async.
pub trait LocalAsyncDrop: Send + Sync + 'static
{
    /// Same as `drop()` but for `async`.
    fn async_drop(&mut self) -> impl std::future::Future<Output = ()>;
}

/// An interface of the `async_clone` for async.
pub trait LocalAsyncClone: Send + Sync + 'static
{
    /// Same as `clone()` but for `async`.
    fn async_clone(&self) -> impl std::future::Future<Output = Self>;
}

/// Use this function to drop [RBuffer] and [WBuffer] obtained via
/// `async_*`.
pub async 
fn async_drop<LAD: LocalAsyncDrop + Send + Sync>(mut lad: LAD)
{
    lad.async_drop().await;

    drop(lad);
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RwBufferError
{
    /// The limit on clones of [RBuffer] was reached.
    TooManyRead,

    /// The limit on clones of [RwBuffer] was reched.
    TooManyBase,

    /// Just retry read operation later.
    ReadTryAgianLater,

    /// Just retry write operation later.
    WriteTryAgianLater,

    /// Just retry cloning [RwBuffer] the base or any other related operation later.
    BaseTryAgainLater,

    /// No more buffers left.
    OutOfBuffers,

    /// Can not downgrade [WBuffer] to [RBuffer]
    DowngradeFailed,

    /// The provided arguments are not valid.
    InvalidArguments,

    /// The buffer is busy.
    Busy,
}

impl fmt::Display for RwBufferError
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        match self
        {
            Self::TooManyRead => 
                write!(f, "TooManyRead: read soft limit reached"),
            Self::TooManyBase => 
                write!(f, "TooManyBase: base soft limit reached"),
            Self::ReadTryAgianLater => 
                write!(f, "ReadTryAgianLater: shared access not available, try again later"),
            Self::WriteTryAgianLater => 
                write!(f, "WriteTryAgianLater: exclusive access not available, try again later"),
            Self::BaseTryAgainLater => 
                write!(f, "BaseTryAgainLater: failed to obtain a clone in reasonable time"),
            Self::OutOfBuffers => 
                write!(f, "OutOfBuffers: no more free bufers are left"),
            Self::DowngradeFailed => 
                write!(f, "DowngradeFailed: can not downgrade exclusive to shared, race condition"),
            Self::InvalidArguments => 
                write!(f, "InvalidArguments: arguments are not valid"),  
            Self::Busy => 
                write!(f, "RwBuffer is busy and cannot be acquired"),       
        }
    }
}

pub type RwBufferRes<T> = Result<T, RwBufferError>;

/// A read only buffer. This instance is [Send] and [Sync]
/// as it does not provide any write access.
#[derive(Debug)]
pub struct RBuffer
{
    /// The inner read only
    inner: NonNull<RwBufferInner>,

    /// Is set to `true` when async_drop was performed earlier.
    a_dropped: bool,
}

unsafe impl Send for RBuffer {}
unsafe impl Sync for RBuffer {}

impl RwBufType for RBuffer {}

impl Eq for RBuffer {}

impl PartialEq for RBuffer
{
    fn eq(&self, other: &Self) -> bool 
    {
        return self.inner == other.inner;
    }
}

impl RBuffer
{
    #[inline]
    fn new(inner: NonNull<RwBufferInner>) -> Self
    {
        return Self{ inner, a_dropped: false };
    }

    #[cfg(test)]
    fn get_flags(&self) -> RwBufferFlags<Self>
    {
        use core::sync::atomic::Ordering;

        let inner = unsafe{ self.inner.as_ref() };

        let flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Relaxed).into();

        return flags;
    }

    /// Borrow the inner buffer as slice.
    pub
    fn as_slice(&self) -> &[u8]
    {
        let inner = unsafe { self.inner.as_ref() };

        return inner.buf.as_ref().unwrap().as_slice();
    }

    /// Attempts to consume the instance and retrive the inner buffer. This means
    /// that the instance will no longer be available.
    /// 
    /// > Safe to call from async.
    ///
    /// The following condition should be satisfied:
    /// 1) No more readers except current instance.
    ///
    /// 2) No base references, item should not contain base references from [RwBuffer].
    /// 
    /// # Returns
    /// 
    /// A [Result] is returned with: 
    /// 
    /// * [Result::Ok] with the consumed inner [Vec]
    /// 
    /// * [Result::Err] with the consumed instance
    pub
    fn try_inner(mut self) -> Result<Vec<u8>, Self>
    {
        let inner = unsafe { self.inner.as_ref() };

        let current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::SeqCst).into();

       // new_flags.unread();

        if current_flags.try_inner_check() == true
        {
            // in theory if at that moment only one read operation left, then no other can occure
            // because the current_flags are obtained with SeqCst, and no read can appear

            let inner = unsafe { self.inner.as_mut() };

            let buf = inner.buf.take().unwrap();
        
            // even if the instance is used with async, this must never block because 
            // instance is already uniq
            drop(self);

            return Ok(buf);
        }

        return Err(self);        
    }

    fn inner(&self) -> &RwBufferInner
    {
        return unsafe { self.inner.as_ref() };
    }
}

impl Deref for RBuffer
{
    type Target = Vec<u8>;

    fn deref(&self) -> &Vec<u8>
    {
        let inner = self.inner();

        return inner.buf.as_ref().unwrap();
    }
}

impl LocalAsyncClone for RBuffer
{
    fn async_clone(&self) -> impl std::future::Future<Output = Self> 
    {
        return 
            std::future::poll_fn(
                |cx|
                {
                    let inner = self.inner();

                    let current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Relaxed).into();
                    let mut new_flags = current_flags.clone();

                    new_flags.read().unwrap();

                    let res = 
                        inner
                            .flags
                            .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

                    if let Ok(_) = res
                    {
                        return Poll::Ready( Self{ inner: self.inner, a_dropped: self.a_dropped } );
                    }

                    return Poll::Pending;
                }
            );
    }
}

impl Clone for RBuffer
{
    /// Attempts to clone the RBuffer incrementing the `read` reference. 
    /// Would block until the clone is obtained. Should not block for long 
    /// time. Because this crate is experimental, it will panic if it will
    /// not be able to obtain clone.
    /// 
    /// # Returns 
    /// 
    /// Returns the new [RBuffer] instance.
    /// 
    /// # Panic
    /// 
    /// Panics if too many references were created. The reference count
    /// is limited to max::u32 - 10. Or will panic if will not be able to obtain 
    /// a clone of the [RBuffer] in reasonable time as this must not block
    /// for a long time.
    fn clone(&self) -> Self 
    {
        let inner = self.inner();

        let mut current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Relaxed).into();
        let mut new_flags = current_flags.clone();

        new_flags.read().unwrap();

        let backoff = Backoff::new();
        let mut parked = false;

        loop
        {
            let res = 
                inner
                    .flags
                    .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

            if let Ok(_) = res
            {
                return Self{ inner: self.inner, a_dropped: self.a_dropped };
            }

            current_flags = res.err().unwrap().into();
            new_flags = current_flags.clone();

            new_flags.read().unwrap();

            if backoff.is_completed() == false
            {
                backoff.snooze();
            }
            else
            {
                if parked == false
                {
                    // last attempt
                    std::thread::park_timeout(Duration::from_millis(1));

                    parked = true;
                }
                else
                {
                    panic!("can not obtain a clone of RBuffer!");
                }
            }
        }
    }
}

impl TryClone for RBuffer
{
    type Error = RwBufferError;

    /// Attempts to clone the RBuffer incrementing the `read` reference.
    /// 
    /// # Returns 
    /// 
    /// Returns the new [Result] where on success a clone of [RBuffer] instance is
    /// returned, otherwise the:
    /// 
    /// * [RwBufferError::ReadTryAgianLater] - is returned if it failed to acquire the read clone
    ///     in reasonable time.
    /// 
    /// * [RwBufferError::TooManyRead] - is returned if limit was reached.
    fn try_clone(&self) -> Result<Self, Self::Error> 
    {
        let inner = self.inner();

        let mut current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::SeqCst).into();
        let mut new_flags = current_flags.clone();

        new_flags.read()?;

        let backoff = Backoff::new();

        loop
        {
            let res = 
                inner
                    .flags
                    .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

            if let Ok(_) = res
            {
                return Ok(Self{ inner: self.inner, a_dropped: self.a_dropped });
            }

            current_flags = res.err().unwrap().into();
            new_flags = current_flags.clone();

            new_flags.read()?;

            if backoff.is_completed() == false
            {
                backoff.snooze();
            }
            else
            {
                break;
            }
        }

        return Err(RwBufferError::ReadTryAgianLater);
    }
}

impl LocalAsyncDrop for RBuffer
{
    fn async_drop(&mut self) -> impl std::future::Future<Output = ()>
    {
        self.a_dropped = true;

        return 
            std::future::poll_fn(
                |cx|
                {
                    let inner = self.inner();

                    let current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::SeqCst).into();
                    let mut new_flags = current_flags.clone();

                    new_flags.unread();

                    let res = 
                        inner
                            .flags
                            .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

                    if let Ok(flags) = res.map(|v| <u64 as Into<RwBufferFlags<Self>>>::into(v))
                    {
                        if flags.is_drop_inplace() == true
                        {
                            // call descrutor
                            unsafe { ptr::drop_in_place(self.inner.as_ptr()) };
                        }

                        return Poll::Ready(());
                    }

                    cx.waker().wake_by_ref();

                    return Poll::Pending;
                }
            );
    }
}

impl Drop for RBuffer
{
    /// Should completly drop the instance (with data) only, if there is no more
    /// readers or it is not referenced in the base.
    /// 
    /// # Panic
    /// 
    /// May panic if it will not be able to drop the instance in reasonable time i.e in 
    /// 1000 attempts.
    fn drop(&mut self)
    {
        if self.a_dropped == true
        {
            return;
        }

        let inner = self.inner();

        let mut current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Relaxed).into();
        let mut new_flags = current_flags.clone();

        new_flags.unread();

        let backoff = Backoff::new();
        
        for _ in 0..1000
        {
            let res = 
                inner
                    .flags
                    .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

            if let Ok(flags) = res.map(|v| <u64 as Into<RwBufferFlags<Self>>>::into(v))
            {
                if flags.is_drop_inplace() == true
                {
                    // call descrutor
                    unsafe { ptr::drop_in_place(self.inner.as_ptr()) };
                }

                return;
            }

            current_flags = res.err().unwrap().into();
            new_flags = current_flags.clone();

            new_flags.unread();

            backoff.snooze();
        }

        // todo... solve this situation somehow
        panic!("assertion trap: RBuffer::drop can not drop RBuffer in reasonable time!");
    }
}

/// A Write and Read buffer. An exclusive instance which can not be copied or
/// clonned. Once writing is complete, the instance can be dropped or downgraded to
/// Read-only instance. This instance is NOT [Send] and [Sync]. 
#[derive(Debug, PartialEq, Eq)]
pub struct WBuffer
{
    /// A pointer to the leaked buffer instance.
    buf: NonNull<RwBufferInner>,

    /// Is set to `true` when `downgrade` is called.
    downgraded: bool,
}

unsafe impl Send for WBuffer{}
unsafe impl Sync for WBuffer{}

impl RwBufType for WBuffer{}

impl WBuffer
{
    #[inline]
    fn new(inner: NonNull<RwBufferInner>) -> Self
    {
        return Self{ buf: inner, downgraded: false };
    }

    /// Attempts to downgrade the `write` instance into the `read` instance by consuming the
    /// [WBuffer]. 
    /// 
    /// Can not be performed vice-versa (at least in this version). In normal conditions
    /// should never return Error.
    /// 
    /// # Returns 
    /// 
    /// A [Result] is returned with the [RBuffer] on success. The [Result::Err] is returned 
    /// if it failed to downgrade instance in resonable time.
    pub
    fn downgrade(mut self) ->  Result<RBuffer, Self>
    {
        let inner = unsafe { self.buf.as_ref() };

        let mut current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Relaxed).into();
        let mut new_flags = current_flags.clone();

        new_flags.downgrade();

        let backoff = Backoff::new();

        while backoff.is_completed() == false
        {
            let res = 
                inner
                    .flags
                    .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

            if let Ok(_) = res
            {
                self.downgraded = true;

                return Ok(RBuffer::new(self.buf.clone()));
            }

            current_flags = res.err().unwrap().into();
            new_flags = current_flags.clone();

            new_flags.downgrade();

            backoff.snooze();
        }

        return Err(self);
    }

    pub
    fn as_slice(&self) -> &[u8]
    {
        let inner = unsafe { self.buf.as_ref() };

        return inner.buf.as_ref().unwrap()
    }
}

impl Deref for WBuffer
{
    type Target = Vec<u8>;

    fn deref(&self) -> &Vec<u8>
    {
        let inner = unsafe { self.buf.as_ref() };

        return inner.buf.as_ref().unwrap();
    }
}

impl DerefMut for WBuffer
{
    fn deref_mut(&mut self) -> &mut Vec<u8>
    {
        let inner = unsafe { self.buf.as_mut() };

        return inner.buf.as_mut().unwrap();
    }
}

impl Drop for WBuffer
{
    /// The instance may perform `drop_in_place` if there is no `base` references.
    fn drop(&mut self)
    {
        if self.downgraded == true
        {
            return;
        }

        let inner = unsafe { self.buf.as_ref() };

        let mut current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Relaxed).into();
        let mut new_flags = current_flags.clone();

        new_flags.unwrite();

        let backoff = Backoff::new();

        for _ in 0..1000
        {
            let res = 
                inner
                    .flags
                    .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

            if let Ok(flags) = res.map(|v| <u64 as Into<RwBufferFlags<Self>>>::into(v))
            {
                if flags.is_drop_inplace() == true
                {
                    // call descrutor
                    unsafe { ptr::drop_in_place(self.buf.as_ptr()) };
                }

                return;
            }

            current_flags = res.err().unwrap().into();
            new_flags = current_flags.clone();

            new_flags.unwrite();

            backoff.snooze();
        }

        // todo... solve this situation somehow

        panic!("assertion trap: WBuffer::drop can not drop RBuffer in reasonable time!");
    }
}

impl LocalAsyncDrop for WBuffer
{
    fn async_drop(&mut self)  -> impl std::future::Future<Output = ()>
    {
        self.downgraded = true;

        return 
            std::future::poll_fn(
                move |cx|
                {
                    let inner = unsafe { self.buf.as_ref() };

                    let current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::SeqCst).into();
                    let mut new_flags = current_flags.clone();

                    new_flags.unwrite();

                    let res = 
                        inner
                            .flags
                            .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

                    if let Ok(flags) = res.map(|v| <u64 as Into<RwBufferFlags<Self>>>::into(v))
                    {
                        if flags.is_drop_inplace() == true
                        {
                            // call descrutor
                            unsafe { ptr::drop_in_place(self.buf.as_ptr()) };
                        }

                        return Poll::Ready(());
                    }

                    cx.waker().wake_by_ref();

                    return Poll::Pending;
                }
            );
    }
}

trait RwBufType {}

/// Internal structure which represents the status. It can not be
/// larger than 8-byte to fit into [AtomicU64].
#[repr(align(8))]
#[derive(Debug, PartialEq, Eq)]
struct RwBufferFlags<TP>
{
    /// A reader refs counter. If larger than 0, no writes possible.
    read: u32, // = 4

    /// An exclusive write lock. When true, no reades should present.
    write: bool, // = 1

    /// A base refs i.e which holds the data.
    /// If this value is zero, means the instance can be dropped in place
    /// when `write` is false and `read` equals 0.
    base: u16, // = 2

    /// Unused
    unused0: u8, // = 1,

    _p: PhantomData<TP>,
}


impl<TP: RwBufType> From<u64> for RwBufferFlags<TP>
{
    fn from(value: u64) -> Self
    {
        return unsafe { mem::transmute(value) };
    }
}

impl<TP: RwBufType> From<RwBufferFlags<TP>> for u64
{
    fn from(value: RwBufferFlags<TP>) -> Self
    {
        return unsafe { mem::transmute(value) };
    }
}

impl<TP: RwBufType> Default for RwBufferFlags<TP>
{
    fn default() -> RwBufferFlags<TP>
    {
        return
            Self
            {
                read: 0,
                write: false,
                base: 1,
                unused0: 0,
                _p: PhantomData
            };
    }
}

impl Copy for RwBufferFlags<WBuffer>{}

impl Clone for RwBufferFlags<WBuffer>
{
    fn clone(&self) -> Self 
    {
        return 
            Self 
            { 
                read: self.read.clone(), 
                write: self.write.clone(), 
                base: self.base.clone(), 
                unused0: self.unused0.clone(), 
                _p: PhantomData
            }
    }
}

impl Copy for RwBufferFlags<RBuffer>{}

impl Clone for RwBufferFlags<RBuffer>
{
    fn clone(&self) -> Self 
    {
        return 
            Self 
            { 
                read: self.read.clone(), 
                write: self.write.clone(), 
                base: self.base.clone(), 
                unused0: self.unused0.clone(), 
                _p: PhantomData
            }
    }
}

impl Copy for RwBufferFlags<RwBuffer>{}

impl Clone for RwBufferFlags<RwBuffer>
{
    fn clone(&self) -> Self 
    {
        return 
            Self 
            { 
                read: self.read.clone(), 
                write: self.write.clone(), 
                base: self.base.clone(), 
                unused0: self.unused0.clone(), 
                _p: PhantomData
            }
    }
}

impl RwBufferFlags<WBuffer>
{
    #[inline]
    fn write(&mut self) -> RwBufferRes<()>
    {
        if self.read == 0
        {
            self.write = true;

            return Ok(());
        }
        else
        {
            return Err(RwBufferError::WriteTryAgianLater);
        }
    }

    #[inline]
    fn downgrade(&mut self) 
    {
        self.write = false;
        self.read += 1;
    }

     #[inline]
    fn unwrite(&mut self)
    {
        self.write = false;
    }
}

impl RwBufferFlags<RBuffer>
{
    #[inline]
    fn try_inner_check(&self) -> bool
    {
        return self.read == 1 && self.write == false && self.base == 0;
    }

    #[inline]
    fn unread(&mut self)
    {
        self.read -= 1;
    }

    #[inline]
    fn read(&mut self) -> RwBufferRes<()>
    {
        if self.write == false
        {
            self.read += 1;

            if self.read <= Self::MAX_READ_REFS
            {
                return Ok(());
            }
            
            return Err(RwBufferError::TooManyRead);
        }

        return Err(RwBufferError::ReadTryAgianLater);
    }
}

impl RwBufferFlags<RwBuffer>
{
    #[inline]
    fn make_pre_unused() -> Self
    {
        return Self{ read: 0, write: false, base: 1, unused0: 0, _p: PhantomData };
    }

    #[inline]
    fn read(&mut self) -> RwBufferRes<()>
    {
        if self.write == false
        {
            self.read += 1;

            if self.read <= Self::MAX_READ_REFS
            {
                return Ok(());
            }
            
            return Err(RwBufferError::TooManyRead);
        }

        return Err(RwBufferError::ReadTryAgianLater);
    }

    #[inline]
    fn write(&mut self) -> RwBufferRes<()>
    {
        if self.read == 0
        {
            self.write = true;

            return Ok(());
        }
        else
        {
            return Err(RwBufferError::WriteTryAgianLater);
        }
    }
    
    #[inline]
    fn base(&mut self) -> RwBufferRes<()>
    {
        self.base += 1;

        if self.base <= Self::MAX_BASE_REFS
        {
            return Ok(());
        }

        return Err(RwBufferError::TooManyBase);
    }

    #[inline]
    fn unbase(&mut self) -> bool
    {
        self.base -= 1;

        return self.base != 0;
    }
}

impl<TP: RwBufType> RwBufferFlags<TP>
{
    /// A soft limit on the amount of references for reading instances.
    pub const MAX_READ_REFS: u32 = u32::MAX - 2;

    /// A soft limit on the amount of references for base instances.
    pub const MAX_BASE_REFS: u16 = u16::MAX - 2;

    
    #[inline]
    fn is_free(&self) -> bool
    {
        return self.write == false && self.read == 0 && self.base == 1;
    }

    #[inline]
    fn is_drop_inplace(&self) -> bool
    {
        return self.read == 0 && self.write == false && self.base == 0;
    }
}

#[derive(Debug)]
pub struct RwBufferInner
{
    /// A [RwBufferFlags] represented as atomic u64.
    flags: AtomicU64,

    /// A buffer.
    buf: Option<Vec<u8>>,
}

impl RwBufferInner
{
    fn new(buf_size: usize) -> Self
    {
        return
            Self
            {
                flags: 
                    AtomicU64::new(RwBufferFlags::<RwBuffer>::default().into()),
                buf: 
                    Some(vec![0_u8; buf_size])
            };
    }
}

/// A base instance which holds the `leaked` pointer to [RwBufferInner].
/// 
/// This instance can provide either an exclusive write access or 
/// multiple read access, but not at the same time. Can be used to store
/// the instance. This instance is [Send] and [Sync] because the insternals
/// are guarded by ordered atomic operations.
#[derive(Debug, PartialEq, Eq)]
pub struct RwBuffer(NonNull<RwBufferInner>);

unsafe impl Send for RwBuffer {}
unsafe impl Sync for RwBuffer {}

impl RwBufType for RwBuffer {}

impl RwBuffer
{
    #[inline]
    fn new(buf_size: usize) -> Self
    {
        let status = Box::new(RwBufferInner::new(buf_size));

        return Self(Box::leak(status).into());
    }

    #[inline]
    fn inner(&self) -> &RwBufferInner
    {
        return unsafe { self.0.as_ref() };
    }

    /// Checks if this instance satisfies the following conditions:
    /// 
    /// * No exclusive write access
    /// 
    /// * No read access
    /// 
    /// * There is only one base reference.
    /// 
    /// But since check everything may have been already changed.
    /// 
    /// # Returns 
    /// 
    /// * - `true` if instance satisfies the conditions above.
    /// 
    /// * - `false` if does not satisfy the conditions above.
    #[inline]
    pub
    fn is_free(&self) -> bool
    {
        let inner = self.inner();

        let flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Relaxed).into();

        return flags.is_free();
    }

    /// Accures the instance, if it satisfy the following conditions:
    /// 
    /// * No exclusive write access
    /// 
    /// * No read access
    /// 
    /// * There is only one base reference.
    /// 
    /// # Returns 
    /// 
    /// The [Result] is retuerned with the clonned [RwBuffer] instance or
    /// [Result::Err] with the following errors:
    /// 
    /// * [RwBufferError::Busy] - the instance have already been taken.
    /// 
    /// * [RwBufferError::TooManyBase] - too many base references are already around.
    #[inline]
    pub(crate) 
    fn acqiure_if_free(&self) -> RwBufferRes<Self>
    {
        let inner = self.inner();

        let current_flags: RwBufferFlags<Self> = RwBufferFlags::make_pre_unused();
        let mut new_flags = current_flags.clone();
        
        new_flags.base()?;

        let res = 
                inner
                    .flags
                    .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

        if let Ok(_) = res
        {
            return Ok(Self(self.0.clone()));
        }

        return Err(RwBufferError::Busy);
    }

    /// Attemts to make an exclusive (write) access to the buffer.
    /// 
    /// Would block for short period of time and return error.
    /// 
    /// # Returns
    /// 
    /// A [Result] in form of [RwBufferRes] is returned with:
    /// 
    /// * [Result::Ok] with the [WBuffer] instance
    /// 
    /// * [Result::Err] may be returned a [RwBufferError::WriteTryAgianLater] in case 
    ///     if the there is/are an active `read` references or acquite exc. lock failed.
    pub
    fn write(&self) -> RwBufferRes<WBuffer>
    {
        let inner = self.inner();

        let mut current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Relaxed).into();
        let mut new_flags = current_flags.clone();

        new_flags.write()?;

        let backoff = Backoff::new();

        while backoff.is_completed() == false
        {
            let res = 
                inner
                    .flags
                    .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

            if let Ok(_) = res
            {
                return Ok(WBuffer::new(self.0.clone()));
            }

            current_flags = res.err().unwrap().into();
            new_flags = current_flags.clone();

            new_flags.write()?;

            backoff.snooze();
        }

        return Err(RwBufferError::WriteTryAgianLater);
    }

    /// Attempts to gain an exclusive access in async way.
    /// 
    /// # Return
    /// 
    /// Return the same result as [RwBuffer::write].
    pub async 
    fn write_async(&self) -> RwBufferRes<WBuffer>
    {
        return 
            std::future::poll_fn(
                |cx|
                {
                    let inner = self.inner();

                    let current_flags: RwBufferFlags<WBuffer> = inner.flags.load(Ordering::SeqCst).into();
                    let mut new_flags = current_flags.clone();

                    if let Err(e) = new_flags.write()
                    {
                        return Poll::Ready(Err(e));
                    }

                    let res = 
                        inner
                            .flags
                            .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

                    if let Ok(_) = res
                    {
                        return Poll::Ready( Ok( WBuffer::new(self.0.clone()) ) );
                    }

                    cx.waker().wake_by_ref();

                    return Poll::Pending;
                }
            )
            .await;
    }
    

    /// Attemts to make a shared (read) access to the buffer.
    /// 
    /// Would block for short period of time and return error.
    /// 
    /// # Returns
    /// 
    /// A [Result] in form of [RwBufferRes] is returned with:
    /// 
    /// * [Result::Ok] with the [RBuffer] instance
    /// 
    /// * [Result::Err] with error type is returned:
    /// 
    /// - [RwBufferError::TooManyRead] is returned when the soft limit of
    ///     references was reached.
    /// 
    /// - [RwBufferError::ReadTryAgianLater] is returned if there is an 
    ///     active exclusive access.
    pub
    fn read(&self) -> RwBufferRes<RBuffer>
    {

        let inner = self.inner();

        let mut current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Relaxed).into();
        let mut new_flags = current_flags.clone();

        new_flags.read()?;

        let backoff = Backoff::new();

        while backoff.is_completed() == false
        {
            let res = 
                inner
                    .flags
                    .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

            if let Ok(_) = res
            {
                return Ok(RBuffer::new(self.0.clone()));
            }

            current_flags = res.err().unwrap().into();
            new_flags = current_flags.clone();

            new_flags.read()?;

            backoff.snooze();
        }

        return Err(RwBufferError::ReadTryAgianLater);
    }

    /// Attempts to gain an shared access in async way.
    /// 
    /// # Return
    /// 
    /// Return the same result as [RwBuffer::read].
    pub async 
    fn read_async(&self) -> RwBufferRes<RBuffer>
    {
        return 
            std::future::poll_fn(
                |cx|
                {
                    let inner = self.inner();

                    let current_flags: RwBufferFlags<RBuffer> = inner.flags.load(Ordering::SeqCst).into();
                    let mut new_flags = current_flags.clone();

                    match new_flags.read()
                    {
                        Ok(_) => {},
                        Err(RwBufferError::TooManyRead) => 
                            return Poll::Ready(Err(RwBufferError::TooManyRead)),
                        Err(RwBufferError::ReadTryAgianLater) =>
                        {
                            cx.waker().wake_by_ref();

                            return Poll::Pending;
                        },
                        Err(e) =>
                            panic!("assertion trap: unknown error {} in Future for AsyncRBuffer", e)
                    }

                    let res = 
                        inner
                            .flags
                            .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

                    if let Ok(_) = res
                    {
                        return Poll::Ready( Ok( RBuffer::new(self.0.clone()) ) );
                    }

                    cx.waker().wake_by_ref();

                    return Poll::Pending;
                }
            )
            .await;
    }

    #[cfg(test)]
    fn get_flags(&self) -> RwBufferFlags<Self>
    {
        let inner = self.inner();

        let flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Acquire).into();

        return flags;
    }

    /// Clones the freshly created instance which is visible for current thread only!
    /// 
    /// # Returns 
    /// 
    /// A [Result] is returned with error [RwBufferError::TooManyBase] if too many copies 
    /// of base are made.
    fn clone_single(&self) -> RwBufferRes<Self>
    {
        let inner = self.inner();

        let mut current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Relaxed).into();

        current_flags.base()?;

        inner.flags.store(current_flags.into(), Ordering::Relaxed);

        return Ok(Self(self.0));
    }
}

impl Clone for RwBuffer
{
    /// Clones the instance and increasing the `base` ref count.
    /// 
    /// Will `panic` if a soft limit of refs were reached.
    fn clone(&self) -> Self
    {
        let inner = self.inner();

        let mut current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Relaxed).into();
        let mut new_flags = current_flags.clone();

        new_flags.base().unwrap();

        let backoff = Backoff::new();
        let mut parked = false;

        loop
        {
            let res = 
                inner
                    .flags
                    .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

            if let Ok(_) = res
            {
                return Self(self.0);
            }

            current_flags = res.err().unwrap().into();
            new_flags = current_flags.clone();

            new_flags.base().unwrap();

            if backoff.is_completed() == false
            {
                backoff.snooze();
            }
            else
            {
                if parked == false
                {
                    // last attempt
                    std::thread::park_timeout(Duration::from_millis(1));

                    parked = true;
                }
                else
                {
                    panic!("can not obtain a clone of RBuffer!");
                }
            }
        }
    }
}

impl TryClone for RwBuffer
{
    type Error = RwBufferError;

    /// Attempts to clone the [RwBuffer] incrementing the `base` reference.
    /// 
    /// # Returns 
    /// 
    /// Returns the new [Result] where on success a clone of [RBuffer] instance is
    /// returned, otherwise the:
    /// 
    /// * [RwBufferError::BaseTryAgainLater] - is returned if it failed to acquire the base clone
    ///     in reasonable time.
    /// 
    /// * [RwBufferError::TooManyBase] - is returned if limit was reached.
    fn try_clone(&self) -> Result<Self, Self::Error> 
    {
        let inner = self.inner();

        let mut current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::SeqCst).into();
        let mut new_flags = current_flags.clone();

        new_flags.base()?;

        let backoff = Backoff::new();

        while backoff.is_completed() == false
        {
            let res = 
                inner
                    .flags
                    .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

            if let Ok(_) = res
            {
                return Ok(Self(self.0));
            }

            current_flags = res.err().unwrap().into();
            new_flags = current_flags.clone();

            new_flags.base()?;

            backoff.snooze();
        }

        return Err(RwBufferError::BaseTryAgainLater);
    }
}

impl Drop for RwBuffer
{
    /// Drops the RwBuffer instance. In case if there is no readers and
    /// writers, then drop immidiatly the inner data.
    /// In case if there is any readers or writing, then drop only wrapper which
    /// is the zero reader.
    fn drop(&mut self)
    {
        let inner = self.inner();

        let mut current_flags: RwBufferFlags<Self> = inner.flags.load(Ordering::Relaxed).into();
        let mut new_flags = current_flags.clone();

        new_flags.unbase();

        let backoff = Backoff::new();

        for _ in 0..1000
        {
            let res = 
                inner
                    .flags
                    .compare_exchange_weak(current_flags.into(), new_flags.into(), Ordering::SeqCst, Ordering::Acquire);

            if let Ok(flags) = res.map(|v| <u64 as Into<RwBufferFlags<Self>>>::into(v))
            {
                if flags.is_drop_inplace() == true
                {
                    // call descrutor
                    unsafe { ptr::drop_in_place(self.0.as_ptr()) };
                }

                return;
            }

            current_flags = res.err().unwrap().into();
            new_flags = current_flags.clone();

            new_flags.unbase();

            backoff.snooze();
        }

        // todo... solve this situation somehow
        panic!("assertion trap: RwBuffer::drop can not drop RwBuffer in reasonable time!");
    }
}

/// An instance which controls the allocation of the new buffers or
/// reusage of already created and free instances. This instance is
/// not thread safe. The external mutex should be used.
#[derive(Debug)]
pub struct RwBuffers
{
    /// A buffer length in bytes. Not aligned.
    buf_len: usize,

    /// A maximum slots for new buffers.
    bufs_cnt_lim: usize,

    /// A list of buffers.
    buffs: VecDeque<RwBuffer>

}

impl RwBuffers
{
    /// Creates new instance wshich holds the base reference in the 
    /// inner storage with the capacity bounds.
    /// 
    /// # Arguments
    /// 
    /// * `buf_len` - a [usize] length of each buffer instance in bytes where
    ///     the payload is located.
    /// 
    /// * `pre_init_cnt` - a [usize] an initial pre allocated slots with created instances.
    /// 
    /// * `bufs_cnt_lim` - a maximum amount of the available slots. Determines the 
    ///     capacity bounds.
    /// 
    /// # Returns
    /// 
    /// A [Result] in form of [RwBufferRes] is returned with:
    /// 
    /// * [Result::Ok] with the [RwBuffers] instance
    /// 
    /// * [Result::Err] with error type is returned:
    /// 
    /// - [RwBufferError::InvalidArguments] is returned when the arguments are
    ///     incorrect. 
    pub
    fn new(buf_len: usize, pre_init_cnt: usize, bufs_cnt_lim: usize) -> RwBufferRes<Self>
    {
        if pre_init_cnt > bufs_cnt_lim
        {
            return Err(RwBufferError::InvalidArguments);
        }
        else if buf_len == 0
        {
            return Err(RwBufferError::InvalidArguments);
        }

        let buffs: VecDeque<RwBuffer> = 
            if pre_init_cnt > 0
            {
                let mut buffs = VecDeque::with_capacity(bufs_cnt_lim);

                for _ in 0..pre_init_cnt
                {
                    buffs.push_back(RwBuffer::new(buf_len));
                }

                buffs
            }
            else
            {
                VecDeque::with_capacity(bufs_cnt_lim)
            };

        return Ok(
            Self
            {
                buf_len: buf_len,
                bufs_cnt_lim: bufs_cnt_lim,
                buffs: buffs,
            }
        )
    }

    /// Same as `new` but without any limits. Unbounded storage.
    /// 
    /// # Arguments
    /// 
    /// * `buf_len` - a [usize] length of each buffer instance in bytes where
    ///     the payload is located.
    /// 
    /// * `pre_init_cnt` - a [usize] an initial pre allocated slots with created instances.
    /// 
    /// # Returns
    /// 
    /// Returns the instance.
    pub
    fn new_unbounded(buf_len: usize, pre_init_cnt: usize) -> Self
    {
        let mut buffs = VecDeque::with_capacity(pre_init_cnt);

        for _ in 0..pre_init_cnt
        {
            buffs.push_back(RwBuffer::new(buf_len));
        }

        return
            Self
            {
                buf_len: buf_len,
                bufs_cnt_lim: 0,
                buffs: buffs,
            };
    }

    /// Allocates either a new buffer or reuse the free. If the instance
    /// is created with bounds then in case if no free slots available
    /// returns error.
    /// 
    /// # Returns
    /// 
    /// A [Result] in form of [RwBufferRes] is returned with:
    /// 
    /// * [Result::Ok] with the [RwBuffers] instance
    /// 
    /// * [Result::Err] with error codes:
    ///  
    /// 
    /// * [RwBufferError::OutOfBuffers] - if limit was reached.
    /// 
    /// * [RwBufferError::TooManyBase] - should not appear, but if would
    ///     it means that there is a bug somewhere in the code.
    pub
    fn allocate(&mut self) -> RwBufferRes<RwBuffer>
    {
        // check the list if any available
        for buf in self.buffs.iter()
        {
            if let Ok(rwbuf) = buf.acqiure_if_free()
            {
                return Ok(rwbuf);
            }
        }

        if self.bufs_cnt_lim == 0 || self.buffs.len() < self.bufs_cnt_lim
        {
            let buf = RwBuffer::new(self.buf_len);
            let c_buf = buf.clone_single()?;

            self.buffs.push_back(buf);

            return Ok(c_buf);
        }

        return Err(RwBufferError::OutOfBuffers);
    }

    /// Allocates a buffer "in place" i.e finds the next allocated but unused
    /// buffer and removes it from the list or alloactes new buffer without
    /// adding it to the list. Should never return error.
    /// 
    /// # Returns
    /// 
    /// A [RwBuffer] is returned.
    pub
    fn allocate_in_place(&mut self) -> RwBuffer
    {
        let mut idx = Option::None;

        for (i, item) in self.buffs.iter().enumerate()
        {
            if let Ok(_) = self.buffs[i].acqiure_if_free()
            {
                idx = Some(i);

                break;
            }
        }

        return 
            idx
                .map_or(
                    RwBuffer::new(self.buf_len), 
                    |f| self.buffs.remove(f).unwrap()
                );

    }

    /// Retains the buffer list by removing any unused buffers as many times
    /// as set in the argument `cnt`. It does not guaranty than the selected 
    /// amount will be freed.
    /// 
    /// # Arguments
    /// 
    /// * `cnt` - how many slots to clean before exit.
    /// 
    /// # Returns 
    /// 
    /// A [usize] is returned which indicates how many instances was removed
    /// before the `cnt` was reached. 
    pub
    fn compact(&mut self, mut cnt: usize) -> usize
    {
        let p_cnt = cnt;

        self
            .buffs
            .retain(
                |buf|
                {
                    if buf.is_free() == true
                    {
                        cnt -= 1;

                        return false;
                    }

                    return true;
                }
            );

        return p_cnt - cnt;
    }

    #[cfg(test)]
    fn get_flags_by_index(&self, index: usize) -> Option<RwBufferFlags<RwBuffer>>
    {
        return Some(self.buffs.get(index)?.get_flags());
    }
}

#[cfg(feature = "std")]
#[cfg(test)]
mod tests;