mpmc-queue/mpmc-queue-wildcard.h

   1 #include <stdatomic.h>
   2 #include <unrelacy.h>
   3 #include "wildcard.h"
   4
   5 template <typename t_element, size_t t_size>
   6 struct mpmc_boundq_1_alt
   7 {
   8 private:
   9
  10         // elements should generally be cache-line-size padded :
  11         t_element               m_array[t_size];
  12
  13         // rdwr counts the reads & writes that have started
  14         atomic<unsigned int>    m_rdwr;
  15         // "read" and "written" count the number completed
  16         atomic<unsigned int>    m_read;
  17         atomic<unsigned int>    m_written;
  18
  19 public:
  20
  21         mpmc_boundq_1_alt()
  22         {
  23                 m_rdwr = 0;
  24                 m_read = 0;
  25                 m_written = 0;
  26         }
  27
  28
  29         /**
  30         @Global_define:
  31         Order_queue<unsigned int*> spec_queue;
  32         */
  33
  34         //-----------------------------------------------------
  35
  36         t_element * read_fetch() {
  37                 unsigned int rdwr = m_rdwr.load(wildcard(1)); // acquire
  38                 unsigned int rd,wr;
  39                 for(;;) {
  40                         rd = (rdwr>>16) & 0xFFFF;
  41                         wr = rdwr & 0xFFFF;
  42
  43                         if ( wr == rd ) { // empty
  44                                 return false;
  45                         }
  46
  47                         // acq_rel
  48                         if ( m_rdwr.compare_exchange_weak(rdwr,rdwr+(1<<16),wildcard(2)) )
  49                                 break;
  50                         else
  51                                 thrd_yield();
  52                 }
  53
  54                 // (*1)
  55                 rl::backoff bo;
  56                 // acquire
  57                 while ( (m_written.load(wildcard(3)) & 0xFFFF) != wr ) {
  58                         thrd_yield();
  59                 }
  60
  61                 t_element * p = & ( m_array[ rd % t_size ] );
  62
  63                 /**
  64                         @Commit_point_Check: true
  65                         @Label: ANY
  66                         @Check:
  67                                 spec_queue.peak() == p
  68                 */
  69                 return p;
  70         }
  71
  72         void read_consume() {
  73                 m_read.fetch_add(1,wildcard(4)); // release
  74                 /**
  75                         @Commit_point_define: true
  76                         @Label: Read_Consume_Success
  77                         @Check:
  78                                 spec_queue.size() > 0
  79                         @Action:
  80                                 spec_queue.remove();
  81                 */
  82         }
  83
  84         //-----------------------------------------------------
  85
  86         t_element * write_prepare() {
  87                 unsigned int rdwr = m_rdwr.load(wildcard(5)); // acquire
  88                 unsigned int rd,wr;
  89                 for(;;) {
  90                         rd = (rdwr>>16) & 0xFFFF;
  91                         wr = rdwr & 0xFFFF;
  92
  93                         if ( wr == ((rd + t_size)&0xFFFF) ) // full
  94                                 return NULL;
  95                         // acq_rel
  96                         if ( m_rdwr.compare_exchange_weak(rdwr,(rd<<16) | ((wr+1)&0xFFFF),wildcard(6)) )
  97                                 break;
  98                         else
  99                                 thrd_yield();
 100                 }
 101
 102                 // (*1)
 103                 rl::backoff bo;
 104                 while ( (m_read.load(wildcard(7)) & 0xFFFF) != rd ) { // acquire
 105                         thrd_yield();
 106                 }
 107
 108
 109                 t_element * p = & ( m_array[ wr % t_size ] );
 110
 111                 /**
 112                         @Commit_point_check: ANY
 113                         @Action: spec_queue.add(p);
 114                 */
 115                 return p;
 116         }
 117
 118         void write_publish()
 119         {
 120                 m_written.fetch_add(1,wildcard(8)); // release
 121         }
 122
 123         //-----------------------------------------------------
 124
 125
 126 };
 127
 128 typedef struct mpmc_boundq_1_alt<int, 4> queue_t;