instance_copy_on_write/

lib.rs

/*-
 * instance-copy-on-write - a synchronization primitive based on copy-on-write.
 * 
 * Copyright (C) 2025 Aleksandr Morozov alex@4neko.org
 * 
 * The instance-copy-on-write crate can be redistributed and/or modified
 * under the terms of either of the following licenses:
 *
 *   1. The MIT License (MIT)
 */

/*!
 An experimental thread access synchronization primitive 
 which is based on CoW Copy-on-Write and also exclusive lock.
 
 By default an `RwLock` based implementation is used, because the 
 `atomics` based implementation is still not ready. It can be activated 
 by enabling the feature prefer_mutex.

 **feature = prefer_mutex**
 
 The `ICoWRead<DATA>` instance with inner value `DATA` obtained from the 
 `ICoW<DATA>` is valid even after the inner value was copied and copy was 
 commited back. All `readers` which will will read the `ICoW` instance after 
 the `commit` will receive a `ICoWRead<DATA>` with updated `DATA`.

 Exclusive copying prevents `readers` from reading while simple `non-exclusive` 
 copy allows to access the inner data for reading and further copying. But, the 
 `non-exclusive` copies are not in sync, so each copy is uniq. The `sequential` 
 commit of copied data is not yet supported.

 The experimental `cow_refcell.rs` was added to support single-thread mode
 pure CoW model which is not really usefull, but this is a path to the future
 `async` implementation.

 *  ```ignore

    let val = 
        ICoW::new(TestStruct::new(1, 2));

    // read only 
    let read1 = val.read();
    //..
    drop(read1);

    // ...

    // copy on write NON-exclusively, read is possible
    let mut transaction = val.clone_copy();

    transaction.start = 5;
    transaction.stop = 6;

    // update, after calling this function, all reades who 
    // read before will still see old version.
    // all reades after, new
    transaction.commit();
    //or
    drop(transaction); // to drop changes


    // exclusive lock, read is also not possible

    let res = val.clone_exclusivly();
    // ..
    // commit changes releasing lock
    commit(val); //or
    drop(val); // to drop changes and release lock

    ```
 */

//extern crate thread_local;

 /// Type of the sync code.
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
 pub enum ICoWLockTypes
 {
    /// Based on atomic and backoff.
    Atomic,

    /// based on the OS RwLock
    RwLock,
 }

impl fmt::Display for ICoWLockTypes
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result
    {
        match self
        {
            Self::Atomic => 
                write!(f, "atomic"),
            Self::RwLock => 
                write!(f, "rwlock"),
        }
    }
}

/// Errors which may be returned.
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub enum ICoWError
{
    /// An attempt to write to the instance using non exclusive copy-on-write operation
    /// while the exclusive is still going.
    ExclusiveLockPending,

    /// Is issued if "exponential backoff has completed and blocking the thread is advised".
    WouldBlock,
}

impl fmt::Display for ICoWError
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        match self
        {
            Self::ExclusiveLockPending => 
                write!(f, "already exlcusivly locked"),
            Self::WouldBlock => 
                write!(f, "cannot perform try operation due to blocking"),
        }
    }
}

extern crate crossbeam_utils;
extern crate crossbeam_deque;

/// A lightweight CoW implementation.
#[cfg(all(target_has_atomic = "32", feature = "prefer_mutex"))]
pub mod cow;

/// A RwLock based copy-on-write.
#[cfg(any(not(target_has_atomic = "32"), not(feature = "prefer_mutex")))]
pub mod cow_mutex;

#[cfg(any(not(target_has_atomic = "32"), not(feature = "prefer_mutex")))]
pub mod cow_refcell;

use std::fmt;

#[cfg(all(target_has_atomic = "32", feature = "prefer_mutex"))]
pub use cow::{ICoW, ICoWRead, ICoWCopy, ICoWLock, ICoWWeak, ICoWWeakRead};

#[cfg(any(not(target_has_atomic = "32"), not(feature = "prefer_mutex")))]
pub use cow_mutex::{ICoW, ICoWRead, ICoWCopy, ICoWLock, ICoWWeak, ICoWWeakRead};

#[cfg(all(not(target_has_atomic = "32"), feature = "prefer_mutex"))]
compile_error!("target CPU does not support 'target_has_atomic'");

pub trait ICowType
{
    fn get_lock_type() -> ICoWLockTypes;
}

#[cfg(test)]
mod test
{
    use std::{sync::{LazyLock, mpsc}, thread, time::{Duration, Instant}};

    use super::*;




