[c++11] multi-threaded lock-free queue testen

[Arjan]

Ik ben reeds enige tijd bezig met de gedachte van een multi-producer, multi-consumer queue.

Hierin ben ik gelukkig niet de enige en er is dan ook een hoop te vinden. Een van de belangrijkste is natuurlijk
boost. Een eigenaardige beperking aan deze boost::lockfree::queue is echter dat het datatype een triviale destructor moet hebben.

Dat vind ik aardig afbreuk doen aan het (wat mij betreft) ideaal beeld van elke thread die een job mag queuen. Die job kan namelijk niet in de uitvoerende thread-memory-space gealloceerd worden. Je kan wel een job maken die dat dan weer doet, maar ook dat vereist weer werk wat imho. niet triviaal is om goed te doen.

Wat mij betreft zou je gewoon een std::function< void() > op een queue moeten pushen and that's it.
Met die insteek ben ik aan de slag gegaan en heb ik uiteindelijk iets geschreven wat dat ook daadwerkelijk kan.

Om dit alles in perspectief te zetten had ik een test setup geschreven zodat ik ook zinnige vergelijkingen kon maken met andere methoden. Wat ik syntactisch probeer te doen lijkt nog het meest op boost::asio, daar heb ik dus ook een vergelijking mee gemaakt.
Daarnaast heb ik ook een vergelijking gemaakt met een standaard vector beschermt met een std::mutex, als een soort sanity check zeg maar.

Over het algemeen zijn de resultaten nogal rooskleurig voor mijn brouwsel, hetgeen mij doet vermoeden dat ik een onbedoelde bias voor mijn setup heb ingebouwd. Ik had dit al eens op codereview.stackoverflow gepost maar daar gebeurt niet zoveel..

Om de bias zoveel mogelijk uit te sluiten heb ik de test aangepast zodat deze direct aansluit op boost. Dit gaf de volgende resultaten op mijn dual core laptop:

boostlockfree:
{
	single producer, single consumer took: 1.10634 seconds
	single producer, multi consumer took: 1.10573 seconds
	multi producer, single consumer took: 0.867554 seconds
	multi producer, multi consumer took: 0.712594 seconds
	total: 4.29545 seconds
}
boostasio:
{
	single producer, single consumer took: 2.14071 seconds
	single producer, multi consumer took: 6.53309 seconds
	multi producer, single consumer took: 6.81575 seconds
	multi producer, multi consumer took: 11.5268 seconds
	total: 27.0172 seconds
}
lock_free::fifo:
{
	single producer, single consumer took: 0.568918 seconds
	single producer, multi consumer took: 0.449427 seconds
	multi producer, single consumer took: 0.509566 seconds
	multi producer, multi consumer took: 0.423657 seconds
	total: 2.00084 seconds
}
mutex_queue:
{
	single producer, single consumer took: 0.33865 seconds
	single producer, multi consumer took: 4.2441 seconds
	multi producer, single consumer took: 4.19272 seconds
	multi producer, multi consumer took: 8.12005 seconds
	total: 16.9422 seconds
}

De implementatie zou nog netter kunnen door mbv. templates de queue fixed size of niet te maken, custom allocators, dat soort dingen. Maar graag hoor ik jullie mening over de test-setup en of de vergelijking eerlijk en representatief is voor daadwerkelijke performance impact.

Hier is de main.cpp

C++:

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

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

#include <lock_free/fifo.h>
#include <lock_free/shared_mutex.h>

#include <boost/asio/io_service.hpp>
#include <boost/lockfree/queue.hpp>


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

typedef function< void() >function_type;

template < typename T >
struct boostlockfree
{
    boostlockfree( size_t r = 1024 ) :
        jobs_( r ) {}

    inline void push_back( T t )
    {
        jobs_.push( t );
    }

    inline bool pop( T &t )
    {
        return jobs_.pop( t );
    }

    boost::lockfree::queue< T > jobs_;
};

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

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

    inline bool pop( T &t )
    {
        static function_type tmp = [](){};
        t = &tmp;
        return service_.run_one() > 0;
    }

    io_service service_;
};

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

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

    inline 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 );
        
        auto tmp = new function_type(
            [data]()
            {
                ++data->consumer_count;
            }
        );
        
        function_type producer = [data,tmp]()
        {
            while ( data->producer_count++ < data->expected )
            {
                data->queue.push_back( tmp );
            }

            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;
            
            delete tmp;
        };

        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[] )
{
    const auto test_count = 1e6;

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

    test< test_data< boostlockfree< function_type* > > >( "boostlockfree", test_count, thread_count );
    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;
}

