soa/gc/testing/gc_test.cc

/* gc_test.cc
   Jeremy Barnes, 23 February 2010
   Copyright (c) 2010 Datacratic.  All rights reserved.

   Test of the garbage collector locking.
*/

#define BOOST_TEST_MAIN
#define BOOST_TEST_DYN_LINK

#include "soa/gc/gc_lock.h"
#include "soa/gc/rcu_lock.h"
#include "jml/utils/string_functions.h"
#include "jml/utils/exc_assert.h"
#include "jml/utils/guard.h"
#include "jml/arch/atomic_ops.h"
#include "jml/arch/thread_specific.h"
#include "jml/arch/rwlock.h"
#include "jml/arch/spinlock.h"
#include "jml/arch/tick_counter.h"
#include <boost/test/unit_test.hpp>
#include <boost/bind.hpp>
#include <iostream>

#include <boost/thread.hpp>
#include <boost/thread/barrier.hpp>


using namespace ML;
using namespace Datacratic;
using namespace std;

// Defined in gc_lock.cc
namespace Datacratic {
extern int32_t gcLockStartingEpoch;
};

struct doInit {
    doInit()
    {
        rcu_init();
    }
} init;

#if 1

BOOST_AUTO_TEST_CASE ( test_gc )
{
    GcLock gc;
    gc.lockShared();

    BOOST_CHECK(gc.isLockedShared());

    bool deferred = false;

    cerr << endl << "before defer" << endl;
    gc.dump();

    gc.defer([&] () { deferred = true; memory_barrier(); });

    cerr << endl << "after defer" << endl;
    gc.dump();

    gc.unlockShared();

    cerr << endl << "after unlock shared" << endl;
    gc.dump();

    BOOST_CHECK(!gc.isLockedShared());
    BOOST_CHECK(deferred);
}

