Rewrite & fix int queue generator.

Limitation to 4096 max entries.
A classic queue may be way faster (too much sharing for thread) but will use much more memory...
To bench

m-genint.h

@@ -38,10 +38,16 @@
    This can be used to map integers to index of ressources in a table.
 */
 
+// TODO: We may want a specialization with constant N value.
+
 typedef unsigned long long genint_limb_t;
 
 typedef struct genint_s {
   size_t n;
+  size_t max;
+  genint_limb_t mask0; 
+  genint_limb_t mask_master;
+  atomic_ullong master;
   atomic_ullong *data;
 } genint_t[1];
 
@@ -53,11 +59,11 @@ typedef struct genint_s {
     assert (s->data != NULL);                                   \
   } while (0)
 
-#define GENINT_ONE ((genint_limb_t)1)
+#define GENINT_ONE  ((genint_limb_t)1)
 
 static inline void genint_init(genint_t s, size_t n)
 {
-  assert (s != NULL && n > 0);
+  assert (s != NULL && n > 0 && n <= GENINT_LIMBSIZE * GENINT_LIMBSIZE );
   const size_t alloc = (n + GENINT_LIMBSIZE - 1) / GENINT_LIMBSIZE;
   atomic_ullong *ptr = M_MEMORY_REALLOC (atomic_ullong, NULL, alloc);
   s->n = n;
@@ -66,8 +72,13 @@ static inline void genint_init(genint_t s, size_t n)
     M_MEMORY_FULL(alloc);
     return;
   }
+  s->max = alloc-1;
+  const size_t index  = s->n % GENINT_LIMBSIZE;
+  s->mask0 = (index == 0) ? -GENINT_ONE : ~((GENINT_ONE<<(GENINT_LIMBSIZE-index))-1);
+  s->mask_master = ((GENINT_ONE << alloc) - 1) << (GENINT_LIMBSIZE-alloc);
+  atomic_init (&s->master, (genint_limb_t)0);
   for(size_t i = 0; i < alloc; i++)
-    atomic_init(&s->data[i], 0UL);
+    atomic_init(&s->data[i], (genint_limb_t)0);
   GENINT_CONTRACT(s);
 }
 
