shared_buffer_rs/

lib.rs

///! 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.

use std::
{
    collections::VecDeque, 
    ops::{Deref, DerefMut}, 
    ptr::{self, NonNull}, 
    sync::atomic::{AtomicU64, Ordering},
    fmt,
};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RwBufferError
{
    TooManyRead,
    TooManyBase,
    ReadTryAgianLater,
    WriteTryAgianLater,
    OutOfBuffers,
    DowngradeFailed,
    InvalidArguments
}

impl fmt::Display for RwBufferError
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        match self
        {
            RwBufferError::TooManyRead => 
                write!(f, "TooManyRead: read soft limit reached"),
            RwBufferError::TooManyBase => 
                write!(f, "TooManyBase: base soft limit reached"),
            RwBufferError::ReadTryAgianLater => 
                write!(f, "ReadTryAgianLater: shared access not available, try again later"),
            RwBufferError::WriteTryAgianLater => 
                write!(f, "WriteTryAgianLater: exclusive access not available, try again later"),
            RwBufferError::OutOfBuffers => 
                write!(f, "OutOfBuffers: no more free bufers are left"),
            RwBufferError::DowngradeFailed => 
                write!(f, "DowngradeFailed: can not downgrade exclusive to shared, race condition"),
            RwBufferError::InvalidArguments => 
                write!(f, "InvalidArguments: arguments are not valid"),                
        }
    }
}

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, PartialEq, Eq)]
pub struct RBuffer(NonNull<RwBufferInner>);

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

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

    #[cfg(test)]
    fn get_flags(&self) -> RwBufferFlags
    {
        let inner = unsafe{ self.0.as_ref() };

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

        return flags;
    }

    /// Borrow the inner buffer as slice.
    pub
    fn as_slice(&self) -> &[u8]
    {
        let inner = unsafe { self.0.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.
    ///
    /// 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.0.as_ref() };

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

        if flags.read == 1 && flags.write == false && flags.base == 0
        {
            let inner = unsafe { self.0.as_mut() };

            let buf = inner.buf.take().unwrap();
        
            drop(self);

            return Ok(buf);
        }
        else
        {
            return Err(self);
        }
    }

    fn inner(&self) -> &RwBufferInner
    {
        return unsafe { self.0.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 Clone for RBuffer
{
    /// Clones the RBuffer incrementing the `read` reference.
    /// 
    /// # Returns 
    /// 
    /// Returns the new [RBuffer] instance.
    /// 
    /// # Panic
    /// 
    /// Panics if too many references were created. The reference count
    /// is limited to max::u32 - 10
    fn clone(&self) -> Self
    {
        let inner = self.inner();

        let mut flags: RwBufferFlags = inner.flags.load(Ordering::Acquire).into();

        if flags.read().unwrap() == false
        {
            panic!("too many read references for RBuffer");
        }

        inner.flags.store(flags.into(), Ordering::Release);

        return Self(self.0);
    }
}

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.
    fn drop(&mut self)
    {
        let inner = self.inner();

        let mut flags: RwBufferFlags = inner.flags.load(Ordering::Acquire).into();

        flags.unread();

        // no one can write at that moment
        if flags.read == 0 && flags.base == 0
        {
            // call descrutor
            unsafe { ptr::drop_in_place(self.0.as_ptr()) };

            return;
        }

        inner.flags.store(flags.into(), Ordering::Release);

        return;
    }
}

/// 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{}

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

    /// Downgrades 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] in form of [RwBufferRes] is returned with:
    /// 
    /// * [Result::Ok] with the [RBuffer] instance
    /// 
    /// * [Result::Err] may be returned a [RwBufferError::DowngradeFailed] in case of a bug
    ///     in the code. This error is an indicator that there is a race condition which
    ///     means there is an error in ordering or CPU does not support strong ordering.
    pub
    fn downgrade(mut self) ->  RwBufferRes<RBuffer>
    {
        let inner = unsafe { self.buf.as_ref() };

        let mut flags: RwBufferFlags = inner.flags.load(Ordering::Acquire).into();

        let res = flags.downgrade();

        inner.flags.store(flags.into(), Ordering::Release);

        if res == true
        {
            self.downgraded = true;

            return Ok(RBuffer::new(self.buf.clone()));
        }
        else
        {
            return Err(RwBufferError::DowngradeFailed);
        }
    }

    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 flags: RwBufferFlags = inner.flags.load(Ordering::Acquire).into();

        flags.unwrite();

        if flags.read == 0 && flags.base == 0
        {
            // call descrutor
            unsafe { ptr::drop_in_place(self.buf.as_ptr()) };

            return;
        }

        inner.flags.store(flags.into(), Ordering::Release);

        return;
    }
}