BOOST_AUTO_TEST_CASE(test_mutual_exclusion)
{
    cerr << "testing mutual exclusion" << endl;

    GcLock lock;
    volatile bool finished = false;
    volatile int numExclusive = 0;
    volatile int numShared = 0;
    int errors = 0;
    int multiShared = 0;
    uint64_t sharedIterations = 0;
    uint64_t exclusiveIterations = 0;

    auto sharedThread = [&] ()
        {
            while (!finished) {
                GcLock::SharedGuard guard(lock);
                ML::atomic_inc(numShared);

                if (numExclusive > 0) {
                    cerr << "exclusive and shared" << endl;
                    ML::atomic_inc(errors);
                }
                if (numShared > 1) {
                    ML::atomic_inc(multiShared);
                }

                ML::atomic_dec(numShared);
                ML::atomic_inc(sharedIterations);
                ML::memory_barrier();
            }
        };

    auto exclusiveThread = [&] ()
        {
            while (!finished) {
                GcLock::ExclusiveGuard guard(lock);
                ML::atomic_inc(numExclusive);

                if (numExclusive > 1) {
                    cerr << "more than one exclusive" << endl;
                    ML::atomic_inc(errors);
                }
                if (numShared > 0) {
                    cerr << "exclusive and shared" << endl;
                    ML::atomic_inc(multiShared);
                }

                ML::atomic_dec(numExclusive);
                ML::atomic_inc(exclusiveIterations);
                ML::memory_barrier();
            }
        };

    lock.getEntry();

    int nthreads = 4;

    {
        cerr << "single shared" << endl;
        sharedIterations = exclusiveIterations = multiShared = finished = 0;
        boost::thread_group tg;
        tg.create_thread(sharedThread);
        sleep(1);
        finished = true;
        tg.join_all();
        BOOST_CHECK_EQUAL(errors, 0);
        cerr << "iterations: shared " << sharedIterations
             << " exclusive " << exclusiveIterations << endl;
        cerr << "multiShared = " << multiShared << endl;
    }

    {
        cerr << "multi shared" << endl;
        sharedIterations = exclusiveIterations = multiShared = finished = 0;
        boost::thread_group tg;
        for (unsigned i = 0;  i < nthreads;  ++i)
            tg.create_thread(sharedThread);
        sleep(1);
        finished = true;
        tg.join_all();
        BOOST_CHECK_EQUAL(errors, 0);
        if (nthreads > 1)
            BOOST_CHECK_GE(multiShared, 0);
        cerr << "iterations: shared " << sharedIterations
             << " exclusive " << exclusiveIterations << endl;
        cerr << "multiShared = " << multiShared << endl;
    }

    {
        cerr << "single exclusive" << endl;
        sharedIterations = exclusiveIterations = multiShared = finished = 0;
        boost::thread_group tg;
        tg.create_thread(exclusiveThread);
        sleep(1);
        finished = true;
        tg.join_all();
        BOOST_CHECK_EQUAL(errors, 0);
        cerr << "iterations: shared " << sharedIterations
             << " exclusive " << exclusiveIterations << endl;
        cerr << "multiShared = " << multiShared << endl;
    }

    {
        cerr << "multi exclusive" << endl;
        sharedIterations = exclusiveIterations = multiShared = finished = 0;
        boost::thread_group tg;
        for (unsigned i = 0;  i < nthreads;  ++i)
            tg.create_thread(exclusiveThread);
        sleep(1);
        finished = true;
        tg.join_all();
        BOOST_CHECK_EQUAL(errors, 0);
        cerr << "iterations: shared " << sharedIterations
             << " exclusive " << exclusiveIterations << endl;
        cerr << "multiShared = " << multiShared << endl;
    }

    {
        cerr << "mixed shared and exclusive" << endl;
        sharedIterations = exclusiveIterations = multiShared = finished = 0;
        boost::thread_group tg;
        for (unsigned i = 0;  i < nthreads;  ++i)
            tg.create_thread(sharedThread);
        for (unsigned i = 0;  i < nthreads;  ++i)
            tg.create_thread(exclusiveThread);
        sleep(1);
        finished = true;
        tg.join_all();
        BOOST_CHECK_EQUAL(errors, 0);
        if (nthreads > 1)
            BOOST_CHECK_GE(multiShared, 0);
        cerr << "iterations: shared " << sharedIterations
             << " exclusive " << exclusiveIterations << endl;
        cerr << "multiShared = " << multiShared << endl;
    }

    {
        cerr << "overflow" << endl;
        gcLockStartingEpoch = 0xFFFFFFF0;
        sharedIterations = exclusiveIterations = multiShared = finished = 0;
        boost::thread_group tg;
        tg.create_thread(sharedThread);
        sleep(1);
        finished = true;
        tg.join_all();
        BOOST_CHECK_EQUAL(errors, 0);
        cerr << "iterations: shared " << sharedIterations
             << " exclusive " << exclusiveIterations << endl;
        cerr << "multiShared = " << multiShared << endl;
    }

    {
        cerr << "INT_MIN to INT_MAX" << endl;
        gcLockStartingEpoch = 0x7FFFFFF0;
        sharedIterations = exclusiveIterations = multiShared = finished = 0;
        boost::thread_group tg;
        tg.create_thread(sharedThread);
        sleep(1);
        finished = true;
        tg.join_all();
        BOOST_CHECK_EQUAL(errors, 0);
        cerr << "iterations: shared " << sharedIterations
             << " exclusive " << exclusiveIterations << endl;
        cerr << "multiShared = " << multiShared << endl;
    }

    {
        cerr << "benign overflow" << endl;
        gcLockStartingEpoch = 0xBFFFFFF0;
        sharedIterations = exclusiveIterations = multiShared = finished = 0;
        boost::thread_group tg;
        tg.create_thread(sharedThread);
        sleep(1);
        finished = true;
        tg.join_all();
        BOOST_CHECK_EQUAL(errors, 0);
        cerr << "iterations: shared " << sharedIterations
             << " exclusive " << exclusiveIterations << endl;
        cerr << "multiShared = " << multiShared << endl;
    }

}

#endif

#define USE_MALLOC 1

template<typename T>
struct Allocator {
    Allocator(int nblocks, T def = T())
        : def(def)
    {
        init(nblocks);
        highestAlloc = nallocs = ndeallocs = 0;
    }

    ~Allocator()
    {
#if ( ! USE_MALLOC )
        delete[] blocks;
        delete[] free;
#endif
    }

    T def;
    T * blocks;
    int * free;
    int nfree;
    int highestAlloc;
    int nallocs;
    int ndeallocs;
    ML::Spinlock lock;

    void init(int nblocks)
    {
#if ( ! USE_MALLOC )
        blocks = new T[nblocks];
        free = new int[nblocks];

        std::fill(blocks, blocks + nblocks, def);

        nfree = 0;
        for (int i = nblocks - 1;  i >= 0;  --i)
            free[nfree++] = i;
#endif
    }

