Testing a lock-free job queue

Question

I've created a lock-free job queue and with the tests I've written, which is also very fast.

That makes me doubt my benchmark procedure, so I'm hoping the collective knowledge will shed some light on the validity of those.

The test basically increments an atomic value (each job is a single increment) until the predefined value is met. In my mind, this test really shows the overhead that the queue imposes because the workload is so simple.

I've tried to create a queue to which I can post jobs and pick up jobs from all threads, a true multiple read/write queue. The tests I've written try to test all use cases, including multi-read/multi-write.

boostasio push/pop functions:

T = function< void() >

void push_back( T t )
{
    service_.post( t );
}

bool pop( T &t )
{
    t = [](){};
    return service_.run_one();
}

mutex_queue:

void push_back( const T &t )
{
    lock_guard< mutex > guard( lock_ );
    data_.push_back( t );
}

bool pop( T &t )
{
    lock_guard< mutex > guard( lock_ );

    if ( index_ == data_.size() ) { return false; }

    t = data_[ index_++ ];
    return true;
}

For details on the lock-free push/pop, I suggest you look at GitHub, since it is a bit extensive to post here.

For completeness, here's the test setup in full:

#include <iostream>
#include <functional>
#include <thread>
#include <sstream>

#include <atomic>
#include <vector>
#include <chrono>

#include <lock_free/fifo.h>

#include <boost/asio/io_service.hpp>


using namespace std;
using namespace chrono;
using namespace boost::asio;

typedef function< void() >function_type;

template < typename T >
struct boostasio
{
    boostasio( size_t r = 1024 ) {}

    void push_back( T t )
    {
        service_.post( t );
    }

    bool pop( T &t )
    {
        t = [](){};
        return service_.run_one();
    }

    io_service service_;
};

template < typename T >
struct mutex_queue
{
    mutex_queue( size_t r = 1024 ) :
        lock_(),
        index_( 0 ),
        data_( r )
    {
        data_.clear();
    }

    void push_back( const T &t )
    {
        lock_guard< mutex > guard( lock_ );
        data_.push_back( t );
    }

    bool pop( T &t )
    {
        lock_guard< mutex > guard( lock_ );

        if ( index_ == data_.size() ) { return false; }

        t = data_[ index_++ ];
        return true;
    }

    mutex lock_;
    size_t index_;
    vector< T > data_;
};

template < typename T >
T to( const string &str )
{
    T result;
    stringstream( str ) >> result;
    return result;
}

template < typename  T >
function_type get_producer( T &&t )
{
    return get< 0 >( t );
}

template < typename  T >
function_type get_consumer( T &&t )
{
    return get< 1 >( t );
}

template < typename  T >
function_type get_result( T &&t )
{
    return get< 2 >( t );
}

template < typename Q >
void test( const string &testname, size_t count, size_t threadcount )
{
    auto create_producer_consumer_result = [=]( const string &name )
    {
        high_resolution_clock::time_point t1 = high_resolution_clock::now();

        auto data = make_shared< Q >( count );

        function_type producer = [data]()
        {
            while ( data->producer_count++ < data->expected )
            {
                data->queue.push_back(
                    [data]()
                    {
                        ++data->consumer_count;
                    }
                );
            }

            if ( data->producer_count >= data->expected )
            {
                --data->producer_count;
            }
        };

        function_type consumer = [data]()
        {
            while ( data->consumer_count < data->expected )
            {
                function_type func;

                while ( data->queue.pop( func ) )
                {
                    func();
                }
            }
        };

        function_type result = [=]()
        {
            high_resolution_clock::time_point t2 = high_resolution_clock::now();

            duration< double > time_span = duration_cast< duration< double > >( t2 - t1 );

            if ( data->expected != data->consumer_count )
            {
                cout << "\texpected: " << data->expected << ", actual: " << data->consumer_count << endl;
            }

            cout << '\t' << name << " took: " << time_span.count() << " seconds" << endl;
        };

        return make_tuple( producer, consumer, result );
    };

    high_resolution_clock::time_point teststart = high_resolution_clock::now();

    cout << testname << ":\n{\n";

    // single producer, single consumer
    {
        auto pcr = create_producer_consumer_result( "single producer, single consumer" );

        get_producer( pcr )();

        get_consumer( pcr )();

        get_result( pcr )();
    }

    // single producer, multi consumer
    {
        auto pcr = create_producer_consumer_result( "single producer, multi consumer" );

        get_producer( pcr )();

        vector< thread > threads;
        size_t c = threadcount;

        while ( c-- )
        {
            threads.push_back( thread( get_consumer( pcr ) ) );
        }

        for ( auto &t : threads )
        {
            t.join();
        }

        get_result( pcr )();
    }

    // multi producer, single consumer
    {
        auto pcr = create_producer_consumer_result( "multi producer, single consumer" );

        vector< thread > threads;
        size_t c = threadcount;

        while ( c-- )
        {
            threads.push_back( thread( get_producer( pcr ) ) );
        }

        for ( auto &t : threads )
        {
            t.join();
        }

        get_consumer( pcr )();

        get_result( pcr )();
    }

    // multi producer, multi consumer
    {
        auto pcr = create_producer_consumer_result( "multi producer, multi consumer" );

        vector< thread > threads;
        size_t c = threadcount / 2;

        while ( c-- )
        {
            threads.push_back( thread( get_producer( pcr ) ) );
            threads.push_back( thread( get_consumer( pcr ) ) );
        }

        for ( auto &t : threads )
        {
            t.join();
        }

        get_result( pcr )();
    }

    duration< double > time_span = duration_cast< duration< double > >( high_resolution_clock::now() - teststart );
    cout << "\ttotal: " << time_span.count() << " seconds\n}" << endl;
}