De rest van de code is hier te vinden:

https://github.com/arjanhouben/lock_free_fifo

[H!GHGuY]

Wij hebben op het werk recent iets gelijkaardigs geschreven en de snelheidwinst kan inderdaad significant zijn. 1 van de belangrijkste punten bij dit soort zaken is cache access.

Ik zou dan ook met een profiler deze code eens grondig onder de loep nemen.

Andere tips: run elke test in een apart process. Zo voorkom je dat de 2-4de test reeds op cache-hot data werken.

[Arjan]

Profiler heb al regelmatig naar gekeken, maar ik zie geen opvallende spikes meer, nog tips waar op te letten?
Vooralsnog lijkt vooral mijn 'push' sneller te zijn dan boost's.

Ik heb de main loop aangepast zodat je kan specificeren welke test je wil runnen zodat ze idd. in een willekeurige volgorde of per process aangeroepen kunnen worden. Overigens maakt het in dit geval geen verschil uit voor het resultaat.

het volgende zul je even moeten plakken in iets wat geen line-wrapping doet..

Running Time	Self		Symbol Name
1756.0ms  100.0%	0,0	 	void* __thread_proxy<tuple<function<void ()> > >(void*)
550.0ms   31.3%	32,0	 	 function::__func<void test<test_data<boostlockfree<function<void ()>*> > >(string)::'lambda'(string const&)::operator()(string const&) const::'lambda1'(), allocator<void test<test_data<boostlockfree<function<void ()>*> > >(string)::'lambda'(string const&)::operator()(string const&) const::'lambda1'()>, void ()>::operator()()
460.0ms   26.1%	460,0	 	  bool boost::lockfree::queue<function<void ()>*, boost::parameter::void_, boost::parameter::void_, boost::parameter::void_>::pop<function<void ()>*>(function<void ()>*&)
58.0ms    3.3%	58,0	 	  function::__func<void test<test_data<boostlockfree<function<void ()>*> > >(string)::'lambda'(string const&)::operator()(string const&) const::'lambda'(), allocator<void test<test_data<boostlockfree<function<void ()>*> > >(string)::'lambda'(string const&)::operator()(string const&) const::'lambda'()>, void ()>::operator()()
513.0ms   29.2%	19,0	 	 function::__func<void test<test_data<lock_free::fifo<function<void ()>*> > >(string)::'lambda'(string const&)::operator()(string const&) const::'lambda1'(), allocator<void test<test_data<lock_free::fifo<function<void ()>*> > >(string)::'lambda'(string const&)::operator()(string const&) const::'lambda1'()>, void ()>::operator()()
453.0ms   25.7%	453,0	 	  lock_free::fifo<function<void ()>*>::pop(function<void ()>*&)
41.0ms    2.3%	41,0	 	  function::__func<void test<test_data<lock_free::fifo<function<void ()>*> > >(string)::'lambda'(string const&)::operator()(string const&) const::'lambda'(), allocator<void test<test_data<lock_free::fifo<function<void ()>*> > >(string)::'lambda'(string const&)::operator()(string const&) const::'lambda'()>, void ()>::operator()()
374.0ms   21.2%	67,0	 	 function::__func<void test<test_data<boostlockfree<function<void ()>*> > >(string)::'lambda'(string const&)::operator()(string const&) const::'lambda0'(), allocator<void test<test_data<boostlockfree<function<void ()>*> > >(string)::'lambda'(string const&)::operator()(string const&) const::'lambda0'()>, void ()>::operator()()
307.0ms   17.4%	307,0	 	  bool boost::lockfree::queue<function<void ()>*, boost::parameter::void_, boost::parameter::void_, boost::parameter::void_>::do_push<false>(function<void ()>* const&)
283.0ms   16.1%	68,0	 	 function::__func<void test<test_data<lock_free::fifo<function<void ()>*> > >(string)::'lambda'(string const&)::operator()(string const&) const::'lambda0'(), allocator<void test<test_data<lock_free::fifo<function<void ()>*> > >(string)::'lambda'(string const&)::operator()(string const&) const::'lambda0'()>, void ()>::operator()()
215.0ms   12.2%	215,0	 	  void lock_free::fifo<function<void ()>*>::push_back<function<void ()>* const&>(function<void ()>* const&&&)
27.0ms    1.5%	27,0	 	 DYLD-STUB$$bool boost::lockfree::queue<function<void ()>*, boost::parameter::void_, boost::parameter::void_, boost::parameter::void_>::pop<function<void ()>*>(function<void ()>*&)
6.0ms    0.3%	6,0	 	 DYLD-STUB$$void lock_free::fifo<function<void ()>*>::push_back<function<void ()>* const&>(function<void ()>* const&&&)
2.0ms    0.1%	2,0	 	 DYLD-STUB$$bool boost::lockfree::queue<function<void ()>*, boost::parameter::void_, boost::parameter::void_, boost::parameter::void_>::do_push<false>(function<void ()>* const&)
1.0ms    0.0%	1,0	 	 DYLD-STUB$$lock_free::fifo<function<void ()>*>::pop(function<void ()>*&)

