ms-queue/my_queue.c

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