coffee/cpp-ipc
1
0
Fork 0
mirror of https://github.com/mutouyun/cpp-ipc.git synced 2025-12-06 16:56:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

using yield when checks read finished failed; fix bugs

Browse Source
This commit is contained in:
mutouyun 2018-11-26 16:53:41 +08:00
parent b3b0cc59f8
commit 453f93a69e
2 changed files with 9 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/circ_elem_array.h
View File

@ -7,6 +7,7 @@
#include <utility> #include <utility>
#include <limits> #include <limits>
#include <algorithm> #include <algorithm>
#include <thread>
namespace ipc { namespace ipc {
@ -102,7 +103,7 @@ public:
} }
void* acquire(void) { void* acquire(void) {
auto el = elem(wt_.fetch_add(1, std::memory_order_acq_rel)); auto el = elem(wt_.load(std::memory_order_consume));
// check read finished by all consumers // check read finished by all consumers
do { do {
uc_t expected = 0; uc_t expected = 0;
@ -110,7 +111,9 @@ public:
expected, static_cast<uc_t>(conn_count()), std::memory_order_acq_rel)) { expected, static_cast<uc_t>(conn_count()), std::memory_order_acq_rel)) {
break; break;
} }
std::this_thread::yield();
} while(1); } while(1);
wt_.fetch_add(1, std::memory_order_release);
return el->data_; return el->data_;
} }
@ -119,7 +122,7 @@ public:
ui_t wi = index_of(el); // get the index of this element (the write index) ui_t wi = index_of(el); // get the index of this element (the write index)
do { do {
bool no_next, cas; bool no_next, cas;
uc_t curr = cr_.load(std::memory_order_relaxed), next; uc_t curr = cr_.load(std::memory_order_consume), next;
do { do {
next = curr; next = curr;
if ((no_next = (index_of(curr) != wi)) /* assign & judge */) { if ((no_next = (index_of(curr) != wi)) /* assign & judge */) {
@ -127,21 +130,21 @@ public:
* commit is not the current commit * commit is not the current commit
* set wf_ for the other producer thread which is commiting * set wf_ for the other producer thread which is commiting
* the element matches cr_ could see it has commited * the element matches cr_ could see it has commited
*/ */
el->head_.wf_.store(1, std::memory_order_release); el->head_.wf_.store(1, std::memory_order_release);
} }
else { else {
/* /*
* commit is the current commit * commit is the current commit
* so we just increase the cursor & go check the next * so we just increase the cursor & go check the next
*/ */
++next; ++next;
el->head_.wf_.store(0, std::memory_order_release); el->head_.wf_.store(0, std::memory_order_release);
} }
/* /*
* it needs to go back and judge again * it needs to go back and judge again
* when cr_ has been changed by the other producer thread * when cr_ has been changed by the other producer thread
*/ */
} while(!(cas = cr_.compare_exchange_weak(curr, next, std::memory_order_acq_rel)) && no_next); } while(!(cas = cr_.compare_exchange_weak(curr, next, std::memory_order_acq_rel)) && no_next);
/* /*
* if compare_exchange failed & !no_next, * if compare_exchange failed & !no_next,
@ -151,7 +154,7 @@ public:
if (no_next || (!cas/* && !no_next*/)) return; if (no_next || (!cas/* && !no_next*/)) return;
/* /*
* check next element has commited or not * check next element has commited or not
*/ */
} while(el = elem(++wi), el->head_.wf_.exchange(0, std::memory_order_acq_rel)); } while(el = elem(++wi), el->head_.wf_.exchange(0, std::memory_order_acq_rel));
} }

1
test/test_circ_elem_array.cpp
View File

@ -5,7 +5,6 @@
#include <new> #include <new>
#include <vector> #include <vector>
#include <unordered_map> #include <unordered_map>
#include <thread>
#include "circ_elem_array.h" #include "circ_elem_array.h"
#include "test.h" #include "test.h"