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         @Happens_before:
  66                 Publish -> Fetch
  67                 Consume -> Prepare
  68         @End
  69         */
  70
  71         //-----------------------------------------------------
  72
  73         /**
  74                 @Begin
  75                 @Interface: Fetch
  76                 @Commit_point_set: Fetch_Empty_Point | Fetch_Succ_Point
  77                 @ID: (call_id_t) __RET__
  78                 //@Check:
  79                         //__RET__ == NULL || has_elem(list, __RET__)
  80                 @End
  81         */
  82         t_element * read_fetch() {
  83                 // Try this new weaker semantics
  84                 unsigned int rdwr = m_rdwr.load(mo_acquire);
  85                 //unsigned int rdwr = m_rdwr.load(mo_relaxed);
  86                 /**
  87                         @Begin
  88                         @Potential_commit_point_define: true
  89                         @Label: Fetch_Potential_Point
  90                         @End
  91                 */
  92                 unsigned int rd,wr;
  93                 for(;;) {
  94                         rd = (rdwr>>16) & 0xFFFF;
  95                         wr = rdwr & 0xFFFF;
  96
  97                         if ( wr == rd ) { // empty
  98
  99                                 /**
 100                                         @Begin
 101                                         @Commit_point_define: true
 102                                         @Potential_commit_point_label: Fetch_Potential_Point
 103                                         @Label: Fetch_Empty_Point
 104                                         @End
 105                                 */
 106
 107                                 return false;
 108                         }
 109
 110                         bool succ = m_rdwr.compare_exchange_weak(rdwr,rdwr+(1<<16),mo_acq_rel);
 111                         /**
 112                                 @Begin
 113                                 @Commit_point_define_check: succ == true
 114                                 @Label: Fetch_Succ_Point
 115                                 @End
 116                         */
 117                         if (succ)
 118                                 break;
 119                         else
 120                                 thrd_yield();
 121                 }
 122
 123                 // (*1)
 124                 rl::backoff bo;
 125                 while ( (m_written.load(mo_acquire) & 0xFFFF) != wr ) {
 126                         thrd_yield();
 127                 }
 128
 129                 t_element * p = & ( m_array[ rd % t_size ] );
 130
 131                 return p;
 132         }
 133
 134         /**
 135                 @Begin
 136                 @Interface: Consume
 137                 @Commit_point_set: Consume_Point
 138                 @ID: (call_id_t) bin
 139                 //@Check:
 140                 //      consume_check(__TID__)
 141                 //@Action:
 142                         //consume(__TID__);
 143                 @End
 144         */
 145         void read_consume(t_element *bin) {
 146                 /**** FIXME: miss ****/
 147                 m_read.fetch_add(1,mo_release);
 148                 /**
 149                         @Begin
 150                         @Commit_point_define_check: true
 151                         @Label: Consume_Point
 152                         @End
 153                 */
 154         }
 155
 156         //-----------------------------------------------------
 157
 158         /**
 159                 @Begin
 160                 @Interface: Prepare
 161                 @Commit_point_set: Prepare_Full_Point | Prepare_Succ_Point
 162                 @ID: (call_id_t) __RET__
 163                 //@Check:
 164                         //prepare_check(__RET__, __TID__)
 165                 //@Action:
 166                         //push_back(list, __RET__);
 167                 @End
 168         */
 169         t_element * write_prepare() {
 170                 // Try weaker semantics
 171                 unsigned int rdwr = m_rdwr.load(mo_acquire);
 172                 //unsigned int rdwr = m_rdwr.load(mo_relaxed);
 173                 /**
 174                         @Begin
 175                         @Potential_commit_point_define: true
 176                         @Label: Prepare_Potential_Point
 177                         @End
 178                 */
 179                 unsigned int rd,wr;
 180                 for(;;) {
 181                         rd = (rdwr>>16) & 0xFFFF;
 182                         wr = rdwr & 0xFFFF;
 183
 184                         if ( wr == ((rd + t_size)&0xFFFF) ) { // full
 185
 186                                 /**
 187                                         @Begin
 188                                         @Commit_point_define: true
 189                                         @Potential_commit_point_label: Prepare_Potential_Point
 190                                         @Label: Prepare_Full_Point
 191                                         @End
 192                                 */
 193                                 return NULL;
 194                         }
 195
 196                         bool succ = m_rdwr.compare_exchange_weak(rdwr,(rd<<16) |
 197                                 ((wr+1)&0xFFFF),mo_acq_rel);
 198                         /**
 199                                 @Begin
 200                                 @Commit_point_define_check: succ == true
 201                                 @Label: Prepare_Succ_Point
 202                                 @End
 203                         */
 204                         if (succ)
 205                                 break;
 206                         else
 207                                 thrd_yield();
 208                 }
 209
 210                 // (*1)
 211                 rl::backoff bo;
 212                 while ( (m_read.load(mo_acquire) & 0xFFFF) != rd ) {
 213                         thrd_yield();
 214                 }
 215
 216                 t_element * p = & ( m_array[ wr % t_size ] );
 217
 218                 return p;
 219         }
 220
 221         /**
 222                 @Begin
 223                 @Interface: Publish
 224                 @Commit_point_set: Publish_Point
 225                 @ID: (call_id_t) bin
 226                 //@Check:
 227                         //publish_check(__TID__)
 228                 //@Action:
 229                         //publish(__TID__);
 230                 @End
 231         */
 232         void write_publish(t_element *bin)
 233         {
 234                 /**** hb violation ****/
 235                 m_written.fetch_add(1,mo_release);
 236                 /**
 237                         @Begin
 238                         @Commit_point_define_check: true
 239                         @Label: Publish_Point
 240                         @End
 241                 */
 242         }
 243
 244         //-----------------------------------------------------
 245
 246
 247 };
 248 /**
 249         @Begin
 250         @Class_end
 251         @End
 252 */