[ Voor 81% gewijzigd door Arjan op 05-08-2014 22:55 ]

[pedorus]

Ik zou toch eens hier naar kijken: http://blog.memsql.com/co...ing-lock-free-algorithms/
Als ik even snel door jouw code heen kijk zie ik waarschijnlijk soortgelijke problemen. Ik zie bijvoorbeeld een mutex in de 'lock free' code?

De kans is zeer groot dat je tijd weet te besparen door een incorrect algoritme. Ander voorbeeldje uit een serie van 3: http://www.drdobbs.com/cp...nse-of-security/210600279

Nog eentje anders?

quote: https://twitter.com/kumpera/status/476482707145625601

Lock-free code is crazy hard. Mono had an infinite loop in its GC allocator that took 2 years to be found. Kudos to @markprobst for the find

[Arjan]

pedorus schreef op woensdag 06 augustus 2014 @ 00:12:
Ik zou toch eens hier naar kijken: http://blog.memsql.com/co...ing-lock-free-algorithms/
Als ik even snel door jouw code heen kijk zie ik waarschijnlijk soortgelijke problemen. Ik zie bijvoorbeeld een mutex in de 'lock free' code?

Die is er niet, je bedoeld wellicht de testcase die de vergelijking maakt met de mutex beveiligde vector?

Er is overigens wel een lock en dat is als je queue vol raakt en er nieuw geheugen klaar gezet moet worden. Dit kun je voorkomen door een fixed size queue te gebruiken of je queue nooit vol te laten lopen. Dit zou wellicht nog verminderd kunnen worden mbv. binning, zoals boost het volgens mij ook doet, maar vooralsnog is er performance-wise geen reden om dat te doen.

De kans is zeer groot dat je tijd weet te besparen door een incorrect algoritme.
[...]

Vandaar dat ik ook feedback vraag op mijn testmethode, mijn implementatie heb ik zeer veel tijd ingestopt, maar ik weet niet hoe ik deze nog steviger op de pijnbank zou moeten leggen.

[Soultaker]

Arjan schreef op zondag 03 augustus 2014 @ 11:33:
Hierin ben ik gelukkig niet de enige en er is dan ook een hoop te vinden. Een van de belangrijkste is natuurlijk
boost. Een eigenaardige beperking aan deze boost::lockfree::queue is echter dat het datatype een triviale destructor moet hebben.

Je kunt ook nog vergelijken met concurrent_queue uit Intel's Thread Building Blocks. Die lijkt ten eerste die beperking niet te hebben, en ten tweede is de performance meestal wel OK. Misschien zuigt Boost gewoon; dat zou ook niet de eerste keer zijn.

[ Voor 4% gewijzigd door Soultaker op 06-08-2014 10:12 ]

[H!GHGuY]

Arjan schreef op dinsdag 05 augustus 2014 @ 22:45:
Profiler heb al regelmatig naar gekeken, maar ik zie geen opvallende spikes meer, nog tips waar op te letten?

Ik bedoel een profiler
Aangezien je op Linux werkt, probeer eens met perf. Die gebruikt CPU performance counters. Daarmee kun je dingen als cache misses, context switches, ... mee meten. Ook wil je statistical profiling wel eens meenemen. Zo kun je hotspots zoeken op zowel instructies, caches en andere.

Arjan schreef op woensdag 06 augustus 2014 @ 00:49:
[...]

