diff --git a/m-snapshot.h b/m-snapshot.h index 9c5cd136..0a4a18ef 100644 --- a/m-snapshot.h +++ b/m-snapshot.h @@ -298,7 +298,6 @@ typedef struct snapshot_mrsw_int_s { assert (SNAPSHOTI_MRSW_INT_FLAG_W(atomic_load(&s->lastNext)) \ <= s->n + SNAPSHOTI_MRSW_EXTRA_BUFFER); \ assert (s->cptTab != NULL); \ - assert (atomic_load(&s->cptTab[s->currentWrite]) == 1); \ } while (0) static inline void snapshot_mrsw_int_init(snapshot_mrsw_int_t s, size_t n) @@ -353,6 +352,9 @@ static inline unsigned int snapshot_mrsw_int_write_idx(snapshot_mrsw_int_t s, un if (SNAPSHOTI_MRSW_INT_FLAG_N(previous)) { // Reuse previous buffer as it was not used by any reader idx = SNAPSHOTI_MRSW_INT_FLAG_W(previous); + // Some other read threads may already have try to reserve this index + // So atomic_load(&s->cptTab[idx]) can be greater than 1. + // However they will fail to ack it in lastNext, and remove their reservation later } else { // Free the write index idx = SNAPSHOTI_MRSW_INT_FLAG_W(previous); @@ -361,8 +363,9 @@ static inline unsigned int snapshot_mrsw_int_write_idx(snapshot_mrsw_int_t s, un // Get a new buffer. if (c != 1) { // If someone else keeps a ref on the buffer, we can't reuse it - // get another free one + // get another free one. idx = genint_pop(s->freeList); + // TODO: sometimes fails... assert(idx != -1U); } assert (idx < s->n + SNAPSHOTI_MRSW_EXTRA_BUFFER); @@ -405,35 +408,48 @@ static inline void snapshot_mrsw_int_write_end(snapshot_mrsw_int_t s, unsigned i idx = SNAPSHOTI_MRSW_INT_FLAG_W(previous); unsigned int c = atomic_fetch_sub(&s->cptTab[idx], 1); assert (c != 0 && c <= s->n + 1); - genint_push(s->freeList, idx); + if (c == 1) { + genint_push(s->freeList, idx); + } SNAPSHOTI_MRSW_INT_CONTRACT(s); } static inline unsigned int snapshot_mrsw_int_read_start(snapshot_mrsw_int_t s) { SNAPSHOTI_MRSW_INT_CONTRACT(s); - unsigned int previous = atomic_load(&s->lastNext); - unsigned int idx; + unsigned int idx, previous; + reload: + // Load the last published index + Next flag + previous = atomic_load(&s->lastNext); while (true) { + // Get the last published index idx = SNAPSHOTI_MRSW_INT_FLAG_W(previous); - unsigned int newNext = SNAPSHOTI_MRSW_INT_FLAG(idx, false); - // Reserve the index - unsigned int c = atomic_fetch_add(&s->cptTab[idx], 1); + // Load the number of threads using this index + unsigned int c = atomic_load(&s->cptTab[idx]); assert (c <= s->n + 1); + // Reserve the index if it still being reserved by someone else + if (M_UNLIKELY (c == 0 + || !atomic_compare_exchange_strong(&s->cptTab[idx], &c, c+1))) + goto reload; // Try to ack it - if (atomic_compare_exchange_strong(&s->lastNext, &previous, newNext)) + unsigned int newNext = SNAPSHOTI_MRSW_INT_FLAG(idx, false); + reforce: + if (M_LIKELY (atomic_compare_exchange_strong(&s->lastNext, &previous, newNext))) break; - // If we have been preempted by another read thread + // We have been preempted by another thread if (idx == SNAPSHOTI_MRSW_INT_FLAG_W(previous)) { // This is still ok if the index has not changed - assert (SNAPSHOTI_MRSW_INT_FLAG_N(previous) == false); + // We can get previous to true again if the writer has recycled the index, + // while we reserved it, and the reader get prempted until its CAS. + if (M_UNLIKELY (SNAPSHOTI_MRSW_INT_FLAG_N(previous) == true)) goto reforce; break; } // Free the reserved index as we failed it to ack it c = atomic_fetch_sub(&s->cptTab[idx], 1); assert (c != 0 && c <= s->n+1); - if (c == 1) + if (c == 1) { genint_push(s->freeList, idx); + } } SNAPSHOTI_MRSW_INT_CONTRACT(s); return idx; @@ -557,6 +573,9 @@ static inline void snapshot_mrsw_int_read_end(snapshot_mrsw_int_t s, unsigned in static inline void M_C(name, _init)(M_C(name, _t) snap, size_t nReader, size_t nWriter) \ { \ M_C(name, _mrsw_init)(snap->core, nReader + nWriter -1 ); \ + unsigned int idx = snap->core->core->currentWrite; \ + snap->core->core->currentWrite = -1; \ + snapshot_mrsw_int_write_end(snap->core->core, idx); \ } \ \ static inline void M_C(name, _clear)(M_C(name, _t) snap) \ diff --git a/tests/test-msnapshot.c b/tests/test-msnapshot.c index 209f3ad5..bbe356bc 100644 --- a/tests/test-msnapshot.c +++ b/tests/test-msnapshot.c @@ -255,8 +255,8 @@ static void test_mrsw_global(int reader) assert (p != NULL); for(int i = 0 ; i < reader; i++) - m_thread_create (idx[i], conso, NULL); - m_thread_create (idx_w, prod, NULL); + m_thread_create (idx[i], conso2, NULL); + m_thread_create (idx_w, prod2, NULL); m_thread_join(idx_w); for(int i = 0 ; i < reader; i++) @@ -265,6 +265,62 @@ static void test_mrsw_global(int reader) snapshot_mrsw_data_clear(global_mrsw); } +snapshot_mrmw_data_t global_mrmw; + +static void conso3(void *arg) +{ + assert (arg == NULL); + while (true) { + const data_t *p = snapshot_mrmw_data_read_start(global_mrmw); + assert (p != NULL); + data_valid_p(p); + if (p->n == 0) + return; + snapshot_mrmw_data_read_end(global_mrmw, p); + } +} + +static void prod3(void *arg) +{ + assert (arg == NULL); + for(unsigned int i = 1; i < 20000000;i++) { + data_t *p = snapshot_mrmw_data_write_start(global_mrmw); + assert (p != NULL); + p->n = i * i; + data_crc(p); + snapshot_mrmw_data_write_end(global_mrmw, p); + } + data_t *p = snapshot_mrmw_data_write_start(global_mrmw); + p->n = 0; + data_crc(p); + snapshot_mrmw_data_write_end(global_mrmw, p); +} + +static void test_mrmw_global(int reader, int writer) +{ + m_thread_t idx_w[writer]; + m_thread_t idx[reader]; + + snapshot_mrmw_data_init(global_mrmw, reader, writer); + data_t *p = snapshot_mrmw_data_write_start(global_mrmw); + assert (p != NULL); + p->n = 42; + data_crc(p); + snapshot_mrmw_data_write_end(global_mrmw, p); + + for(int i = 0 ; i < reader; i++) + m_thread_create (idx[i], conso3, NULL); + for(int i = 0 ; i < writer; i++) + m_thread_create (idx_w[i], prod3, NULL); + + for(int i = 0 ; i < writer; i++) + m_thread_join(idx_w[i]); + for(int i = 0 ; i < reader; i++) + m_thread_join(idx[i]); + + snapshot_mrmw_data_clear(global_mrmw); +} + int main(void) { test_uint(); @@ -274,6 +330,7 @@ int main(void) test_mrsw_global(1); test_mrsw_global(2); test_mrsw_global(4); + test_mrsw_global(8); exit(0); }