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         nonatomic<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() : m_rdwr(0), m_read(0), m_written(0)
  21         {
  22         }
  23
  24         //-----------------------------------------------------
  25
  26         nonatomic<t_element> * read_fetch() {
  27                 unsigned int rdwr = m_rdwr.load(mo_acquire);
  28                 unsigned int rd,wr;
  29                 for(;;) {
  30                         rd = (rdwr>>16) & 0xFFFF;
  31                         wr = rdwr & 0xFFFF;
  32
  33                         if ( wr == rd ) // empty
  34                                 return false;
  35
  36                         if ( m_rdwr.compare_exchange_weak(rdwr,rdwr+(1<<16),mo_acq_rel) )
  37                                 break;
  38                 }
  39
  40                 // (*1)
  41                 rl::backoff bo;
  42                 while ( (m_written.load(mo_acquire) & 0xFFFF) != wr ) {
  43                         bo.yield();
  44                 }
  45
  46                 nonatomic<t_element> * p = & ( m_array[ rd % t_size ] );
  47
  48                 return p;
  49         }
  50
  51         void read_consume() {
  52                 m_read.fetch_add(1,mo_release);
  53         }
  54
  55         //-----------------------------------------------------
  56
  57         nonatomic<t_element> * write_prepare() {
  58                 unsigned int rdwr = m_rdwr.load(mo_acquire);
  59                 unsigned int rd,wr;
  60                 for(;;) {
  61                         rd = (rdwr>>16) & 0xFFFF;
  62                         wr = rdwr & 0xFFFF;
  63
  64                         if ( wr == ((rd + t_size)&0xFFFF) ) // full
  65                                 return NULL;
  66
  67                         if ( m_rdwr.compare_exchange_weak(rdwr,(rd<<16) | ((wr+1)&0xFFFF),mo_acq_rel) )
  68                                 break;
  69                 }
  70
  71                 // (*1)
  72                 rl::backoff bo;
  73                 while ( (m_read.load(mo_acquire) & 0xFFFF) != rd ) {
  74                         bo.yield();
  75                 }
  76
  77                 nonatomic<t_element> * p = & ( m_array[ wr % t_size ] );
  78
  79                 return p;
  80         }
  81
  82         void write_publish()
  83         {
  84                 m_written.fetch_add(1,mo_release);
  85         }
  86
  87         //-----------------------------------------------------
  88
  89
  90 };