lib/Buffer.hpp

@@ -6,7 +6,6 @@
 #include <vector>
 #include <algorithm>
 #include <cassert>
-#include <cstring>
 
 namespace reactor {
 
@@ -106,13 +105,13 @@ public:
 
 	void appendInt16(uint16_t x)
 	{
-		uint16_t be = hton16(x);
+		uint16_t be = htons(x);
 		append(reinterpret_cast<const char*>(&be), sizeof(be));
 	}
 
 	void appendInt32(uint32_t x)
 	{
-		uint32_t be = hton32(x);
+		uint32_t be = htonl(x);
 		append(reinterpret_cast<const char*>(&be), sizeof(be));
 	}
 
@@ -162,14 +161,14 @@ public:
 	{
 		assert(readableBytes() >= sizeof(uint16_t));
 		uint16_t be = *reinterpret_cast<const uint16_t*>(peek());
-		return ntoh16(be);
+		return ntohs(be);
 	}
 
 	uint32_t peekInt32() const
 	{
 		assert(readableBytes() >= sizeof(uint32_t));
 		uint32_t be = *reinterpret_cast<const uint32_t*>(peek());
-		return ntoh32(be);
+		return ntohl(be);
 	}
 
 	uint64_t peekInt64() const
@@ -194,13 +193,13 @@ public:
 
 	void prependInt16(uint16_t x)
 	{
-		uint16_t be = hton16(x);
+		uint16_t be = htons(x);
 		prepend(reinterpret_cast<const char*>(&be), sizeof(be));
 	}
 
 	void prependInt32(uint32_t x)
 	{
-		uint32_t be = hton32(x);
+		uint32_t be = htonl(x);
 		prepend(reinterpret_cast<const char*>(&be), sizeof(be));
 	}
 

lib/Utilities.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <future>
 #include <string>
 #include <memory>
 #include <functional>
@@ -8,92 +9,19 @@
 #include <mutex>
 #include <thread>
 #include <condition_variable>
-#include <queue>
-#include <stop_token>
 #include <iostream>
 #include <fstream>
 #include <chrono>
 #include <type_traits>
 #include <algorithm>
+#include <cstring>
+#include <netinet/in.h>
 #include <sstream>
-#include <ranges>
-#include <bit>
-#include <syncstream>
-#include <future>
+#include <queue>
 
-namespace reactor
-{
+namespace reactor {
 
-#if defined(__GNUC__) || defined(__clang__)
-	#define bswap_16(x) __builtin_bswap16(x)
-	#define bswap_32(x) __builtin_bswap32(x)
-	#define bswap_64(x) __builtin_bswap64(x)
-#else
-	// Generic fallback implementations
-	inline uint16_t bswap_16(uint16_t val)
-	{
-		return (val << 8) | (val >> 8);
-	}
-
-	inline uint32_t bswap_32(uint32_t val)
-	{
-		val = ((val << 8) & 0xFF00FF00) | ((val >> 8) & 0xFF00FF);
-		return (val << 16) | (val >> 16);
-	}
-
-	inline uint64_t bswap_64(uint64_t val)
-	{
-		uint64_t temp = val;
-		char* ptr = reinterpret_cast<char*>(&temp);
-		std::reverse(ptr, ptr + sizeof(uint64_t));
-		return temp;
-	}
-#endif
-
-// C++20 Network byte order utilities
-inline uint16_t hton16(uint16_t host_uint16)
-{
-	if constexpr (std::endian::native == std::endian::little) {
-		return bswap_16(host_uint16);
-	} else {
-		return host_uint16;
-	}
-}
-
-inline uint32_t hton32(uint32_t host_uint32)
-{
-	if constexpr (std::endian::native == std::endian::little) {
-		return bswap_32(host_uint32);
-	} else {
-		return host_uint32;
-	}
-}
-
-inline uint64_t hton64(uint64_t host_uint64)
-{
-	if constexpr (std::endian::native == std::endian::little) {
-		return bswap_64(host_uint64);
-	} else {
-		return host_uint64;
-	}
-}
-
-inline uint16_t ntoh16(uint16_t net_uint16)
-{
-	return hton16(net_uint16);
-}
-
-inline uint32_t ntoh32(uint32_t net_uint32)
-{
-	return hton32(net_uint32);
-}
-
-inline uint64_t ntoh64(uint64_t net_uint64)
-{
-	return hton64(net_uint64);
-}
-
-// NonCopyable base class (unchanged, follows modern practice)
+// NonCopyable base class
 class NonCopyable
 {
 protected:
@@ -105,38 +33,197 @@ protected:
 	NonCopyable& operator=(NonCopyable&&) noexcept = default;
 };
 
-// Object Pool (unchanged, this is a standard pattern)
+// Network byte order utilities
+inline uint64_t hton64(uint64_t n)
+{
+	static const int one = 1;
+	static const char sig = *(char*)&one;
+	if (sig == 0) return n;
+	char* ptr = reinterpret_cast<char*>(&n);
+	std::reverse(ptr, ptr + sizeof(uint64_t));
+	return n;
+}
+
+inline uint64_t ntoh64(uint64_t n) { return hton64(n); }
+
+// Lock-free MPSC queue using type erasure
+class LockFreeQueue : public NonCopyable
+{
+private:
+	struct Node
+	{
+		Node() = default;
+
+		template<typename T>
+		Node(T&& data) : data_(std::make_unique<ConcreteTask<T>>(std::forward<T>(data))) {}
+
+		struct Task {
+			virtual ~Task() = default;
+			virtual void call() = 0;
+		};
+
+		template<typename F>
+		struct ConcreteTask : Task {
+			F func_;
+			ConcreteTask(F&& f) : func_(std::forward<F>(f)) {}
+			void call() override { func_(); }
+		};
+
+		std::unique_ptr<Task> data_;
+		std::atomic<Node*> next_{nullptr};
+	};
+
+	std::atomic<Node*> head_;
+	std::atomic<Node*> tail_;
+
+public:
+	LockFreeQueue() : head_(new Node), tail_(head_.load()) {}
+
+	~LockFreeQueue()
+	{
+		while (dequeue()) {}
+		delete head_.load();
+	}
+
+	template<typename F>
+	void enqueue(F&& input)
+	{
+		Node* node = new Node(std::forward<F>(input));
+		Node* prevhead = head_.exchange(node, std::memory_order_acq_rel);
+		prevhead->next_.store(node, std::memory_order_release);
+	}
+
+	bool dequeue()
+	{
+		Node* tail = tail_.load(std::memory_order_relaxed);
+		Node* next = tail->next_.load(std::memory_order_acquire);
+
+		if (next == nullptr) return false;
+
+		if (next->data_) {
+			next->data_->call();
+		}
+		tail_.store(next, std::memory_order_release);
+		delete tail;
+		return true;
+	}
+
+	bool empty()
+	{
+		Node* tail = tail_.load(std::memory_order_relaxed);
+		Node* next = tail->next_.load(std::memory_order_acquire);
+		return next == nullptr;
+	}
+};
+
+// Type-safe lock-free queue for other use cases
+template<typename T>
+class LockFreeQueueTyped : public NonCopyable
+{
+private:
+	struct Node
+	{
+		std::atomic<T*> data_{nullptr};
+		std::atomic<Node*> next_{nullptr};
+	};
+
+	std::atomic<Node*> head_;
+	std::atomic<Node*> tail_;
+
+public:
+	LockFreeQueueTyped()
+	{
+		Node* dummy = new Node;
+		head_.store(dummy);
+		tail_.store(dummy);
+	}
+
+	~LockFreeQueueTyped()
+	{
+		T output;
+		while (dequeue(output)) {}
+		delete head_.load();
+	}
+
+	void enqueue(T&& input)
+	{
+		Node* newNode = new Node;
+		T* data = new T(std::move(input));
+		newNode->data_.store(data);
+
+		Node* prevHead = head_.exchange(newNode);
+		prevHead->next_.store(newNode);
+	}
+
+	void enqueue(const T& input)
+	{
+		Node* newNode = new Node;
+		T* data = new T(input);
+		newNode->data_.store(data);
+
+		Node* prevHead = head_.exchange(newNode);
+		prevHead->next_.store(newNode);
+	}
+
+	bool dequeue(T& output)
+	{
+		Node* tail = tail_.load();
+		Node* next = tail->next_.load();
+
+		if (next == nullptr) {
+			return false;
+		}
+
+		T* data = next->data_.load();
+		if (data == nullptr) {
+			return false;
+		}
+
+		output = *data;
+		delete data;
+		tail_.store(next);
+		delete tail;
+		return true;
+	}
+
+	bool empty()
+	{
+		Node* tail = tail_.load();
+		Node* next = tail->next_.load();
+		return next == nullptr;
+	}
+};
+
+// Object Pool
 template<typename T>
 class ObjectPool : public NonCopyable, public std::enable_shared_from_this<ObjectPool<T>>
 {
 private:
-	std::vector<std::unique_ptr<T>> objects_;
+	std::vector<T*> objects_;
 	std::mutex mutex_;
 
 public:
 	std::shared_ptr<T> getObject()
 	{
-		static_assert(!std::is_pointer_v<T>, "ObjectPool type cannot be a pointer");
+		static_assert(!std::is_pointer_v<T>, "ObjectPool type cannot be pointer");
 
-		std::unique_ptr<T> p = nullptr;
+		T* p = nullptr;
 		{
 			std::lock_guard<std::mutex> lock(mutex_);
 			if (!objects_.empty()) {
-				p = std::move(objects_.back());
+				p = objects_.back();
 				objects_.pop_back();
 			}
 		}
 
-		if (!p) {
-			p = std::make_unique<T>();
-		}
+		if (!p) p = new T;
 
 		std::weak_ptr<ObjectPool<T>> weakPtr = this->shared_from_this();
-		return std::shared_ptr<T>(p.release(), [weakPtr](T* ptr) {
+		return std::shared_ptr<T>(p, [weakPtr](T* ptr) {
 			auto self = weakPtr.lock();
 			if (self) {
 				std::lock_guard<std::mutex> lock(self->mutex_);
-				self->objects_.push_back(std::unique_ptr<T>(ptr));
+				self->objects_.push_back(ptr);
 			} else {
 				delete ptr;
 			}
@@ -144,6 +231,7 @@ public:
 	}
 };
 
+// Simple Logger
 enum class LogLevel { TRACE, DEBUG, INFO, WARN, ERROR, FATAL };
 
 class Logger : public NonCopyable
@@ -151,6 +239,7 @@ class Logger : public NonCopyable
 private:
 	static inline LogLevel level_ = LogLevel::INFO;
 	static inline std::unique_ptr<std::ofstream> file_;
+	static inline std::mutex mutex_;
 
 	std::ostringstream stream_;
 	LogLevel msgLevel_;
@@ -192,11 +281,12 @@ public:
 	{
 		if (msgLevel_ >= level_) {
 			stream_ << "\n";
-			// std::osyncstream handles synchronized, atomic writes
+			std::lock_guard<std::mutex> lock(mutex_);
 			if (file_ && file_->is_open()) {
-				std::osyncstream(*file_) << stream_.str();
+				*file_ << stream_.str();
+				file_->flush();
 			} else {
-				std::osyncstream(std::cout) << stream_.str();
+				std::cout << stream_.str();
 			}
 		}
 	}
@@ -213,30 +303,39 @@ public:
 	static void setLevel(LogLevel level) { level_ = level; }
 	static void setLogFile(const std::string& filename)
 	{
+		std::lock_guard<std::mutex> lock(mutex_);
 		file_ = std::make_unique<std::ofstream>(filename, std::ios::app);
 	}
 };
 
-// C++20 Concurrent Task Queue using jthread and stop_token
 class ConcurrentTaskQueue : public NonCopyable
 {
 private:
-	std::vector<std::jthread> threads_;
+	std::vector<std::thread> threads_;
 	std::queue<std::function<void()>> taskQueue_;
 	std::mutex mutex_;
-	std::condition_variable_any taskCond_;
-	std::stop_source stopSource_;
+	std::condition_variable taskCond_;
+	std::atomic<bool> stop_;
 	std::string name_;
 
-	void workerThread(std::stop_token token)
+	/*
+	 * Worker thread function.
+	 * Waits for tasks and executes them.
+	 */
+	void workerThread()
 	{
-		while (!token.stop_requested()) {
+		while (true)
+		{
 			std::function<void()> task;
 			{
 				std::unique_lock<std::mutex> lock(mutex_);
-				taskCond_.wait(lock, token, [this] { return !taskQueue_.empty(); });
+				taskCond_.wait(lock, [this]
+				{
+					return stop_.load() || !taskQueue_.empty();
+				});
 
-				if (token.stop_requested() && taskQueue_.empty()) {
+				if (stop_.load() && taskQueue_.empty())
+				{
 					return;
 				}
 
@@ -249,26 +348,36 @@ private:
 
 public:
 	ConcurrentTaskQueue(size_t threadNum, const std::string& name = "ConcurrentTaskQueue")
-		: name_(name)
+		: stop_(false), name_(name)
 	{
-		for (size_t i = 0; i < threadNum; ++i) {
-			threads_.emplace_back(&ConcurrentTaskQueue::workerThread, this, stopSource_.get_token());
+		for (size_t i = 0; i < threadNum; ++i)
+		{
+			threads_.emplace_back(&ConcurrentTaskQueue::workerThread, this);
 		}
 	}
 
 	~ConcurrentTaskQueue()
 	{
-		stopSource_.request_stop();
+		{
+			std::lock_guard<std::mutex> lock(mutex_);
+			stop_ = true;
+		}
 		taskCond_.notify_all();
-		// std::jthread destructors automatically join
+		for (auto& t : threads_)
+		{
+			if (t.joinable()) t.join();
+		}
 	}
 
+	/*
+	 * Add a task to the queue.
+	 */
 	template<typename F>
 	void runTaskInQueue(F&& task)
 	{
 		{
 			std::lock_guard<std::mutex> lock(mutex_);
-			if (stopSource_.stop_requested()) return;
+			if (stop_) return;
 			taskQueue_.emplace(std::forward<F>(task));
 		}
 		taskCond_.notify_one();
@@ -296,7 +405,7 @@ public:
 	}
 };
 
-// Logging macros (unchanged)
+// Logging macros
 #define LOG_TRACE reactor::Logger(reactor::LogLevel::TRACE)
 #define LOG_DEBUG reactor::Logger(reactor::LogLevel::DEBUG)
 #define LOG_INFO  reactor::Logger(reactor::LogLevel::INFO)
@@ -312,13 +421,18 @@ void hashCombine(std::size_t& seed, const T& value)
 	seed ^= hasher(value) + 0x9e3779b9 + (seed << 6) + (seed >> 2);
 }
 
-// C++20 string splitting using ranges
 inline std::vector<std::string> splitString(const std::string& s, const std::string& delimiter)
 {
 	std::vector<std::string> result;
-	for (const auto& range : std::views::split(s, delimiter)) {
-		result.emplace_back(range.begin(), range.end());
+	size_t start = 0;
+	size_t end = s.find(delimiter);
+
+	while (end != std::string::npos) {
+		result.push_back(s.substr(start, end - start));
+		start = end + delimiter.length();
+		end = s.find(delimiter, start);
 	}
+	result.push_back(s.substr(start));
 	return result;
 }
 

tests/test_utilities.cpp

@@ -5,8 +5,87 @@
 #include <chrono>
 #include <vector>
 #include <future>
-#include <latch> // For std::latch
-#include <atomic>
+
+void test_lock_free_queue()
+{
+	std::cout << "Testing lock-free queue...\n";
+
+	reactor::LockFreeQueueTyped<int> queue;
+	assert(queue.empty());
+
+	queue.enqueue(1);
+	queue.enqueue(2);
+	queue.enqueue(3);
+	assert(!queue.empty());
+
+	int val;
+	assert(queue.dequeue(val) && val == 1);
+	assert(queue.dequeue(val) && val == 2);
+	assert(queue.dequeue(val) && val == 3);
+	assert(queue.empty());
+	assert(!queue.dequeue(val));
+
+	std::cout << "✓ Lock-free queue basic operations passed\n";
+}
+
+void test_lock_free_queue_concurrent()
+{
+	std::cout << "Testing lock-free queue concurrency...\n";
+
+	reactor::LockFreeQueueTyped<int> queue;
+	constexpr int num_items = 50;  // Reduced further
+	constexpr int num_producers = 1;  // Simplified to 1 producer
+	constexpr int num_consumers = 1;  // Simplified to 1 consumer
+
+	std::vector<std::thread> producers;
+	std::vector<std::thread> consumers;
+	std::atomic<int> consumed_count{0};
+	std::atomic<bool> stop_consumers{false};
+
+	for (int p = 0; p < num_producers; ++p) {
+		producers.emplace_back([&queue, p, num_items]() {
+			for (int i = 0; i < num_items; ++i) {
+				queue.enqueue(p * num_items + i);
+				std::this_thread::sleep_for(std::chrono::milliseconds(1));
+			}
+		});
+	}
+
+	for (int c = 0; c < num_consumers; ++c) {
+		consumers.emplace_back([&queue, &consumed_count, &stop_consumers]() {
+			int val;
+			while (!stop_consumers) {
+				if (queue.dequeue(val)) {
+					consumed_count++;
+				} else {
+					std::this_thread::sleep_for(std::chrono::milliseconds(1));
+				}
+			}
+		});
+	}
+
+	for (auto& p : producers) {
+		p.join();
+	}
+
+	// Wait longer for all items to be consumed
+	auto start = std::chrono::steady_clock::now();
+	while (consumed_count < num_producers * num_items) {
+		std::this_thread::sleep_for(std::chrono::milliseconds(10));
+		auto now = std::chrono::steady_clock::now();
+		if (std::chrono::duration_cast<std::chrono::seconds>(now - start).count() > 5) {
+			break; // Timeout after 5 seconds
+		}
+	}
+
+	stop_consumers = true;
+	for (auto& c : consumers) {
+		c.join();
+	}
+
+	assert(consumed_count == num_producers * num_items);
+	std::cout << "✓ Lock-free queue concurrency passed\n";
+}
 
 void test_object_pool()
 {
@@ -15,6 +94,8 @@ void test_object_pool()
 	struct TestObject
 	{
 		int value = 42;
+		TestObject() { std::cout << "TestObject constructed\n"; }
+		~TestObject() { std::cout << "TestObject destructed\n"; }
 	};
 
 	auto pool = std::make_shared<reactor::ObjectPool<TestObject>>();
@@ -60,26 +141,24 @@ void test_concurrent_task_queue()
 {
 	std::cout << "Testing concurrent task queue...\n";
 
-	reactor::ConcurrentTaskQueue queue(4, "TestQueue");
+	reactor::ConcurrentTaskQueue queue(2, "TestQueue");
 	assert(queue.getName() == "TestQueue");
 
-	constexpr int num_tasks = 50;
 	std::atomic<int> counter{0};
-	std::latch all_tasks_done(num_tasks); // C++20 latch for synchronization
+	std::promise<void> all_done;
+	auto future = all_done.get_future();
 
-	for (int i = 0; i < num_tasks; ++i)
-	{
-		queue.runTaskInQueue([&counter, &all_tasks_done]()
-		{
-			counter++;
-			// Signal that one task is complete
-			all_tasks_done.count_down();
+	for (int i = 0; i < 10; ++i) {
+		queue.runTaskInQueue([&counter, i, &all_done]() {
+			std::this_thread::sleep_for(std::chrono::milliseconds(10));
+			if (++counter == 10) {
+				all_done.set_value();
+			}
 		});
 	}
 
-	// Wait for all tasks to complete
-	all_tasks_done.wait();
-	assert(counter.load() == num_tasks);
+	future.wait();
+	assert(counter == 10);
 
 	std::cout << "✓ Concurrent task queue passed\n";
 }
@@ -89,15 +168,15 @@ void test_sync_task()
 	std::cout << "Testing sync task execution...\n";
 
 	reactor::ConcurrentTaskQueue queue(1, "SyncTest");
-	bool task_executed = false;
 
-	queue.syncTaskInQueue([&task_executed]()
-	{
+	// Use shared_ptr to avoid move issues
+	auto executed_flag = std::make_shared<bool>(false);
+	queue.syncTaskInQueue([executed_flag]() {
 		std::this_thread::sleep_for(std::chrono::milliseconds(50));
-		task_executed = true;
+		*executed_flag = true;
 	});
 
-	assert(task_executed);
+	assert(*executed_flag);
 	std::cout << "✓ Sync task execution passed\n";
 }
 
@@ -160,8 +239,10 @@ void test_non_copyable()
 
 int main()
 {
-	std::cout << "=== Utilities Tests (C++20 Version) ===\n";
+	std::cout << "=== Utilities Tests ===\n";
 
+	test_lock_free_queue();
+	test_lock_free_queue_concurrent();
 	test_object_pool();
 	test_logger();
 	test_concurrent_task_queue();