benchmark/spsc-bugfix/queue.h

   1 #ifndef _QUEUE_H
   2 #define _QUEUE_H
   3
   4 #include <unrelacy.h>
   5 #include <atomic>
   6
   7 #include <spec_lib.h>
   8 #include <stdlib.h>
   9 #include <cdsannotate.h>
  10 #include <specannotation.h>
  11 #include <model_memory.h>
  12 #include "common.h"
  13
  14 #include "eventcount.h"
  15
  16 /**
  17         @Begin
  18         @Class_begin
  19         @End
  20 */
  21 template<typename T>
  22 class spsc_queue
  23 {
  24
  25 public:
  26
  27
  28         spsc_queue()
  29         {
  30
  31                 /**
  32                         @Begin
  33                         @Entry_point
  34                         @End
  35                 */
  36
  37                 node* n = new node ();
  38                 head = n;
  39                 tail = n;
  40         }
  41
  42         ~spsc_queue()
  43         {
  44                 //RL_ASSERT(head == tail);
  45                 //delete ((node*)head($));
  46                 delete ((node*)head);
  47         }
  48
  49         /**
  50                 @Begin
  51                 @Options:
  52                         LANG = CPP;
  53                         CLASS = spsc_queue;
  54                 @Global_define:
  55                 @DeclareVar:
  56                         IntegerList *__queue;
  57                 @InitVar:
  58                         __queue = createIntegerList();
  59                 @Finalize:
  60                         if (__queue)
  61                                 destroyIntegerList(__queue);
  62                         return true;
  63                 @Happens_before: Enqueue -> Dequeue
  64                 @Commutativity: Enqueue <-> Dequeue: true
  65                 @Commutativity: Dequeue <-> Dequeue: !_Method1.__RET__ || !_Method2.__RET__
  66
  67                 @End
  68         */
  69
  70
  71         /**
  72                 @Begin
  73                 @Interface: Enqueue
  74                 @Commit_point_set: Enqueue_Point
  75                 @ID: data
  76                 @Action:
  77                         push_back(__queue, data);
  78                         //model_print("Enqueue: val=%d\n", val);
  79                 @End
  80         */
  81         void enqueue(T data)
  82         {
  83                 node* n = new node (data);
  84                 /********** DR & SPEC (sequential) **********/
  85                 //head($)->next.store(n, std::memory_order_release);
  86                 head->next.store(n, std::memory_order_release);
  87                 /**
  88                         @Begin
  89                         @Commit_point_define_check: true
  90                         @Label: Enqueue_Point
  91                         @End
  92                 */
  93                 head = n;
  94                 // #Mark delete this
  95                 ec.signal();
  96         }
  97         /**
  98                 @Begin
  99                 @Interface: Dequeue
 100                 @Commit_point_set: Dequeue_Point
 101                 @ID: __RET__ ? __RET__ : 0
 102                 @Action:
 103                         int elem = 0;
 104                         if (__RET__) {
 105                                 elem = front(__queue);
 106                                 pop_front(__queue);
 107                         }
 108                         //model_print("Dequeue: __RET__=%d, retVal=%d, elem=%d, \n", __RET__, *retVal, elem);
 109                         //model_print("Result: %d\n", __RET__ ? *retVal == elem : true);
 110                 @Post_check: __RET__ ? __RET__ == elem : true
 111                 @End
 112         */
 113         T dequeue()
 114         {
 115                 T data = try_dequeue();
 116                 while (0 == data)
 117                 {
 118                         int cmp = ec.get();
 119                         data = try_dequeue();
 120                         if (data)
 121                                 break;
 122                         ec.wait(cmp);
 123                         data = try_dequeue();
 124                         if (data)
 125                                 break;
 126                 }
 127                 return data;
 128         }
 129
 130 private:
 131         struct node
 132         {
 133                 std::atomic<node*> next;
 134                 //rl::var<T> data;
 135                 T data;
 136
 137                 node(T data = T())
 138                         : data(data)
 139                 {
 140                         next = 0;
 141                 }
 142         };
 143
 144         //rl::var<node*> head;
 145         //rl::var<node*> tail;
 146         node *head;
 147         node *tail;
 148
 149
 150
 151         eventcount ec;
 152
 153         T try_dequeue()
 154         {
 155                 //node* t = tail($);
 156                 node *t = tail;
 157                 /********** DR & SPEC (sequential) **********/
 158                 node* n = t->next.load(std::memory_order_acquire);
 159                 /**
 160                         @Begin
 161                         @Commit_point_define_check: n != 0
 162                         @Label: Dequeue_Point
 163                         @End
 164                 */
 165                 if (0 == n)
 166                         return 0;
 167                 //T data = n->data($);
 168                 T data = n->data;
 169                 delete (t);
 170                 tail = n;
 171                 return data;
 172         }
 173 };
 174 /**
 175         @Begin
 176         @Class_end
 177         @End
 178 */
 179
 180 #endif