    #[test]
    fn test_3()
    {
        #[derive(Debug, Clone)]
        struct Test { s: String }

        let icow = ICoW::new(Test{ s: "test".into() });

        for _ in 0..10
        {
            let s = Instant::now();

            let read0 = icow.read();

            let e = s.elapsed();

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

    }

    #[test]
    fn test_33()
    {
        #[derive(Debug, Clone)]
        struct Test { s: String };

        let icow = ICoW::new(Test{ s: "test".into() });

        let write_ex = icow.try_clone_copy_exclusivly().unwrap();

        {
        let write_ex_err = icow.try_clone_copy_exclusivly();
        assert_eq!(write_ex_err.is_none(), true);
        }

        {
        let write_ex_err = icow.try_clone_copy();
        assert_eq!(write_ex_err.is_none(), true);
        }
        {
        let read1 = icow.try_read();
        assert_eq!(read1.is_none(), true);
        }

        drop(write_ex);

        drop(icow);

    }

    #[test]
    fn test_4()
    {
        #[derive(Debug, Clone)]
        struct Test { s: u32 }

        let icow = ICoW::new(Test{ s: 1 });

        let read0 = icow.read();
        let read1 = icow.read();

        let mut excl_write = icow.try_clone_copy_exclusivly().unwrap();

        excl_write.s = 5;


        excl_write.commit();

        assert_eq!(read0.s, 1);

        let read3 = icow.read();
        assert_eq!(read3.s, 5);

        let mut writing = icow.try_clone_copy().unwrap();
        writing.s = 4;

        writing.commit().unwrap();

        assert_eq!(read0.s, 1);
        assert_eq!(read3.s, 5);

        let read4 = icow.read();
        assert_eq!(read4.s, 4);
    }

    #[test]
    fn test_parking_lot_reader_wakeup()
    {
        #[derive(Debug, Clone)]
        struct Test { s: u32 }

        let icow = ICoW::new(Test{ s: 1 });

        let read0 = icow.read();
        let read2 = icow.read();

        let c_icow = icow.clone();
        let (se, rc) = mpsc::channel::<()>();
        let handler0 = 
            std::thread::spawn(move || 
                {
                    let mut lock0 = c_icow.try_clone_copy_exclusivly().unwrap();

                    se.send(()).unwrap();

                    lock0.s = 5;

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

                    lock0.commit();
                }
            );

        rc.recv().unwrap();

        let s = Instant::now();

        let read1 = icow.read();
        
        let e = s.elapsed();

        println!("thread wait: {:?}", e);

        assert_eq!(e.as_secs(), 2);

        handler0.join().unwrap();

        drop(read0);
        drop(read1);
        drop(read2);
    }

    #[test]
    fn text_9_exclu_lock_pend()
    {
        #[derive(Debug, Clone)]
        struct Test { s: u32 }

        let icow = ICoW::new(Test{ s: 1 });

            let mut write0 = icow.clone_copy();
            write0.s = 1222;

        let c_icow = icow.clone();
        let (se, rc) = mpsc::channel::<()>();
        let handler0 = 
            std::thread::spawn(move || 
                {
                    let mut lock0 = 
                        loop 
                        {
                            let res = c_icow.try_clone_copy_exclusivly();

                            if res.is_none() == true
                            {
                                continue;
                            }
                            
                            break res.unwrap();
                        };

                    se.send(()).unwrap();

                    lock0.s = 5;

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

                    lock0.commit();
                }
            );


        rc.recv().unwrap();

        let s = Instant::now();

        let commit_res = write0.commit();
        assert_eq!(commit_res.is_err(), true);
        assert_eq!(commit_res.as_ref().err().unwrap().0, ICoWError::ExclusiveLockPending);
        write0 = commit_res.err().unwrap().1;

        write0.commit_blocking(true).unwrap();
        
        let e = s.elapsed();

        println!("thread wait: {:?}", e);

        assert_eq!(e.as_secs(), 1);

        handler0.join().unwrap();

        return;
    }

    #[test]
    fn test_5()
    {
        #[derive(Debug, Clone)]
        struct Test { s: u32 }

        let icow = ICoW::new(Test{ s: 1 });

        let read0 = icow.read();
        let read2 = icow.read();
        
        let c_icow = icow.clone();

        let (se, rc) = mpsc::channel::<()>();
        let handler0 = 
            std::thread::spawn(move || 
                {
                    let mut lock0 = c_icow.try_clone_copy_exclusivly().unwrap();

                    se.send(()).unwrap();

                    lock0.s = 5;

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

                    lock0.commit();
                }
            );

        rc.recv().unwrap();

        let s = Instant::now();

        let read1 = icow.read();
        
        let e = s.elapsed();

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

        
        assert_eq!(read1.s, 5);
        assert_eq!(read0.s, 1);

        handler0.join().unwrap();

        let weak0 = read0.weak();
        let weak1 = read1.weak();

        drop(read0);
        drop(read1);

        assert_eq!(weak0.upgrade().is_some(), true);
        assert_eq!(weak1.upgrade().is_some(), true);

        drop(read2);
        assert_eq!(weak0.upgrade().is_none(), true);
        assert_eq!(weak1.upgrade().is_some(), true);
    }

    #[test]
    fn test_5_read_lock()
    {
        #[derive(Debug, Clone)]
        struct Test { s: u32 }

        let icow = ICoW::new(Test{ s: 1 });
        
        let c_icow = icow.clone();

        let (se, rc) = mpsc::channel::<()>();
        let (se2, rc2) = mpsc::channel::<()>();
        let handler0 = 
            std::thread::spawn(move || 
                {
                    rc2.recv().unwrap();

                    let mut lock0 = 
                        loop 
                        {
                            let res = c_icow.try_clone_copy_exclusivly();

                            if res.is_none() == true
                            {
                                continue;
                            }
                            
                            break res.unwrap();
                        };

                    se.send(()).unwrap();

                    lock0.s = 5;

                    lock0.commit();
                }
            );

        let mut reads = Vec::<ICoWRead<'_, Test>>::with_capacity(20000);  
        let mut sent = false;

        let s = Instant::now();

        for i in 0..10000
        {
            let read0 = icow.read();
            if read0.s == 5
            {
                println!("break at {}", i);
                break;
            }

            reads.push(read0);

            if sent == false
            {
                se2.send(());
                sent = true;
            }
        }
        
        let e = s.elapsed();

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

        assert_eq!(reads[0].s, 1);

        handler0.join().unwrap();
    }

    #[test]
    fn test_6()
    {
        #[derive(Debug, Clone)]
        struct Test { s: u32 }

        let icow = ICoW::new(Test{ s: 1 });

        let read0 = icow.read();
        let read2 = icow.read();
        
        let c_icow = icow.clone();

        let (se, rc) = mpsc::channel::<()>();
        let handler0 = 
            std::thread::spawn(move || 
                {
                    let read2 = c_icow.read();

                    let mut lock0 = c_icow.try_clone_copy_exclusivly().unwrap();

                    se.send(()).unwrap();

                    lock0.s = 5;

                    std::thread::sleep(Duration::from_nanos(50));
                    lock0.commit();

                    let read3 = c_icow.read();

                    assert_eq!(read2.s, 1);
                    assert_eq!(read3.s, 5);
                }
            );

        rc.recv().unwrap();

        for _ in 0..100000000
        {
            let read1 = icow.read();

            if read1.s == 1
            {
                continue;
            }
            else
            {
                break;
            }
        }

        let read1 = icow.read();
        assert_eq!(read1.item.s, 5);

        handler0.join().unwrap();

        return;
    }

    #[test]
    fn test_6_drop()
    {
        #[derive(Debug, Clone)]
        struct Test { s: u32 }

        let icow = ICoW::new(Test{ s: 1 });

        let read0 = icow.read();
        let read2 = icow.read();
        
        let c_icow = icow.clone();

        let (se, rc) = mpsc::channel::<()>();
        let handler0 = 
            std::thread::spawn(move || 
                {
                    let read2 = c_icow.read();

                    let mut lock0 = c_icow.try_clone_copy_exclusivly().unwrap();

                    rc.recv().unwrap();

                    lock0.s = 5;

                    std::thread::sleep(Duration::from_millis(1500));
                    lock0.commit();

                    let read3 = c_icow.read();

                    assert_eq!(read2.s, 1);
                    assert_eq!(read3.s, 5);
                }
            );

        drop(read0);
        let r2 = read2.clone_copy_into_inner();

        se.send(()).unwrap();

        drop(icow);

        assert_eq!(r2.s, 1);

        handler0.join().unwrap();

        return;
    }


    #[test]
    fn test_7()
    {
        #[derive(Debug, Clone)]
        struct TestStruct { s: u32 }

        impl TestStruct
        {
            fn new(s: u32) -> Self
            {
                return Self{ s: s };
            }
        }

        static VAL: LazyLock<ICoW<TestStruct>> = 
            LazyLock::new(|| ICoW::new(TestStruct::new(1))); 
        

        
        
            let borrow1 = VAL.read();
            assert_eq!(borrow1.item.s, 1);

            let (mpsc_send, mpsc_rcv) = mpsc::channel::<u64>();
            let (mpsc_send2, mpsc_rcv2) = mpsc::channel::<u64>();

            let thread1 = 
                std::thread::spawn(move ||
                    {
                        for _ in 0..1000
                        {
                            let _ = mpsc_rcv2.recv();
                            let borrow1 = VAL.read();

                            let mut transaction = VAL.try_clone_copy_exclusivly().unwrap();

                            transaction.s = 5;

                            

                            std::thread::sleep(Duration::from_nanos(1001));
                            transaction.commit();
                            let borrow2 = VAL.read();
                            

                            assert_eq!(borrow1.item.s, 1);

                            assert_eq!(borrow2.item.s, 5);

                            let _ = mpsc_send.send(1);
                        }
                    }
                );
            
            

            for x in 0..1000
            {
                println!("{}", x);
                mpsc_send2.send(1).unwrap();
                let _ = mpsc_rcv.recv();

                let borrow1 = VAL.read();
                assert_eq!(borrow1.s, 5);

                let mut transaction = VAL.try_clone_copy_exclusivly().unwrap();
                transaction.s = 1;
                transaction.commit();

                let borrow1 = VAL.read();
                assert_eq!(borrow1.s, 1);
                
            }
            
            

            thread1.join().unwrap();

            
    }


    #[test]
    fn test_8()
    {
        #[derive(Debug, Clone)]
        struct Test { s: u32 }

        let icow = ICoW::new(Test{ s: 1 });

        for _ in 0..20
        {
            let read0 = icow.read();
            let read2 = icow.read();
            
            let c_icow = icow.clone();

            //let (se, rc) = mpsc::channel::<()>();
            let handler0 = 
                std::thread::spawn(move || 
                    {
                        let read2 = c_icow.read();

                        let mut lock0 = c_icow.clone_copy();

                        //se.send(()).unwrap();

                        lock0.s = 5;

                        std::thread::sleep(Duration::from_micros(1));
                        lock0.commit_blocking(false).unwrap();

                        let read3 = c_icow.read();

                        assert_eq!(read2.s, 1);
                        assert_eq!(read3.s, 5);
                    }
                );

            //rc.recv().unwrap();

            for i in 0..1000000000
            {
                let read1 = icow.read();

                if read1.s == 1
                {
                    continue;
                }
                else
                {
                    println!("{}", i);
                    break;
                }
            }

            let read1 = icow.read();
            assert_eq!(read1.s, 5);

            handler0.join().unwrap();

            let mut lock0 = icow.try_clone_copy().unwrap();
            lock0.s = 1;
            lock0.commit_blocking(false).unwrap();
        }

        return;
    }

    #[test]
    fn test_9()
    {
        #[derive(Debug, Clone)]
        struct Test { s: u32 }

        let icow = ICoW::new(Test{ s: 1 });

        let weak_reader = icow.reader();

        let weak_r0 = weak_reader.aquire().unwrap();

        assert_eq!(weak_r0.s, 1);

        drop(weak_r0);

        let mut copy1 = icow.clone_copy();
        copy1.s = 8;
        copy1.commit_blocking(true).unwrap();

        let weak_r0 = weak_reader.aquire().unwrap();

        assert_eq!(weak_r0.s, 8);

        drop(weak_r0);

        // drop icow, check that the aquire() returns None because the data and
        // icow instance have gone.
        drop(icow);

        let weak_r0 = weak_reader.aquire();
        assert_eq!(weak_r0.is_none(), true);

    }

    #[test]
    fn test_10()
    {
        #[derive(Debug, Clone)]
        struct Test { s: u32 }

        let icow = ICoW::new(Test{ s: 1 });

        let weak_reader = icow.reader();

        let (tx, rx) = mpsc::channel::<()>();
        let (tx2, rx2) = mpsc::channel::<()>();

        let thread_hndl = 
            thread::spawn(move ||
                {
                    let weak_r0 = weak_reader.aquire().unwrap();

                    assert_eq!(weak_r0.s, 1);

                    drop(weak_r0);

                    tx.send(()).unwrap();

                    // ----
                    let _ = rx2.recv().unwrap();

                    let weak_r0 = weak_reader.aquire().unwrap();

                    assert_eq!(weak_r0.item_updated(), true);
                    assert_eq!(weak_r0.s, 8);

                    drop(weak_r0);

                    tx.send(()).unwrap();
                    // ----
                    let _ = rx2.recv().unwrap();

                    // check that icow dropped
                    let weak_r0 = weak_reader.aquire();
                    assert_eq!(weak_r0.is_none(), true);
                }
            );

        let _ = rx.recv().unwrap();

        let mut copy1 = icow.clone_copy();
        copy1.s = 8;
        copy1.commit_blocking(true).unwrap();

        tx2.send(()).unwrap();
        let _ = rx.recv().unwrap();

        // drop icow, check that the aquire() returns None because the data and
        // icow instance have gone.
        drop(icow);

        tx2.send(()).unwrap();

        thread_hndl.join().unwrap();

        return;

    }
}