cds/container/chase-lev-deque.h

   1 #ifndef _CHASE_LEV_DEQUE_H
   2 #define _CHASE_LEV_DEQUE_H
   3
   4 #include <atomic>
   5 #include <cds/sync/backoff.h>
   6 #include <cstdlib>
   7 #include <inttypes.h>
   8 #include <iostream>
   9
  10 namespace cds_others {
  11
  12 #define EMPTY 0xffffffff
  13 #define ABORT 0xfffffffe
  14
  15 using std::memory_order_seq_cst;
  16 using std::memory_order_release;
  17 using std::memory_order_acquire;
  18 using std::memory_order_acq_rel;
  19 using std::memory_order_relaxed;
  20 using std::atomic_int;
  21 using std::atomic_size_t;
  22 using std::atomic_uintptr_t;
  23 using std::size_t;
  24
  25 class ChaseLevDeque {
  26 private:
  27   atomic_size_t top;
  28   atomic_size_t bottom;
  29   atomic_uintptr_t array; /* Atomic(Array *) */
  30
  31 public:
  32   struct Array {
  33     atomic_size_t size;
  34     atomic_int buffer[];
  35   };
  36
  37   ChaseLevDeque() {
  38     Array *a = (Array *)calloc(1, sizeof(Array) + 2 * sizeof(atomic_int));
  39     array.store((uintptr_t)a, memory_order_relaxed);
  40     top.store(0, memory_order_relaxed);
  41     bottom.store(0, memory_order_relaxed);
  42     a->size.store(2, memory_order_relaxed);
  43   }
  44
  45   int take() {
  46     size_t b = bottom.load(memory_order_relaxed) - 1;
  47     Array *a = (Array *)array.load(memory_order_relaxed);
  48     bottom.store(b, memory_order_relaxed);
  49     atomic_thread_fence(memory_order_seq_cst);
  50     size_t t = top.load(memory_order_relaxed);
  51     int x;
  52     if (t <= b) {
  53       /* Non-empty queue. */
  54       x = a->buffer[b % a->size.load(memory_order_relaxed)].load(
  55           memory_order_relaxed);
  56       if (t == b) {
  57         /* Single last element in queue. */
  58         if (!top.compare_exchange_strong(t, t + 1, memory_order_seq_cst,
  59                                          memory_order_relaxed))
  60           /* Failed race. */
  61           x = EMPTY;
  62         bottom.store(b + 1, memory_order_relaxed);
  63       }
  64     } else { /* Empty queue. */
  65       x = EMPTY;
  66       bottom.store(b + 1, memory_order_relaxed);
  67     }
  68     return x;
  69   }
  70
  71   void resize() {
  72     Array *a = (Array *)array.load(memory_order_relaxed);
  73     size_t size = a->size.load(memory_order_relaxed);
  74     size_t new_size = size << 1;
  75     Array *new_a =
  76         (Array *)calloc(1, new_size * sizeof(atomic_int) + sizeof(Array));
  77     size_t t = top.load(memory_order_relaxed);
  78     size_t b = bottom.load(memory_order_relaxed);
  79     new_a->size.store(new_size, memory_order_relaxed);
  80     size_t i;
  81     for (i = t; i < b; i++) {
  82       new_a->buffer[i % new_size].store(
  83           a->buffer[i % size].load(memory_order_relaxed), memory_order_relaxed);
  84     }
  85     array.store((uintptr_t)new_a, memory_order_release);
  86     //    std::cout << "Resize to " << new_size << "\n";
  87   }
  88
  89   void push(int x) {
  90     size_t b = bottom.load(memory_order_relaxed);
  91     size_t t = top.load(memory_order_acquire);
  92     Array *a = (Array *)array.load(memory_order_relaxed);
  93     if (b - t > a->size.load(memory_order_relaxed) - 1) /* Full queue. */ {
  94       resize();
  95       // Bug in paper...should have next line...
  96       a = (Array *)array.load(memory_order_relaxed);
  97     }
  98     a->buffer[b % a->size.load(memory_order_relaxed)].store(
  99         x, memory_order_relaxed);
 100     atomic_thread_fence(memory_order_release);
 101     bottom.store(b + 1, memory_order_relaxed);
 102   }
 103
 104   int steal() {
 105     size_t t = top.load(memory_order_acquire);
 106     atomic_thread_fence(memory_order_seq_cst);
 107     size_t b = bottom.load(memory_order_acquire);
 108     int x = EMPTY;
 109     if (t < b) {
 110       /* Non-empty queue. */
 111       Array *a = (Array *)array.load(memory_order_acquire);
 112       x = a->buffer[t % a->size.load(memory_order_relaxed)].load(
 113           memory_order_relaxed);
 114       if (!top.compare_exchange_strong(t, t + 1, memory_order_seq_cst,
 115                                        memory_order_relaxed))
 116         /* Failed race. */
 117         return ABORT;
 118     }
 119     return x;
 120   }
 121 };
 122
 123 } // namespace cds_others
 124
 125 #endif