benchmark/mpmc-queue/mpmc-queue.h

   1 #include <stdatomic.h>
   2 #include <unrelacy.h>
   3
   4 template <typename t_element, size_t t_size>
   5 struct mpmc_boundq_1_alt
   6 {
   7 private:
   8
   9         // elements should generally be cache-line-size padded :
  10         t_element               m_array[t_size];
  11
  12         // rdwr counts the reads & writes that have started
  13         atomic<unsigned int>    m_rdwr;
  14         // "read" and "written" count the number completed
  15         atomic<unsigned int>    m_read;
  16         atomic<unsigned int>    m_written;
  17
  18 public:
  19
  20         mpmc_boundq_1_alt()
  21         {
  22                 m_rdwr = 0;
  23                 m_read = 0;
  24                 m_written = 0;
  25         }
  26
  27
  28         /**
  29                 @Global_define:
  30                 @Options:
  31                         LANG = CPP;
  32                         CLASS = mpmc_boundq_1_alt;
  33                 @DeclareStruct:
  34                         typedef struct elem {
  35                                 t_element* pos;
  36                                 boolean written;
  37                                 call_id_t id;
  38                         } elem;
  39                 @DeclareVar:
  40                         spec_list *list;
  41                         id_tag_t *tag;
  42                 @InitVar:
  43                         list = new_spec_list();
  44                         tag = new_id_tag();
  45                 @DefineFunc:
  46                         elem* new_elem(t_element *pos, call_id_t id) {
  47                                 elem *e = (elem*) MODEL_MALLOC(sizeof(elem));
  48                                 e->pos = pos;
  49                                 e->written = false;
  50                                 e->id = id;
  51                         }
  52                 @DefineFunc:
  53                         elem* get_elem(t_element *pos) {
  54                                 for (int i = 0; i < size(list); i++) {
  55                                         elem *e = (elem*) elem_at_index(list, i);
  56                                         if (e->pos == pos) {
  57                                                 return e;
  58                                         }
  59                                 }
  60                                 return NULL;
  61                         }
  62                 @DefineFunc:
  63                         int has_elem(t_element *pos) {
  64                                 for (int i = 0; i < size(list); i++) {
  65                                         elem *existing = (elem*) elem_at_index(list, i);
  66                                         if (pos == existing->pos) {
  67                                                 return i;
  68                                         }
  69                                 }
  70                                 return -1;
  71                         }
  72                 @DefineFunc:
  73                         void prepare(t_element *pos) {
  74                                 elem *e = new_elem(
  75                                 push_back(list, e);
  76                         }
  77                 @DefineFunc:
  78                         void publish(t_element *pos) {
  79                                 elem *e = new_elem(
  80                                 push_back(list, e);
  81                         }
  82                 @DefineFunc:
  83                         void consume_elem(t_element *pos) {
  84                                 int idx = has_elem(pos);
  85                                 if (idx == -1)
  86                                         return;
  87                                 remove_at_index(list, idx);
  88                         }
  89
  90         */
  91
  92         //-----------------------------------------------------
  93
  94         t_element * read_fetch() {
  95                 unsigned int rdwr = m_rdwr.load(mo_acquire);
  96                 unsigned int rd,wr;
  97                 for(;;) {
  98                         rd = (rdwr>>16) & 0xFFFF;
  99                         wr = rdwr & 0xFFFF;
 100
 101                         if ( wr == rd ) { // empty
 102                                 return false;
 103                         }
 104
 105                         if ( m_rdwr.compare_exchange_weak(rdwr,rdwr+(1<<16),mo_acq_rel) )
 106                                 break;
 107                         else
 108                                 thrd_yield();
 109                 }
 110
 111                 // (*1)
 112                 rl::backoff bo;
 113                 while ( (m_written.load(mo_acquire) & 0xFFFF) != wr ) {
 114                         thrd_yield();
 115                 }
 116
 117                 t_element * p = & ( m_array[ rd % t_size ] );
 118
 119                 return p;
 120         }
 121
 122         void read_consume() {
 123                 m_read.fetch_add(1,mo_release);
 124         }
 125
 126         //-----------------------------------------------------
 127
 128         t_element * write_prepare() {
 129                 unsigned int rdwr = m_rdwr.load(mo_acquire);
 130                 unsigned int rd,wr;
 131                 for(;;) {
 132                         rd = (rdwr>>16) & 0xFFFF;
 133                         wr = rdwr & 0xFFFF;
 134
 135                         if ( wr == ((rd + t_size)&0xFFFF) ) // full
 136                                 return NULL;
 137
 138                         if ( m_rdwr.compare_exchange_weak(rdwr,(rd<<16) | ((wr+1)&0xFFFF),mo_acq_rel) )
 139                                 break;
 140                         else
 141                                 thrd_yield();
 142                 }
 143
 144                 // (*1)
 145                 rl::backoff bo;
 146                 while ( (m_read.load(mo_acquire) & 0xFFFF) != rd ) {
 147                         thrd_yield();
 148                 }
 149
 150
 151                 t_element * p = & ( m_array[ wr % t_size ] );
 152
 153                 return p;
 154         }
 155
 156         void write_publish()
 157         {
 158                 m_written.fetch_add(1,mo_release);
 159         }
 160
 161         //-----------------------------------------------------
 162
 163
 164 };