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 (testcase2.c) ****/
 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 (testcase1.c) ****/
 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                         // FIXME: Theoretically we have a bug here, it is not so likely to
 218                         // happen!!! The following acq_rel can not guarantee that the load
 219                         // acquire on m_read to read the most recent value. However, it can
 220                         // decrease the chance of the load acquire of m_read to read a too
 221                         // old value. Same as the case in read_fetch(). Maybe making things
 222                         // SC can fix it. We can run testcase3.c to expose this bug (set the
 223                         // array size  = 1, the MASK = 1).
 224
 225                         /**** Inadmissibility (testcase3.cc, MASK = 1, size = 1) ****/
 226                         bool succ = m_rdwr.compare_exchange_weak(rdwr,(rd<<16) |
 227                                 ((wr+1)&MASK),mo_acq_rel);
 228                         //printf("wr=%d\n", (wr+1)&MASK);
 229                         if (succ) {
 230                                 break;
 231                         } else {
 232                                 /**
 233                                         @Begin
 234                                         @Commit_point_clear: true
 235                                         @Label: PrepareClear1
 236                                         @End
 237                                 */
 238
 239                                 /**
 240                                         @Begin
 241                                         @Commit_point_define_check: true
 242                                         @Label: PrepareReadRW2
 243                                         @End
 244                                 */
 245                                 thrd_yield();
 246                         }
 247                 }
 248
 249                 // (*1)
 250                 rl::backoff bo;
 251                 while (true) {
 252                         /**** Inadmissibility (testcase1.c) ****/
 253                         int read = m_read.load(mo_acquire);
 254                         if ((read & MASK) != rd)
 255                                 thrd_yield();
 256                         else
 257                                 break;
 258                 }
 259
 260                 t_element * p = & ( m_array[ wr % t_size ] );
 261
 262                 // This is also just a hack to tell the CDSChecker that we have a memory
 263                 // operation here
 264                 arr.load(memory_order_relaxed);
 265                 /**
 266                         @Begin
 267                         @Commit_point_clear: true
 268                         @Label: PrepareClear2
 269                         @End
 270                 */
 271                 /**
 272                         @Begin
 273                         @Commit_point_define_check: true
 274                         @Label: PrepareReadPointer
 275                         @End
 276                 */
 277
 278                 return p;
 279         }
 280
 281         /**
 282                 @Begin
 283                 @Interface: Publish
 284                 @Commit_point_set: PublishFetchAdd
 285                 @ID: (call_id_t) bin
 286                 //@Action: model_print("Publish: %d\n", bin);
 287                 @End
 288         */
 289         void write_publish(t_element *bin)
 290         {
 291                 /**** Inadmissibility (testcase2.c) ****/
 292                 int tmp = m_written.fetch_add(1,mo_release);
 293                 /**
 294                         @Begin
 295                         @Commit_point_define_check: true
 296                         @Label: PublishFetchAdd
 297                         @End
 298                 */
 299                 printf("publish: m_written=%d\n", tmp + 1);
 300         }
 301
 302         //-----------------------------------------------------
 303
 304
 305 };
 306 /**
 307         @Begin
 308         @Class_end
 309         @End
 310 */