Merge pull request #14 from DNedic/feat/multithreaded_tests

Multi-threaded tests for more data structures

tests/CMakeLists.txt

@@ -22,6 +22,12 @@ add_executable(tests
     mpmc/priority_queue.cpp
 )
 
+if (NOT DEFINED TEST_MT_TRANSFER_CNT)
+    set(TEST_MT_TRANSFER_CNT 10240)
+endif()
+
+target_compile_definitions(tests PRIVATE TEST_MT_TRANSFER_CNT=${TEST_MT_TRANSFER_CNT})
+
 # Required in order to test the std::span API as well
 target_compile_features(tests PRIVATE cxx_std_20)
 

tests/README.md

@@ -0,0 +1,35 @@
+# Tests
+
+The library contains tests for all data structures and their respective features.
+Each data structure has it's own test file, split into `spsc` and `mpmc` folders.
+
+## Building and running
+
+In order to build tests, simply run CMake in the library root:
+```
+cmake -B build
+cmake --build build
+```
+
+> **Note:** Due to `std::span` tests, a compiler with C++20 support is required for building tests
+
+After that, you can run tests either with `ctest`:
+```
+ctest --output-on-failure --test-dir build/tests
+```
+or by executing the `build/tests/tests` binary.
+
+## Multi-threaded tests
+Apart from regular unit tests, the library also contains multi-threaded tests.
+As these tests are not deterministic by their nature, and can give false negatives, the number of elements copied is parametrized.
+
+To define the number of elements to transfer in multi-threaded tests, pass the `TEST_MT_TRANSFER_CNT` variable to CMake:
+```
+cmake -DTEST_MT_TRANSFER_CNT=100000 -B build
+```
+
+## Writing tests
+If adding a new feature, or fixing a bug, it is necessary to add tests in order to avoid future regressions.
+You can take a look at existing tests for examples.
+
+[Catch2](https://github.com/catchorg/Catch2) is used as the testing framework, you can read the documentation of the library [here](https://github.com/catchorg/Catch2/blob/devel/docs/tutorial.md#writing-tests).

tests/spsc/bipartite_buf.cpp

@@ -1,5 +1,6 @@
 #include <algorithm>
 #include <catch2/catch_test_macros.hpp>
+#include <thread>
 
 #include "lockfree.hpp"
 
@@ -234,3 +235,52 @@ TEST_CASE("std::span API test", "[bb_std_span_api]") {
 
     REQUIRE(read_pair.first == read_span.data());
 }
+
+TEST_CASE("Multithreaded read/write multiple", "[bb_multithread_multiple]") {
+    std::vector<std::thread> threads;
+    lockfree::spsc::BipartiteBuf<unsigned int, 1024U> bb;
+    std::vector<unsigned int> written;
+    std::vector<unsigned int> read;
+
+    const size_t data_size = 59; // Intentionally a prime number
+
+    // consumer
+    threads.emplace_back([&]() {
+        size_t read_count = 0;
+        do {
+            std::pair<unsigned int *, size_t> read_region = bb.ReadAcquire();
+            if (read_region.second) {
+                read.insert(read.end(), &read_region.first[0],
+                            &read_region.first[read_region.second]);
+                bb.ReadRelease(read_region.second);
+                read_count += read_region.second;
+            }
+        } while (read_count < TEST_MT_TRANSFER_CNT);
+    });
+
+    // producer
+    threads.emplace_back([&]() {
+        unsigned int data[data_size] = {0};
+        for (unsigned int i = 0; i < data_size; i++) {
+            data[i] = rand();
+        }
+
+        size_t write_count = 0;
+        do {
+            unsigned int *write_region = bb.WriteAcquire(data_size);
+            if (write_region != nullptr) {
+                std::copy(&data[0], &data[data_size], write_region);
+                bb.WriteRelease(data_size);
+                written.insert(written.end(), &data[0], &data[data_size]);
+                write_count += data_size;
+            }
+        } while (write_count < TEST_MT_TRANSFER_CNT);
+    });
+
+    for (auto &t : threads) {
+        t.join();
+    }
+
+    REQUIRE(
+        std::equal(std::begin(written), std::end(written), std::begin(read)));
+}

tests/spsc/queue.cpp

@@ -1,5 +1,6 @@
 #include <algorithm>
 #include <math.h>
+#include <thread>
 
 #include <catch2/catch_test_macros.hpp>
 
@@ -87,3 +88,40 @@ TEST_CASE("Optional API", "[q_optional_api]") {
 
     REQUIRE(queue.PopOptional() == -1024);
 }
+
+TEST_CASE("Multithreaded read/write", "[q_multithread]") {
+    std::vector<std::thread> threads;
+    lockfree::spsc::Queue<uint64_t, 1024U> queue;
+    std::vector<uint64_t> written;
+    std::vector<uint64_t> read;
+
+    // consumer
+    threads.emplace_back([&]() {
+        uint64_t element = 0;
+        do {
+            bool read_success = queue.Pop(element);
+            if (read_success) {
+                read.push_back(element);
+            }
+        } while (element < TEST_MT_TRANSFER_CNT);
+    });
+
+    // producer
+    threads.emplace_back([&]() {
+        uint64_t element = 0;
+        do {
+            bool push_success = queue.Push(element);
+            if (push_success) {
+                written.push_back(element);
+                element++;
+            }
+        } while (element < TEST_MT_TRANSFER_CNT + 1);
+    });
+
+    for (auto &t : threads) {
+        t.join();
+    }
+
+    REQUIRE(
+        std::equal(std::begin(written), std::end(written), std::begin(read)));
+}

tests/spsc/ring_buf.cpp

@@ -324,7 +324,7 @@ TEST_CASE("Peek std::span", "[rb_peek_span]") {
                        std::begin(test_data_read)));
 }
 
