mfbt/tests/TestSPSCQueue.cpp

   1 /* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* vim: set ts=8 sts=2 et sw=2 tw=80: */
   3 /* This Source Code Form is subject to the terms of the Mozilla Public
   4  * License, v. 2.0. If a copy of the MPL was not distributed with this
   5  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
   6
   7 #include "mozilla/SPSCQueue.h"
   8 #include "mozilla/PodOperations.h"
   9 #include <vector>
  10 #include <iostream>
  11 #include <thread>
  12 #include <chrono>
  13 #include <memory>
  14 #include <string>
  15
  16 #ifdef _WIN32
  17 #  include <windows.h>
  18 #endif
  19
  20 using namespace mozilla;
  21
  22 /* Generate a monotonically increasing sequence of numbers. */
  23 template <typename T>
  24 class SequenceGenerator {
  25  public:
  26   SequenceGenerator() = default;
  27   void Get(T* aElements, size_t aCount) {
  28     for (size_t i = 0; i < aCount; i++) {
  29       aElements[i] = static_cast<T>(mIndex);
  30       mIndex++;
  31     }
  32   }
  33   void Rewind(size_t aCount) { mIndex -= aCount; }
  34
  35  private:
  36   size_t mIndex = 0;
  37 };
  38
  39 /* Checks that a sequence is monotonically increasing. */
  40 template <typename T>
  41 class SequenceVerifier {
  42  public:
  43   SequenceVerifier() = default;
  44   void Check(T* aElements, size_t aCount) {
  45     for (size_t i = 0; i < aCount; i++) {
  46       if (aElements[i] != static_cast<T>(mIndex)) {
  47         std::cerr << "Element " << i << " is different. Expected "
  48                   << static_cast<T>(mIndex) << ", got " << aElements[i] << "."
  49                   << std::endl;
  50         MOZ_RELEASE_ASSERT(false);
  51       }
  52       mIndex++;
  53     }
  54   }
  55
  56  private:
  57   size_t mIndex = 0;
  58 };
  59
  60 const int BLOCK_SIZE = 127;
  61
  62 template <typename T>
  63 void TestRing(int capacity) {
  64   SPSCQueue<T> buf(capacity);
  65   std::unique_ptr<T[]> seq(new T[capacity]);
  66   SequenceGenerator<T> gen;
  67   SequenceVerifier<T> checker;
  68
  69   int iterations = 1002;
  70
  71   while (iterations--) {
  72     gen.Get(seq.get(), BLOCK_SIZE);
  73     int rv = buf.Enqueue(seq.get(), BLOCK_SIZE);
  74     MOZ_RELEASE_ASSERT(rv == BLOCK_SIZE);
  75     PodZero(seq.get(), BLOCK_SIZE);
  76     rv = buf.Dequeue(seq.get(), BLOCK_SIZE);
  77     MOZ_RELEASE_ASSERT(rv == BLOCK_SIZE);
  78     checker.Check(seq.get(), BLOCK_SIZE);
  79   }
  80 }
  81
  82 void Delay() {
  83   // On Windows and x86 Android, the timer resolution is so bad that, even if
  84   // we used `timeBeginPeriod(1)`, any nonzero sleep from the test's inner loops
  85   // would make this program take far too long.
  86 #ifdef _WIN32
  87   Sleep(0);
  88 #elif defined(ANDROID)
  89   std::this_thread::sleep_for(std::chrono::microseconds(0));
  90 #else
  91   std::this_thread::sleep_for(std::chrono::microseconds(10));
  92 #endif
  93 }
  94
  95 template <typename T>
  96 void TestRingMultiThread(int capacity) {
  97   SPSCQueue<T> buf(capacity);
  98   SequenceVerifier<T> checker;
  99   std::unique_ptr<T[]> outBuffer(new T[capacity]);
 100
 101   std::thread t([&buf, capacity] {
 102     int iterations = 1002;
 103     std::unique_ptr<T[]> inBuffer(new T[capacity]);
 104     SequenceGenerator<T> gen;
 105
 106     while (iterations--) {
 107       Delay();
 108       gen.Get(inBuffer.get(), BLOCK_SIZE);
 109       int rv = buf.Enqueue(inBuffer.get(), BLOCK_SIZE);
 110       MOZ_RELEASE_ASSERT(rv <= BLOCK_SIZE);
 111       if (rv != BLOCK_SIZE) {
 112         gen.Rewind(BLOCK_SIZE - rv);
 113       }
 114     }
 115   });
 116
 117   int remaining = 1002;
 118
 119   while (remaining--) {
 120     Delay();
 121     int rv = buf.Dequeue(outBuffer.get(), BLOCK_SIZE);
 122     MOZ_RELEASE_ASSERT(rv <= BLOCK_SIZE);
 123     checker.Check(outBuffer.get(), rv);
 124   }
 125
 126   t.join();
 127 }
 128
 129 template <typename T>
 130 void BasicAPITest(T& ring) {
 131   MOZ_RELEASE_ASSERT(ring.Capacity() == 128);
 132
 133   MOZ_RELEASE_ASSERT(ring.AvailableRead() == 0);
 134   MOZ_RELEASE_ASSERT(ring.AvailableWrite() == 128);
 135
 136   int rv = ring.EnqueueDefault(63);
 137
 138   MOZ_RELEASE_ASSERT(rv == 63);
 139   MOZ_RELEASE_ASSERT(ring.AvailableRead() == 63);
 140   MOZ_RELEASE_ASSERT(ring.AvailableWrite() == 65);
 141
 142   rv = ring.EnqueueDefault(65);
 143
 144   MOZ_RELEASE_ASSERT(rv == 65);
 145   MOZ_RELEASE_ASSERT(ring.AvailableRead() == 128);
 146   MOZ_RELEASE_ASSERT(ring.AvailableWrite() == 0);
 147
 148   rv = ring.Dequeue(nullptr, 63);
 149
 150   MOZ_RELEASE_ASSERT(ring.AvailableRead() == 65);
 151   MOZ_RELEASE_ASSERT(ring.AvailableWrite() == 63);
 152
 153   rv = ring.Dequeue(nullptr, 65);
 154
 155   MOZ_RELEASE_ASSERT(ring.AvailableRead() == 0);
 156   MOZ_RELEASE_ASSERT(ring.AvailableWrite() == 128);
 157 }
 158
 159 const size_t RING_BUFFER_SIZE = 128;
 160 const size_t ENQUEUE_SIZE = RING_BUFFER_SIZE / 2;
 161
 162 void TestResetAPI() {
 163   SPSCQueue<float> ring(RING_BUFFER_SIZE);
 164   std::thread t([&ring] {
 165     std::unique_ptr<float[]> inBuffer(new float[ENQUEUE_SIZE]);
 166     int rv = ring.Enqueue(inBuffer.get(), ENQUEUE_SIZE);
 167     MOZ_RELEASE_ASSERT(rv > 0);
 168   });
 169
 170   t.join();
 171
 172   ring.ResetThreadIds();
 173
 174   // Enqueue with a different thread. We have reset the thread ID
 175   // in the ring buffer, this should work.
 176   std::thread t2([&ring] {
 177     std::unique_ptr<float[]> inBuffer(new float[ENQUEUE_SIZE]);
 178     int rv = ring.Enqueue(inBuffer.get(), ENQUEUE_SIZE);
 179     MOZ_RELEASE_ASSERT(rv > 0);
 180   });
 181
 182   t2.join();
 183 }
 184
 185 void TestMove() {
 186   const size_t ELEMENT_COUNT = 16;
 187   struct Thing {
 188     Thing() : mStr("") {}
 189     explicit Thing(const std::string& aStr) : mStr(aStr) {}
 190     Thing(Thing&& aOtherThing) {
 191       mStr = std::move(aOtherThing.mStr);
 192       // aOtherThing.mStr.clear();
 193     }
 194     Thing& operator=(Thing&& aOtherThing) {
 195       mStr = std::move(aOtherThing.mStr);
 196       return *this;
 197     }
 198     std::string mStr;
 199   };
 200
 201   std::vector<Thing> vec_in;
 202   std::vector<Thing> vec_out;
 203
 204   for (uint32_t i = 0; i < ELEMENT_COUNT; i++) {
 205     vec_in.push_back(Thing(std::to_string(i)));
 206     vec_out.push_back(Thing());
 207   }
 208
 209   SPSCQueue<Thing> queue(ELEMENT_COUNT);
 210
 211   int rv = queue.Enqueue(&vec_in[0], ELEMENT_COUNT);
 212   MOZ_RELEASE_ASSERT(rv == ELEMENT_COUNT);
 213
 214   // Check that we've moved the std::string into the queue.
 215   for (uint32_t i = 0; i < ELEMENT_COUNT; i++) {
 216     MOZ_RELEASE_ASSERT(vec_in[i].mStr.empty());
 217   }
 218
 219   rv = queue.Dequeue(&vec_out[0], ELEMENT_COUNT);
 220   MOZ_RELEASE_ASSERT(rv == ELEMENT_COUNT);
 221
 222   for (uint32_t i = 0; i < ELEMENT_COUNT; i++) {
 223     MOZ_RELEASE_ASSERT(std::stoul(vec_out[i].mStr) == i);
 224   }
 225 }
 226
 227 int main() {
 228   const int minCapacity = 199;
 229   const int maxCapacity = 1277;
 230   const int capacityIncrement = 27;
 231
 232   SPSCQueue<float> q1(128);
 233   BasicAPITest(q1);
 234   SPSCQueue<char> q2(128);
 235   BasicAPITest(q2);
 236
 237   for (uint32_t i = minCapacity; i < maxCapacity; i += capacityIncrement) {
 238     TestRing<uint32_t>(i);
 239     TestRingMultiThread<uint32_t>(i);
 240     TestRing<float>(i);
 241     TestRingMultiThread<float>(i);
 242   }
 243
 244   TestResetAPI();
 245   TestMove();
 246
 247   return 0;
 248 }