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         /* Note: needed to add this init manually */
  35         atomic_init(&q->nodes[0].next, 0);
  36
  37         /* initialize queue */
  38         head = MAKE_POINTER(1, 0);
  39         tail = MAKE_POINTER(1, 0);
  40         next = MAKE_POINTER(0, 0); // (NULL, 0)
  41
  42         atomic_init(&q->head, head);
  43         atomic_init(&q->tail, tail);
  44         atomic_init(&q->nodes[1].next, next);
  45
  46         /* initialize avail list */
  47         for (i = 2; i < MAX_NODES; i++) {
  48                 next = MAKE_POINTER(i + 1, 0);
  49                 atomic_init(&q->nodes[i].next, next);
  50         }
  51
  52         next = MAKE_POINTER(0, 0); // (NULL, 0)
  53         atomic_init(&q->nodes[MAX_NODES].next, next);
  54 }
  55
  56 void enqueue(queue_t *q, unsigned int val)
  57 {
  58         int success = 0;
  59         unsigned int node;
  60         pointer tail;
  61         pointer next;
  62         pointer tmp;
  63
  64         /* Get a fresh node, set its value field, then set it's next reference to null */
  65         node = new_node();
  66         store_32(&q->nodes[node].value, val);
  67         tmp = atomic_load_explicit(&q->nodes[node].next, relaxed);
  68         set_ptr(&tmp, 0); // NULL
  69         atomic_store_explicit(&q->nodes[node].next, tmp, relaxed);
  70
  71         while (!success) {
  72                 //tail isn't actually synchronization point
  73                 tail = atomic_load_explicit(&q->tail, relaxed);
  74                 //acquire reference...
  75                 next = atomic_load_explicit(&q->nodes[get_ptr(tail)].next, memory_order_seq_cst);
  76                 if (tail == atomic_load_explicit(&q->tail, memory_order_seq_cst)) {
  77                         if (get_ptr(next) == 0) { // == NULL
  78                                 pointer value = MAKE_POINTER(node, get_count(next) + 1);
  79                                 success = atomic_compare_exchange_strong_explicit(&q->nodes[get_ptr(tail)].next,
  80                                                 &next, value, memory_order_seq_cst, memory_order_seq_cst);
  81                         }
  82                         if (!success) {
  83                                 unsigned int ptr = get_ptr(atomic_load_explicit(&q->nodes[get_ptr(tail)].next, memory_order_relaxed));
  84                                 pointer value = MAKE_POINTER(ptr,
  85                                                 get_count(tail) + 1);
  86                                 atomic_compare_exchange_strong_explicit(&q->tail,
  87                                                 &tail, value,
  88                                                 memory_order_seq_cst, memory_order_seq_cst);
  89                         }
  90                 }
  91         }
  92         atomic_compare_exchange_strong_explicit(&q->tail,
  93                         &tail,
  94                         MAKE_POINTER(node, get_count(tail) + 1),
  95                         memory_order_seq_cst, memory_order_seq_cst);
  96 }
  97
  98 unsigned int dequeue(queue_t *q)
  99 {
 100         unsigned int value;
 101         int success = 0;
 102         pointer head;
 103         pointer tail;
 104         pointer next;
 105
 106         while (!success) {
 107                 head = atomic_load_explicit(&q->head, memory_order_seq_cst);
 108                 tail = atomic_load_explicit(&q->tail, memory_order_seq_cst);
 109                 next = atomic_load_explicit(&q->nodes[get_ptr(head)].next, memory_order_seq_cst);
 110                 if (atomic_load_explicit(&q->head, memory_order_seq_cst) == head) {
 111                         if (get_ptr(head) == get_ptr(tail)) {
 112                                 if (get_ptr(next) == 0) { // NULL
 113                                         return 0; // NULL
 114                                 }
 115                                 atomic_compare_exchange_strong_explicit(&q->tail,
 116                                                 &tail,
 117                                                 MAKE_POINTER(get_ptr(next), get_count(tail) + 1),
 118                                                 memory_order_seq_cst, memory_order_seq_cst);
 119                         } else {
 120                                 value = load_32(&q->nodes[get_ptr(next)].value);
 121                                 success = atomic_compare_exchange_strong_explicit(&q->head,
 122                                                 &head,
 123                                                 MAKE_POINTER(get_ptr(next), get_count(head) + 1),
 124                                                 memory_order_seq_cst, memory_order_seq_cst);
 125                         }
 126                 }
 127         }
 128         reclaim(get_ptr(head));
 129         return value;
 130 }