    T * alloc()
    {
#if USE_MALLOC
        ML::atomic_inc(nallocs);
        ML::atomic_max(highestAlloc, nallocs - ndeallocs);
        return new T(def);
#else
        boost::lock_guard<ML::Spinlock> guard(lock);
        if (nfree == 0)
            throw ML::Exception("none free");
        int i = free[nfree - 1];
        highestAlloc = std::max(highestAlloc, i);
        T * result = blocks + i;
        --nfree;
        ++nallocs;
        return result;
#endif
    }

    void dealloc(T * value)
    {
        if (!value) return;
        *value = def;
#if USE_MALLOC
        delete value;
        ML::atomic_inc(ndeallocs);
        return;
#else
        boost::lock_guard<ML::Spinlock> guard(lock);
        int i = value - blocks;
        free[nfree++] = i;
        ++ndeallocs;
#endif
    }

    static void doDealloc(void * thisptr, void * blockPtr_, void * blockVar_)
    {
        int * & blockVar = *reinterpret_cast<int **>(blockVar_);
        int * blockPtr = reinterpret_cast<int *>(blockPtr_);
        ExcAssertNotEqual(blockVar, blockPtr);
        //blockVar = 0;
        //ML::memory_barrier();
        //cerr << "blockPtr = " << blockPtr << endl;
        //int * blockPtr = reinterpret_cast<int *>(block);
        reinterpret_cast<Allocator *>(thisptr)->dealloc(blockPtr);
    }

    static void doDeallocAll(void * thisptr, void * blocksPtr_, void * numBlocks_)
    {
        size_t numBlocks = reinterpret_cast<size_t>(numBlocks_);
        int ** blocksPtr = reinterpret_cast<int **>(blocksPtr_);
        Allocator * alloc = reinterpret_cast<Allocator *>(thisptr);

        for (unsigned i = 0;  i != numBlocks;  ++i) {
            if (blocksPtr[i])
                alloc->dealloc(blocksPtr[i]);
        }

        delete[] blocksPtr;
    }
};

template<typename Lock>
struct TestBase {
    TestBase(int nthreads, int nblocks, int nSpinThreads = 0)
        : finished(false),
          nthreads(nthreads),
          nblocks(nblocks),
          nSpinThreads(nSpinThreads),
          allocator(1024 * 1024, -1),
          nerrors(0),
          allBlocks(nthreads)
    {
        for (unsigned i = 0;  i < nthreads;  ++i) {
            allBlocks[i] = new int *[nblocks];
            std::fill(allBlocks[i], allBlocks[i] + nblocks, (int *)0);
        }
    }

    ~TestBase()
    {
        for (unsigned i = 0;  i < nthreads;  ++i)
            delete[] allBlocks[i];
    }

    volatile bool finished;
    int nthreads;
    int nblocks;
    int nSpinThreads;
    Allocator<int> allocator;
    Lock gc;
    uint64_t nerrors;

    /* All of the blocks are published here.  Any pointer which is read from
       here by another thread should always refer to exactly the same
       value.
    */
    vector<int **> allBlocks;

    void checkVisible(int threadNum, unsigned long long start)
    {
        // We're reading from someone else's pointers, so we need to lock here
        gc.enterCS();
        //gc.lockShared();

        for (unsigned i = 0;  i < nthreads;  ++i) {
            for (unsigned j = 0;  j < nblocks;  ++j) {
                //int * val = allBlocks[i][j];
                int * val = (int *)(rcu_dereference_sym(allBlocks[i][j]));
                if (val) {
                    int atVal = *val;
                    if (atVal != i) {
                        cerr << ML::format("%.6f thread %d: invalid value read "
                                "from thread %d block %d: %d\n",
                                (ticks() - start) / ticks_per_second, threadNum,
                                i, j, atVal);
                        ML::atomic_inc(nerrors);
                        //abort();
                    }
                }
            }
        }

        gc.exitCS();
        //gc.unlockShared();
    }

    void doReadThread(int threadNum)
    {
        gc.getEntry();
        unsigned long long start = ticks();
        while (!finished) {
            checkVisible(threadNum, start);
        }
    }

    void doSpinThread()
    {
        while (!finished) {
        }
    }

