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 #include "common.h"
  11
  12 /**
  13         @Begin
  14         @Class_begin
  15         @End
  16 */
  17 template <typename t_element, size_t t_size>
  18 struct mpmc_boundq_1_alt
  19 {
  20 private:
  21
  22         // elements should generally be cache-line-size padded :
  23         t_element               m_array[t_size];
  24
  25         // rdwr counts the reads & writes that have started
  26         atomic<unsigned int>    m_rdwr;
  27         // "read" and "written" count the number completed
  28         atomic<unsigned int>    m_read;
  29         atomic<unsigned int>    m_written;
  30
  31 public:
  32
  33         mpmc_boundq_1_alt()
  34         {
  35         /**
  36                 @Begin
  37                         @Entry_point
  38                         @End
  39                 */
  40                 m_rdwr = 0;
  41                 m_read = 0;
  42                 m_written = 0;
  43         }
  44
  45
  46         /**
  47                 @Begin
  48                 @Options:
  49                         LANG = CPP;
  50                         CLASS = mpmc_boundq_1_alt;
  51                 @Global_define:
  52                 @DeclareVar:
  53                         id_tag_t *tag;
  54                 @InitVar:
  55                         tag = NULL;
  56         @Happens_before:
  57                 Publish -> Fetch
  58         @End
  59         */
  60
  61         //-----------------------------------------------------
  62
  63         /**
  64                 @Begin
  65                 @Interface: Fetch
  66                 @Commit_point_set: Fetch_Succ_Point | Fetch_Fail_Point
  67                 @ID: (call_id_t) __RET__
  68                 @End
  69         */
  70         t_element * read_fetch() {
  71                 unsigned int rdwr = m_rdwr.load(mo_acquire);
  72                 /**
  73                         @Begin
  74                         @Commit_point_define_check: (unsigned int) ((rdwr>>16) & 0xFFFF) == (unsigned int) (rdwr & 0xFFFF)
  75                         @Label: Fetch_Fail_Point
  76                         @End
  77                 */
  78                 unsigned int rd,wr;
  79                 for(;;) {
  80                         rd = (rdwr>>16) & 0xFFFF;
  81                         wr = rdwr & 0xFFFF;
  82
  83                         //model_print("cond: %d\n", (unsigned int) ((rdwr>>16) & 0xFFFF) ==
  84                         //(unsigned int) (rdwr & 0xFFFF));
  85                         //model_print("cond: %d\n", wr == rd);
  86                         if ( wr == rd ) { // empty
  87                                 /**
  88                                         //@Begin
  89                                         @Commit_point_define: true
  90                                         @Potential_commit_point_label: Fetch_Potential_Point
  91                                         @Label: Fetch_Fail_Point
  92                                         @End
  93                                 */
  94                                 model_print("in cond\n");
  95                                 return false;
  96                         }
  97
  98                         bool succ = m_rdwr.compare_exchange_weak(rdwr,rdwr+(1<<16),mo_acq_rel);
  99
 100                         if (succ)
 101                                 break;
 102                         else
 103                                 thrd_yield();
 104                 }
 105
 106                 // (*1)
 107                 rl::backoff bo;
 108
 109                 while ( true ) {
 110                         if ((m_written.load(mo_acquire) & 0xFFFF) == wr) {
 111                                 /**
 112                                         @Begin
 113                                         @Commit_point_define_check: true
 114                                         @Label: Fetch_Succ_Point
 115                                         @End
 116                                 */
 117                                 break;
 118                         }
 119                         thrd_yield();
 120                 }
 121
 122                 t_element * p = & ( m_array[ rd % t_size ] );
 123
 124                 return p;
 125         }
 126
 127         void read_consume() {
 128                 m_read.fetch_add(1,mo_release);
 129         }
 130
 131         //-----------------------------------------------------
 132
 133         t_element * write_prepare() {
 134                 unsigned int rdwr = m_rdwr.load(mo_acquire);
 135                 unsigned int rd,wr;
 136                 for(;;) {
 137                         rd = (rdwr>>16) & 0xFFFF;
 138                         wr = rdwr & 0xFFFF;
 139
 140                         if ( wr == ((rd + t_size)&0xFFFF) ) { // full
 141                                 return NULL;
 142                         }
 143
 144                         bool succ = m_rdwr.compare_exchange_weak(rdwr,(rd<<16) |
 145                                 ((wr+1)&0xFFFF),mo_acq_rel);
 146                         if (succ)
 147                                 break;
 148                         else
 149                                 thrd_yield();
 150                 }
 151
 152                 // (*1)
 153                 rl::backoff bo;
 154                 while ( (m_read.load(mo_acquire) & 0xFFFF) != rd ) {
 155                         thrd_yield();
 156                 }
 157
 158                 t_element * p = & ( m_array[ wr % t_size ] );
 159
 160                 return p;
 161         }
 162
 163         /**
 164                 @Begin
 165                 @Interface: Publish
 166                 @Commit_point_set: Publish_Point
 167                 @ID: (uint64_t) elem
 168                 @End
 169         */
 170         void write_publish(t_element *elem)
 171         {
 172                 m_written.fetch_add(1,mo_release);
 173                 /**
 174                         @Begin
 175                         @Commit_point_define_check: true
 176                         @Label: Publish_Point
 177                         @End
 178                 */
 179         }
 180
 181         //-----------------------------------------------------
 182
 183
 184 };
 185 /**
 186         @Begin
 187         @Class_end
 188         @End
 189 */