CoopAllTheThings/tests/log_ring_test.cpp

// Unit test for the lossy MPSC log ring (common/include/coop/log_ring.hpp). Two parts:
//   1. Deterministic single-threaded wrap: pushing more than `capacity` lines drops the oldest and
//      drains the rest in order.
//   2. Threaded torn-read guard: many producer threads hammer a small ring while one consumer drains
//      a small ring (so slots wrap constantly). Each line's text is a unique token repeated across the
//      whole buffer; a torn read (a slot overwritten mid-copy) would mix two tokens, which the consumer
//      detects. The ring is lossy by design, so we assert lines are never TORN -- not that none are
//      lost. Without the seqlock copy+recheck (and the producer's in-progress marker), this trips.
#include <atomic>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <string>
#include <thread>
#include <vector>

#include "coop/log_ring.hpp"

using namespace coop;

namespace
{
int g_failures = 0;
void check(bool ok, const char* what)
{
	std::printf("%s %s\n", ok ? "  ok:" : "FAIL:", what);
	if (!ok)
	{
		++g_failures;
	}
}

// A line whose whole text is one token ("T<thread>S<seq>") repeated, space-separated. A torn read
// (two generations mixed in one slot) yields tokens that aren't all equal.
std::string make_line(unsigned thread, unsigned long long seq)
{
	char token[32];
	std::snprintf(token, sizeof(token), "T%02uS%010llu", thread, seq);
	std::string s;
	while (s.size() + std::strlen(token) + 1 < kLogMsgLen - 1)
	{
		s += token;
		s += ' ';
	}
	return s;
}

// True if every space-separated token in `text` is identical (i.e. the line wasn't torn).
bool line_consistent(const char* text)
{
	std::string first;
	std::string cur;
	for (const char* p = text;; ++p)
	{
		if (*p == ' ' || *p == '\0')
		{
			if (!cur.empty())
			{
				if (first.empty())
				{
					first = cur;
				}
				else if (cur != first)
				{
					return false;
				}
				cur.clear();
			}
			if (*p == '\0')
			{
				break;
			}
		}
		else
		{
			cur.push_back(*p);
		}
	}
	return !first.empty();
}

std::vector<std::uint8_t> make_ring(std::uint32_t capacity)
{
	std::vector<std::uint8_t> buf(log_ring_total_size(capacity), 0);
	log_ring_init(*reinterpret_cast<LogRing*>(buf.data()), capacity);
	return buf;
}
} // namespace

int main()
{
	// 1. Deterministic wrap: capacity 8, push 11 -> the oldest 3 drop, drain 3..10 in order.
	{
		auto buf = make_ring(8);
		auto& ring = *reinterpret_cast<LogRing*>(buf.data());
		for (unsigned long long n = 0; n < 11; ++n)
		{
			const std::string line = make_line(0, n);
			log_ring_push(ring, 1234, LogLevel_Info, n, line.c_str());
		}
		std::vector<std::string> got;
		std::uint64_t cursor = 0;
		log_ring_drain(ring, cursor, [&](const LogRecord& rec) { got.push_back(rec.text); });
		check(got.size() == 8, "wrap: drains exactly capacity lines after overflow");
		bool ordered = true;
		for (std::size_t k = 0; k < got.size(); ++k)
		{
			ordered = ordered && got[k] == make_line(0, 3 + k); // oldest 3 (0,1,2) dropped
		}
		check(ordered, "wrap: keeps the newest `capacity` lines, in order, oldest dropped");
	}

	// 2. Threaded torn-read guard: small ring, several producers, one slow-ish consumer.
	{
		constexpr std::uint32_t kCap = 32;
		auto buf = make_ring(kCap);
		auto& ring = *reinterpret_cast<LogRing*>(buf.data());

		std::atomic<bool> stop{false};
		std::atomic<long long> produced{0};
		std::vector<std::thread> producers;
		for (unsigned t = 0; t < 4; ++t)
		{
			producers.emplace_back([&, t] {
				unsigned long long n = 0;
				while (!stop.load(std::memory_order_relaxed))
				{
					const std::string line = make_line(t, n++);
					log_ring_push(ring, t, LogLevel_Info, n, line.c_str());
					produced.fetch_add(1, std::memory_order_relaxed);
				}
			});
		}

		std::atomic<long long> consumed{0};
		std::atomic<long long> torn{0};
		std::thread consumer([&] {
			std::uint64_t cursor = 0;
			auto drain = [&] {
				log_ring_drain(ring, cursor, [&](const LogRecord& rec) {
					consumed.fetch_add(1, std::memory_order_relaxed);
					if (!line_consistent(rec.text))
					{
						torn.fetch_add(1, std::memory_order_relaxed);
					}
				});
			};
			while (!stop.load(std::memory_order_relaxed))
			{
				drain();
				std::this_thread::sleep_for(std::chrono::microseconds(50)); // fall behind so slots wrap
			}
			drain();
		});

		std::this_thread::sleep_for(std::chrono::milliseconds(1500));
		stop.store(true, std::memory_order_relaxed);
		for (auto& p : producers)
		{
			p.join();
		}
		consumer.join();

		std::printf("   produced=%lld consumed=%lld torn=%lld\n", produced.load(), consumed.load(), torn.load());
		check(produced.load() > 100000, "threaded: producers ran a real workload");
		check(consumed.load() > 0, "threaded: consumer received lines (ring wrapped, most are lost -- ok)");
		check(torn.load() == 0, "threaded: no torn line ever emitted (seqlock copy+recheck holds)");
	}

	std::printf(g_failures == 0 ? "PASS log_ring_test\n" : "FAILED log_ring_test (%d)\n", g_failures);
	return g_failures == 0 ? 0 : 1;
}