shared_priority_mutex.hpp

Go to the documentation of this file.

 
#ifndef BOOST_FAIRNESS_SHARED_PRIORITY_MUTEX_HPP
#define BOOST_FAIRNESS_SHARED_PRIORITY_MUTEX_HPP
#include <array>
#include <atomic>
#include <boost/fairness/priority_t.hpp>
#include <boost/fairness/spinlock_priority_mutex.hpp>
#include <boost/fairness/detail/wait_ops.hpp>
 
namespace boost::fairness{
 
    #define WAIT 0
    #define PROCEED 1
 
    template<size_t N = 1>
    requires (N >= 1 && N <= BOOST_FAIRNESS_MAXIMUM_PRIORITY)
    class shared_priority_mutex{
 
        using Thread_cnt_t = uint32_t;
 
        struct threadPriority{
            Thread_cnt_t writers_waiting{};
            Thread_cnt_t readers_waiting{};
        };
 
        template<size_t S = 1>
        requires (S >= 1 && S <= BOOST_FAIRNESS_MAXIMUM_PRIORITY)
        struct waitingFlags{
            std::array<std::atomic<uint32_t>, S> writers{};
            std::array<std::atomic<uint32_t>, S> readers{};
        };
 
        public:
 
        shared_priority_mutex() = default;
 
        shared_priority_mutex(const shared_priority_mutex&) = delete;
 
        shared_priority_mutex& operator=(const shared_priority_mutex&) = delete;
 
        shared_priority_mutex(shared_priority_mutex&&) = delete;
 
        shared_priority_mutex& operator=(shared_priority_mutex&&) = delete;
 
        ~shared_priority_mutex() = default;
 
        void lock(Priority_t const priority = 0){
 
            internalMutex_.lock(priority);
 
            ++priorities_[priority].writers_waiting;
 
            ++totalWritersWaiting_;
 
            for (;;)
            {
 
                if (
                    totalCurrentReaders_ == 0 &&
                    !lockOwned_ &&
                    find_first_priority_() >= priority
                ){
                    --priorities_[priority].writers_waiting;
 
                    --totalWritersWaiting_;
 
                    lockOwned_ = true;
 
                    internalMutex_.unlock();
 
                    return;
                }
 
                internalMutex_.unlock();
 
                detail::wait(waitingFlags_.writers[priority], WAIT);
 
                internalMutex_.lock(priority);
 
            }
        }
 
        void unlock(){
 
            Priority_t p;
 
            internalMutex_.lock();
 
            lockOwned_ = false;
 
            p = find_first_priority_();
 
            if (p == BOOST_FAIRNESS_MAXIMUM_PRIORITY)
            {
                internalMutex_.unlock();
 
                return;
            }
 
            if (totalWritersWaiting_ == 0)
            {
                allow_all_readers_();
 
                internalMutex_.unlock();
 
                notify_all_readers_();
 
                return;
            }
 
            reset_(p);
 
            internalMutex_.unlock();
 
            notify_priority_(p);
 
        }
 
        [[nodiscard]] bool try_lock(Priority_t const priority = 0){
 
            internalMutex_.lock(priority);
 
            if (lockOwned_ ||
            totalCurrentReaders_ > 0 ||
            find_first_priority_() < priority)
            {
 
                internalMutex_.unlock();
 
                return false;
            }
 
            lockOwned_ = true;
 
            internalMutex_.unlock();
 
            return true;
 
        }
 
        void lock_shared(Priority_t priority = 0){
 
            internalMutex_.lock(priority);
 
            ++priorities_[priority].readers_waiting;
 
            for(;;){
 
                if (!lockOwned_ && find_first_priority_with_writers_() >= priority)
                {
                    ++totalCurrentReaders_;
 
                    --priorities_[priority].readers_waiting;
 
                    internalMutex_.unlock();
 
                    return;
                }
 
                internalMutex_.unlock();
 
                detail::wait(waitingFlags_.readers[priority], WAIT);
 
                internalMutex_.lock(priority);
 
            }
        }
 
        void unlock_shared(){
 
            Priority_t p;
 
            internalMutex_.lock();
 
            --totalCurrentReaders_;
 
            p = find_first_priority_();
 
            if (p == BOOST_FAIRNESS_MAXIMUM_PRIORITY)
            {
                internalMutex_.unlock();
 
                return;
            }
 
            if (totalWritersWaiting_ == 0)
            {
                allow_all_readers_();
 
                internalMutex_.unlock();
 
                notify_all_readers_();
 
                return;
            }
 
            if (totalCurrentReaders_ == 0)
            {
                reset_(p);
 
                internalMutex_.unlock();
 
                notify_priority_(p);
 
                return;
            }
 
            internalMutex_.unlock();
 
        }
 
        [[nodiscard]] bool try_lock_shared(Priority_t priority = 0){
 
            internalMutex_.lock(priority);
 
            if (lockOwned_ || find_first_priority_with_writers_() < priority){
 
                internalMutex_.unlock();
 
                return false;
            }
 
            ++totalCurrentReaders_;
 
            internalMutex_.unlock();
 
            return true;
            
        }
 
        private:
 
        alignas(BOOST_FAIRNESS_HARDWARE_DESTRUCTIVE_SIZE) spinlock_priority_mutex<N> internalMutex_;
        alignas(BOOST_FAIRNESS_HARDWARE_DESTRUCTIVE_SIZE) waitingFlags<N> waitingFlags_;
        std::array<threadPriority, N> priorities_{};
        Thread_cnt_t totalCurrentReaders_{};
        Thread_cnt_t totalWritersWaiting_{};
        bool lockOwned_{};
 
        Priority_t find_first_priority_() const {
            for (Priority_t i = 0; i <  N; ++i){
                if (priorities_[i].writers_waiting+priorities_[i].readers_waiting > 0)
                    return i;
            }
            return BOOST_FAIRNESS_MAXIMUM_PRIORITY;
        }
 
        Priority_t find_first_priority_with_writers_() const {
            for (Priority_t i = 0; i <  N; ++i){
                if (priorities_[i].writers_waiting > 0)
                    return i;
            }
            return BOOST_FAIRNESS_MAXIMUM_PRIORITY;
        }
 
        void notify_all_readers_(){
            for (Priority_t i = 0; i < N; ++i){
                waitingFlags_.readers[i].store(PROCEED);
                detail::notify_all(waitingFlags_.readers[i]);
            }
        }
 
        void notify_priority_(Priority_t const p){
            detail::notify_one(waitingFlags_.writers[p]);
            detail::notify_all(waitingFlags_.readers[p]);
        }
 
        void reset_(Priority_t const p){
            for (Priority_t i = 0; i <  N; ++i){
                waitingFlags_.writers[i].store(WAIT);
                waitingFlags_.readers[i].store(WAIT);
            }
            waitingFlags_.writers[p].store(PROCEED);
            waitingFlags_.readers[p].store(PROCEED);
        }
 
        void allow_all_readers_(){
            for (Priority_t i = 0; i < N; ++i){
                waitingFlags_.readers[i].store(PROCEED);
                waitingFlags_.writers[i].store(WAIT);
            }
        }
    };
 
    #undef WAIT
    #undef PROCEED
 
}
#endif // BOOST_FAIRNESS_SHARED_PRIORITY_MUTEX_HPP