Reactor/tests/test_threading.cpp

#include "../lib/EventLoopThread.hpp"
#include <cassert>
#include <iostream>
#include <map>
#include <thread>
#include <chrono>
#include <atomic>
#include <vector>
#include <future>

void test_event_loop_thread_basic()
{
	std::cout << "Testing basic EventLoopThread...\n";

	reactor::EventLoopThread loop_thread("TestThread");
	auto loop = loop_thread.getLoop();

	assert(loop != nullptr);
	assert(loop_thread.name() == "TestThread");

	std::atomic<bool> task_executed{false};
	loop->runInLoop([&task_executed]() {
		task_executed = true;
	});

	std::this_thread::sleep_for(std::chrono::milliseconds(50));
	assert(task_executed);

	std::cout << "✓ Basic EventLoopThread passed\n";
}

void test_event_loop_thread_timer()
{
	std::cout << "Testing EventLoopThread timer...\n";

	reactor::EventLoopThread loop_thread("TimerThread");
	auto loop = loop_thread.getLoop();

	std::atomic<int> timer_count{0};
	auto timer_id = loop->runEvery(reactor::Duration(20), [&timer_count]() {
		timer_count++;
	});

	std::this_thread::sleep_for(std::chrono::milliseconds(85));
	loop->cancel(timer_id);

	int final_count = timer_count;
	assert(final_count >= 3 && final_count <= 5);

	std::cout << "✓ EventLoopThread timer passed (count: " << final_count << ")\n";
}

void test_multiple_event_loop_threads()
{
	std::cout << "Testing multiple EventLoopThreads...\n";

	constexpr int num_threads = 3;
	std::vector<std::unique_ptr<reactor::EventLoopThread>> threads;
	std::vector<std::atomic<int>*> counters;

	for (int i = 0; i < num_threads; ++i) {
		std::string name = "Thread-" + std::to_string(i);
		threads.push_back(std::make_unique<reactor::EventLoopThread>(name));
		counters.push_back(new std::atomic<int>{0});
	}

	for (int i = 0; i < num_threads; ++i) {
		auto loop = threads[i]->getLoop();
		auto counter = counters[i];

		for (int j = 0; j < 10; ++j) {
			loop->queueInLoop([counter]() {
				(*counter)++;
			});
		}
	}

	std::this_thread::sleep_for(std::chrono::milliseconds(100));

	for (int i = 0; i < num_threads; ++i) {
		assert(*counters[i] == 10);
		delete counters[i];
	}

	std::cout << "✓ Multiple EventLoopThreads passed\n";
}

void test_event_loop_thread_pool_basic()
{
	std::cout << "Testing basic EventLoopThreadPool...\n";

	reactor::EventLoopThreadPool pool(3, "PoolThread");

	assert(pool.size() == 3);
	assert(pool.getBaseName() == "PoolThread");

	auto loops = pool.getAllLoops();
	assert(loops.size() == 3);

	for (auto loop : loops) {
		assert(loop != nullptr);
	}

	std::cout << "✓ Basic EventLoopThreadPool passed\n";
}

void test_thread_pool_round_robin()
{
	std::cout << "Testing thread pool round robin...\n";

	reactor::EventLoopThreadPool pool(3, "RoundRobin");

	std::vector<reactor::EventLoop*> selected_loops;
	for (int i = 0; i < 9; ++i) {
		selected_loops.push_back(pool.getNextLoop());
	}

	for (int i = 0; i < 3; ++i) {
		assert(selected_loops[i] == selected_loops[i + 3]);
		assert(selected_loops[i] == selected_loops[i + 6]);
	}

	std::cout << "✓ Thread pool round robin passed\n";
}

void test_thread_pool_task_distribution()
{
	std::cout << "Testing thread pool task distribution...\n";

	reactor::EventLoopThreadPool pool(3, "TaskDist");

	std::vector<std::atomic<int>*> counters(3);
	for (int i = 0; i < 3; ++i) {
		counters[i] = new std::atomic<int>{0};
	}

	std::map<reactor::EventLoop*, int> loop_to_index;
	auto loops = pool.getAllLoops();
	for (int i = 0; i < 3; ++i) {
		loop_to_index[loops[i]] = i;
	}

	constexpr int tasks_per_loop = 10;
	for (int i = 0; i < 3 * tasks_per_loop; ++i) {
		auto loop = pool.getNextLoop();
		int index = loop_to_index[loop];

		loop->queueInLoop([counter = counters[index]]() {
			(*counter)++;
		});
	}

	std::this_thread::sleep_for(std::chrono::milliseconds(100));

	for (int i = 0; i < 3; ++i) {
		assert(*counters[i] == tasks_per_loop);
		delete counters[i];
	}

	std::cout << "✓ Thread pool task distribution passed\n";
}

