[boost] [interprocess] support for a new synchronisation mechanism

Subject: [boost] [interprocess] support for a new synchronisation mechanism
From: Dudi Likvornik (likdudi_at_[hidden])
Date: 2019-03-06 17:12:37

Hi,

My name is David, I would like to suggest a new synchronisation mechanism
based on linux futex.

as far as I saw, its not being supported as of now with in the interprocess
communication library.

One suggestion is to support a mutex, here is an example, possible partial
implementation, based upon the following paper:

http://dept-info.labri.fr/~denis/Enseignement/2008-IR/Articles/01-futex.pdf

Mutex::Mutex(): m_word(UNLOCK) {}

    void Mutex::lock()

    {

        int lockState = UNLOCK;

        if(std::atomic_compare_exchange_strong(&m_word, &lockState,
static_cast<int>(LOCK_SINGLE)) == false)

        {

            if(lockState != LOCK_MANY)

                lockState = std::atomic_exchange(&m_word,
static_cast<int>(LOCK_MANY));

            while(lockState != UNLOCK)

            {

#if defined(__linux)

                syscall(SYS_futex, &m_word, FUTEX_WAIT, LOCK_MANY, nullptr,
nullptr, 0);

#else

                throw Exception(__CORE_SOURCE, "wait is not being supported
by current platform");

#endif

                lockState = std::atomic_exchange(&m_word,
static_cast<int>(LOCK_MANY));

            }

        }

    }

    void Mutex::unlock()

    {

        if(std::atomic_fetch_sub(&m_word, 1) == LOCK_SINGLE)

            return;

        m_word = UNLOCK;

#if defined(__linux)

        syscall(SYS_futex, &m_word, FUTEX_WAKE, 1, nullptr, nullptr, 0);

#else

        throw Exception(__CORE_SOURCE, "wake is not being supported by
current platform");

#endif

    }

    bool Mutex::try_lock()

    {

        int lockState = UNLOCK;

        return std::atomic_compare_exchange_strong(&m_word, &lockState,
static_cast<int>(LOCK_SINGLE));

    }

—————————————————————————————————————————————————————————

And a conditional variable:

Condition::Condition(): m_command(SLEEP), m_waitersCount(0){}

    void Condition::signal(Command command)

    {

        m_wakeLock.lock();

        if(m_waitersCount == 0)

        {

            m_wakeLock.unlock();

            return;

        }

        m_command.store(command, std::memory_order_relaxed);

#if defined(__linux)

        syscall(SYS_futex, &m_command, FUTEX_WAKE, command == NOTIFY_ONE ?
1 : std::numeric_limits<int>::max(), nullptr, nullptr, 0);

#else

        throw Exception(__CORE_SOURCE, "wake is not being supported by
current platform");

#endif

        while(m_command.load(std::memory_order_relaxed) != SLEEP &&
m_waitersCount.load(std::memory_order_relaxed) != 0){}

        m_wakeLock.unlock();

    }

    void Condition::wait(std::unique_lock<core::Mutex> &lock)

    {

        m_waitersCount++;

        lock.unlock();

        while(true)

        {

            while(m_command.load(std::memory_order_relaxed) == SLEEP)

#if defined(__linux)

                syscall(SYS_futex, &m_command, FUTEX_WAIT, SLEEP, nullptr,
nullptr, 0);

#else

                throw Exception(__CORE_SOURCE, "wait is not being supported
by current platform");

#endif

            int fromCommand_one = NOTIFY_ONE;

            if(m_command.compare_exchange_strong(fromCommand_one,
static_cast<int>(SLEEP), std::memory_order_relaxed))

            {

                m_waitersCount--;

                lock.lock();

                return;

            }

            else if(m_command.load(std::memory_order_relaxed) == NOTIFY_ALL)

            {

                if(m_waitersCount.fetch_sub(1, std::memory_order_relaxed)
== 1)

                    m_command.store(SLEEP, std::memory_order_relaxed);

                lock.lock();

                return;

            }

        }

    }

    ConditionVariable::ConditionVariable() {}

    void ConditionVariable::wait(std::unique_lock<core::Mutex> &lock)

    {

        m_condition.wait(lock);

    }

    void ConditionVariable::notify_one()

    {

        m_condition.signal(Condition::NOTIFY_ONE);

    }

    void ConditionVariable::notify_all()

    {

        m_condition.signal(Condition::NOTIFY_ALL);

    }

class ConditionVariable

    {

    public:

        ConditionVariable();

        ConditionVariable(const ConditionVariable&) = delete;

        ConditionVariable& operator=(const ConditionVariable&) = delete;

        void wait(std::unique_lock<Mutex>& lock);

        template<typename Predict>

        void wait(std::unique_lock<Mutex>& lock, const Predict& predict)

        {

            while(predict() == false)

                m_condition.wait(lock);

        }

        void notify_one();

        void notify_all();

    private:

        Condition m_condition;

    };

Your thoughts of the idea, if any.

p.s.

Please be soft, its my first post ever :).