Die is er niet, je bedoeld wellicht de testcase die de vergelijking maakt met de mutex beveiligde vector?

Ik zag toch ook echt een shared_mutex. Je kan het dus niet echt lock-free noemen.
Met je IDs als atomic ben je echter al op het goede pad. Er zijn trouwens enkele interessante papers te vinden.
Bijvoorbeeld: Cache-Aware Lock-Free Queues for Multiple Producers/Consumers and Weak Memory Consistency

Er is overigens wel een lock en dat is als je queue vol raakt en er nieuw geheugen klaar gezet moet worden. Dit kun je voorkomen door een fixed size queue te gebruiken of je queue nooit vol te laten lopen. Dit zou wellicht nog verminderd kunnen worden mbv. binning, zoals boost het volgens mij ook doet, maar vooralsnog is er performance-wise geen reden om dat te doen.

Je kan ook efficiente semphores gebruiken.

[Arjan]

Soultaker schreef op woensdag 06 augustus 2014 @ 10:12:
[...]

Je kunt ook nog vergelijken met concurrent_queue uit Intel's Thread Building Blocks. Die lijkt ten eerste die beperking niet te hebben, en ten tweede is de performance meestal wel OK. Misschien zuigt Boost gewoon; dat zou ook niet de eerste keer zijn.

Die ga ik erbij zetten, zal de resultaten posten.

H!GHGuY schreef op woensdag 06 augustus 2014 @ 11:27:
[...]

Ik bedoel een profiler
Aangezien je op Linux werkt, probeer eens met perf. Die gebruikt CPU performance counters. Daarmee kun je dingen als cache misses, context switches, ... mee meten. Ook wil je statistical profiling wel eens meenemen. Zo kun je hotspots zoeken op zowel instructies, caches en andere.

Ik werk voornamelijk op osx, instruments kan dit volgens mij ook. Anders kan ik op windows nog eens vtune proberen. Nadeel is echter dat ik de resultaten uit dergelijke metingen niet goed op waarde kan schatten. Neem nou een cache miss, die gaan vrijwel zeker voorkomen in multithreaded code, maar welke waarde is te hoog? Ik heb daar wel eens naar gekeken, maar nooit het gevoel gehad er zinnig op te kunnen sturen.

Ik zag toch ook echt een shared_mutex. Je kan het dus niet echt lock-free noemen.

Dat is een naming error helaas. Die shared mutex is namelijk geen mutex, alleen een klasse die nog zo heet.. Het is een atomic counter die ervoor zorgt dat gedurende de resize er geen meerdere threads tegelijk bezig zijn, ik begrijp de verwarring


Hierbij nog de resultaten van intel erbij:

boostlockfree:
{
	single producer, single consumer took: 0.977331 seconds
	single producer, multi consumer took: 0.967351 seconds
	multi producer, single consumer took: 0.833409 seconds
	multi producer, multi consumer took: 0.627882 seconds
	total: 3.91037 seconds
}
boostasio:
{
	single producer, single consumer took: 1.9213 seconds
	single producer, multi consumer took: 6.83095 seconds
	multi producer, single consumer took: 6.53116 seconds
	multi producer, multi consumer took: 11.7241 seconds
	total: 27.0086 seconds
}
lock_free::fifo:
{
	single producer, single consumer took: 0.753901 seconds
	single producer, multi consumer took: 0.510071 seconds
	multi producer, single consumer took: 0.619403 seconds
	multi producer, multi consumer took: 0.468217 seconds
	total: 2.40044 seconds
}
mutex_queue:
{
	single producer, single consumer took: 0.320638 seconds
	single producer, multi consumer took: 4.16108 seconds
	multi producer, single consumer took: 4.14933 seconds
	multi producer, multi consumer took: 8.17982 seconds
	total: 16.8621 seconds
}
inteltbb:
{
	single producer, single consumer took: 0.661875 seconds
	single producer, multi consumer took: 1.53375 seconds
	multi producer, single consumer took: 2.00534 seconds
	multi producer, multi consumer took: 1.07156 seconds
	total: 5.27277 seconds
}

[ Voor 29% gewijzigd door Arjan op 06-08-2014 21:00 ]

[H!GHGuY]

Arjan schreef op woensdag 06 augustus 2014 @ 20:45:

