tor-browser

TestSPSCQueue.cpp (8064B)

---

/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set ts=8 sts=2 et sw=2 tw=80: */
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */

#include "mozilla/SPSCQueue.h"
#include "mozilla/PodOperations.h"
#include <vector>
#include <iostream>
#include <thread>
#include <chrono>
#include <memory>
#include <string>

#ifdef _WIN32
#  include <windows.h>
#endif

using namespace mozilla;

/* Generate a monotonically increasing sequence of numbers. */
template <typename T>
class SequenceGenerator {
public:
 SequenceGenerator() = default;
 void Get(T* aElements, size_t aCount) {
   for (size_t i = 0; i < aCount; i++) {
     aElements[i] = static_cast<T>(mIndex);
     mIndex++;
   }
 }
 void Rewind(size_t aCount) { mIndex -= aCount; }

private:
 size_t mIndex = 0;
};

/* Checks that a sequence is monotonically increasing. */
template <typename T>
class SequenceVerifier {
public:
 SequenceVerifier() = default;
 void Check(T* aElements, size_t aCount) {
   for (size_t i = 0; i < aCount; i++) {
     if (aElements[i] != static_cast<T>(mIndex)) {
       std::cerr << "Element " << i << " is different. Expected "
                 << static_cast<T>(mIndex) << ", got " << aElements[i] << "."
                 << std::endl;
       MOZ_RELEASE_ASSERT(false);
     }
     mIndex++;
   }
 }

private:
 size_t mIndex = 0;
};

const int BLOCK_SIZE = 127;

template <typename T>
void TestRing(int capacity) {
 SPSCQueue<T> buf(capacity);
 std::unique_ptr<T[]> seq(new T[capacity]);
 SequenceGenerator<T> gen;
 SequenceVerifier<T> checker;

 int iterations = 1002;

 while (iterations--) {
   gen.Get(seq.get(), BLOCK_SIZE);
   int rv = buf.Enqueue(seq.get(), BLOCK_SIZE);
   MOZ_RELEASE_ASSERT(rv == BLOCK_SIZE);
   PodZero(seq.get(), BLOCK_SIZE);
   rv = buf.Dequeue(seq.get(), BLOCK_SIZE);
   MOZ_RELEASE_ASSERT(rv == BLOCK_SIZE);
   checker.Check(seq.get(), BLOCK_SIZE);
 }
}

void Delay() {
 // On Windows and x86 Android, the timer resolution is so bad that, even if
 // we used `timeBeginPeriod(1)`, any nonzero sleep from the test's inner loops
 // would make this program take far too long.
#ifdef _WIN32
 Sleep(0);
#elif defined(ANDROID)
 std::this_thread::sleep_for(std::chrono::microseconds(0));
#else
 std::this_thread::sleep_for(std::chrono::microseconds(10));
#endif
}

template <typename T>
void TestRingMultiThread(int capacity) {
 SPSCQueue<T> buf(capacity);
 SequenceVerifier<T> checker;
 std::unique_ptr<T[]> outBuffer(new T[capacity]);

 std::thread t([&buf, capacity] {
   int iterations = 1002;
   std::unique_ptr<T[]> inBuffer(new T[capacity]);
   SequenceGenerator<T> gen;

   while (iterations--) {
     Delay();
     gen.Get(inBuffer.get(), BLOCK_SIZE);
     int rv = buf.Enqueue(inBuffer.get(), BLOCK_SIZE);
     MOZ_RELEASE_ASSERT(rv <= BLOCK_SIZE);
     if (rv != BLOCK_SIZE) {
       gen.Rewind(BLOCK_SIZE - rv);
     }
   }
 });

 int remaining = 1002;

 while (remaining--) {
   Delay();
   int rv = buf.Dequeue(outBuffer.get(), BLOCK_SIZE);
   MOZ_RELEASE_ASSERT(rv <= BLOCK_SIZE);
   checker.Check(outBuffer.get(), rv);
 }

 t.join();
}

template <typename T>
void BasicAPITest(T& ring) {
 MOZ_RELEASE_ASSERT(ring.Capacity() == 128);

 MOZ_RELEASE_ASSERT(ring.AvailableRead() == 0);
 MOZ_RELEASE_ASSERT(ring.AvailableWrite() == 128);

 int rv = ring.EnqueueDefault(63);

 MOZ_RELEASE_ASSERT(rv == 63);
 MOZ_RELEASE_ASSERT(ring.AvailableRead() == 63);
 MOZ_RELEASE_ASSERT(ring.AvailableWrite() == 65);

 rv = ring.EnqueueDefault(65);

 MOZ_RELEASE_ASSERT(rv == 65);
 MOZ_RELEASE_ASSERT(ring.AvailableRead() == 128);
 MOZ_RELEASE_ASSERT(ring.AvailableWrite() == 0);

 rv = ring.Dequeue(nullptr, 63);

 MOZ_RELEASE_ASSERT(ring.AvailableRead() == 65);
 MOZ_RELEASE_ASSERT(ring.AvailableWrite() == 63);

 rv = ring.Dequeue(nullptr, 65);

 MOZ_RELEASE_ASSERT(ring.AvailableRead() == 0);
 MOZ_RELEASE_ASSERT(ring.AvailableWrite() == 128);
}

const size_t RING_BUFFER_SIZE = 128;
const size_t ENQUEUE_SIZE = RING_BUFFER_SIZE / 2;

