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