Ik werk voornamelijk op osx, instruments kan dit volgens mij ook. Anders kan ik op windows nog eens vtune proberen. Nadeel is echter dat ik de resultaten uit dergelijke metingen niet goed op waarde kan schatten. Neem nou een cache miss, die gaan vrijwel zeker voorkomen in multithreaded code, maar welke waarde is te hoog? Ik heb daar wel eens naar gekeken, maar nooit het gevoel gehad er zinnig op te kunnen sturen.

In de meeste gevallen gaat het over de relatieve hoeveelheden + waar ze zich voordoen.

Als ik met niet vergis kan perf (en oprofile) de code aangeven waar de cache misses zich voordoen. Daar kun je dan zien waar de meeste zich voordoen en waar je kan optimizen.

De absolute waardes zijn pas interessant als je cache-aware of cache-oblivious algoritmes wil schrijven om dan het aantal cache misses per operatie te gaan schatten of verifieren.

[pedorus]

In het kader van toch eens testen met g++-4.8 -O2 op een i7-3930K:

boostlockfree:
{
        single producer, single consumer took: 0.0859925 seconds
        single producer, multi consumer took: 0.173336 seconds
        multi producer, single consumer took: 0.457049 seconds
        multi producer, multi consumer took: 0.307004 seconds
        total: 1.08722 seconds
}
boostasio:
{
        single producer, single consumer took: 0.285684 seconds
        single producer, multi consumer took: 0.362252 seconds
        multi producer, single consumer took: 0.481476 seconds
        multi producer, multi consumer took: 0.607431 seconds
        total: 1.73702 seconds
}
lock_free::fifo:
{
        single producer, single consumer took: 0.134983 seconds
        single producer, multi consumer took: 0.302158 seconds
        multi producer, single consumer took: 0.172903 seconds
        multi producer, multi consumer took: 0.323649 seconds
        total: 0.934493 seconds
}
mutex_queue:
{
        single producer, single consumer took: 0.05627 seconds
        single producer, multi consumer took: 0.146091 seconds
        multi producer, single consumer took: 0.174592 seconds
        multi producer, multi consumer took: 0.260636 seconds
        total: 0.638694 seconds
}

In voorkomende gevallen loopt de boostasio test vast op multi/multi. Lijkt me ernstig.

Kortom, heel veel "optimalisaties" ten opzichte van zo'n simpele mutex. En meestal is zo'n queue natuurlijk minder bezet dan in deze tests.

En die mutex heeft extra voordelen natuurlijk. Zo hebben we niet potentiële issues zoals deze (zet breakpoint op 41, resume bij 'exclusive' lock in andere thread):

C++: shared_mutex.h

            void lock_shared()
            {
                if ( ++lock_required_ & locked )
                {
                    --lock_required_;
                    
                    wait_for_non_exclusive();
                    
                    ++lock_required_;
                }
            }

Of de eigenschap dat het geheugen gemakkelijk vol kan lopen. Of de eigenschap dat als het geheugen volloopt, de queue na een throw vastloopt en een verkeerde allocatie in een (ongelockte) errorhandler probeert te doen, enz. Of een ander probleem dat er ongetwijfeld in zit.

Ik denk trouwens dat op een dual core die yield()'s nogal uitmaken, dat geeft een wat oneerlijke vergelijking.

[Arjan]

pedorus schreef op donderdag 07 augustus 2014 @ 01:24:
In het kader van toch eens testen met g++-4.8 -O2 op een i7-3930K:
[...]
In voorkomende gevallen loopt de boostasio test vast op multi/multi. Lijkt me ernstig.

Kortom, heel veel "optimalisaties" ten opzichte van zo'n simpele mutex. En meestal is zo'n queue natuurlijk minder bezet dan in deze tests.

En die mutex heeft extra voordelen natuurlijk. Zo hebben we niet potentiële issues zoals deze (zet breakpoint op 41, resume bij 'exclusive' lock in andere thread):

C++: shared_mutex.h

            void lock_shared()
            {
                if ( ++lock_required_ & locked )
                {
                    --lock_required_;
                    
                    wait_for_non_exclusive();
                    
                    ++lock_required_;
                }
            }

regel 41 weghalen en de if naar een while veranderen zou dat probleem oplossen, er is dan wel een preference voor een exclusive lock, maar dat maakt in dit gebruik niet uit

Of de eigenschap dat het geheugen gemakkelijk vol kan lopen.

