Waitset for Non-Blocking Algorithm's by Chris Thomasson

/*
Portable Waitset for Non-Blocking Algorithm's
Copyright (C) 2010  Christopher Michael Thomasson

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/





//#define RL_DEBUGBREAK_ON_ASSERT
//#define RL_MSVC_OUTPUT
//#define RL_FORCE_SEQ_CST
#define RL_GC
#include <relacy/relacy_std.hpp>
#include <cstdio>
#include <cstddef>
#include <climits>




#if ! defined (NDEBUG)
#   define DBG_PRINTF(e) std::printf e
#else
#   define DBG_PRINTF(e) ((void)0)
#endif




#define MEMBAR_SEQ_CST() std::atomic_thread_fence(std::memory_order_seq_cst)
#define MEMBAR_CONSUME() std::atomic_thread_fence(std::memory_order_consume)
#define MEMBAR_ACQUIRE() std::atomic_thread_fence(std::memory_order_acquire)
#define MEMBAR_ACQ_REL() std::atomic_thread_fence(std::memory_order_acq_rel)
#define MEMBAR_RELEASE() std::atomic_thread_fence(std::memory_order_release)




class waitset
{
    std::mutex m_mutex;
    std::condition_variable m_cond;
    std::atomic<bool> m_waitbit;
    VAR_T(unsigned) m_waiters;



public:
    waitset()
    :   m_waitbit(false),
        m_waiters(0)
    {

    }


    ~waitset()
    {
        bool waitbit = m_waitbit.load(std::memory_order_relaxed);

        unsigned waiters = VAR(m_waiters);

        RL_ASSERT(! waitbit && ! waiters);
    }



private:
    void prv_signal(bool waitbit, bool broadcast)
    {
        if (! waitbit) return;

        m_mutex.lock($);

        unsigned waiters = VAR(m_waiters);

        if (waiters < 2 || broadcast)
        {
            m_waitbit.store(false, std::memory_order_relaxed);
        }

        m_mutex.unlock($);

        if (waiters)
        {
            if (! broadcast)
            {
                m_cond.notify_one($);
            }

            else
            {
                m_cond.notify_all($);
            }
        }
    }



public:
    void wait_begin()
    {
        m_mutex.lock($);

        m_waitbit.store(true, std::memory_order_relaxed);

        MEMBAR_SEQ_CST();
    }


    bool wait_try_begin()
    {
        if (! m_mutex.try_lock($)) return false;

        m_waitbit.store(true, std::memory_order_relaxed);

        MEMBAR_SEQ_CST();

        return true;
    }


    void wait_cancel()
    {
        unsigned waiters = VAR(m_waiters);

        if (! waiters)
        {
            m_waitbit.store(false, std::memory_order_relaxed);
        }

        m_mutex.unlock($);
    }


    void wait_commit()
    {
        MEMBAR_SEQ_CST();

        ++VAR(m_waiters);

        m_cond.wait(m_mutex, $);

        if (! --VAR(m_waiters))
        {
            m_waitbit.store(false, std::memory_order_relaxed);
        }

        m_mutex.unlock($);
    }



public:
    void signal()
    {
        MEMBAR_SEQ_CST();

        bool waitbit = m_waitbit.load(std::memory_order_relaxed);

        prv_signal(waitbit, false);
    }


    void broadcast()
    {
        MEMBAR_SEQ_CST();

        bool waitbit = m_waitbit.load(std::memory_order_relaxed);

        prv_signal(waitbit, true);
    }
};




struct node
{
    VAR_T(int) m_state;
    std::atomic<node*> m_next;
    node(int state) : m_state(state) {}
};




// I have the garbage collector turned on in Relacy.
// That's why this non-blocking stack algorihtm works...
// ;^)
class nbstack
{
    std::atomic<node*> m_head;



public:
    nbstack()
    :   m_head(NULL)
    {

    }


    ~nbstack()
    {
        node* head = m_head.load(std::memory_order_relaxed);

        RL_ASSERT(! head);
    }



public:
    void push(node* n)
    {
        node* head = m_head.load(std::memory_order_relaxed);

        do
        {
            n->m_next.store(head, std::memory_order_relaxed);

            MEMBAR_RELEASE();
        }

        while (! m_head.compare_exchange_weak(
                            head,
                            n,
                            std::memory_order_relaxed));
    }


    node* try_flush()
    {
        node* head = m_head.exchange(NULL, std::memory_order_relaxed);

        if (head) MEMBAR_CONSUME();

        return head;
    }


    node* get_head()
    {
        node* head = m_head.load(std::memory_order_relaxed);

        if (head) MEMBAR_CONSUME();

        return head;
    }


    node* try_pop()
    {
        node* head = m_head.load(std::memory_order_relaxed);
        node* xchg;

        do
        {
            if (! head) return NULL;

            MEMBAR_CONSUME();

            xchg = head->m_next.load(std::memory_order_relaxed);
        }

        while (! m_head.compare_exchange_weak(
                            head,
                            xchg,
                            std::memory_order_relaxed));

        return head;
    }
};




#define ITERS 8
#define PRODUCERS 2
#define CONSUMERS PRODUCERS
#define THREADS (PRODUCERS + CONSUMERS)


struct waitset_test
:   rl::test_suite<waitset_test, THREADS>
{
    nbstack g_nbstack;
    waitset g_waitset;



    void produce(node* n)
    {
        g_nbstack.push(n);

        g_waitset.signal();
    }



    node* consume()
    {
        node* n;

        while (! (n = g_nbstack.try_pop()))
        {
            g_waitset.wait_begin();

            if ((n = g_nbstack.try_pop()))
            {
                g_waitset.wait_cancel();

                break;
            }

            g_waitset.wait_commit();
        }

        return n;
    }



    void thread(unsigned int tidx)
    {
        if (tidx < PRODUCERS)
        {
            // producers
            DBG_PRINTF(("(%u):producer:enter\n", tidx));

            for (unsigned i = 0; i < ITERS; ++i)
            {
                int state = (i + 1) << tidx;

                node* n = new node(state);

                DBG_PRINTF(("(%u):producer:produced(%p)(%d)\n",
                            tidx, (void*)n, state));

                produce(n);
            }

            DBG_PRINTF(("(%u):producer:leave\n", tidx));
        }

        else
        {
            // consumers
            DBG_PRINTF(("(%u):consumer:enter\n", tidx));

            for (unsigned i = 0; i < ITERS; ++i)
            {
                node* n = consume();

                int state = VAR(n->m_state);

                DBG_PRINTF(("(%u):consumer:consumed(%p)(%d)\n",
                            tidx, (void*)n, state));

                // delete n; we are in a GC... No need to delete nodes!  ;^)
            }

            DBG_PRINTF(("(%u):consumer:leave\n", tidx));
        }
    }
};








int main()
{
    {
        rl::test_params p;

        p.iteration_count = 30000000;
        //p.execution_depth_limit = 10000;
        //p.search_type = rl::sched_bound;
        //p.search_type = rl::fair_full_search_scheduler_type;
        //p.search_type = rl::fair_context_bound_scheduler_type;
        rl::simulate<waitset_test>(p);
    }

    std::getchar();

    return 0;
}