ms-queue/my_queue.c

   1 #include <threads.h>
   2 #include <stdlib.h>
   3 #include "librace.h"
   4
   5 #include "my_queue.h"
   6
   7 #define relaxed memory_order_relaxed
   8 #define release memory_order_release
   9 #define acquire memory_order_acquire
  10
  11 static unsigned int *node_nums;
  12
  13 static unsigned int new_node()
  14 {
  15         return node_nums[get_thread_num()];
  16 }
  17
  18 static void reclaim(unsigned int node)
  19 {
  20         node_nums[get_thread_num()] = node;
  21 }
  22
  23 void init_queue(queue_t *q, int num_threads)
  24 {
  25         unsigned int i;
  26         pointer head;
  27         pointer tail;
  28         pointer next;
  29
  30         node_nums = malloc(num_threads * sizeof(*node_nums));
  31         for (i = 0; i < num_threads; i++)
  32                 node_nums[i] = 2 + i;
  33
  34         /* initialize queue */
  35         head = MAKE_POINTER(1, 0);
  36         tail = MAKE_POINTER(1, 0);
  37         next = MAKE_POINTER(0, 0); // (NULL, 0)
  38
  39         atomic_init(&q->head, head);
  40         atomic_init(&q->tail, tail);
  41         atomic_init(&q->nodes[1].next, next);
  42
  43         /* initialize avail list */
  44         for (i = 2; i < MAX_NODES; i++) {
  45                 next = MAKE_POINTER(i + 1, 0);
  46                 atomic_init(&q->nodes[i].next, next);
  47         }
  48
  49         next = MAKE_POINTER(0, 0); // (NULL, 0)
  50         atomic_init(&q->nodes[MAX_NODES].next, next);
  51 }
  52
  53 void enqueue(queue_t *q, unsigned int val)
  54 {
  55         int success = 0;
  56         unsigned int node;
  57         pointer tail;
  58         pointer next;
  59         pointer tmp;
  60
  61         node = new_node();
  62         store_32(&q->nodes[node].value, val);
  63         tmp = atomic_load_explicit(&q->nodes[node].next, relaxed);
  64         set_ptr(&tmp, 0); // NULL
  65         atomic_store_explicit(&q->nodes[node].next, tmp, relaxed);
  66
  67         while (!success) {
  68                 tail = atomic_load_explicit(&q->tail, acquire);
  69                 next = atomic_load_explicit(&q->nodes[get_ptr(tail)].next, acquire);
  70                 if (tail == atomic_load_explicit(&q->tail, relaxed)) {
  71                         if (get_ptr(next) == 0) { // == NULL
  72                                 pointer value = MAKE_POINTER(node, get_count(next) + 1);
  73                                 success = atomic_compare_exchange_strong_explicit(&q->nodes[get_ptr(tail)].next,
  74                                                 &next, value, memory_order_acq_rel, memory_order_acq_rel);
  75                         }
  76                         if (!success) {
  77                                 unsigned int ptr = get_ptr(atomic_load_explicit(&q->nodes[get_ptr(tail)].next, memory_order_seq_cst));
  78                                 pointer value = MAKE_POINTER(ptr,
  79                                                 get_count(tail) + 1);
  80                                 atomic_compare_exchange_strong_explicit(&q->tail,
  81                                                 &tail, value,
  82                                                 memory_order_acq_rel, memory_order_acq_rel);
  83                                 thrd_yield();
  84                         }
  85                 }
  86         }
  87         atomic_compare_exchange_strong_explicit(&q->tail,
  88                         &tail,
  89                         MAKE_POINTER(node, get_count(tail) + 1),
  90                         memory_order_acq_rel, memory_order_acq_rel);
  91 }
  92
  93 unsigned int dequeue(queue_t *q)
  94 {
  95         unsigned int value;
  96         int success = 0;
  97         pointer head;
  98         pointer tail;
  99         pointer next;
 100
 101         while (!success) {
 102                 head = atomic_load_explicit(&q->head, acquire);
 103                 tail = atomic_load_explicit(&q->tail, acquire);
 104                 next = atomic_load_explicit(&q->nodes[get_ptr(head)].next, acquire);
 105                 if (atomic_load_explicit(&q->head, relaxed) == head) {
 106                         if (get_ptr(head) == get_ptr(tail)) {
 107                                 if (get_ptr(next) == 0) { // NULL
 108                                         return 0; // NULL
 109                                 }
 110                                 atomic_compare_exchange_strong_explicit(&q->tail,
 111                                                 &tail,
 112                                                 MAKE_POINTER(get_ptr(next), get_count(tail) + 1),
 113                                                 memory_order_acq_rel, memory_order_acq_rel);
 114                                 thrd_yield();
 115                         } else {
 116                                 value = load_32(&q->nodes[get_ptr(next)].value);
 117                                 success = atomic_compare_exchange_strong_explicit(&q->head,
 118                                                 &head,
 119                                                 MAKE_POINTER(get_ptr(next), get_count(head) + 1),
 120                                                 memory_order_acq_rel, memory_order_acq_rel);
 121                                 if (!success)
 122                                         thrd_yield();
 123                         }
 124                 }
 125         }
 126         reclaim(get_ptr(head));
 127         return value;
 128 }