File size: 3,210 Bytes
f86fb3b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | #include <gtest/gtest.h>
#include <thread>
#include <vector>
#include <atomic>
#include "../include/core/spsc_queue.hpp"
#include "../include/core/mpsc_queue.hpp"
#include "../include/core/thread_utils.hpp"
using namespace quant::core;
TEST(LockFreeQueuesTest, SPSCSingleThreadBasic) {
LockFreeSPSCQueue<int, 4> queue;
EXPECT_TRUE(queue.empty());
EXPECT_EQ(queue.size(), 0);
EXPECT_TRUE(queue.push(10));
EXPECT_TRUE(queue.push(20));
EXPECT_TRUE(queue.push(30));
EXPECT_TRUE(queue.push(40));
// Capacity 4 is full: 5th push must return false
EXPECT_FALSE(queue.push(50));
EXPECT_EQ(queue.size(), 4);
int val = 0;
EXPECT_TRUE(queue.pop(val));
EXPECT_EQ(val, 10);
EXPECT_TRUE(queue.push(50));
EXPECT_TRUE(queue.pop(val));
EXPECT_EQ(val, 20);
EXPECT_TRUE(queue.pop(val));
EXPECT_EQ(val, 30);
EXPECT_TRUE(queue.pop(val));
EXPECT_EQ(val, 40);
EXPECT_TRUE(queue.pop(val));
EXPECT_EQ(val, 50);
EXPECT_FALSE(queue.pop(val));
EXPECT_TRUE(queue.empty());
}
TEST(LockFreeQueuesTest, SPSCConcurrentStress) {
constexpr size_t TOTAL_ITEMS = 500000;
LockFreeSPSCQueue<uint64_t, 65536> queue;
std::vector<uint64_t> received;
received.reserve(TOTAL_ITEMS);
std::thread producer([&]() {
for (uint64_t i = 1; i <= TOTAL_ITEMS; ++i) {
while (!queue.push(i)) {
ThreadUtils::cpu_pause();
}
}
});
std::thread consumer([&]() {
for (size_t i = 0; i < TOTAL_ITEMS; ++i) {
uint64_t item;
while (!queue.pop(item)) {
ThreadUtils::cpu_pause();
}
received.push_back(item);
}
});
producer.join();
consumer.join();
ASSERT_EQ(received.size(), TOTAL_ITEMS);
for (size_t i = 0; i < TOTAL_ITEMS; ++i) {
ASSERT_EQ(received[i], i + 1);
}
}
TEST(LockFreeQueuesTest, MPSCConcurrentMultiProducer) {
constexpr size_t PRODUCERS = 4;
constexpr size_t PER_PRODUCER = 100000;
constexpr size_t TOTAL_ITEMS = PRODUCERS * PER_PRODUCER;
LockFreeMPSCQueue<uint64_t, 65536> queue;
std::atomic<bool> start_flag{false};
std::vector<std::thread> producers;
for (size_t p = 0; p < PRODUCERS; ++p) {
producers.emplace_back([&, p]() {
while (!start_flag.load(std::memory_order_acquire)) {}
for (size_t i = 0; i < PER_PRODUCER; ++i) {
uint64_t val = (p << 32) | i;
while (!queue.push(val)) {
ThreadUtils::cpu_pause();
}
}
});
}
std::vector<uint64_t> received;
received.reserve(TOTAL_ITEMS);
std::thread consumer([&]() {
while (!start_flag.load(std::memory_order_acquire)) {}
while (received.size() < TOTAL_ITEMS) {
uint64_t item;
if (queue.pop(item)) {
received.push_back(item);
} else {
ThreadUtils::cpu_pause();
}
}
});
start_flag.store(true, std::memory_order_release);
for (auto& th : producers) th.join();
consumer.join();
EXPECT_EQ(received.size(), TOTAL_ITEMS);
}
|