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         t_element * read_fetch() {
  30                 // FIXME: We can have a relaxed for sure here since the next CAS
  31                 // will fix the problem
  32                 unsigned int rdwr = m_rdwr.load(mo_acquire);
  33                 unsigned int rd,wr;
  34                 for(;;) {
  35                         rd = (rdwr>>16) & 0xFFFF;
  36                         wr = rdwr & 0xFFFF;
  37
  38                         if ( wr == rd ) // empty
  39                                 return NULL;
  40
  41                         if ( m_rdwr.compare_exchange_weak(rdwr,rdwr+(1<<16),mo_acq_rel) )
  42                                 break;
  43                         else
  44                                 thrd_yield();
  45                 }
  46
  47                 // (*1)
  48                 rl::backoff bo;
  49                 while ( (m_written.load(mo_acquire) & 0xFFFF) != wr ) {
  50                         thrd_yield();
  51                 }
  52
  53                 t_element * p = & ( m_array[ rd % t_size ] );
  54
  55                 return p;
  56         }
  57
  58         void read_consume() {
  59                 m_read.fetch_add(1,mo_release);
  60         }
  61
  62         //-----------------------------------------------------
  63
  64         t_element * write_prepare() {
  65                 // FIXME: We can have a relaxed for sure here since the next CAS
  66                 // will fix the problem
  67                 unsigned int rdwr = m_rdwr.load(mo_acquire);
  68                 unsigned int rd,wr;
  69                 for(;;) {
  70                         rd = (rdwr>>16) & 0xFFFF;
  71                         wr = rdwr & 0xFFFF;
  72
  73                         if ( wr == ((rd + t_size)&0xFFFF) ) // full
  74                                 return NULL;
  75
  76                         if ( m_rdwr.compare_exchange_weak(rdwr,(rd<<16) | ((wr+1)&0xFFFF),mo_acq_rel) )
  77                                 break;
  78                         else
  79                                 thrd_yield();
  80                 }
  81
  82                 // (*1)
  83                 rl::backoff bo;
  84                 while ( (m_read.load(mo_acquire) & 0xFFFF) != rd ) {
  85                         thrd_yield();
  86                 }
  87
  88                 t_element * p = & ( m_array[ wr % t_size ] );
  89
  90                 return p;
  91         }
  92
  93         void write_publish()
  94         {
  95                 m_written.fetch_add(1,mo_release);
  96         }
  97
  98         //-----------------------------------------------------
  99
 100
 101 };