void test_empty_thread_pool()
{
	std::cout << "Testing empty thread pool...\n";

	reactor::EventLoopThreadPool pool(0, "EmptyPool");

	assert(pool.size() == 0);
	assert(pool.getNextLoop() == nullptr);
	assert(pool.getAllLoops().empty());

	std::cout << "✓ Empty thread pool passed\n";
}

void test_thread_pool_concurrent_access()
{
	std::cout << "Testing thread pool concurrent access...\n";

	reactor::EventLoopThreadPool pool(4, "ConcurrentAccess");

	std::atomic<int> total_tasks{0};
	std::vector<std::thread> client_threads;

	for (int t = 0; t < 8; ++t) {
		client_threads.emplace_back([&pool, &total_tasks]() {
			for (int i = 0; i < 25; ++i) {
				auto loop = pool.getNextLoop();
				assert(loop != nullptr);

				loop->queueInLoop([&total_tasks]() {
					total_tasks++;
				});
			}
		});
	}

	for (auto& thread : client_threads) {
		thread.join();
	}

	std::this_thread::sleep_for(std::chrono::milliseconds(200));
	assert(total_tasks == 8 * 25);

	std::cout << "✓ Thread pool concurrent access passed\n";
}

void test_thread_pool_with_timers()
{
	std::cout << "Testing thread pool with timers...\n";

	reactor::EventLoopThreadPool pool(2, "TimerPool");

	std::atomic<int> timer_count{0};
	std::vector<uint64_t> timer_ids;

	auto loops = pool.getAllLoops();
	for (auto loop : loops) {
		auto timer_id = loop->runEvery(reactor::Duration(30), [&timer_count]() {
			timer_count++;
		});
		timer_ids.push_back(timer_id);
	}

	std::this_thread::sleep_for(std::chrono::milliseconds(100));

	for (int i = 0; i < loops.size(); ++i) {
		loops[i]->cancel(timer_ids[i]);
	}

	int final_count = timer_count;
	assert(final_count >= 4 && final_count <= 8);

	std::cout << "✓ Thread pool with timers passed (count: " << final_count << ")\n";
}

void test_thread_synchronization()
{
	std::cout << "Testing thread synchronization...\n";

	reactor::EventLoopThread loop_thread("SyncThread");
	auto loop = loop_thread.getLoop();

	std::vector<int> results;
	std::mutex results_mutex;

	std::vector<std::future<void>> futures;

	for (int i = 0; i < 10; ++i) {
		std::promise<void> promise;
		auto future = promise.get_future();

		loop->queueInLoop([i, &results, &results_mutex, promise = std::move(promise)]() mutable {
			{
				std::lock_guard<std::mutex> lock(results_mutex);
				results.push_back(i);
			}
			promise.set_value();
		});

		futures.push_back(std::move(future));
	}

	for (auto& future : futures) {
		future.wait();
	}

	assert(results.size() == 10);

	std::cout << "✓ Thread synchronization passed\n";
}

void test_thread_pool_destruction()
{
	std::cout << "Testing thread pool destruction...\n";

	std::atomic<int> destructor_count{0};

	{
		reactor::EventLoopThreadPool pool(2, "DestroyTest");

		auto loops = pool.getAllLoops();
		for (auto loop : loops) {
			loop->queueInLoop([&destructor_count]() {
				destructor_count++;
			});
		}

		std::this_thread::sleep_for(std::chrono::milliseconds(50));
	}

	assert(destructor_count == 2);
	std::cout << "✓ Thread pool destruction passed\n";
}

int main()
{
	std::cout << "=== Threading Tests ===\n";

	test_event_loop_thread_basic();
	test_event_loop_thread_timer();
	test_multiple_event_loop_threads();
	test_event_loop_thread_pool_basic();
	test_thread_pool_round_robin();
	test_thread_pool_task_distribution();
	test_empty_thread_pool();
	test_thread_pool_concurrent_access();
	test_thread_pool_with_timers();
	test_thread_synchronization();
	test_thread_pool_destruction();

	std::cout << "All threading tests passed! ✓\n";
	return 0;
}