/// 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
{
    /// 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
}

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

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

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

impl RwBufferFlags
{
    /// 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 base(&mut self) -> bool
    {
        self.base += 1;

        return self.base <= Self::MAX_BASE_REFS;
    }

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

        return self.base != 0;
    }

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

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

            return true;
        }
        else
        {
            return false;
        }
    }

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

            return Ok(self.read <= Self::MAX_READ_REFS);
        }

        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 unwrite(&mut self)
    {
        self.write = false;
    }
}

#[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::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 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.
    /// 
    /// # 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 = inner.flags.load(Ordering::Acquire).into();

        return flags.write == false && flags.read == 0 && flags.base == 1;
    }

    /// Accures the instance, if it satisfy the following conditions:
    /// 
    /// * No exclusive write access
    /// 
    /// * No read access
    /// 
    /// * There is only one base reference.
    /// 
    /// # Returns 
    /// 
    /// * - `true` if instance satisfies the conditions above.
    /// 
    /// * - `false` if does not satisfy the conditions above.
    #[inline]
    pub
    fn accure_if_free(&self) -> bool
    {
        let inner = self.inner();

        let mut flags: RwBufferFlags = inner.flags.load(Ordering::Acquire).into();

        let res =
            if flags.write == false && flags.read == 0 && flags.base == 1
            {
                let _ = flags.base();

                true
            }
            else
            {
                false
            };

        inner.flags.store(flags.into(), Ordering::Release);

        return res;
    }

    /// Attemts to make an exclusive (write) access to the buffer.
    /// 
    /// # 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.
    pub
    fn write(&self) -> RwBufferRes<WBuffer>
    {
        let inner = self.inner();

        let mut flags: RwBufferFlags = inner.flags.load(Ordering::Acquire).into();

        let res = flags.write();

        inner.flags.store(flags.into(), Ordering::Release);

        res?;

        return Ok(WBuffer::new(self.0.clone()));
    }

    /// Attemts to make a shared (read) access to the buffer.
    /// 
    /// # 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 flags: RwBufferFlags = inner.flags.load(Ordering::Acquire).into();

        let res = flags.read();

        inner.flags.store(flags.into(), Ordering::Release);

        if res? == false
        {
            return Err(RwBufferError::TooManyRead);
        }

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

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

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

        return flags;
    }
}

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 flags: RwBufferFlags = inner.flags.load(Ordering::Acquire).into();

        if flags.base() == false
        {
            panic!("too many base references for RBuffer");
        }

        inner.flags.store(flags.into(), Ordering::Release);

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

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 flags: RwBufferFlags = inner.flags.load(Ordering::Acquire).into();

        let unbased = flags.unbase();

        if flags.read == 0 && flags.write == false && unbased == false
        {
            // call descrutor
            unsafe { ptr::drop_in_place(self.0.as_ptr()) };

            return;
        }

        inner.flags.store(flags.into(), Ordering::Release);
    }
}

/// 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 [RwBufferError::OutOfBuffers] error.
    pub
    fn allocate(&mut self) -> RwBufferRes<RwBuffer>
    {
        // check the list if any available
        for buf in self.buffs.iter()
        {
            if buf.is_free() == true
            {
                return Ok(buf.clone());
            }
        }

        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();

            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
    {
        for i in 0..self.buffs.len()
        {
            if self.buffs[i].accure_if_free() == true
            {
                return self.buffs.remove(i).unwrap();
            }
        }

        let buf = RwBuffer::new(self.buf_len);

        return buf;
    }

    /// 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>
    {
        return Some(self.buffs.get(index)?.get_flags());
    }
}


#[cfg(test)]
mod tests
{
    use std::time::{Duration, Instant};

    use super::*;

    #[test]
    fn simple_test()
    {
        let mut bufs = RwBuffers::new(4096, 1, 2).unwrap();

        let buf0_res = bufs.allocate();
        assert_eq!(buf0_res.is_ok(), true, "{:?}", buf0_res.err().unwrap());

        let buf0 = buf0_res.unwrap();

        let buf0_w = buf0.write();
        assert_eq!(buf0_w.is_ok(), true, "{:?}", buf0_w.err().unwrap());
        assert_eq!(buf0.read(), Err(RwBufferError::ReadTryAgianLater));
        drop(buf0_w);

        let buf0_r = buf0.read();
        assert_eq!(buf0_r.is_ok(), true, "{:?}", buf0_r.err().unwrap());
        assert_eq!(buf0.write(), Err(RwBufferError::WriteTryAgianLater));

        let buf0_1 = buf0.clone();
        assert_eq!(buf0_1.write(), Err(RwBufferError::WriteTryAgianLater));

        let flags0 = buf0.get_flags();
        let flags0_1 = buf0_1.get_flags();

        assert_eq!(flags0, flags0_1);
        assert_eq!(flags0.base, 3);
        assert_eq!(flags0.read, 1);
        assert_eq!(flags0.write, false);
    }

