Backport simplifications from cdeque-c11 to cdeque-native

Author: nhatminhle

libworkstream_df/cbuffer-native.h

@@ -27,16 +27,19 @@ typedef struct cbuffer{
 static inline cbuffer_p
 cbuffer_alloc (size_t log_size)
 {
+  void *p;
   cbuffer_p cbuffer;
-  if (posix_memalign ((void **)&cbuffer, 64, sizeof (cbuffer_t)))
+
+  if (posix_memalign (&p, 64, sizeof *cbuffer))
     wstream_df_fatal ("Out of memory ...");
 
+  cbuffer = p;
   cbuffer->log_size = log_size;
   cbuffer->size = (1 << log_size);
   cbuffer->modulo_mask = cbuffer->size - 1;
-  if (posix_memalign ((void **)&cbuffer->array, 64,
-		       sizeof (wstream_df_type) * cbuffer->size))
+  if (posix_memalign (&p, 64, sizeof *cbuffer->array * cbuffer->size))
     wstream_df_fatal ("Out of memory ...");
+  cbuffer->array = p;
 
   return cbuffer;
 }
@@ -50,12 +53,6 @@ cbuffer_free (cbuffer_p cbuffer)
     free (cbuffer);
 }
 
-static inline size_t
-cbuffer_size (cbuffer_p cbuffer)
-{
-  return cbuffer->size;
-}
-
 static inline wstream_df_type
 cbuffer_get (cbuffer_p cbuffer, size_t i)
 {
@@ -68,11 +65,11 @@ cbuffer_set (cbuffer_p cbuffer, size_t i, wstream_df_type elem)
   cbuffer->array[i & cbuffer->modulo_mask] = elem;
 }
 
-static inline void
-cbuffer_grow (cbuffer_p volatile *cbuffer, size_t bottom, size_t top)
+static inline cbuffer_p
+cbuffer_grow (cbuffer_p old_cbuffer, size_t bottom, size_t top,
+	      cbuffer_p volatile *pnew)
 {
-  cbuffer_p new_cbuffer = cbuffer_alloc ((*cbuffer)->log_size + 1);
-  cbuffer_p old_cbuffer = *cbuffer;
+  cbuffer_p new_cbuffer = cbuffer_alloc (old_cbuffer->log_size + 1);
 
   size_t old_buffer_size = old_cbuffer->size;
   size_t old_top_pos = top & old_cbuffer->modulo_mask;
@@ -81,43 +78,45 @@ cbuffer_grow (cbuffer_p volatile *cbuffer, size_t bottom, size_t top)
   size_t new_top_pos = top & new_cbuffer->modulo_mask;
   size_t new_bot_pos = bottom & new_cbuffer->modulo_mask;
 
-  /* If no wrap-around for old buffer, new one can't have one if size
-     is doubled.  */
   if (old_top_pos < old_bot_pos)
-    memcpy (&new_cbuffer->array[new_top_pos],
-	    &(*cbuffer)->array[old_top_pos],
-	    (old_bot_pos - old_top_pos) * sizeof (wstream_df_type));
-
-  /* If old buffer wraps around, then either new one wraps around at
-     same place or it just doesn't.  */
+    {
+      /* If no wrap-around for old buffer, new one can't have one if
+	 size is doubled.  */
+      memcpy (&new_cbuffer->array[new_top_pos],
+	      &old_cbuffer->array[old_top_pos],
+	      (old_bot_pos - old_top_pos) * sizeof (wstream_df_type));
+    }
   else
     {
+      /* If old buffer wraps around, then either new one wraps around
+	 at same place or it just doesn't.  */
+
       /* No wrap around in new buffer?  */
-      int no_wrap_around = (new_top_pos < new_bot_pos) ? 1 : 0;
+      int no_wrap_around = new_top_pos < new_bot_pos;
 
       memcpy (&new_cbuffer->array[new_top_pos],
-	      &(*cbuffer)->array[old_top_pos],
+	      &old_cbuffer->array[old_top_pos],
 	      (old_buffer_size - old_top_pos) * sizeof (wstream_df_type));
       memcpy (&new_cbuffer->array[no_wrap_around * old_buffer_size],
-	      (*cbuffer)->array,
+	      old_cbuffer->array,
 	      (old_bot_pos) * sizeof (wstream_df_type));
     }
 
   store_store_fence ();
-
-  *cbuffer = new_cbuffer;
-
+  *pnew = new_cbuffer;
   store_store_fence ();
 
   /* XXX(nhatle): Race condition with steal() on freed buffer? */
   cbuffer_free (old_cbuffer);
+
+  return new_cbuffer;
 }
 
 static inline void
 print_cbuffer (cbuffer_p cbuffer)
 {
   size_t i;
-  for (i = 0; i < cbuffer_size (cbuffer); i++)
+  for (i = 0; i < cbuffer->size; i++)
     printf ("%p,", cbuffer_get (cbuffer, i));
   printf ("\n");
 }

