mpmc-queue/mpmc-queue.cc

   1 #include <threads.h>
   2 #include "mpmc-queue.h"
   3
   4 template <typename t_element>
   5 t_element * mpmc_boundq_1_alt<t_element>::read_fetch() {
   6         // FIXME: We can have a relaxed for sure here since the next CAS
   7         // will fix the problem
   8         unsigned int rdwr = m_rdwr.load(mo_acquire);
   9         unsigned int rd,wr;
  10         for(;;) {
  11                 rd = (rdwr>>16) & 0xFFFF;
  12                 wr = rdwr & 0xFFFF;
  13
  14                 if ( wr == rd ) // empty
  15                         return NULL;
  16
  17                 if ( m_rdwr.compare_exchange_weak(rdwr,rdwr+(1<<16),mo_acq_rel) )
  18                         break;
  19                 else
  20                         thrd_yield();
  21         }
  22
  23         // (*1)
  24         rl::backoff bo;
  25         /**********    Detected Admissibility (testcase1)    **********/
  26         while ( (m_written.load(mo_acquire) & 0xFFFF) != wr ) {
  27                 thrd_yield();
  28         }
  29         /** @OPDefine: true */
  30
  31         t_element * p = & ( m_array[ rd % t_size ] );
  32
  33         return p;
  34 }
  35
  36
  37 template <typename t_element>
  38 void mpmc_boundq_1_alt<t_element>::read_consume(t_element *bin) {
  39         /**********    Detected Admissibility (testcase2)    **********/
  40     // run with -Y
  41         m_read.fetch_add(1,mo_release);
  42         /** @OPDefine: true */
  43 }
  44
  45
  46 template <typename t_element>
  47 t_element * mpmc_boundq_1_alt<t_element>::write_prepare() {
  48         // FIXME: We can have a relaxed for sure here since the next CAS
  49         // will fix the problem
  50         unsigned int rdwr = m_rdwr.load(mo_acquire);
  51         unsigned int rd,wr;
  52         for(;;) {
  53                 rd = (rdwr>>16) & 0xFFFF;
  54                 wr = rdwr & 0xFFFF;
  55
  56                 if ( wr == ((rd + t_size)&0xFFFF) ) // full
  57                         return NULL;
  58
  59                 if ( m_rdwr.compare_exchange_weak(rdwr,(rd<<16) | ((wr+1)&0xFFFF),mo_acq_rel) )
  60                         break;
  61                 else
  62                         thrd_yield();
  63         }
  64
  65         // (*1)
  66         rl::backoff bo;
  67         /**********    Detected Admissibility (testcase2)    **********/
  68     // run with -Y
  69         while ( (m_read.load(mo_acquire) & 0xFFFF) != rd ) {
  70                 thrd_yield();
  71         }
  72         /** @OPDefine: true */
  73
  74         t_element * p = & ( m_array[ wr % t_size ] );
  75
  76         return p;
  77 }
  78
  79 template <typename t_element>
  80 void mpmc_boundq_1_alt<t_element>::write_publish(t_element *bin)
  81 {
  82         /**********    Detected Admissibility (testcase1)    **********/
  83         m_written.fetch_add(1,mo_release);
  84         /** @OPDefine: true */
  85 }
  86
  87 /** @DeclareState:
  88         @Commutativity: write_prepare <-> write_publish (M1->C_RET == M2->bin)
  89         @Commutativity: write_publish <-> read_fetch (M1->bin == M2->C_RET)
  90         @Commutativity: read_fetch <-> read_consume (M1->C_RET == M2->bin)
  91         @Commutativity: read_consume <-> write_prepare (M1->bin == M2->C_RET) */
  92
  93
  94 mpmc_boundq_1_alt<int32_t>* createMPMC(int size) {
  95         return new mpmc_boundq_1_alt<int32_t>(size);
  96 }
  97
  98 void destroyMPMC(mpmc_boundq_1_alt<int32_t> *q) {
  99         delete q;
 100 }
 101
 102 /** @PreCondition: */
 103 int32_t * read_fetch(mpmc_boundq_1_alt<int32_t> *q) {
 104         return q->read_fetch();
 105 }
 106
 107 /** @PreCondition: */
 108 void read_consume(mpmc_boundq_1_alt<int32_t> *q, int32_t *bin) {
 109         q->read_consume(bin);
 110 }
 111
 112 /** @PreCondition: */
 113 int32_t * write_prepare(mpmc_boundq_1_alt<int32_t> *q) {
 114         return q->write_prepare();
 115 }
 116
 117 /** @PreCondition: */
 118 void write_publish(mpmc_boundq_1_alt<int32_t> *q, int32_t *bin) {
 119         q->write_publish(bin);
 120 }