center alloc all msgs.

lichao

2021-05-13 db322f33ba13592f2492317e3f1a070454c97059

 src/robust.h

@@ -19,6 +19,7 @@
#ifndef ROBUST_Q31RCWYU
#define ROBUST_Q31RCWYU

#include "bh_util.h"
#include "log.h"
#include <atomic>
#include <chrono>
@@ -37,8 +38,6 @@

using namespace std::chrono;
using namespace std::chrono_literals;
constexpr uint64_t MaskBits(int nbits) { return (uint64_t(1) << nbits) - 1; }

void QuickSleep();

class CasMutex
@@ -99,7 +98,7 @@
public:
   typedef uint64_t id_t;
   FMutex(id_t id) :
       id_(id), fd_(Open(id_))
       id_(id), fd_(Open(id_)), count_(0)
   {
      if (fd_ == -1) { throw "error create mutex!"; }
   }
@@ -117,11 +116,10 @@
   }
   static int Open(id_t id) { return open(GetPath(id).c_str(), O_CREAT | O_RDONLY, 0666); }
   static int Close(int fd) { return close(fd); }
   void FLock();
   void FUnlock();
   id_t id_;
   int fd_;
   std::mutex mtx_;
   std::atomic<int32_t> count_;
};

union semun {
@@ -310,5 +308,36 @@
   AData buf[capacity];
};

template <class Int>
class AtomicQueue<0, Int>
{
   typedef Int Data;
   typedef std::atomic<Data> AData;
   static_assert(sizeof(Data) == sizeof(AData));

public:
   AtomicQueue() { memset(this, 0, sizeof(*this)); }
   bool push(const Data d, bool try_more = false)
   {
      auto cur = buf.load();
      return Empty(cur) && buf.compare_exchange_strong(cur, Enc(d));
   }
   bool pop(Data &d, bool try_more = false)
   {
      Data cur = buf.load();
      bool r = !Empty(cur) && buf.compare_exchange_strong(cur, 0);
      if (r) { d = Dec(cur); }
      return r;
   }
   uint32_t head() const { return 0; }
   uint32_t tail() const { return 0; }

private:
   static inline bool Empty(const Data d) { return (d & 1) == 0; } // lowest bit 1 means data ok.
   static inline Data Enc(const Data d) { return (d << 1) | 1; }   // lowest bit 1 means data ok.
   static inline Data Dec(const Data d) { return d >> 1; }         // lowest bit 1 means data ok.
   AData buf;
};

} // namespace robust
#endif // end of include guard: ROBUST_Q31RCWYU