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

still have bugs

Browse Source
This commit is contained in:
mutouyun 2018-11-26 21:36:19 +08:00
parent 2d86727be1
commit db35146542
1 changed files with 26 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/circ_elem_array.h
View File

@ -37,7 +37,8 @@ template <std::size_t DataSize>
class elem_array : private elem_array_head { class elem_array : private elem_array_head {
struct head_t { struct head_t {
ac_t rf_; // read flag ac_t rf_; // read flag
ac_t wf_; // write flag std::atomic_bool wf_; // write flag
std::atomic_flag acq_; // acquire flag
}; };
public: public:
@ -102,9 +103,16 @@ public:
} }
void* acquire(void) { void* acquire(void) {
auto el = elem(wt_.load(std::memory_order_consume)); elem_t* el;
while (1) {
// searching an available element
el = elem(wt_.fetch_add(1, std::memory_order_acquire));
if (el->head_.acq_.test_and_set(std::memory_order_release)) {
std::this_thread::yield();
continue;
}
// check read finished by all consumers // check read finished by all consumers
do { while(1) {
uc_t expected = 0; uc_t expected = 0;
std::atomic_thread_fence(std::memory_order_acquire); std::atomic_thread_fence(std::memory_order_acquire);
if (el->head_.rf_.compare_exchange_weak( if (el->head_.rf_.compare_exchange_weak(
@ -112,8 +120,10 @@ public:
break; break;
} }
std::this_thread::yield(); std::this_thread::yield();
} while(1); }
wt_.fetch_add(1, std::memory_order_release); el->head_.acq_.clear(std::memory_order_release);
break;
}
return el->data_; return el->data_;
} }
@ -131,7 +141,7 @@ public:
* 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(true, std::memory_order_release);
} }
else { else {
/* /*
@ -139,7 +149,7 @@ public:
* 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(false, std::memory_order_release);
} }
/* /*
* it needs to go back and judge again * it needs to go back and judge again
@ -155,7 +165,7 @@ public:
/* /*
* 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(false, std::memory_order_acq_rel));
} }
uc_t cursor(void) const { uc_t cursor(void) const {