void TestResetAPI() {
 SPSCQueue<float> ring(RING_BUFFER_SIZE);
 std::thread p([&ring] {
   std::unique_ptr<float[]> inBuffer(new float[ENQUEUE_SIZE]);
   int rv = ring.Enqueue(inBuffer.get(), ENQUEUE_SIZE);
   MOZ_RELEASE_ASSERT(rv > 0);
 });

 p.join();

 std::thread c([&ring] {
   std::unique_ptr<float[]> outBuffer(new float[ENQUEUE_SIZE]);
   int rv = ring.Dequeue(outBuffer.get(), ENQUEUE_SIZE);
   MOZ_RELEASE_ASSERT(rv > 0);
 });

 c.join();

 // Enqueue with a different thread. We reset the thread ID in the ring buffer,
 // this should work.
 std::thread p2([&ring] {
   ring.ResetProducerThreadId();
   std::unique_ptr<float[]> inBuffer(new float[ENQUEUE_SIZE]);
   int rv = ring.Enqueue(inBuffer.get(), ENQUEUE_SIZE);
   MOZ_RELEASE_ASSERT(rv > 0);
 });

 p2.join();

 // Dequeue with a different thread. We reset the thread ID in the ring buffer,
 // this should work.
 std::thread c2([&ring] {
   ring.ResetConsumerThreadId();
   std::unique_ptr<float[]> outBuffer(new float[ENQUEUE_SIZE]);
   int rv = ring.Dequeue(outBuffer.get(), ENQUEUE_SIZE);
   MOZ_RELEASE_ASSERT(rv > 0);
 });

 c2.join();

 // Similarly, but do the Enqueues without a Dequeue in between, since a
 // Dequeue could affect memory ordering.
 std::thread p4;
 std::thread p3([&] {
   ring.ResetProducerThreadId();
   std::unique_ptr<float[]> inBuffer(new float[ENQUEUE_SIZE]);
   int rv = ring.Enqueue(inBuffer.get(), ENQUEUE_SIZE);
   MOZ_RELEASE_ASSERT(rv > 0);
   p4 = std::thread([&ring] {
     ring.ResetProducerThreadId();
     std::unique_ptr<float[]> inBuffer(new float[ENQUEUE_SIZE]);
     int rv = ring.Enqueue(inBuffer.get(), ENQUEUE_SIZE);
     MOZ_RELEASE_ASSERT(rv > 0);
   });
 });

 p3.join();
 p4.join();

 std::thread c4;
 std::thread c3([&] {
   ring.ResetConsumerThreadId();
   std::unique_ptr<float[]> outBuffer(new float[ENQUEUE_SIZE]);
   int rv = ring.Dequeue(outBuffer.get(), ENQUEUE_SIZE);
   MOZ_RELEASE_ASSERT(rv > 0);
   c4 = std::thread([&ring] {
     ring.ResetConsumerThreadId();
     std::unique_ptr<float[]> outBuffer(new float[ENQUEUE_SIZE]);
     int rv = ring.Dequeue(outBuffer.get(), ENQUEUE_SIZE);
     MOZ_RELEASE_ASSERT(rv > 0);
   });
 });

 c3.join();
 c4.join();
}

void TestMove() {
 const size_t ELEMENT_COUNT = 16;
 struct Thing {
   Thing() : mStr("") {}
   explicit Thing(const std::string& aStr) : mStr(aStr) {}
   Thing(Thing&& aOtherThing) {
     mStr = std::move(aOtherThing.mStr);
     // aOtherThing.mStr.clear();
   }
   Thing& operator=(Thing&& aOtherThing) {
     mStr = std::move(aOtherThing.mStr);
     return *this;
   }
   std::string mStr;
 };

 std::vector<Thing> vec_in;
 std::vector<Thing> vec_out;

 for (uint32_t i = 0; i < ELEMENT_COUNT; i++) {
   vec_in.push_back(Thing(std::to_string(i)));
   vec_out.push_back(Thing());
 }

 SPSCQueue<Thing> queue(ELEMENT_COUNT);

 int rv = queue.Enqueue(&vec_in[0], ELEMENT_COUNT);
 MOZ_RELEASE_ASSERT(rv == ELEMENT_COUNT);

 // Check that we've moved the std::string into the queue.
 for (uint32_t i = 0; i < ELEMENT_COUNT; i++) {
   MOZ_RELEASE_ASSERT(vec_in[i].mStr.empty());
 }

 rv = queue.Dequeue(&vec_out[0], ELEMENT_COUNT);
 MOZ_RELEASE_ASSERT(rv == ELEMENT_COUNT);

 for (uint32_t i = 0; i < ELEMENT_COUNT; i++) {
   MOZ_RELEASE_ASSERT(std::stoul(vec_out[i].mStr) == i);
 }
}

int main() {
 const int minCapacity = 199;
 const int maxCapacity = 1277;
 const int capacityIncrement = 27;

 SPSCQueue<float> q1(128);
 BasicAPITest(q1);
 SPSCQueue<char> q2(128);
 BasicAPITest(q2);

 for (uint32_t i = minCapacity; i < maxCapacity; i += capacityIncrement) {
   TestRing<uint32_t>(i);
   TestRingMultiThread<uint32_t>(i);
   TestRing<float>(i);
   TestRingMultiThread<float>(i);
 }

 TestResetAPI();
 TestMove();

 return 0;
}