Commit ec7de3cf authored by Dominik Charousset's avatar Dominik Charousset

use atomic head/tail pointer in prod/consumer list

this patch makes the head and tail pointer of
util::producer_consumer_list atomic in order to have
a thread-safe implementation of the empty() member function
parent 2d9a9802
...@@ -42,7 +42,7 @@ ...@@ -42,7 +42,7 @@
#if !defined(_GLIBCXX_USE_SCHED_YIELD) && !defined(__clang__) #if !defined(_GLIBCXX_USE_SCHED_YIELD) && !defined(__clang__)
#include <time.h> #include <time.h>
namespace std { namespace this_thread { namespace { namespace std { namespace this_thread { namespace {
inline void yield() noexcept { inline void yield() throw {
timespec req; timespec req;
req.tv_sec = 0; req.tv_sec = 0;
req.tv_nsec = 1; req.tv_nsec = 1;
...@@ -87,13 +87,26 @@ class producer_consumer_list { ...@@ -87,13 +87,26 @@ class producer_consumer_list {
typedef value_type* pointer; typedef value_type* pointer;
typedef const value_type* const_pointer; typedef const value_type* const_pointer;
struct node { class node {
public:
pointer value; pointer value;
std::atomic<node*> next; std::atomic<node*> next;
node(pointer val) : value(val), next(nullptr) { } node(pointer val) : value(val), next(nullptr) { }
private:
static constexpr size_type payload_size = static constexpr size_type payload_size =
sizeof(pointer) + sizeof(std::atomic<node*>); sizeof(pointer) + sizeof(std::atomic<node*>);
static constexpr size_type pad_size = (CPPA_CACHE_LINE_SIZE * ((payload_size / CPPA_CACHE_LINE_SIZE) + 1)) - payload_size;
static constexpr size_type cline_size = CPPA_CACHE_LINE_SIZE;
static constexpr size_type pad_size =
(cline_size * ((payload_size / cline_size) + 1)) - payload_size;
// avoid false sharing // avoid false sharing
char pad[pad_size]; char pad[pad_size];
...@@ -105,11 +118,11 @@ class producer_consumer_list { ...@@ -105,11 +118,11 @@ class producer_consumer_list {
"sizeof(node*) >= CPPA_CACHE_LINE_SIZE"); "sizeof(node*) >= CPPA_CACHE_LINE_SIZE");
// for one consumer at a time // for one consumer at a time
node* m_first; std::atomic<node*> m_first;
char m_pad1[CPPA_CACHE_LINE_SIZE - sizeof(node*)]; char m_pad1[CPPA_CACHE_LINE_SIZE - sizeof(node*)];
// for one producers at a time // for one producers at a time
node* m_last; std::atomic<node*> m_last;
char m_pad2[CPPA_CACHE_LINE_SIZE - sizeof(node*)]; char m_pad2[CPPA_CACHE_LINE_SIZE - sizeof(node*)];
// shared among producers // shared among producers
...@@ -119,7 +132,9 @@ class producer_consumer_list { ...@@ -119,7 +132,9 @@ class producer_consumer_list {
public: public:
producer_consumer_list() { producer_consumer_list() {
m_first = m_last = new node(nullptr); auto ptr = new node(nullptr);
m_first = ptr;
m_last = ptr;
m_consumer_lock = false; m_consumer_lock = false;
m_producer_lock = false; m_producer_lock = false;
} }
...@@ -127,7 +142,7 @@ class producer_consumer_list { ...@@ -127,7 +142,7 @@ class producer_consumer_list {
~producer_consumer_list() { ~producer_consumer_list() {
while (m_first) { while (m_first) {
node* tmp = m_first; node* tmp = m_first;
m_first = tmp->next; m_first = tmp->next.load();
delete tmp; delete tmp;
} }
} }
...@@ -140,7 +155,7 @@ class producer_consumer_list { ...@@ -140,7 +155,7 @@ class producer_consumer_list {
std::this_thread::yield(); std::this_thread::yield();
} }
// publish & swing last forward // publish & swing last forward
m_last->next = tmp; m_last.load()->next = tmp;
m_last = tmp; m_last = tmp;
// release exclusivity // release exclusivity
m_producer_lock = false; m_producer_lock = false;
...@@ -154,7 +169,7 @@ class producer_consumer_list { ...@@ -154,7 +169,7 @@ class producer_consumer_list {
} }
// only one consumer allowed // only one consumer allowed
node* first = m_first; node* first = m_first;
node* next = m_first->next; node* next = m_first.load()->next;
if (next) { if (next) {
// queue is not empty // queue is not empty
result = next->value; // take it out of the node result = next->value; // take it out of the node
...@@ -176,9 +191,7 @@ class producer_consumer_list { ...@@ -176,9 +191,7 @@ class producer_consumer_list {
} }
bool empty() const { bool empty() const {
// this seems to be a non-thread-safe implementation, // atomically compares first and last pointer without locks
// however, any 'race condition' that might occur
// only means we cannot assume an empty list
return m_first == m_last; return m_first == m_last;
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment