Improve tests & fix bugs.

Still one bug remain.

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, previous;
-  unsigned int idx;
+ 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);
+    // Load the number of threads using this index
-    // Reserve the index
+    unsigned int c = atomic_load(&s->cptTab[idx]);
-    unsigned int c = atomic_fetch_add(&s->cptTab[idx], 1);
     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)		\

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[i], conso2, NULL);
-  m_thread_create (idx_w, prod, 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);
 }