    void allocThreadDefer(int threadNum)
    {
        gc.getEntry();
        try {
            uint64_t nErrors = 0;

            int ** blocks = allBlocks[threadNum];

            while (!finished) {

                int ** oldBlocks = new int * [nblocks];

                //gc.enterCS();

                for (unsigned i = 0;  i < nblocks;  ++i) {
                    int * block = allocator.alloc();
                    if (*block != -1) {
                        cerr << "old block was allocated" << endl;
                        ++nErrors;
                    }
                    *block = threadNum;
                    ML::memory_barrier();
                    //rcu_set_pointer_sym((void **)&blocks[i], block);
                    int * oldBlock = blocks[i];
                    blocks[i] = block;
                    ML::memory_barrier();
                    oldBlocks[i] = oldBlock;
                }

                gc.defer(Allocator<int>::doDeallocAll, &allocator, oldBlocks,
                         (void *)(size_t)nblocks);

                //gc.exitCS();
            }


            int * oldBlocks[nblocks];

            for (unsigned i = 0;  i < nblocks;  ++i) {
                oldBlocks[i] = blocks[i];
                blocks[i] = 0;
            }

            gc.visibleBarrier();

            //cerr << "at end" << endl;

            for (unsigned i = 0;  i < nblocks;  ++i)
                allocator.dealloc(oldBlocks[i]);

            //cerr << "nErrors = " << nErrors << endl;
        } catch (...) {
            static ML::Spinlock lock;
            lock.acquire();
            //cerr << "threadnum " << threadNum << " inEpoch "
            //     << gc.getEntry().inEpoch << endl;
            gc.dump();
            abort();
        }
    }

    void allocThreadSync(int threadNum)
    {
        gc.getEntry();
        try {
            uint64_t nErrors = 0;

            int ** blocks = allBlocks[threadNum];
            int * oldBlocks[nblocks];

            while (!finished) {

                for (unsigned i = 0;  i < nblocks;  ++i) {
                    int * block = allocator.alloc();
                    if (*block != -1) {
                        cerr << "old block was allocated" << endl;
                        ++nErrors;
                    }
                    *block = threadNum;
                    int * oldBlock = blocks[i];
                    blocks[i] = block;
                    oldBlocks[i] = oldBlock;
                }

                ML::memory_barrier();
                gc.visibleBarrier();

                for (unsigned i = 0;  i < nblocks;  ++i)
                    if (oldBlocks[i]) *oldBlocks[i] = 1234;

                for (unsigned i = 0;  i < nblocks;  ++i)
                    if (oldBlocks[i]) allocator.dealloc(oldBlocks[i]);
            }

            for (unsigned i = 0;  i < nblocks;  ++i) {
                oldBlocks[i] = blocks[i];
                blocks[i] = 0;
            }

            gc.visibleBarrier();

            for (unsigned i = 0;  i < nblocks;  ++i)
                allocator.dealloc(oldBlocks[i]);

            //cerr << "nErrors = " << nErrors << endl;
        } catch (...) {
            static ML::Spinlock lock;
            lock.acquire();
            //cerr << "threadnum " << threadNum << " inEpoch "
            //     << gc.getEntry().inEpoch << endl;
            gc.dump();
            abort();
        }
    }

    void run(boost::function<void (int)> allocFn,
             int runTime = 1)
    {
        gc.getEntry();
        boost::thread_group tg;

        for (unsigned i = 0;  i < nthreads;  ++i)
            tg.create_thread(boost::bind<void>(&TestBase::doReadThread, this, i));

        for (unsigned i = 0;  i < nthreads;  ++i)
            tg.create_thread(boost::bind<void>(allocFn, i));

        for (unsigned i = 0;  i < nSpinThreads;  ++i)
            tg.create_thread(boost::bind<void>(&TestBase::doSpinThread, this));

        sleep(runTime);

        finished = true;

        tg.join_all();

        gc.deferBarrier();

        gc.dump();

        BOOST_CHECK_EQUAL(allocator.nallocs, allocator.ndeallocs);
        BOOST_CHECK_EQUAL(nerrors, 0);

        cerr << "allocs " << allocator.nallocs
             << " deallocs " << allocator.ndeallocs << endl;
        cerr << "highest " << allocator.highestAlloc << endl;

        cerr << "gc.currentEpoch() = " << gc.currentEpoch() << endl;
    }
};

#if 1
BOOST_AUTO_TEST_CASE ( test_gc_sync_many_threads_contention )
{
    cerr << "testing contention synchronized GcLock with many threads" << endl;

    int nthreads = 8;
    int nSpinThreads = 16;
    int nblocks = 2;

    TestBase<GcLock> test(nthreads, nblocks, nSpinThreads);
    test.run(boost::bind(&TestBase<GcLock>::allocThreadSync, &test, _1));
}
#endif

