You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
283 lines
6.7 KiB
283 lines
6.7 KiB
#include <thread>
|
|
#include "catch2/catch_all.hpp"
|
|
#include "../LibSnugLog/include/disruptor.h"
|
|
|
|
struct strategy {
|
|
static constexpr overflow_response_t on_overflow = overflow_response_t::must_wait;
|
|
void wait() {}
|
|
};
|
|
|
|
struct strategy_that_overflows {
|
|
static constexpr overflow_response_t on_overflow = overflow_response_t::must_overflow;
|
|
void wait() {}
|
|
};
|
|
|
|
TEST_CASE("Disruptor works sequentially") {
|
|
std::array<char, 8192> buffer{};
|
|
disruptor<strategy> v{buffer.data(), buffer.size()};
|
|
|
|
SECTION("117") {
|
|
auto W = v.reserve_write(100);
|
|
v[W.start] = 117;
|
|
v.conclude_write(W);
|
|
auto R = v.reserve_read();
|
|
REQUIRE(v[R.start]== 117);
|
|
v.conclude_read(R);
|
|
}
|
|
|
|
SECTION("12") {
|
|
{
|
|
auto W = v.reserve_write(6);
|
|
v[W.start] = 12;
|
|
v.conclude_write(W);
|
|
auto R = v.reserve_read();
|
|
REQUIRE(v[R.start]== 12);
|
|
v.conclude_read(R);
|
|
}
|
|
|
|
{
|
|
auto W = v.reserve_write(6);
|
|
v[W.start] = 8;
|
|
v.conclude_write(W);
|
|
auto R = v.reserve_read();
|
|
REQUIRE(v[R.start]== 8);
|
|
v.conclude_read(R);
|
|
}
|
|
}
|
|
|
|
SECTION("Disruptor loop around") {
|
|
std::multiset<char> mset;
|
|
for(int i = 0; i != 255; i++) {
|
|
auto W = v.reserve_write(100);
|
|
v[W.start] = (char)i;
|
|
for(size_t idx = W.start; idx != W.end; idx = (idx+1)%v.size()) {
|
|
v[idx] = (char)i;
|
|
}
|
|
v.conclude_write(W);
|
|
auto R = v.reserve_read();
|
|
for(size_t idx = R.start; idx != R.end; idx = (idx+1)%v.size()) {
|
|
mset.insert(v[idx]);
|
|
}
|
|
v.conclude_read(R);
|
|
}
|
|
for(int i = 0; i != 255; i++) {
|
|
REQUIRE(mset.count((char)i) == 100);
|
|
}
|
|
}
|
|
|
|
SECTION("Disruptor concurrent odd vs even") {
|
|
std::atomic<bool> trigger = false;
|
|
std::multiset<char> mset;
|
|
std::stringstream continuity;
|
|
|
|
int acc = 0;
|
|
for(int i = 0; i<= 255; i++) {
|
|
acc+=i;
|
|
}
|
|
|
|
std::thread reader([&](){
|
|
int cnt = 0;
|
|
while (cnt != acc) {
|
|
auto R = v.reserve_read();
|
|
for (size_t idx = R.start; idx != R.end; idx = (idx + 1) % v.size()) {
|
|
mset.insert(v[idx]);
|
|
continuity << (char)v[idx];
|
|
}
|
|
v.conclude_read(R);
|
|
cnt += (R.end > R.start) * (R.end - R.start)
|
|
+ (R.end < R.start) * (v.size() - R.start + R.end);
|
|
}
|
|
});
|
|
|
|
std::thread even([&]() {
|
|
while(!trigger.load());
|
|
for (int i = 2; i <= 255; i += 2) {
|
|
auto W = v.reserve_write(i);
|
|
v[W.start] = (char) i;
|
|
for (size_t idx = W.start; idx != W.end; idx = (idx + 1) % v.size()) {
|
|
v[idx] = (char) i;
|
|
}
|
|
v.conclude_write(W);
|
|
}
|
|
});
|
|
|
|
std::thread odd([&]() {
|
|
while(!trigger.load());
|
|
for (int i = 1; i <= 255; i += 2) {
|
|
auto W = v.reserve_write(i);
|
|
v[W.start] = (char) i;
|
|
for (size_t idx = W.start; idx != W.end; idx = (idx + 1) % v.size()) {
|
|
v[idx] = (char) i;
|
|
}
|
|
v.conclude_write(W);
|
|
}
|
|
});
|
|
|
|
// byte received count test
|
|
trigger.store(true);
|
|
reader.join(); even.join(); odd.join();
|
|
for(int i = 1; i <= 255; i++) {
|
|
REQUIRE(mset.count((char)i) == i);
|
|
}
|
|
|
|
|
|
// Continuity tests
|
|
int changes = 0;
|
|
auto str = continuity.str();
|
|
char current = *str.begin();
|
|
auto it = str.begin();
|
|
for(;it != str.end();) {
|
|
while(it != str.end() && *it == current) {++it;}
|
|
changes += 1;
|
|
current = *it;
|
|
}
|
|
REQUIRE(changes == 255);
|
|
}
|
|
}
|
|
|
|
TEST_CASE("Fails if buffer too small") {
|
|
REQUIRE_THROWS_AS(disruptor<strategy>(nullptr, page_size), disruptor_exception);
|
|
}
|
|
|
|
TEST_CASE("Fails if buffer size is 0") {
|
|
REQUIRE_THROWS_AS(disruptor<strategy>(nullptr, 0), disruptor_exception);
|
|
}
|
|
|
|
TEST_CASE("Fails if buffer too small") {
|
|
REQUIRE_THROWS_AS(disruptor<strategy_that_overflows>(nullptr, page_size), disruptor_exception);
|
|
}
|
|
|
|
TEST_CASE("Fails if buffer size is 0") {
|
|
REQUIRE_THROWS_AS(disruptor<strategy_that_overflows>(nullptr, 0), disruptor_exception);
|
|
}
|
|
|
|
TEST_CASE("Disruptor works on overflow mode if things are not too contentious", "[long][unstable]") {
|
|
std::array<char, 8192> buffer{};
|
|
disruptor<strategy_that_overflows> v{buffer.data(), buffer.size()};
|
|
|
|
SECTION("117") {
|
|
auto W = v.reserve_write(100);
|
|
v[W.start] = 117;
|
|
v.conclude_write(W);
|
|
auto R = v.reserve_read();
|
|
REQUIRE(v[R.start]== 117);
|
|
v.conclude_read(R);
|
|
}
|
|
|
|
SECTION("12") {
|
|
{
|
|
auto W = v.reserve_write(6);
|
|
v[W.start] = 12;
|
|
v.conclude_write(W);
|
|
auto R = v.reserve_read();
|
|
REQUIRE(v[R.start]== 12);
|
|
v.conclude_read(R);
|
|
}
|
|
|
|
{
|
|
auto W = v.reserve_write(6);
|
|
v[W.start] = 8;
|
|
v.conclude_write(W);
|
|
auto R = v.reserve_read();
|
|
REQUIRE(v[R.start]== 8);
|
|
v.conclude_read(R);
|
|
}
|
|
}
|
|
|
|
SECTION("Disruptor loop around") {
|
|
std::multiset<char> mset;
|
|
for(int i = 0; i != 255; i++) {
|
|
auto W = v.reserve_write(100);
|
|
v[W.start] = (char)i;
|
|
for(size_t idx = W.start; idx != W.end; idx = (idx+1)%v.size()) {
|
|
v[idx] = (char)i;
|
|
}
|
|
v.conclude_write(W);
|
|
auto R = v.reserve_read();
|
|
for(size_t idx = R.start; idx != R.end; idx = (idx+1)%v.size()) {
|
|
mset.insert(v[idx]);
|
|
}
|
|
v.conclude_read(R);
|
|
}
|
|
for(int i = 0; i != 255; i++) {
|
|
REQUIRE(mset.count((char)i) == 100);
|
|
}
|
|
}
|
|
|
|
SECTION("Disruptor concurrent odd vs even") {
|
|
std::atomic<bool> trigger = false;
|
|
std::multiset<char> mset;
|
|
std::stringstream continuity;
|
|
|
|
int acc = 0;
|
|
for(int i = 0; i<= 255; i++) {
|
|
acc+=i;
|
|
}
|
|
|
|
using namespace std::chrono_literals;
|
|
|
|
std::thread reader([&](){
|
|
int cnt = 0;
|
|
auto start = std::chrono::high_resolution_clock::now();
|
|
while (std::chrono::high_resolution_clock::now() - start < 150ms) {
|
|
auto R = v.reserve_read();
|
|
for (size_t idx = R.start; idx != R.end; idx = (idx + 1) % v.size()) {
|
|
mset.insert(v[idx]);
|
|
continuity << (char)v[idx];
|
|
}
|
|
v.conclude_read(R);
|
|
cnt += (R.end > R.start) * (R.end - R.start)
|
|
+ (R.end < R.start) * (v.size() - R.start + R.end);
|
|
}
|
|
});
|
|
|
|
std::thread even([&]() {
|
|
while(!trigger.load());
|
|
for (int i = 2; i <= 255; i += 2) {
|
|
auto W = v.reserve_write(i);
|
|
v[W.start] = (char) i;
|
|
for (size_t idx = W.start; idx != W.end; idx = (idx + 1) % v.size()) {
|
|
v[idx] = (char) i;
|
|
}
|
|
v.conclude_write(W);
|
|
std::this_thread::sleep_for(50us);
|
|
}
|
|
});
|
|
|
|
std::thread odd([&]() {
|
|
while(!trigger.load());
|
|
for (int i = 1; i <= 255; i += 2) {
|
|
auto W = v.reserve_write(i);
|
|
v[W.start] = (char) i;
|
|
for (size_t idx = W.start; idx != W.end; idx = (idx + 1) % v.size()) {
|
|
v[idx] = (char) i;
|
|
}
|
|
v.conclude_write(W);
|
|
std::this_thread::sleep_for(50us);
|
|
}
|
|
});
|
|
|
|
// byte received count test
|
|
trigger.store(true);
|
|
reader.join();
|
|
even.join();
|
|
odd.join();
|
|
uint16_t cnt = 0;
|
|
for(int i = 1; i <= 255; i++) {
|
|
cnt += (mset.count((char)i) == i);
|
|
}
|
|
REQUIRE(cnt >= 100);
|
|
|
|
// Continuity tests
|
|
int changes = 0;
|
|
auto str = continuity.str();
|
|
char current = *str.begin();
|
|
auto it = str.begin();
|
|
for(;it != str.end();) {
|
|
while(it != str.end() && *it == current) {++it;}
|
|
changes += 1;
|
|
current = *it;
|
|
}
|
|
REQUIRE(changes >= 100);
|
|
}
|
|
}
|