-TEST_CASE("Multithreaded read/write", "[rb_multi]") {
+TEST_CASE("Multithreaded read/write", "[rb_multithread]") {
     std::vector<std::thread> threads;
     lockfree::spsc::RingBuf<uint64_t, 1024U> rb;
     std::vector<uint64_t> written;
@@ -338,7 +338,7 @@ TEST_CASE("Multithreaded read/write", "[rb_multi]") {
             if (read_success) {
                 read.push_back(data[0]);
             }
-        } while (data[0] < 2047);
+        } while (data[0] < TEST_MT_TRANSFER_CNT);
     });
     // producer
     threads.emplace_back([&]() {
@@ -349,11 +349,57 @@ TEST_CASE("Multithreaded read/write", "[rb_multi]") {
                 written.push_back(cnt);
                 cnt++;
             }
-        } while (cnt < 2048);
+        } while (cnt < TEST_MT_TRANSFER_CNT + 1);
     });
     for (auto &t : threads) {
         t.join();
     }
     REQUIRE(
         std::equal(std::begin(written), std::end(written), std::begin(read)));
 }
+
+TEST_CASE("Multithreaded read/write multiple", "[rb_multithread_multiple]") {
+    std::vector<std::thread> threads;
+    lockfree::spsc::RingBuf<unsigned int, 1024U> rb;
+    std::vector<unsigned int> written;
+    std::vector<unsigned int> read;
+
+    const size_t data_size = 59; // Intentionally a prime number
+
+    // consumer
+    threads.emplace_back([&]() {
+        unsigned int data[data_size] = {0};
+        size_t read_count = 0;
+        do {
+            bool read_success = rb.Read(data, data_size);
+            if (read_success) {
+                read.insert(read.end(), &data[0], &data[data_size]);
+                read_count += data_size;
+            }
+        } while (read_count < TEST_MT_TRANSFER_CNT);
+    });
+
+    // producer
+    threads.emplace_back([&]() {
+        unsigned int data[data_size] = {0};
+        for (unsigned int i = 0; i < data_size; i++) {
+            data[i] = rand();
+        }
+
+        size_t write_count = 0;
+        do {
+            bool write_success = rb.Write(data, data_size);
+            if (write_success) {
+                written.insert(written.end(), &data[0], &data[data_size]);
+                write_count += data_size;
+            }
+        } while (write_count < TEST_MT_TRANSFER_CNT);
+    });
+
+    for (auto &t : threads) {
+        t.join();
+    }
+
+    REQUIRE(
+        std::equal(std::begin(written), std::end(written), std::begin(read)));
+}