libworkstream_df/cdeque-c11.c.h

@@ -1,3 +1,6 @@
+#include <stdatomic.h>
+#include <stddef.h>
+
 #include "papi-defs.h"
 #include "wstream_df.h"
 #include "error.h"
@@ -32,8 +35,8 @@ cdeque_take (cdeque_p cdeque)
 {
   _PAPI_P1B;
   size_t bottom, top;
-  cbuffer_p buffer;
   wstream_df_type task;
+  cbuffer_p buffer;
 
   bottom = atomic_load_explicit (&cdeque->bottom, relaxed) - 1;
   buffer = atomic_load_explicit (&cdeque->cbuffer, relaxed);

libworkstream_df/cdeque-native.c.h

@@ -13,15 +13,16 @@ cdeque_push_bottom (cdeque_p cdeque, wstream_df_type elem)
   size_t bottom = cdeque->bottom;
   size_t top = cdeque->top;
 
-  XLOG ("cdeque_push_bottom with elem: %d\n", elem);
-  if (bottom >= top + cdeque_size (cdeque))
-    cdeque_grow (cdeque, bottom, top);
+  cbuffer_p buffer = cdeque->cbuffer;
 
-  cdeque_set (cdeque, bottom, elem);
+  XLOG ("cdeque_push_bottom with elem: %d\n", elem);
+  if (bottom >= top + buffer->size)
+    buffer = cbuffer_grow (buffer, bottom, top, &cdeque->cbuffer);
 
+  cbuffer_set (buffer, bottom, elem);
   store_store_fence ();
-
   cdeque->bottom = bottom + 1;
+
   _PAPI_P0E;
 }
 
@@ -32,13 +33,9 @@ cdeque_take (cdeque_p cdeque)
   _PAPI_P1B;
   size_t bottom, top;
   wstream_df_type task;
+  cbuffer_p buffer;
 
-  if (cdeque->bottom == 0)
-    {
-      _PAPI_P1E;
-      return NULL;
-    }
-
+  buffer = cdeque->cbuffer;
 #if LLSC_OPTIMIZATION && defined(__arm__)
   do
     bottom = load_linked (&cdeque->bottom) - 1;
@@ -49,27 +46,29 @@ cdeque_take (cdeque_p cdeque)
   cdeque->bottom = bottom;
   store_load_fence ();
 #endif
+
   top = cdeque->top;
 
-  if (bottom < top)
+  if (bottom == (size_t) -1 || bottom < top)
     {
-      cdeque->bottom = top;
+      cdeque->bottom = bottom + 1;
       _PAPI_P1E;
       return NULL;
     }
 
-  task = cdeque_get (cdeque, bottom);
+  task = cbuffer_get (buffer, bottom);
 
   if (bottom > top)
     {
       _PAPI_P1E;
       return task;
     }