BOOST_AUTO_TEST_CASE ( test_gc_deferred_contention )
{
    cerr << "testing contended deferred GcLock" << endl;

    int nthreads = 8;
    int nSpinThreads = 0;//16;
    int nblocks = 2;

    TestBase<GcLock> test(nthreads, nblocks, nSpinThreads);
    test.run(boost::bind(&TestBase<GcLock>::allocThreadDefer, &test, _1));
}


#if 1

BOOST_AUTO_TEST_CASE ( test_gc_sync )
{
    cerr << "testing synchronized GcLock" << endl;

    int nthreads = 2;
    int nblocks = 2;

    TestBase<GcLock> test(nthreads, nblocks);
    test.run(boost::bind(&TestBase<GcLock>::allocThreadSync, &test, _1));
}

BOOST_AUTO_TEST_CASE ( test_gc_sync_many_threads )
{
    cerr << "testing synchronized GcLock with many threads" << endl;

    int nthreads = 8;
    int nblocks = 2;

    TestBase<GcLock> test(nthreads, nblocks);
    test.run(boost::bind(&TestBase<GcLock>::allocThreadSync, &test, _1));
}

BOOST_AUTO_TEST_CASE ( test_rcu_sync )
{
    cerr << "testing synchronized RCU" << endl;

    int nthreads = 2;
    int nblocks = 2;

    TestBase<RcuLock> test(nthreads, nblocks);
    test.run(boost::bind(&TestBase<RcuLock>::allocThreadSync, &test, _1));
}

BOOST_AUTO_TEST_CASE ( test_gc_deferred )
{
    cerr << "testing deferred GcLock" << endl;

    int nthreads = 2;
    int nblocks = 2;

    TestBase<GcLock> test(nthreads, nblocks);
    test.run(boost::bind(&TestBase<GcLock>::allocThreadDefer, &test, _1));
}

BOOST_AUTO_TEST_CASE ( test_rcu_deferred )
{
    cerr << "testing deferred RCU" << endl;

    int nthreads = 2;
    int nblocks = 2;

    TestBase<RcuLock> test(nthreads, nblocks);
    test.run(boost::bind(&TestBase<RcuLock>::allocThreadDefer, &test, _1));
}


struct SharedGcLockProxy : public SharedGcLock {
    static const char* name;
    SharedGcLockProxy() :
        SharedGcLock(GC_OPEN, name)
    {}
};
const char* SharedGcLockProxy::name = "gc_test.dat";

BOOST_AUTO_TEST_CASE( test_shared_lock_sync )
{
    cerr << "testing contention synchronized GcLock with shared lock" << endl;

    SharedGcLock lockGuard(GC_CREATE, SharedGcLockProxy::name);
    Call_Guard unlink_guard([&] { lockGuard.unlink(); });

    int nthreads = 8;
    int nSpinThreads = 16;
    int nblocks = 2;

    TestBase<SharedGcLockProxy> test(nthreads, nblocks, nSpinThreads);
    test.run(boost::bind(
                    &TestBase<SharedGcLockProxy>::allocThreadSync, &test, _1));

}

BOOST_AUTO_TEST_CASE( test_shared_lock_defer )
{
    cerr << "testing contended deferred GcLock with shared lock" << endl;

    SharedGcLock lockGuard(GC_CREATE, SharedGcLockProxy::name);
    Call_Guard unlink_guard([&] { lockGuard.unlink(); });

    int nthreads = 8;
    int nSpinThreads = 16;
    int nblocks = 2;

    TestBase<SharedGcLockProxy> test(nthreads, nblocks, nSpinThreads);
    test.run(boost::bind(
                    &TestBase<SharedGcLockProxy>::allocThreadSync, &test, _1));
}

BOOST_AUTO_TEST_CASE ( test_defer_race )
{
    cerr << "testing defer race" << endl;
    GcLock gc;

    boost::thread_group tg;

    volatile bool finished = false;

    int nthreads = 0;

    volatile int numStarted = 0;

    auto doTestThread = [&] ()
        {
            while (!finished) {
                ML::atomic_inc(numStarted);
                while (numStarted != nthreads) ;

                gc.deferBarrier();

                ML::atomic_dec(numStarted);
                while (numStarted != 0) ;
            }
        };


    for (unsigned i = 0;  i < nthreads;  ++i)
        tg.create_thread(doTestThread);

    int runTime = 1;

    sleep(runTime);

    finished = true;

    tg.join_all();
}

#endif