template < typename T >
struct test_data
{
    test_data( size_t e ) :
    expected( e ),
    queue(),
    producer_count( 0 ),
    consumer_count( 0 ) { }

    const size_t expected;
    T queue;
    atomic_size_t producer_count;
    atomic_size_t consumer_count;
};

int main( int argc, char *argv[] )
{
    constexpr auto test_count = 1e6;

    const auto thread_count = argc > 1 ? to< size_t >( argv[ 1 ] ) : 16;

    test< test_data< boostasio< function_type > > >( "boostasio", test_count, thread_count );
    test< test_data< lock_free::fifo< function_type > > >( "lock_free::fifo", test_count, thread_count );
    test< test_data< mutex_queue< function_type > > >( "mutex_queue", test_count, thread_count );

    return 0;
}

Here are some results on a machine which has 8 cores (+8 HT) and running with 16 threads:

boostasio:
{
  single producer, single consumer took: 0.711752 seconds
  single producer, multi consumer took: 5.03024 seconds
  multi producer, single consumer took: 4.16782 seconds
  multi producer, multi consumer took: 8.45779 seconds
  total: 18.3679 seconds
}
lock_free::fifo:
{
  single producer, single consumer took: 0.356197 seconds
  single producer, multi consumer took: 1.12591 seconds
  multi producer, single consumer took: 0.575264 seconds
  multi producer, multi consumer took: 1.24645 seconds
  total: 3.304 seconds
}
mutex_queue:
{
  single producer, single consumer took: 0.363318 seconds
  single producer, multi consumer took: 2.77809 seconds
  multi producer, single consumer took: 2.72058 seconds
  multi producer, multi consumer took: 5.11961 seconds
  total: 10.9818 seconds
}

Answer 1

One thing you should do is not use so many (if any) using namespace statements. This does make it a bit easier to write code, but it also can cause many problems with methods being confused with each other. For a detailed discussion of this, check out this question on SO.

Also, you have many extra returns in your #include statements which make it look like many groups instead of just one.

Finally, your indentation is really good, which is mandatory for readability. However, you do not follow standard C++ style for your braces. C++ uses the K&R style, which sets the braces for functions on the next line, but braces inside functions, such as those used in ifs, loops, and switch statements, start on the same line, like this:

void my_function()
{
    while (true) {
        std::cout<<"Something here...\n";
    }

    if (true) {
        std::cout<<"Something else here...\n";
    }
}

Answer 2

(I'm only superficially familiar with boost asio, so I'll ignore it for the rest of my post. But looking at your benchmark results, most of what I'm about to say about mutex_queue might hold as well.)

Only your multi/multi case covers read/write concurrency, as in all other cases the consumers aren't started until the producers have finished. But otherwise, the general benchmark mechanic seems sound.

The performance characteristics you posted for mutex_queue are along the lines of what I would expect: basically measuring the overhead of contested locks over uncontested locks. Slightly less than 2.4us per contested lock sounds reasonable.

The push_back side of lock_free::fifo looks fairly efficient, with the only real thumb-twiddling occuring when the queue needs a resize. The pop side slightly less so, as marking an element as read is essentially a batch process under read concurrency. I also believe there's a small bug in this part. If the first element in a batch (the one for which id == read_ holds) was marked done by it's producer due to an exception, increase_read will never be called.

I'd say that the real issue here, if you can call it that, is that your comparison is just horribly unfair to mutex_queue. You typically use a fully-locked queue when the only thing that matters is that things happen in a correct and predictable manner, and all costs be damned.

mutex_queue and lock_free::fifo don't offer the same FIFO constraint under read/write concurrency. From a consumer's perspective, mutex_queue orders jobs by start-of-posting. lock_free::fifo only applies that ordering at end-of-posting, potentially reordering the queue. If producers were to tag jobs with an increasing ID after acquiring the lock, a mutex_queue consumer would be guaranteed an increasing ID with every pop, whereas a lock_free::fifo consumer might occasionally see a decreasing ID from a 'slower' producer.

My recommendation would be to compare lock_free::fifo to a lock-based queue that's actually optimized for the multi-read/multi-write scenario. At the very least something that uses element-level locking instead of queue-level locking where possible.

Answer 3

Your code sometimes does not follow C++ Coding Conventions. Take a look at here:

struct test_data
{
    test_data( size_t e ) :
    expected( e ),
    queue(),
    producer_count( 0 ),
    consumer_count( 0 ) { }

According to conventions:

Consider natural language rules for spacing Readability is the goal. In general, "natural language" rules are advocated, including:

• no space before ",", ";";
• a space (or EOL) after.

You have a space before ;.

Also:

Consider having no spaces after (-bracket and before )-bracket
Consider having no spaces after [-bracket and before ]-bracket

You have extra spaces all over the place. To me, it makes the code look not natural. If you like how you write code, you can keep how you write it, but these are some suggestions.