test/stress/misc/deque_driver.cpp

   1 #include "common.h"
   2 #include <cds/misc/chase-lev-deque.h>
   3 #include <cds_test/stress_test.h>
   4 #include <cstdlib>
   5 #include <ctime>
   6 #include <iostream>
   7 #include <memory>
   8 #include <thread>
   9
  10 using namespace std;
  11
  12 namespace {
  13
  14 typedef cds_others::ChaseLevDeque Deque;
  15 static size_t s_nDequeStealerThreadCount = 3;
  16 static size_t s_nDequeMainPassCount = 100000000;
  17
  18 class ChaseLevDequeTest : public cds_test::stress_fixture {
  19 protected:
  20   static Deque *deque;
  21   static atomic_int terminate_stealer;
  22   static ullong *sums;
  23   static ullong *succ_counts;
  24   static ullong push_sum;
  25   static ullong push_count;
  26
  27   static void SetUpTestCase() {
  28     cds_test::config const &cfg = get_config("Misc");
  29     GetConfig(DequeStealerThreadCount);
  30     GetConfig(DequeMainPassCount);
  31   }
  32
  33   static void StealerThread(int index) {
  34     while (!terminate_stealer.load(memory_order_relaxed)) {
  35       while (true) {
  36         int res = deque->steal();
  37         if (res != EMPTY && res != ABORT) {
  38           sums[index] += res;
  39           succ_counts[index]++;
  40         } else {
  41           break;
  42         }
  43       }
  44     }
  45   }
  46
  47   static void MainThread(int index, int push_percentage) {
  48     for (ullong i = 0; i < s_nDequeMainPassCount; i++) {
  49       if ((::rand() % 100) < push_percentage) {
  50         int item = ::rand() % 100;
  51         deque->push(item);
  52         push_sum += item;
  53         push_count++;
  54       } else {
  55         int res = deque->take();
  56         if (res != EMPTY) {
  57           sums[index] += res;
  58           succ_counts[index]++;
  59         }
  60       }
  61     }
  62     while (true) {
  63       int res = deque->take();
  64       if (res != EMPTY) {
  65         sums[index] += res;
  66         succ_counts[index]++;
  67       } else {
  68         break;
  69       }
  70     }
  71   }
  72 };
  73
  74 atomic_int ChaseLevDequeTest::terminate_stealer;
  75 ullong *ChaseLevDequeTest::sums;
  76 ullong *ChaseLevDequeTest::succ_counts;
  77 ullong ChaseLevDequeTest::push_count;
  78 ullong ChaseLevDequeTest::push_sum;
  79 Deque *ChaseLevDequeTest::deque;
  80
  81 TEST_F(ChaseLevDequeTest, DequePushPopTake) {
  82   deque = new Deque();
  83   push_sum = 0;
  84   sums = (ullong *)calloc(1, sizeof(ullong) * (s_nDequeStealerThreadCount + 1));
  85   succ_counts =
  86       (ullong *)calloc(1, sizeof(ullong) * (s_nDequeStealerThreadCount + 1));
  87   srand(time(NULL));
  88
  89   // Stealer threads
  90   std::unique_ptr<std::thread[]> threads(
  91       new std::thread[s_nDequeStealerThreadCount]);
  92   for (ullong i = 0; i < s_nDequeStealerThreadCount; i++) {
  93     threads[i] = std::thread(StealerThread, i);
  94   }
  95
  96   for (int i = 90; i > 0; i -= 10) {
  97     MainThread(s_nDequeStealerThreadCount, i);
  98   }
  99
 100   terminate_stealer.store(1, memory_order_relaxed);
 101   for (ullong i = 0; i < s_nDequeStealerThreadCount; i++) {
 102     threads[i].join();
 103   }
 104
 105   // Result analysis
 106   ullong received_sum = 0;
 107   ullong overall_count = 0;
 108   for (ullong i = 0; i <= s_nDequeStealerThreadCount; i++) {
 109     received_sum += sums[i];
 110     overall_count += succ_counts[i];
 111   }
 112   if (overall_count != push_count || received_sum != push_sum) {
 113     cout << "Incorrect deque\n";
 114     cout << "Push sum: " << push_sum << "\n";
 115     cout << "Received sum:" << received_sum << "\n";
 116     cout << "Push count=" << push_count << "\n";
 117     cout << "Received count:" << overall_count << "\n";
 118   }
 119 }
 120
 121 } // namespace