Dat het geheugen kan vollopen is natuurlijk inherent aan het bijhouden van een lijst. Het grootste probleem is wanneer de jobs nooit helemaal leeg raken, dat gaat een keer falen idd.

Of de eigenschap dat als het geheugen volloopt, de queue na een throw vastloopt en een verkeerde allocatie in een (ongelockte) errorhandler probeert te doen, enz.

Waar gebeurd dit? Als er iets tijdens de push throwed wordt de bijbehorende job op done gezet.

Of een ander probleem dat er ongetwijfeld in zit.
Ik denk trouwens dat op een dual core die yield()'s nogal uitmaken, dat geeft een wat oneerlijke vergelijking.

op windows/osx maakt het niet zoveel uit. Ik heb ook getest op dual xeon quad core, met HT zit je dan op 16 cores. bij linux lijkt zelfs in een VM de mutex variant de snelste.. dat is wel opmerkelijk

[ Voor 18% gewijzigd door Arjan op 07-08-2014 11:51 ]

[pedorus]

Arjan schreef op donderdag 07 augustus 2014 @ 11:03:
regel 41 weghalen en de if naar een while veranderen zou dat probleem oplossen, er is dan wel een preference voor een exclusive lock, maar dat maakt in dit gebruik niet uit

Fixen is in dit geval makkelijk, het is het spotten van dit soort issues dat erg lastig is.

Waar gebeurd dit? Als er iets tijdens de push throwed wordt de bijbehorende job op done gezet.

En daar is de lock out of scope. Als de throw vanaf resize() komt, dan is het zelfs vrijwel zeker dat het hier mis gaat. Als de resize() faalt, dan hangen na de throw alle producers oneindig in een loopje. En vanwege de manier waarop de queue werkt is het niet ondenkbaar dat die resize een keer gaat falen als het redelijk druk is met producers.

Eigenlijk werk ik zelf trouwens altijd met queues met een maximaal aantal elementen om geheugenproblemen te voorkomen. Op een gegeven moment is stoppen beter dan crashen.

op windows/osx maakt het niet zoveel uit. Ik heb ook getest op dual xeon quad core, met HT zit je dan op 16 cores. bij linux lijkt zelfs in een VM de mutex variant de snelste.. dat is wel opmerkelijk

Ik vermoed dat je de andere queues toch wat kan verbeteren door in je testcode bij de consumer een yield te doen bij falen op de dual core. Met een dual processor situatie zit je met het concept dat het geheugen eigenlijk wordt gekoppeld aan een specifieke processor, wat de boel wel eens zou kunnen vertragen voor sommige queues, afhankelijk van op welke processors de threads draaien.

Ook een mooie om testen: https://github.com/fsaintjacques/disruptor--

Dan nog zijn de testscenarios vaak niet erg representatief omdat er geen werk plaatsvind. In de praktijk denk ik dat de queue bij de meeste taken erg weinig uitmaakt.

[Arjan]

pedorus schreef op donderdag 07 augustus 2014 @ 13:29:
[...]

Fixen is in dit geval makkelijk, het is het spotten van dit soort issues dat erg lastig is.

Ja dat klopt, dat was een nice catch

En daar is de lock out of scope. Als de throw vanaf resize() komt, dan is het zelfs vrijwel zeker dat het hier mis gaat. Als de resize() faalt, dan hangen na de throw alle producers oneindig in een loopje. En vanwege de manier waarop de queue werkt is het niet ondenkbaar dat die resize een keer gaat falen als het redelijk druk is met producers.

wederom, goed gevonden!

Eigenlijk werk ik zelf trouwens altijd met queues met een maximaal aantal elementen om geheugenproblemen te voorkomen. Op een gegeven moment is stoppen beter dan crashen.

[...]

Ik vermoed dat je de andere queues toch wat kan verbeteren door in je testcode bij de consumer een yield te doen bij falen op de dual core. Met een dual processor situatie zit je met het concept dat het geheugen eigenlijk wordt gekoppeld aan een specifieke processor, wat de boel wel eens zou kunnen vertragen voor sommige queues, afhankelijk van op welke processors de threads draaien.

In mijn tests maakt het niks uit. Ik had het volgende gedaan:

C++:

                while ( data->consumer_count < data->expected )
            {
                function_type *func;

                while ( data->queue.pop( func ) )
                {
                    (*func)();
                }
                this_thread::yield();
            }

