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