libcruft-util/test/parallel/queue.cpp

#include <cruft/util/tap.hpp>

#include <cruft/util/parallel/queue.hpp>
#include <cruft/util/thread/flag.hpp>
#include <cruft/util/thread/semaphore.hpp>
#include <cruft/util/view.hpp>

#include <algorithm>

#include <thread>


static constexpr uint32_t slots = 4;
static constexpr int parallelism = 8;
static constexpr int chunk_size = 1<<12;

cruft::thread::flag start;

using queue_t = cruft::parallel::queue<int,slots>;


///////////////////////////////////////////////////////////////////////////////
// push numbers in the range [first,first+chunk_size) to the queue
void
produce (queue_t &dst, const int first)
{
    start.wait ();

    for (int i = first; i < first + chunk_size; ++i)
        while (!dst.push (i))
            ;
}


//-----------------------------------------------------------------------------
// pop chunk_size elements from the queue to a vector
void
consume (queue_t &src, cruft::view<int*> dst)
{
    (void)dst;
    start.wait ();

    int last;

    for (int i = 0; i < chunk_size; ++i) {
        while (!src.pop (last))
            ;
        dst[i] = last;
    }
}


///////////////////////////////////////////////////////////////////////////////
// * create n producers which push the c integers over [n*c,n*(c+1)) into a queue
// * create n consumers which pop c integers from a queue into a store
// * ensure all integers we expected are in the store
//
// we use an artificially small queue size to force contention.
//
// to stress test we increase the threads, increase the data length, and
// decrease the queue size.
int
main ()
{
    cruft::TAP::logger tap;
    queue_t src;

    // start n consumers, and n producers that fill an array with integers
    std::vector<int> dst (parallelism * chunk_size);
    std::vector<std::thread> consumers;
    std::vector<std::thread> producers;

    for (int i = 0; i < parallelism; ++i) {
        consumers.emplace_back (
            consume,
            std::ref (src),
            cruft::view{
                dst.data() + i * chunk_size,
                chunk_size
            }
        );

        producers.emplace_back (produce, std::ref (src), i * chunk_size);
    }

    start.notify_all ();

    // wait for everyone to complete
    for (auto &t: producers)
        t.join ();
    for (auto &t: consumers)
        t.join ();

    // sort the result and ensure all values are present
    std::sort (dst.begin (), dst.end ());

    int missing = 0;
    for (int i = 0; i < parallelism * chunk_size; ++i)
        if (dst[i] != i)
            ++missing;

    tap.expect_eq (missing, 0, "no lost values in parallel produce/consume queue");
    return tap.status ();
}