    #[test]
    fn simple_test_dopped_in_place()
    {
        let mut bufs = RwBuffers::new(4096, 1, 2).unwrap();

        let buf0_res = bufs.allocate();
        assert_eq!(buf0_res.is_ok(), true, "{:?}", buf0_res.err().unwrap());

        let buf0 = buf0_res.unwrap();

        println!("{:?}", buf0.get_flags());

        let buf0_w = buf0.write();
        assert_eq!(buf0_w.is_ok(), true, "{:?}", buf0_w.err().unwrap());
        assert_eq!(buf0.read(), Err(RwBufferError::ReadTryAgianLater));

        drop(buf0);

        let buf0_flags = bufs.get_flags_by_index(0);
        assert_eq!(buf0_flags.is_some(), true, "no flags");
        let buf0_flags = buf0_flags.unwrap();

        println!("{:?}", buf0_flags);

        assert_eq!(buf0_flags.base, 1);
        assert_eq!(buf0_flags.read, 0);
        assert_eq!(buf0_flags.write, true);

        drop(buf0_w.unwrap());

        let buf0_flags = bufs.get_flags_by_index(0);
        assert_eq!(buf0_flags.is_some(), true, "no flags");
        let buf0_flags = buf0_flags.unwrap();

        println!("{:?}", buf0_flags);

        assert_eq!(buf0_flags.base, 1);
        assert_eq!(buf0_flags.read, 0);
        assert_eq!(buf0_flags.write, false);

    }

    #[test]
    fn simple_test_dropped_in_place_downgrade()
    {
        let mut bufs = RwBuffers::new(4096, 1, 2).unwrap();

        let buf0_res = bufs.allocate();
        assert_eq!(buf0_res.is_ok(), true, "{:?}", buf0_res.err().unwrap());

        let buf0 = buf0_res.unwrap();

        println!("{:?}", buf0.get_flags());

        let buf0_w = buf0.write();
        assert_eq!(buf0_w.is_ok(), true, "{:?}", buf0_w.err().unwrap());
        assert_eq!(buf0.read(), Err(RwBufferError::ReadTryAgianLater));

        drop(buf0);

        let buf0_rd = buf0_w.unwrap().downgrade();
        assert_eq!(buf0_rd.is_ok(), true, "{:?}", buf0_rd.err().unwrap());

        let buf0_flags = bufs.get_flags_by_index(0);
        assert_eq!(buf0_flags.is_some(), true, "no flags");
        let buf0_flags = buf0_flags.unwrap();

        println!("{:?}", buf0_flags);

        assert_eq!(buf0_flags.base, 1);
        assert_eq!(buf0_flags.read, 1);
        assert_eq!(buf0_flags.write, false);

    }

    #[test]
    fn simple_test_drop_in_place_downgrade()
    {
        let mut bufs = RwBuffers::new(4096, 1, 2).unwrap();

        let buf0 = bufs.allocate_in_place();

        println!("{:?}", buf0.get_flags());

        let buf0_w = buf0.write();
        assert_eq!(buf0_w.is_ok(), true, "{:?}", buf0_w.err().unwrap());
        assert_eq!(buf0.read(), Err(RwBufferError::ReadTryAgianLater));

        drop(buf0);

        let buf0_rd = buf0_w.unwrap().downgrade();
        assert_eq!(buf0_rd.is_ok(), true, "{:?}", buf0_rd.err().unwrap());

        let buf0_flags = bufs.get_flags_by_index(0);
        assert_eq!(buf0_flags.is_some(), false, "flags");

        let buf0_rd = buf0_rd.unwrap();
        let buf0_flags = buf0_rd.get_flags();

        println!("{:?}", buf0_flags);

        assert_eq!(buf0_flags.base, 0);
        assert_eq!(buf0_flags.read, 1);
        assert_eq!(buf0_flags.write, false);
    }