+
   /* One compare and swap when the deque has one single element.  */
   if (!compare_and_swap (&cdeque->top, top, top+1))
     task = NULL;
-
   cdeque->bottom = top + 1;
+
   _PAPI_P1E;
   return task;
 }
@@ -81,15 +80,18 @@ cdeque_steal (cdeque_p remote_cdeque)
   _PAPI_P2B;
   size_t bottom, top;
   wstream_df_type elem;
+  cbuffer_p buffer;
 
   top = remote_cdeque->top;
+
 #if defined(__arm__)
   /* Block until the value read from top has been propagated to all
      other threads. */
   store_load_fence ();
 #else
   load_load_fence (top);
 #endif
+
 #if LLSC_OPTIMIZATION && defined(__arm__)
   bottom = load_linked (&remote_cdeque->bottom);
   if (!store_conditional (&remote_cdeque->bottom, bottom))
@@ -98,6 +100,7 @@ cdeque_steal (cdeque_p remote_cdeque)
 #else
   bottom = remote_cdeque->bottom;
 #endif
+
   load_load_fence (bottom);
 
   XLOG ("cdeque_steal with bottom %d, top %d\n", bottom, top);
@@ -108,7 +111,8 @@ cdeque_steal (cdeque_p remote_cdeque)
       return NULL;
     }
 
-  elem = cdeque_get (remote_cdeque, top);
+  buffer = remote_cdeque->cbuffer;
+  elem = cbuffer_get (buffer, top);
 #if defined(__arm__)
   /* Do not reorder the previous load with the load from the CAS. */
   load_load_fence ((uintptr_t) elem);

libworkstream_df/cdeque-native.h

@@ -23,34 +23,6 @@ typedef struct cdeque {
   cbuffer_p volatile cbuffer __attribute__ ((aligned (64)));
 } cdeque_t, *cdeque_p;
 
-/* Get the size/capacity of the deque CDEQUE.  */
-static inline size_t
-cdeque_size (cdeque_p cdeque)
-{
-  return cbuffer_size (cdeque->cbuffer);
-}
-
-/* Extend the size of the deque CDEQUE, with bottom BOTTOM and top TOP.  */
-static inline void
-cdeque_grow (cdeque_p cdeque, size_t bottom, size_t top)
-{
-  cbuffer_grow (&cdeque->cbuffer, bottom, top);
-}
-
-/* Set the deque at position POS with element ELEM.  */
-static inline void
-cdeque_set (cdeque_p cdeque, size_t pos, wstream_df_type elem)
-{
-  cbuffer_set (cdeque->cbuffer, pos, elem);
-}
-
-/* Get the element from the deque CDEQUE at position POS.  */
-static inline wstream_df_type
-cdeque_get (cdeque_p cdeque, size_t pos)
-{
-  return cbuffer_get (cdeque->cbuffer, pos);
-}
-
 static inline void
 cdeque_init (cdeque_p cdeque, size_t log_size)
 {
@@ -64,10 +36,9 @@ static inline cdeque_p
 cdeque_alloc (size_t log_size)
 {
   cdeque_p cdeque = (cdeque_p) malloc (sizeof (cdeque_t));
-  cdeque->bottom = 0;
-  cdeque->top = 0;
-  cdeque->cbuffer = cbuffer_alloc (log_size);
-
+  if (cdeque == NULL)
+    wstream_df_fatal ("Out of memory ...");
+  cdeque_init (cdeque, log_size);
   return cdeque;
 }
 
@@ -84,8 +55,9 @@ static inline void
 print_cdeque (cdeque_p cdeque)
 {
   size_t i;
+  cbuffer_p buffer = cdeque->cbuffer;
   for (i = cdeque->top; i < cdeque->bottom; i++)
-    printf ("%p,", cdeque_get (cdeque, i));
+    printf ("%p,", cbuffer_get (buffer, i));
   printf ("\n");
 }