@@ -81,29 +92,41 @@ static inline void genint_clear(genint_t s)
 static inline size_t genint_pop(genint_t s)
 {
   GENINT_CONTRACT(s);
-  const size_t max = (s->n + GENINT_LIMBSIZE - 1) / GENINT_LIMBSIZE;
+  // First read master to see which limb is not full.
-  const size_t index  = s->n % GENINT_LIMBSIZE;
+  genint_limb_t master = atomic_load(&s->master);
-  const genint_limb_t mask0 = (index == 0) ? -GENINT_ONE : (GENINT_ONE<<index)-1;
+  while (master != s->mask_master) {
-  // O(n) approach, but with a really low constant factor
+    // Let's get the index i of the first not full limb.
-  for(size_t i = 0; i < max; i++)
+    unsigned int i = m_core_clz(~master);
-    {
+    // An compute the mask of this limb being full
-      const size_t mask = (i+1 == max) ? mask0 : -GENINT_ONE;
+    const size_t mask = (i == s->max) ? s->mask0 : -GENINT_ONE;
-      unsigned int bit;
+    unsigned int bit;
-      genint_limb_t next, org = atomic_load(&s->data[i]);
+    // Let's load this limb,
-      do {
+    genint_limb_t next, org = atomic_load(&s->data[i]);
-        if (org == mask)
+    do {
-          goto next_element;
+      // If it is now full, we have been prempted by another. Let's restart.
-        // At least one bit is free in the limb
+      if (M_UNLIKELY (org == mask))
-        bit = GENINT_LIMBSIZE - m_core_clz(org);
+        goto next_element;
-        assert ((org & (GENINT_ONE<<bit)) == 0);
+      // At least one bit is free in the limb
-        // Set & reserve the integer as being used.
+      bit = GENINT_LIMBSIZE - 1 - m_core_clz(~org);
-        next = org | (GENINT_ONE << bit);
+      assert ((org & (GENINT_ONE<<bit)) == 0);
-      } while (!atomic_compare_exchange_weak (&s->data[i], &org, next));
+      assert (i * GENINT_LIMBSIZE + GENINT_LIMBSIZE - 1 - bit < s->n);
-      // Return the new number
+      // Set & reserve the integer as being used.
-      GENINT_CONTRACT(s);
+      next = org | (GENINT_ONE << bit);
-      return i * GENINT_LIMBSIZE + bit;
+    } while (!atomic_compare_exchange_weak (&s->data[i], &org, next));
-    next_element: (void)1;
+    // If the limb is now full, try to update master
+    while (M_UNLIKELY(next == mask
+                      && !atomic_compare_exchange_weak (&s->master, &master, master|(GENINT_ONE << (GENINT_LIMBSIZE - 1 - i) ))
+                      && ((master & (GENINT_ONE << (GENINT_LIMBSIZE - 1 - i))) == 0) )) {
+      // Fail to update. Reload limb to check if it is still full.
+      next = atomic_load(&s->data[i]);
     }
+    // Return the new number
+    GENINT_CONTRACT(s);
+    return i * GENINT_LIMBSIZE + GENINT_LIMBSIZE - 1 - bit;
+  next_element:
+    // Reload master
+    master = atomic_load(&s->master);
+  }
   GENINT_CONTRACT(s);
   return -(size_t)1; // No more ressource available
 }
@@ -113,12 +136,22 @@ static inline void genint_push(genint_t s, size_t n)
   GENINT_CONTRACT(s);
   assert (n < s->n);
   size_t i    = n / GENINT_LIMBSIZE;
-  size_t bit  = n % GENINT_LIMBSIZE;
+  size_t bit  = GENINT_LIMBSIZE - 1 - (n % GENINT_LIMBSIZE);
   genint_limb_t next, org = atomic_load(&s->data[i]);
   do {
-    assert ((org & (GENINT_ONE<<bit)) != 0);
+    assert ((org & (GENINT_ONE << bit)) != 0);
     next = org & (~(GENINT_ONE << bit));
   } while (!atomic_compare_exchange_weak (&s->data[i], &org, next));
+  genint_limb_t master = atomic_load(&s->master);
+  genint_limb_t mask_i = GENINT_ONE << (GENINT_LIMBSIZE - 1 - i);
+  if (M_UNLIKELY (master & mask_i)) {
+    while (!atomic_compare_exchange_weak (&s->master, &master, master & (~mask_i)) && (master & mask_i)) {
+      const size_t mask = (i == s->max) ? s->mask0 : -GENINT_ONE;
+      org = atomic_load(&s->data[i]);
+      if (org == mask)
+        return;
+    }
+  }
   GENINT_CONTRACT(s);
 }
 

tests/test-mgenint.c

@@ -21,6 +21,8 @@
  * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
 #include <stdio.h>
+#include "m-mutex.h"
+
 #include "m-genint.h"
 
 static void test(size_t n)
@@ -46,10 +48,42 @@ static void test(size_t n)
   genint_clear(s);
 }
 
+genint_t global;
+
+static void conso(void *p)
+{
+  size_t n = *(size_t*)p;
+  for(int i = 0; i < 100000; i++) {
+    unsigned int j = genint_pop(global);
+    assert (j == -1U || j < n);
+    if (j != -1U)
+      genint_push(global, j);
+  }
+}
+
+static void test2(size_t n)
+{
+  m_thread_t idx[4];
+
+  genint_init(global, n);
+
+  for(int i = 0; i < 4; i++) {
+    m_thread_create (idx[i], conso, (void*)&n);
+  }
+  for(int i = 0; i < 4;i++) {
+    m_thread_join(idx[i]);
+  }
+
+  genint_init(global, n);
+}
+
 int main(void)
 {
   for(size_t n = 1; n < 256; n++) {
     test(n);
   }
+  for(size_t n = 1; n < 256; n+=17) {
+    test2(n);
+  }
   exit(0);
 }