benchmark/mpmc-queue/mpmc-queue.h

   1 #include <stdatomic.h>
   2 #include <unrelacy.h>
   3 #include <common.h>
   4
   5 #include <spec_lib.h>
   6 #include <stdlib.h>
   7 #include <cdsannotate.h>
   8 #include <specannotation.h>
   9 #include <model_memory.h>
  10
  11 /**
  12         @Begin
  13         @Class_begin
  14         @End
  15 */
  16 template <typename t_element, size_t t_size>
  17 struct mpmc_boundq_1_alt
  18 {
  19 private:
  20
  21         unsigned int MASK;
  22
  23         atomic<t_element*> arr;
  24
  25         // elements should generally be cache-line-size padded :
  26         t_element               m_array[t_size];
  27
  28         // rdwr counts the reads & writes that have started
  29         atomic<unsigned int>    m_rdwr;
  30         // "read" and "written" count the number completed
  31         atomic<unsigned int>    m_read;
  32         atomic<unsigned int>    m_written;
  33
  34 public:
  35
  36         mpmc_boundq_1_alt(int _MASK = 0xFFFF)
  37         {
  38         /**
  39                 @Begin
  40                         @Entry_point
  41                         @End
  42                 */
  43                 m_rdwr = 0;
  44                 m_read = 0;
  45                 m_written = 0;
  46                 // For this we want MASK = 1; MASK wrap around
  47                 MASK = _MASK; //0x11;
  48                 arr = m_array;
  49         }
  50
  51
  52         /**
  53                 @Begin
  54                 @Options:
  55                         LANG = CPP;
  56                         CLASS = mpmc_boundq_1_alt;
  57                 @Global_define:
  58                 @Happens_before:
  59                         Publish -> Fetch
  60                         Consume -> Prepare
  61                 @Commutativity: Prepare <-> Prepare: _Method1.__RET__ !=
  62                 _Method2.__RET__ || !_Method1.__RET__ || !_Method2.__RET__
  63                 @Commutativity: Prepare <-> Publish: _Method1.__RET__ != _Method2.bin ||
  64                 !_Method1.__RET__
  65                 @Commutativity: Prepare <-> Fetch: _Method1.__RET__ != _Method2.__RET__
  66                 || !_Method1.__RET__ || !_Method2.__RET__
  67                 @Commutativity: Prepare <-> Consume : _Method1.__RET__ != _Method2.bin || !_Method1.__RET__
  68
  69                 @Commutativity: Publish <-> Publish: _Method1.bin != _Method2.bin
  70                 @Commutativity: Publish <-> Fetch: _Method1.bin != _Method2.__RET__ ||
  71                 !_Method2.__RET__
  72                 @Commutativity: Publish <-> Consume : _Method1.bin != _Method2.bin
  73
  74                 @Commutativity: Fetch <-> Fetch: _Method1.__RET__ != _Method2.__RET__ ||
  75                 !_Method1.__RET__ || !_Method2.__RET__
  76                 @Commutativity: Fetch <-> Consume : _Method1.__RET__ != _Method2.bin || !_Method1.__RET__
  77
  78                 @Commutativity: Consume <-> Consume : _Method1.bin != _Method2.bin
  79
  80         @End
  81         */
  82
  83         //-----------------------------------------------------
  84
  85         /**
  86                 @Begin
  87                 @Interface: Fetch
  88                 @Commit_point_set: FetchReadRW1 | FetchReadRW2 | FetchReadPointer
  89                 @ID: (call_id_t) __RET__
  90                 //@Action: model_print("Fetch: %d\n", __RET__);
  91                 @End
  92         */
  93         t_element * read_fetch() {
  94                 // Since we have a lool to CAS the value of m_rdwr, this can be relaxed
  95                 unsigned int rdwr = m_rdwr.load(mo_acquire);
  96                 //unsigned int rdwr = m_rdwr.load(mo_relaxed);
  97                 /**
  98                         @Begin
  99                         @Commit_point_define_check: true
 100                         @Label: FetchReadRW1
 101                         @End
 102                 */
 103                 unsigned int rd,wr;
 104                 for(;;) {
 105                         rd = (rdwr>>16) & MASK;
 106                         wr = rdwr & MASK;
 107
 108                         if ( wr == rd ) { // empty
 109                                 return false;
 110                         }
 111                         /**** Inadmissibility (testcase2.cc, MASK = 1, size = 1) ****/
 112                         bool succ =
 113                                 m_rdwr.compare_exchange_weak(rdwr,rdwr+(1<<16),mo_acq_rel);
 114                         if (succ) {
 115                                 break;
 116                         } else {
 117                                 /**
 118                                         @Begin
 119                                         @Commit_point_clear: true
 120                                         @Label: FetchClear1
 121                                         @End
 122                                 */
 123
 124                                 /**
 125                                         @Begin
 126                                         @Commit_point_define_check: true
 127                                         @Label: FetchReadRW2
 128                                         @End
 129                                 */
 130                                 thrd_yield();
 131                         }
 132                 }
 133
 134                 // (*1)
 135                 rl::backoff bo;
 136                 while (true) {
 137                         /**** Inadmissibility ****/
 138                         int written = m_written.load(mo_acquire);
 139                         if ((written & MASK) != wr) {
 140                                 thrd_yield();
 141                         } else {
 142                                 printf("Fetch: m_written=%d\n", written);
 143                                 break;
 144                         }
 145                 }
 146                 t_element * p = & ( m_array[ rd % t_size ] );
 147
 148                 // This is just a hack to tell the CDSChecker that we have a memory
 149                 // operation here
 150                 arr.load(memory_order_relaxed);
 151                 /**
 152                         @Begin
 153                         @Commit_point_clear: true
 154                         @Label: FetchClear2
 155                         @End
 156                 */
 157                 /**
 158                         @Begin
 159                         @Commit_point_define_check: true
 160                         @Label: FetchReadPointer
 161                         @End
 162                 */
 163
 164                 return p;
 165         }
 166
 167         /**
 168                 @Begin
 169                 @Interface: Consume
 170                 @Commit_point_set: ConsumeFetchAdd
 171                 @ID: (call_id_t) bin
 172                 //@Action: model_print("Consume: %d\n", bin);
 173                 @End
 174         */
 175         void read_consume(t_element *bin) {
 176                 /**** Inadmissibility ****/
 177                 m_read.fetch_add(1,mo_release);
 178                 /**
 179                         @Begin
 180                         @Commit_point_define_check: true
 181                         @Label: ConsumeFetchAdd
 182                         @End
 183                 */
 184         }
 185
 186         //-----------------------------------------------------
 187
 188         /**
 189                 @Begin
 190                 @Interface: Prepare
 191                 @Commit_point_set: PrepareReadRW1 | PrepareReadRW2 | PrepareReadPointer
 192                 @ID: (call_id_t) __RET__
 193                 //@Action: model_print("Prepare: %d\n", __RET__);
 194                 @End
 195         */
 196         t_element * write_prepare() {
 197                 // Try weaker semantics
 198                 // Since we have a lool to CAS the value of m_rdwr, this can be relaxed
 199                 unsigned int rdwr = m_rdwr.load(mo_acquire);
 200                 //unsigned int rdwr = m_rdwr.load(mo_relaxed);
 201                 /**
 202                         @Begin
 203                         @Commit_point_define_check: true
 204                         @Label: PrepareReadRW1
 205                         @End
 206                 */
 207                 unsigned int rd,wr;
 208                 for(;;) {
 209                         rd = (rdwr>>16) & MASK;
 210                         wr = rdwr & MASK;
 211                         //printf("write_prepare: rd=%d, wr=%d\n", rd, wr);
 212
 213                         if ( wr == ((rd + t_size)&MASK) ) { // full
 214                                 return NULL;
 215                         }
 216
 217                         /**** Inadmissibility (testcase3.cc, MASK = 1, size = 1) ****/
 218                         bool succ = m_rdwr.compare_exchange_weak(rdwr,(rd<<16) |
 219                                 ((wr+1)&MASK),mo_acq_rel);
 220                         //printf("wr=%d\n", (wr+1)&MASK);
 221                         if (succ) {
 222                                 break;
 223                         } else {
 224                                 /**
 225                                         @Begin
 226                                         @Commit_point_clear: true
 227                                         @Label: PrepareClear1
 228                                         @End
 229                                 */
 230
 231                                 /**
 232                                         @Begin
 233                                         @Commit_point_define_check: true
 234                                         @Label: PrepareReadRW2
 235                                         @End
 236                                 */
 237                                 thrd_yield();
 238                         }
 239                 }
 240
 241                 // (*1)
 242                 rl::backoff bo;
 243                 while (true) {
 244                         /**** Inadmissibility ****/
 245                         int read = m_read.load(mo_acquire);
 246                         if ((read & MASK) != rd)
 247                                 thrd_yield();
 248                         else
 249                                 break;
 250                 }
 251
 252                 t_element * p = & ( m_array[ wr % t_size ] );
 253
 254                 // This is also just a hack to tell the CDSChecker that we have a memory
 255                 // operation here
 256                 arr.load(memory_order_relaxed);
 257                 /**
 258                         @Begin
 259                         @Commit_point_clear: true
 260                         @Label: PrepareClear2
 261                         @End
 262                 */
 263                 /**
 264                         @Begin
 265                         @Commit_point_define_check: true
 266                         @Label: PrepareReadPointer
 267                         @End
 268                 */
 269
 270                 return p;
 271         }
 272
 273         /**
 274                 @Begin
 275                 @Interface: Publish
 276                 @Commit_point_set: Publish_W_RMW
 277                 @ID: (call_id_t) bin
 278                 //@Action: model_print("Publish: %d\n", bin);
 279                 @End
 280         */
 281         void write_publish(t_element *bin)
 282         {
 283                 /**** Inadmissibility ****/
 284                 int tmp = m_written.fetch_add(1,mo_release);
 285                 /**
 286                         @Begin
 287                         @Commit_point_define_check: true
 288                         @Label: Publish_W_RMW
 289                         @End
 290                 */
 291                 printf("publish: m_written=%d\n", tmp + 1);
 292         }
 293
 294         //-----------------------------------------------------
 295
 296
 297 };
 298 /**
 299         @Begin
 300         @Class_end
 301         @End
 302 */