    #[test]
    fn timing_test()
    {
        let mut bufs = RwBuffers::new(4096, 1, 2).unwrap();

        for _ in 0..10
        {
            let inst = Instant::now();
            let buf0_res = bufs.allocate_in_place();
            let end = inst.elapsed();

            println!("alloc: {:?}", end);
            drop(buf0_res);
        }

        let buf0_res = bufs.allocate();
        assert_eq!(buf0_res.is_ok(), true, "{:?}", buf0_res.err().unwrap());

        let buf0 = buf0_res.unwrap();

        for _ in 0..10
        {
            let inst = Instant::now();
            let buf0_w = buf0.write();
            let end = inst.elapsed();

            println!("write: {:?}", end);

            assert_eq!(buf0_w.is_ok(), true, "{:?}", buf0_w.err().unwrap());
            assert_eq!(buf0.read(), Err(RwBufferError::ReadTryAgianLater));
            drop(buf0_w);
        }

        for _ in 0..10
        {
            let inst = Instant::now();
            let buf0_r = buf0.read();
            let end = inst.elapsed();

            println!("read: {:?}", end);

            assert_eq!(buf0_r.is_ok(), true, "{:?}", buf0_r.err().unwrap());
            assert_eq!(buf0.write(), Err(RwBufferError::WriteTryAgianLater));
            drop(buf0_r);
        }
    }

    #[test]
    fn simple_test_mth()
    {
        let mut bufs = RwBuffers::new(4096, 1, 3).unwrap();

        let buf0 = bufs.allocate().unwrap();

        let buf0_rd = buf0.write().unwrap().downgrade().unwrap();

        let join1=
            std::thread::spawn(move ||
                {
                    println!("{:?}", buf0_rd);

                    std::thread::sleep(Duration::from_secs(2));

                    return;
                }
            );

        let buf1_rd = buf0.read().unwrap();

        let join2=
            std::thread::spawn(move ||
                {
                    println!("{:?}", buf1_rd);

                    std::thread::sleep(Duration::from_secs(2));

                    return;
                }
            );

        let flags = buf0.get_flags();

        assert_eq!(flags.base, 2);
        assert_eq!(flags.read, 2);
        assert_eq!(flags.write, false);

        let _ = join1.join();
        let _ = join2.join();

        let flags = buf0.get_flags();

        assert_eq!(flags.base, 2);
        assert_eq!(flags.read, 0);
        assert_eq!(flags.write, false);
    }

    #[test]
    fn test_try_into_read()
    {
        let mut bufs = RwBuffers::new(4096, 1, 2).unwrap();

        let buf0 = bufs.allocate_in_place();

        println!("{:?}", buf0.get_flags());

        let buf0_w = buf0.write();
        assert_eq!(buf0_w.is_ok(), true, "{:?}", buf0_w.err().unwrap());
        assert_eq!(buf0.read(), Err(RwBufferError::ReadTryAgianLater));

        drop(buf0);

        let buf0_rd = buf0_w.unwrap().downgrade();
        assert_eq!(buf0_rd.is_ok(), true, "{:?}", buf0_rd.err().unwrap());

        let buf0_flags = bufs.get_flags_by_index(0);
        assert_eq!(buf0_flags.is_some(), false, "flags");

        let buf0_rd = buf0_rd.unwrap();
        let buf0_flags = buf0_rd.get_flags();

        println!("{:?}", buf0_flags);

        assert_eq!(buf0_flags.base, 0);
        assert_eq!(buf0_flags.read, 1);
        assert_eq!(buf0_flags.write, false);

        let inst = Instant::now();
        let ve = buf0_rd.try_inner();
        let end = inst.elapsed();

        println!("try inner: {:?}", end);
        assert_eq!(ve.is_ok(), true);


    }

    #[tokio::test]
    async fn test_multithreading()
    {

        let mut bufs = RwBuffers::new(4096, 1, 3).unwrap();

        let buf0 = bufs.allocate().unwrap();

        let mut buf0_write = buf0.write().unwrap();
        
        buf0_write.as_mut_slice()[0] = 5;
        buf0_write.as_mut_slice()[1] = 4;

        println!("{}", buf0_write[0]);

        let buf0_r = buf0_write.downgrade().unwrap();

        let join1=
            tokio::task::spawn(async move
                {
                    println!("thread[1]:{}", buf0_r[0]);

                    tokio::time::sleep(Duration::from_millis(200)).await;

                    return;
                }
            );

        let buf0_r = buf0.read().unwrap();

        // drop base
        drop(buf0);

        let join2=
            tokio::task::spawn(async move
                {
                    println!("thread[2]: {}", buf0_r[0]);
                    println!("thread[2]: {}", buf0_r[1]);

                    tokio::time::sleep(Duration::from_millis(200)).await;

                    return;
                }
            );

        let _ = join1.await;
        let _ = join2.await;

        return;
    }
}