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