test/stress/sequential/sequential_queue.cpp

   1 /*
   2     This file is a part of libcds - Concurrent Data Structures library
   3
   4     (C) Copyright Maxim Khizhinsky (libcds.dev@gmail.com) 2006-2017
   5
   6     Source code repo: http://github.com/khizmax/libcds/
   7     Download: http://sourceforge.net/projects/libcds/files/
   8
   9     Redistribution and use in source and binary forms, with or without
  10     modification, are permitted provided that the following conditions are met:
  11
  12     * Redistributions of source code must retain the above copyright notice,
  13    this
  14       list of conditions and the following disclaimer.
  15
  16     * Redistributions in binary form must reproduce the above copyright notice,
  17       this list of conditions and the following disclaimer in the documentation
  18       and/or other materials provided with the distribution.
  19
  20     THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  21     AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  22     IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  23    ARE
  24     DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
  25     FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  26     DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
  27     SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
  28     CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  29    LIABILITY,
  30     OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
  31    USE
  32     OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  33 */
  34
  35 #include "../misc/common.h"
  36 #include "../queue/queue_type.h"
  37
  38 #include <algorithm>
  39 #include <type_traits>
  40 #include <vector>
  41
  42 // Sequential queue push/pop test
  43 namespace {
  44
  45 static size_t s_nEnqueueCount = 4000000;
  46 // Call 's_nEnqueueStride' number of enqueue before dequeue.
  47 static size_t s_nEnqueueStride = 10000;
  48 static size_t s_nQueueSize = 100000;
  49 static size_t s_nVyukovQueuePushCount = 300000;
  50
  51 struct old_value {
  52   size_t nNo;
  53 };
  54
  55 template <class Value = old_value>
  56 class sequential_queue : public cds_test::stress_fixture {
  57 protected:
  58   using value_type = Value;
  59
  60   template <typename Queue>
  61   void test(Queue &q, size_t enqueue_count = s_nEnqueueCount) {
  62     value_type v;
  63     v.nNo = 0;
  64     size_t push_failure = 0;
  65     size_t pop_sum = 0;
  66
  67     while (v.nNo < enqueue_count) {
  68       size_t curr_push_count =
  69           std::min(enqueue_count - v.nNo, s_nEnqueueStride);
  70       for (size_t i = 0; i < curr_push_count; i++) {
  71         if (q.push(v))
  72           ++v.nNo;
  73         else
  74           ++push_failure;
  75       }
  76
  77       value_type res;
  78       while (q.pop(res)) {
  79         pop_sum += res.nNo;
  80       }
  81     }
  82
  83     if (push_failure) {
  84       std::cout << "Sequential queue push error count: " << push_failure
  85                 << "\n";
  86     }
  87     size_t supposed_sum = enqueue_count * (enqueue_count - 1) / 2;
  88     if (pop_sum != supposed_sum) {
  89       std::cout << "Sequential queue pop sum: " << pop_sum
  90                 << " != " << supposed_sum << "\n";
  91     }
  92   }
  93
  94 public:
  95   static void SetUpTestCase() {
  96     cds_test::config const &cfg = get_config("SequentialQueue");
  97     GetConfig(EnqueueCount);
  98     GetConfig(EnqueueStride);
  99     GetConfig(QueueSize);
 100     GetConfig(VyukovQueuePushCount);
 101   }
 102 };
 103
 104 using simple_sequential_queue = sequential_queue<>;
 105
 106 #define CDSSTRESS_Sequential_Queue_F(test_fixture, type_name)                  \
 107   TEST_F(test_fixture, type_name) {                                            \
 108     typedef queue::Types<value_type>::type_name queue_type;                    \
 109     queue_type queue(s_nQueueSize);                                            \
 110     test(queue, s_nVyukovQueuePushCount);                                      \
 111   }
 112
 113 #define CDSSTRESS_Sequential_VyukovQueue(test_fixture)                         \
 114   CDSSTRESS_Sequential_Queue_F(test_fixture, VyukovMPMCCycleQueue_dyn);        \
 115   CDSSTRESS_Sequential_Queue_F(test_fixture, VyukovMPMCCycleQueue_dyn_ic);
 116
 117 CDSSTRESS_Sequential_VyukovQueue(simple_sequential_queue);
 118
 119 #undef CDSSTRESS_Sequential_Queue_F
 120 #define CDSSTRESS_Sequential_Queue_F(test_fixture, type_name)                  \
 121   TEST_F(test_fixture, type_name) {                                            \
 122     typedef queue::Types<value_type>::type_name queue_type;                    \
 123     queue_type queue;                                                          \
 124     test(queue);                                                               \
 125   }
 126
 127 #define CDSSTRESS_Sequential_MSQueue(test_fixture)                             \
 128   CDSSTRESS_Sequential_Queue_F(test_fixture, MSQueue_HP);                      \
 129   CDSSTRESS_Sequential_Queue_F(test_fixture, MSQueue_DHP);
 130
 131 #define CDSSTRESS_Sequential_MoirQueue(test_fixture)                           \
 132   CDSSTRESS_Sequential_Queue_F(test_fixture, MoirQueue_HP);                    \
 133   CDSSTRESS_Sequential_Queue_F(test_fixture, MoirQueue_DHP);
 134
 135 #define CDSSTRESS_Sequential_OptimsticQueue(test_fixture)                      \
 136   CDSSTRESS_Sequential_Queue_F(test_fixture, OptimisticQueue_HP);              \
 137   CDSSTRESS_Sequential_Queue_F(test_fixture, OptimisticQueue_DHP);
 138
 139 #define CDSSTRESS_Sequential_BasketQueue(test_fixture)                         \
 140   CDSSTRESS_Sequential_Queue_F(test_fixture, BasketQueue_HP);                  \
 141   CDSSTRESS_Sequential_Queue_F(test_fixture, BasketQueue_DHP);
 142
 143 #define CDSSTRESS_Sequential_RWQueue(test_fixture)                             \
 144   CDSSTRESS_Sequential_Queue_F(test_fixture, RWQueue_Spin);
 145
 146 #define CDSSTRESS_Sequential_WeakRingBuffer(test_fixture)                      \
 147   CDSSTRESS_Sequential_Queue_F(test_fixture, WeakRingBuffer_dyn);
 148
 149 CDSSTRESS_Sequential_MSQueue(simple_sequential_queue);
 150 CDSSTRESS_Sequential_MoirQueue(simple_sequential_queue);
 151 CDSSTRESS_Sequential_BasketQueue(simple_sequential_queue);
 152 CDSSTRESS_Sequential_OptimsticQueue(simple_sequential_queue);
 153 CDSSTRESS_Sequential_RWQueue(simple_sequential_queue);
 154 //CDSSTRESS_Sequential_WeakRingBuffer(simple_sequential_queue);
 155
 156 // ********************************************************************
 157 // SegmentedQueue test
 158
 159 class sequential_segmented_queue
 160     : public sequential_queue<>,
 161       public ::testing::WithParamInterface<size_t> {
 162   typedef sequential_queue<> base_class;
 163
 164 protected:
 165   template <typename Queue> void test() {
 166     size_t quasi_factor = GetParam();
 167     Queue q(quasi_factor);
 168     base_class::test(q);
 169   }
 170
 171 public:
 172   static std::vector<size_t> get_test_parameters() {
 173     cds_test::config const &cfg =
 174         cds_test::stress_fixture::get_config("SequentialQueue");
 175     bool bIterative = cfg.get_bool("SegmentedQueue_Iterate", false);
 176     size_t quasi_factor = cfg.get_size_t("SegmentedQueue_SegmentSize", 256);
 177
 178     std::vector<size_t> args;
 179     if (bIterative && quasi_factor > 4) {
 180       for (size_t qf = 4; qf <= quasi_factor; qf *= 2)
 181         args.push_back(qf);
 182     } else {
 183       if (quasi_factor > 2)
 184         args.push_back(quasi_factor);
 185       else
 186         args.push_back(2);
 187     }
 188
 189     return args;
 190   }
 191 };
 192
 193 #undef CDSSTRESS_Sequential_Queue_F
 194 #define CDSSTRESS_Sequential_Queue_F(test_fixture, type_name)                  \
 195   TEST_P(test_fixture, type_name) {                                            \
 196     typedef typename queue::Types<value_type>::type_name queue_type;           \
 197     test<queue_type>();                                                        \
 198   }
 199
 200 #define CDSSTRESS_Sequential_SegmentedQueue(test_fixture)                      \
 201   CDSSTRESS_Sequential_Queue_F(test_fixture, SegmentedQueue_HP_spin);          \
 202   CDSSTRESS_Sequential_Queue_F(test_fixture, SegmentedQueue_HP_spin_padding);  \
 203   CDSSTRESS_Sequential_Queue_F(test_fixture, SegmentedQueue_DHP_spin);
 204
 205 CDSSTRESS_Sequential_SegmentedQueue(sequential_segmented_queue)
 206
 207 #ifdef CDSTEST_GTEST_INSTANTIATE_TEST_CASE_P_HAS_4TH_ARG
 208     static std::string
 209     get_test_parameter_name(testing::TestParamInfo<size_t> const &p) {
 210   return std::to_string(p.param);
 211 }
 212 INSTANTIATE_TEST_CASE_P(
 213     SQ, sequential_segmented_queue,
 214     ::testing::ValuesIn(sequential_segmented_queue::get_test_parameters()),
 215     get_test_parameter_name);
 216 #else
 217     INSTANTIATE_TEST_CASE_P(
 218         SQ, sequential_segmented_queue,
 219         ::testing::ValuesIn(sequential_segmented_queue::get_test_parameters()));
 220 #endif
 221
 222 // ********************************************************************
 223 // SPSC WeakRingBuffer
 224 static size_t s_nBufferSize = 1024 * 1024;
 225 static size_t s_nPushCount = 1000000;
 226
 227 class sequential_weak_ring_buffer : public cds_test::stress_fixture {
 228 protected:
 229   typedef size_t value_type;
 230
 231   template <typename Queue> void test(Queue &m_Queue) {
 232     size_t const nPushCount = s_nPushCount;
 233     size_t m_nPushed = 0;
 234     size_t m_nPushFailed = 0;
 235     size_t m_nPopped = 0;
 236     size_t m_nBadValue = 0;
 237     size_t m_nPopFrontFailed = 0;
 238     size_t m_nPopEmpty = 0;
 239
 240     for (size_t i = 0; i < nPushCount; ++i) {
 241       size_t len = rand(1024) + 64;
 242       void *buf = m_Queue.back(len);
 243       if (buf) {
 244         memset(buf, len % 256, len);
 245         m_Queue.push_back();
 246         m_nPushed += len;
 247       } else
 248         ++m_nPushFailed;
 249     }
 250
 251     while (true) {
 252       auto buf = m_Queue.front();
 253       if (buf.first) {
 254         m_nPopped += buf.second;
 255
 256         uint8_t val = static_cast<uint8_t>(buf.second % 256);
 257         uint8_t const *p = reinterpret_cast<uint8_t *>(buf.first);
 258         for (uint8_t const *pEnd = p + buf.second; p < pEnd; ++p) {
 259           if (*p != val) {
 260             ++m_nBadValue;
 261             break;
 262           }
 263         }
 264
 265         if (!m_Queue.pop_front())
 266           ++m_nPopFrontFailed;
 267       } else {
 268         ++m_nPopEmpty;
 269         if (m_Queue.empty())
 270           break;
 271       }
 272     }
 273   }
 274
 275 public:
 276   static void SetUpTestCase() {
 277     cds_test::config const &cfg = get_config("SequentialRingBuffer");
 278
 279     s_nBufferSize = cfg.get_size_t("BufferSize", s_nBufferSize);
 280     s_nPushCount = cfg.get_size_t("PushCount", s_nPushCount);
 281
 282     if (s_nBufferSize < 1024 * 64)
 283       s_nBufferSize = 1024 * 64;
 284     if (s_nPushCount == 0u)
 285       s_nPushCount = 1024;
 286   }
 287 };
 288
 289 #undef CDSSTRESS_Sequential_Queue_F
 290 #define CDSSTRESS_Sequential_Queue_F(test_fixture, type_name)                  \
 291   TEST_F(test_fixture, type_name) {                                            \
 292     typedef queue::Types<value_type>::type_name queue_type;                    \
 293     queue_type queue(s_nBufferSize);                                           \
 294     test(queue);                                                               \
 295   }
 296
 297 CDSSTRESS_WeakRingBuffer_void(sequential_weak_ring_buffer)
 298
 299 #undef CDSSTRESS_Queue_F
 300
 301 } // namespace