Index: doc/theses/thierry_delisle_PhD/code/readyQ_proto/links.hpp =================================================================== --- doc/theses/thierry_delisle_PhD/code/readyQ_proto/links.hpp (revision 6a8208cbc93288ea3ca61f2a04619465c12cf317) +++ doc/theses/thierry_delisle_PhD/code/readyQ_proto/links.hpp (revision 59f3f616c737817ce7af766fa1bcf77a79d695d9) @@ -117,5 +117,5 @@ } - long long ts() const { + unsigned long long ts() const { return before._links.ts; } Index: doc/theses/thierry_delisle_PhD/code/readyQ_proto/links2.hpp =================================================================== --- doc/theses/thierry_delisle_PhD/code/readyQ_proto/links2.hpp (revision 6a8208cbc93288ea3ca61f2a04619465c12cf317) +++ doc/theses/thierry_delisle_PhD/code/readyQ_proto/links2.hpp (revision 59f3f616c737817ce7af766fa1bcf77a79d695d9) @@ -56,9 +56,11 @@ template class mpsc_queue : private mcs_queue { - node_t * volatile head; + node_t * volatile _head; public: - mpsc_queue(): mcs_queue(), head(nullptr) {} + mpsc_queue(): mcs_queue(), _head(nullptr) {} inline bool empty() const { return mcs_queue::empty(); } + + node_t * head() const { return _head; } // Added a new element to the queue @@ -66,5 +68,5 @@ inline node_t * push(node_t * elem) { node_t * prev = mcs_queue::push(elem); - if (!prev) head = elem; + if (!prev) _head = elem; return prev; } @@ -75,5 +77,5 @@ // NOT Multi-Thread Safe inline node_t * pop(node_t *& next) { - node_t * elem = head; + node_t * elem = _head; // If head is empty just return if (!elem) return nullptr; @@ -81,5 +83,5 @@ // If there is already someone in the list, then it's easy if (elem->_links.next) { - head = next = elem->_links.next; + _head = next = elem->_links.next; // force memory sync __atomic_thread_fence(__ATOMIC_SEQ_CST); @@ -93,5 +95,5 @@ // at the CAS in advance and therefore can write to head // after that point, it could overwrite the write in push - head = nullptr; + _head = nullptr; next = mcs_queue::advance(elem); @@ -99,5 +101,5 @@ // it is the only way we can guarantee we are not overwriting // a write made in push - if (next) head = next; + if (next) _head = next; } Index: doc/theses/thierry_delisle_PhD/code/readyQ_proto/utils.hpp =================================================================== --- doc/theses/thierry_delisle_PhD/code/readyQ_proto/utils.hpp (revision 6a8208cbc93288ea3ca61f2a04619465c12cf317) +++ doc/theses/thierry_delisle_PhD/code/readyQ_proto/utils.hpp (revision 59f3f616c737817ce7af766fa1bcf77a79d695d9) @@ -11,5 +11,5 @@ #include -#include +// #include // class Random { Index: doc/theses/thierry_delisle_PhD/code/readyQ_proto/work_stealing.hpp =================================================================== --- doc/theses/thierry_delisle_PhD/code/readyQ_proto/work_stealing.hpp (revision 6a8208cbc93288ea3ca61f2a04619465c12cf317) +++ doc/theses/thierry_delisle_PhD/code/readyQ_proto/work_stealing.hpp (revision 59f3f616c737817ce7af766fa1bcf77a79d695d9) @@ -15,5 +15,5 @@ #include "snzi.hpp" -#include +// #include using namespace std; @@ -28,7 +28,28 @@ template struct __attribute__((aligned(128))) localQ_t { - mpsc_queue queue = {}; - spinlock_t lock = {}; - bool needs_help = true; + #ifdef NO_MPSC + intrusive_queue_t list; + + inline auto ts() { return list.ts(); } + inline auto lock() { return list.lock.lock(); } + inline auto try_lock() { return list.lock.try_lock(); } + inline auto unlock() { return list.lock.unlock(); } + + inline auto push( node_t * node ) { return list.push( node ); } + inline auto pop() { return list.pop(); } + #else + mpsc_queue queue = {}; + spinlock_t _lock = {}; + + inline auto ts() { auto h = queue.head(); return h ? h->_links.ts : 0ull; } + inline auto lock() { return _lock.lock(); } + inline auto try_lock() { return _lock.try_lock(); } + inline auto unlock() { return _lock.unlock(); } + + inline auto push( node_t * node ) { return queue.push( node ); } + inline auto pop() { return queue.pop(); } + #endif + + }; @@ -44,5 +65,6 @@ work_stealing(unsigned _numThreads, unsigned) : numThreads(_numThreads * nqueues) - , lists(new intrusive_queue_t[numThreads]) + , lists(new localQ_t[numThreads]) + // , lists(new intrusive_queue_t[numThreads]) , times(new timestamp_t[numThreads]) // , snzi( std::log2( numThreads / 2 ), 2 ) @@ -58,10 +80,12 @@ __attribute__((noinline, hot)) void push(node_t * node) { - // node->_links.ts = rdtscl(); - node->_links.ts = 1; + node->_links.ts = rdtscl(); + // node->_links.ts = 1; auto & list = *({ unsigned i; - do { + #ifdef NO_MPSC + do { + #endif tls.stats.push.attempt++; // unsigned r = tls.rng1.next(); @@ -72,10 +96,14 @@ i = tls.my_queue + (r % nqueues); } - } while(!lists[i].lock.try_lock()); + #ifdef NO_MPSC + } while(!lists[i].try_lock()); + #endif &lists[i]; }); list.push( node ); - list.lock.unlock(); + #ifdef NO_MPSC + list.unlock(); + #endif // tls.rng2.set_raw_state( tls.rng1.get_raw_state()); // count++; @@ -84,17 +112,36 @@ __attribute__((noinline, hot)) node_t * pop() { - if( tls.myfriend == outside ) { - auto r = tls.rng1.next(); - tls.myfriend = r % numThreads; - times[tls.myfriend].val = 0; - } - else if(times[tls.myfriend].val == 0) { - node_t * n = try_pop(tls.myfriend, tls.stats.pop.help); - tls.stats.help++; - tls.myfriend = outside; - if(n) return n; - } - if(tls.my_queue != outside) { + // if( tls.myfriend == outside ) { + // auto r = tls.rng1.next(); + // tls.myfriend = r % numThreads; + // // assert(lists[(tls.it % nqueues) + tls.my_queue].ts() >= lists[((tls.it + 1) % nqueues) + tls.my_queue].ts()); + // tls.mytime = std::min(lists[(tls.it % nqueues) + tls.my_queue].ts(), lists[((tls.it + 1) % nqueues) + tls.my_queue].ts()); + // // times[tls.myfriend].val = 0; + // // lists[tls.myfriend].val = 0; + // } + // // else if(times[tls.myfriend].val == 0) { + // // else if(lists[tls.myfriend].val == 0) { + // else if(times[tls.myfriend].val < tls.mytime) { + // // else if(times[tls.myfriend].val < lists[(tls.it % nqueues) + tls.my_queue].ts()) { + // node_t * n = try_pop(tls.myfriend, tls.stats.pop.help); + // tls.stats.help++; + // tls.myfriend = outside; + // if(n) return n; + // } + // if( tls.myfriend == outside ) { + // auto r = tls.rng1.next(); + // tls.myfriend = r % numThreads; + // tls.mytime = lists[((tls.it + 1) % nqueues) + tls.my_queue].ts(); + // } + // else { + // if(times[tls.myfriend].val + 1000 < tls.mytime) { + // node_t * n = try_pop(tls.myfriend, tls.stats.pop.help); + // tls.stats.help++; + // if(n) return n; + // } + // tls.myfriend = outside; + // } + node_t * n = local(); if(n) return n; @@ -112,6 +159,8 @@ private: inline node_t * local() { - // unsigned i = (tls.rng2.prev() % 4) + tls.my_queue; unsigned i = (--tls.it % nqueues) + tls.my_queue; + node_t * n = try_pop(i, tls.stats.pop.local); + if(n) return n; + i = (--tls.it % nqueues) + tls.my_queue; return try_pop(i, tls.stats.pop.local); } @@ -153,10 +202,9 @@ // If we can't get the lock, move on - if( !list.lock.try_lock() ) { stat.elock++; return nullptr; } - + if( !list.try_lock() ) { stat.elock++; return nullptr; } // If list is empty, unlock and retry if( list.ts() == 0 ) { - list.lock.unlock(); + list.unlock(); stat.eempty++; return nullptr; @@ -164,10 +212,15 @@ auto node = list.pop(); - list.lock.unlock(); + list.unlock(); stat.success++; - times[i].val = 1; //node.first->_links.ts; - // count--; - // _mm_stream_si64((long long int*)×[i].val, node.first->_links.ts); - return node.first; + #ifdef NO_MPSC + // times[i].val = 1; + times[i].val = node.first->_links.ts; + // lists[i].val = node.first->_links.ts; + return node.first; + #else + times[i].val = node->_links.ts; + return node; + #endif } @@ -191,4 +244,5 @@ unsigned my_queue = calc_preferred(); unsigned myfriend = outside; + unsigned long long int mytime = 0; #if defined(READ) unsigned it = 0; @@ -211,5 +265,6 @@ private: const unsigned numThreads; - std::unique_ptr []> lists; + std::unique_ptr []> lists; + // std::unique_ptr []> lists; std::unique_ptr times; __attribute__((aligned(128))) std::atomic_size_t count;