Chris M Thomasson

/*
Word-Based Portable Proxy Garbage Collector
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
#include <relacy/relacy_std.hpp>
#include <cstdio>
#include <cstddef>


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


struct node
{
    std::atomic<node*> m_next;
    VAR_T(node*) m_defer_next;
};


template<size_t T_defer_limit>
class proxy
{
    static void prv_destroy(node* n)
    {
        while (n)
        {
            node* next = VAR(n->m_defer_next);
            delete n;
            n = next;
        }
    }


public:
    class collector
    {
        friend class proxy;


    private:
        std::atomic<node*> m_defer;
        std::atomic<unsigned int> m_defer_count;
        std::atomic<unsigned int> m_count;


    public:
        collector()
        :   m_defer(NULL),
            m_defer_count(0),
            m_count(0)
        {

        }


        ~collector()
        {
            prv_destroy(m_defer.load(std::memory_order_relaxed));
        }
    };


private:
    std::atomic<unsigned int> m_current;
    std::atomic<bool> m_quiesce;
    VAR_T(node*) m_defer;
    collector m_collectors[2];


public:
    proxy()
    :   m_current(0),
        m_quiesce(false),
        m_defer(NULL)
    {

    }

    ~proxy()
    {
        prv_destroy(VAR(m_defer));
    }


private:
    void prv_quiesce_begin()
    {
        // Try to begin the quiescence process.
        if (! m_quiesce.exchange(true, std::memory_order_acquire))
        {
            // advance the current collector and grab the old one.
            unsigned int old = m_current.load(std::memory_order_relaxed) & 0xFU;
            old = m_current.exchange((old + 1) & 1, std::memory_order_acq_rel);
            collector& c = m_collectors[old & 0xFU];

            // decode reference count.
            unsigned int refs = old & 0xFFFFFFF0U;

            // verify reference count and previous collector index.
            RL_ASSERT(! (refs & 0x10U) && (old & 0xFU) == (&c - m_collectors));

            // increment and generate an odd reference count.
            if (c.m_count.fetch_add(refs + 0x10U, std::memory_order_release) == -refs)
            {
                // odd reference count and drop-to-zero condition detected!
                prv_quiesce_complete(c);
            }
        }
    }


    void prv_quiesce_complete(collector& c)
    {
        // the collector `c' is now in a quiescent state! :^)
        std::atomic_thread_fence(std::memory_order_acquire);

        // maintain the back link and obtain "fresh" objects from
        // this collection.
        node* n = VAR(m_defer);
        VAR(m_defer) = c.m_defer.load(std::memory_order_relaxed);
        c.m_defer.store(0, std::memory_order_relaxed);

        // verify and reset the reference count.
        RL_ASSERT(c.m_count.load(std::memory_order_relaxed) == 0x10U);
        c.m_count.store(0, std::memory_order_relaxed);
        c.m_defer_count.store(0, std::memory_order_relaxed);

        // release the quiesce lock.
        m_quiesce.store(false, std::memory_order_release);

        // destroy nodes.
        prv_destroy(n);
    }


public:
    collector& acquire()
    {
        // increment the master count _and_ obtain current collector.
        unsigned int current =
            m_current.fetch_add(0x20U, std::memory_order_acquire);

        // decode the collector index.
        return m_collectors[current & 0xFU];
    }


    void release(collector& c)
    {
        // decrement the collector.
        unsigned int count =
            c.m_count.fetch_sub(0x20U, std::memory_order_release);

        // check for the completion of the quiescence process.
        if ((count & 0xFFFFFFF0U) == 0x30U)
        {
            // odd reference count and drop-to-zero condition detected!
            prv_quiesce_complete(c);
        }
    }


    collector& sync(collector& c)
    {
        // check if the `c' is in the middle of a quiescence process.
        if (c.m_count.load(std::memory_order_relaxed) & 0x10U)
        {
            // drop `c' and get the next collector.
            release(c);

            return acquire();
        }

        return c;
    }


    void collect()
    {
        prv_quiesce_begin();
    }


    void collect(collector& c, node* n)
    {
        if (! n) return;

        // link node into the defer list.
        node* prev = c.m_defer.exchange(n, std::memory_order_relaxed);
        VAR(n->m_defer_next) = prev;

        // bump the defer count and begin quiescence process if over
        // the limit.
        unsigned int count =
            c.m_defer_count.fetch_add(1, std::memory_order_relaxed) + 1;

        if (count >= (T_defer_limit / 2))
        {
            prv_quiesce_begin();
        }
    }
};


// you're basic lock-free stack...
// well, minus ABA counter and DWCAS of course!  ;^)
class stack
{
    std::atomic<node*> m_head;


public:
    stack()
    :   m_head(NULL)
    {

    }


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

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

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


    node* flush()
    {
        return m_head.exchange(NULL, std::memory_order_acquire);
    }


    node* get_head()
    {
        return m_head.load(std::memory_order_acquire);
    }


    node* pop()
    {
        node* head = m_head.load(std::memory_order_acquire);
        node* xchg;

        do
        {
            if (! head) return NULL;

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

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

        return head;
    }
};


#define ITERS 15
#define DEFER 6
#define WRITERS 2
#define READERS 3
#define REAPERS 1
#define THREADS (WRITERS + READERS + REAPERS)


struct proxy_test
:   rl::test_suite<proxy_test, THREADS>
{
    typedef proxy<DEFER> proxy_type;


    proxy_type g_proxy;
    stack g_stack;
    unsigned g_writers; // for proper test termination only.


    void before()
    {
        g_writers = WRITERS;
    }


    void thread(unsigned int tidx)
    {
        rl::backoff b;

        if (tidx < READERS)
        {
            proxy_type::collector* c = &g_proxy.acquire();

            // readers.
            while (g_writers)
            {

                node* n = g_stack.get_head();

                while (n)
                {
                    node* next = n->m_next.load(std::memory_order_relaxed);
                    n = next;
                }

                c = &g_proxy.sync(*c);

                b.yield($);
            }

            g_proxy.release(*c);
        }

        else if (tidx < WRITERS + READERS)
        {
            // writers.
            unsigned count = 0;

            for (unsigned int i = 0; i < ITERS; ++i)
            {
                g_stack.push(new node);

                if (! (i % 2))
                {
                    proxy_type::collector& c = g_proxy.acquire();
                    g_proxy.collect(c, g_stack.pop());
                    g_proxy.release(c);
                    b.yield($);
                }
            }

            for (unsigned int i = count; i < ITERS; ++i)
            {
                proxy_type::collector& c = g_proxy.acquire();
                g_proxy.collect(c, g_stack.pop());
                g_proxy.release(c);
            }

            --g_writers;
        }

        else if (tidx < WRITERS + READERS + REAPERS)
        {
            // reapers.
            while (g_writers)
            {
                g_proxy.collect();
                b.yield($);
            }
        }
    }
};


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

        p.iteration_count = 10000000;
        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<proxy_test>(p);
    }

    return 0;
}