ms-queue/my_queue.c

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