output/spsc-bugfix/eventcount.h

   1 #ifndef _EVENTCOUNT_H
   2 #define _EVENTCOUNT_H
   3
   4 #include <atomic>
   5 #include <mutex>
   6 #include <condition_variable>
   7
   8 class eventcount
   9 {
  10 public:
  11         eventcount() : waiters(0)
  12         {
  13                 count = 0;
  14         }
  15
  16         void signal_relaxed()
  17         {
  18                 unsigned cmp = count.load(std::memory_order_relaxed);
  19                 signal_impl(cmp);
  20         }
  21
  22         void signal()
  23         {
  24                                                 unsigned cmp = count.fetch_add(0, std::memory_order_seq_cst);
  25                 signal_impl(cmp);
  26         }
  27
  28         unsigned get()
  29         {
  30                 unsigned cmp = count.fetch_or(0x80000000,
  31 std::memory_order_seq_cst);
  32                 return cmp & 0x7FFFFFFF;
  33         }
  34
  35         void wait(unsigned cmp)
  36         {
  37                 unsigned ec = count.load(std::memory_order_seq_cst);
  38                 if (cmp == (ec & 0x7FFFFFFF))
  39                 {
  40                         guard.lock($);
  41                         ec = count.load(std::memory_order_seq_cst);
  42                         if (cmp == (ec & 0x7FFFFFFF))
  43                         {
  44                                 waiters += 1;
  45                                 cv.wait(guard);
  46                         }
  47                         guard.unlock($);
  48                 }
  49         }
  50
  51 private:
  52         std::atomic<unsigned> count;
  53         rl::var<unsigned> waiters;
  54         std::mutex guard;
  55         std::condition_variable cv;
  56
  57         void signal_impl(unsigned cmp)
  58         {
  59                 if (cmp & 0x80000000)
  60                 {
  61                         guard.lock($);
  62                         while (false == count.compare_exchange_weak(cmp,
  63                                 (cmp + 1) & 0x7FFFFFFF, std::memory_order_relaxed));
  64                         unsigned w = waiters($);
  65                         waiters = 0;
  66                         guard.unlock($);
  67                         if (w)
  68                                 cv.notify_all($);
  69                 }
  70         }
  71 };
  72
  73 #endif
  74