add robust FMutex, works fine; use boost circular.

lichao

2021-05-08 28f06bc49a4d8d69f1ea2f767863b7921d12f155

 utest/robust_test.cpp

@@ -1,5 +1,6 @@
#include "robust.h"
#include "util.h"
#include <boost/circular_buffer.hpp>

using namespace robust;

@@ -18,7 +19,7 @@
BOOST_AUTO_TEST_CASE(QueueTest)
{
   const int nthread = 100;
   const uint64_t nmsg = 1000 * 1000 * 100;
   const uint64_t nmsg = 1000 * 1000 * 10;

   SharedMemory &shm = TestShm();
   shm.Remove();
@@ -33,7 +34,12 @@
      BOOST_CHECK_EQUAL((u64 & 255), i);
   }

#if 1
   uint64_t correct_total = nmsg * (nmsg - 1) / 2;
   std::atomic<uint64_t> total(0);
   std::atomic<uint64_t> nwrite(0);
   std::atomic<uint64_t> writedone(0);

#if 0
   typedef AtomicQueue<4> Rcb;

   Rcb tmp;
@@ -48,18 +54,8 @@
   BOOST_CHECK(tmp.tail() == 1);

   ShmObject<Rcb> rcb(shm, "test_rcb");
#else
   typedef Circular<int64_t> Rcb;
   ShmObject<Rcb> rcb(shm, "test_rcb", 64, shm.get_segment_manager());
#endif
   bool try_more = true;

   const int nsize = sizeof(Rcb);

   bool try_more = false;
   uint64_t correct_total = nmsg * (nmsg - 1) / 2;
   std::atomic<uint64_t> total(0);
   std::atomic<uint64_t> nwrite(0);
   std::atomic<uint64_t> writedone(0);
   auto Writer = [&]() {
      uint64_t n = 0;
      while ((n = nwrite++) < nmsg) {
@@ -82,6 +78,58 @@
      }
   };

#else
   typedef Circular<int64_t> Rcb;
   ShmObject<Rcb> rcb(shm, "test_rcb", 16, shm.get_segment_manager());

   typedef FMutex Mutex;
   // typedef SemMutex Mutex;
   Mutex mtx(123);
   auto Writer = [&]() {
      uint64_t n = 0;
      while ((n = nwrite++) < nmsg) {
         auto Write = [&]() {
            robust::Guard<Mutex> lk(mtx);
            if (rcb->full()) {
               return false;
            } else {
               rcb->push_back(n);
               return true;
            }
            // return rcb->push_back(n);
         };
         while (!Write()) {
            // MySleep();
         }
         ++writedone;
      }
   };
   std::atomic<uint64_t> nread(0);
   auto Reader = [&]() {
      while (nread.load() < nmsg) {
         int64_t d;
         auto Read = [&]() {
            robust::Guard<Mutex> lk(mtx);
            if (rcb->empty()) {
               return false;
            } else {
               d = rcb->front();
               rcb->pop_front();
               return true;
            }
            // return rcb->pop_front(d);
         };
         if (Read()) {
            ++nread;
            total += d;
         } else {
            // MySleep();
         }
      }
   };

#endif

   auto status = [&]() {
      auto next = steady_clock::now();
      uint32_t lw = 0;
@@ -102,7 +150,8 @@
   {
      ThreadManager threads;
      boost::timer::auto_cpu_timer timer;
      printf("Testing Robust Buffer, msgs %ld, queue size: %d, threads: %d \n", nmsg, Rcb::capacity, nthread);
      // printf("Testing Robust Buffer, msgs %ld, queue size: %d, threads: %d \n", nmsg, Rcb::capacity, nthread);
      printf("Testing Robust Buffer, msgs %ld, queue size: %d, threads: %d \n", nmsg, 16, nthread);
      for (int i = 0; i < nthread; ++i) {
         threads.Launch(Reader);
         threads.Launch(Writer);
@@ -116,7 +165,8 @@

BOOST_AUTO_TEST_CASE(MutexTest)
{
   typedef robust::Mutex RobustMutex;
   // typedef robust::MFMutex RobustMutex;
   typedef robust::SemMutex RobustMutex;

   for (int i = 0; i < 20; ++i) {
      int size = i;
@@ -131,7 +181,9 @@

   const std::string mtx_name("test_mutex");
   const std::string int_name("test_int");
   auto mtx = shm.FindOrCreate<RobustMutex>(mtx_name);
   // auto mtx = shm.FindOrCreate<RobustMutex>(mtx_name, 12345);
   RobustMutex rmtx(12345);
   auto mtx = &rmtx;
   auto pi = shm.FindOrCreate<int>(int_name, 100);

   std::mutex m;
@@ -142,29 +194,48 @@
      printf("int : %d, add1: %d\n", old, ++*pi);
   }

   {
      const int ntimes = 1000 * 1000;
      RobustMutex mutex;
   auto LockSpeed = [](auto &mutex, const std::string &name) {
      const int ntimes = 1000 * 1;
      auto Lock = [&]() {
         for (int i = 0; i < ntimes; ++i) {
            mutex.lock();
            mutex.unlock();
         }
      };

      printf("\nTesting %s lock/unlock %d times\n", name.c_str(), ntimes);
      {
         boost::timer::auto_cpu_timer timer;
         printf("test lock/unlock %d times: ", ntimes);
         printf("1 thread: ");
         Lock();
      }
      {
      auto InThread = [&](int nthread) {
         boost::timer::auto_cpu_timer timer;
         printf("test lock/unlock %d times, 2 thread: ", ntimes);
         std::thread t1(Lock), t2(Lock);
         t1.join();
         t2.join();
      }
   }
         printf("%d threads: ", nthread);
         std::vector<std::thread> vt;
         for (int i = 0; i < nthread; ++i) {
            vt.emplace_back(Lock);
         }
         for (auto &t : vt) {
            t.join();
         }
      };
      InThread(4);
      InThread(16);
      InThread(100);
      InThread(1000);
   };
   NullMutex null_mtx;
   std::mutex std_mtx;
   CasMutex cas_mtx;
   FMutex mfmtx(3);
   boost::interprocess::interprocess_mutex ipc_mutex;
   SemMutex sem_mtx(3);
   LockSpeed(null_mtx, "null mutex");
   LockSpeed(std_mtx, "std::mutex");
   // LockSpeed(cas_mtx, "CAS mutex");
   LockSpeed(ipc_mutex, "boost ipc mutex");
   LockSpeed(mfmtx, "mutex+flock");
   LockSpeed(sem_mtx, "sem mutex");

   auto TryLock = [&]() {
      if (mtx->try_lock()) {
@@ -183,6 +254,7 @@
   if (mtx) {
      printf("mtx exists\n");
      if (TryLock()) {
         // Sleep(10s);
         auto op = [&]() {
            if (TryLock()) {
               Unlock();