Ook een mooie om testen: https://github.com/fsaintjacques/disruptor--

Heb je een stukje example code voor die? Ik zie zo 123 niet hoe ik die erin moet lepelen.

Dan nog zijn de testscenarios vaak niet erg representatief omdat er geen werk plaatsvind. In de praktijk denk ik dat de queue bij de meeste taken erg weinig uitmaakt.

Juist omdat er geen werk plaatsvind is een dergelijke test representatief voor de hoeveelheid overhead, tenminste dat was mijn insteek voor deze tests.

[pedorus]

Arjan schreef op donderdag 07 augustus 2014 @ 18:19:
Heb je een stukje example code voor die? Ik zie zo 123 niet hoe ik die erin moet lepelen.

Zie nu dat dit alleen een message queue is en geen worker queue. Daarnaast is de implementatie op https://github.com/redjack/varon-t wellicht beter. Het idee is in ieder geval om zowel geen locks als geen CAS te gebruiken.

Deze lijkt me interessanter voor een snel vergelijk: https://github.com/krizha...b/master/lockfree_rb_q.cc

Ik kwam trouwens ook nog een java queue tegen uit de tijd dat codereview.stackexchange wel iets deed: http://codereview.stackex...mplementation-thread-safe

Juist omdat er geen werk plaatsvind is een dergelijke test representatief voor de hoeveelheid overhead, tenminste dat was mijn insteek voor deze tests.

De latency die een kruispunt tijdens de spits geeft lijkt me niet heel representatief voor datzelfde kruispunt in midden in de nacht. Overigens zou dat in principe een klein voordeel kunnen geven aan lock-free vanwege de overhead van een lock.

[H!GHGuY]

Onze queue is ook deels gebaseerd op de paper van disruptor, maar gaat op sommige stukken nog verder en houdt nog beter rekening met cache access en de specifiek processor waarop we werken.

We hadden dan ook nog wel wat andere requirements en ideëen. Er zitten ook nog wat ideëen in van die andere paper die ik aanhaalde (hoewel we die paper niet kenden tijdens onze implementatie.

Belangrijk is om shared state volledig in de shared cache te proberen houden en zo weinig mogelijk cacheline bouncing te krijgen. Atomic operations doen dit over het algemeen, maar dan moet je nog altijd opletten met je queue data zelf.

[Zoijar]

Wanneer maakt het eigenlijk uit dat je queue zo ontzettend snel is dat de laatste cycle eruit geperst wordt? Als het werk zo kort is dat de queue overhead veel uitmaakt, dan is het wellicht beter om grotere jobs te sturen. Of helemaal geen queue. Als het werk duur is, dan valt de queue optimizen onder veel tijd en complexiteit besteden aan 0.1% execution tijd 10% te verbeteren.

En wat meet je dan? Throughput of latency? Latency zou voor ons bijvoorbeeld veel belangrijker zijn. (dus hoeveel cycles ben ik kwijt om 1 bericht door een meestal lege queue te persen, mogelijk niet cache hot)

[ Voor 10% gewijzigd door Zoijar op 08-08-2014 11:19 ]

[H!GHGuY]

Zoijar schreef op vrijdag 08 augustus 2014 @ 11:17:
Wanneer maakt het eigenlijk uit dat je queue zo ontzettend snel is dat de laatste cycle eruit geperst wordt? Als het werk zo kort is dat de queue overhead veel uitmaakt, dan is het wellicht beter om grotere jobs te sturen. Of helemaal geen queue. Als het werk duur is, dan valt de queue optimizen onder veel tijd en complexiteit besteden aan 0.1% execution tijd 10% te verbeteren.

En wat meet je dan? Throughput of latency? Latency zou voor ons bijvoorbeeld veel belangrijker zijn. (dus hoeveel cycles ben ik kwijt om 1 bericht door een meestal lege queue te persen, mogelijk niet cache hot)

Als je een gigantische applicatie hebt die quasi volledig message-driven is, dan is een performante queue zeer belangrijk. Niet elke applicatie kan z'n message workload zomaar kiezen. Wij hebben grote messages, maar evengoed kleine. En je kan niet altijd bepalen welke (en hoeveel ervan) je voor de kiezen krijgt.
Trouwens deed onze vorige queue implementatie het op bepaalde workloads nogal slecht, dus was het meer dan 0.1% -> 0.09% krijgen...