diff --git a/CMakeLists.txt b/CMakeLists.txt index 4a6ee81..ece3592 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,11 +1,11 @@ -cmake_minimum_required(VERSION 3.15) +cmake_minimum_required(VERSION 3.20) project(KV_Database LANGUAGES CXX) set(CMAKE_EXPORT_COMPILE_COMMANDS ON) -# C++17 Configuration -set(CMAKE_CXX_STANDARD 17) +# C++23 Configuration +set(CMAKE_CXX_STANDARD 23) set(CMAKE_CXX_STANDARD_REQUIRED ON) # Turn our core logic into a LIBRARY so both the App and the Tests can use it @@ -17,10 +17,7 @@ add_library(KV_Core src/Worker/SnapshotScheduler.cpp src/Limit/Limiter.cpp src/UserSession/UserSession.cpp - src/UserSession/UserSession.h - src/UserSession/Connection.h src/Worker/UserSessionBackgroundWorker.cpp - src/Worker/UserSessionBackgroundWorker.h src/Networking/EpollEventLoop.cpp src/Networking/IocpEventLoop.cpp ) diff --git a/README.md b/README.md index 87c0606..80e7619 100644 --- a/README.md +++ b/README.md @@ -6,8 +6,8 @@ _Redis-inspired · event-driven · cross-platform · containerized_ -[![C++](https://img.shields.io/badge/C%2B%2B-17-00599C?style=flat-square&logo=cplusplus&logoColor=white)](https://en.cppreference.com/w/cpp/17) -[![CMake](https://img.shields.io/badge/CMake-3.15%2B-064F8C?style=flat-square&logo=cmake&logoColor=white)](https://cmake.org/) +[![C++](https://img.shields.io/badge/C%2B%2B-23-00599C?style=flat-square&logo=cplusplus&logoColor=white)](https://en.cppreference.com/w/cpp/23) +[![CMake](https://img.shields.io/badge/CMake-3.20%2B-064F8C?style=flat-square&logo=cmake&logoColor=white)](https://cmake.org/) [![Docker](https://img.shields.io/badge/Docker-ready-2496ED?style=flat-square&logo=docker&logoColor=white)](https://www.docker.com/) [![Platform](https://img.shields.io/badge/platform-Linux%20%7C%20Windows-informational?style=flat-square&logo=linux&logoColor=white)](#) [![Tests](https://img.shields.io/badge/tests-GoogleTest-4285F4?style=flat-square&logo=google&logoColor=white)](https://github.com/google/googletest) @@ -148,8 +148,8 @@ ByteForge uses the **Token Bucket** algorithm for rate limiting — the same app ### Prerequisites -- C++17 compiler -- CMake 3.15+ +- C++23 compiler +- CMake 3.20+ - Docker (optional) - [nmap/ncat](https://nmap.org/download.html) for testing @@ -183,13 +183,13 @@ The server starts on port `6625` by default. Type `stop` to shut it down cleanly Build the image: ```bash -docker build -t byteforce:latest . +docker build -t byteforge:latest . ``` Run the container: ```bash -docker run -d -p 6625:6625 --name my-byteforce-db byteforce-db:latest +docker run -d -p 6625:6625 --name my-byteforge-db byteforge:latest ``` Connect using: diff --git a/src/Limit/Limiter.h b/src/Limit/Limiter.h index b98e1b4..8f1b51f 100644 --- a/src/Limit/Limiter.h +++ b/src/Limit/Limiter.h @@ -15,6 +15,28 @@ */ class RateLimiter { + +public: + /** + * @brief Constructs the rate limiter. + * @param maxTokens Max burst capacity per IP (default: 10). + * @param refillRate Tokens refilled per second per IP (default: 5). + * @param globalLimit Max total requests per second (default: 1000). + */ + RateLimiter(double maxTokens = 10, double refillRate = 5, int globalLimit = 1000); + + /** + * @brief Destroys the rate limiter and clears all state. + */ + ~RateLimiter(); + + /** + * @brief Checks if a request from the given IP should be allowed. + * @param ip The raw client IP address + * @return true if allowed, false if rate limited. + */ + bool isAllowed(uint32_t ip); + private: /** * @brief Token bucket state for a single IP address. @@ -26,7 +48,7 @@ class RateLimiter }; std::unordered_map ipRecords; /**< Per-IP bucket state. */ - std::mutex mapMutex; /**< Protects ipRecords from concurrent access. */ + std::mutex mapMutex; /**< Protects ipRecords from concurrent access. */ std::atomic globalCount; /**< Total requests in current global window. */ std::chrono::steady_clock::time_point globalWindowStart; /**< Start of current global window. */ @@ -49,27 +71,6 @@ class RateLimiter * allowed, false if the IP is rate limited. */ bool isAllowedPrivate(uint32_t ip); - -public: - /** - * @brief Constructs the rate limiter. - * @param maxTokens Max burst capacity per IP (default: 10). - * @param refillRate Tokens refilled per second per IP (default: 5). - * @param globalLimit Max total requests per second (default: 1000). - */ - RateLimiter(double maxTokens = 10, double refillRate = 5, int globalLimit = 1000); - - /** - * @brief Destroys the rate limiter and clears all state. - */ - ~RateLimiter(); - - /** - * @brief Checks if a request from the given IP should be allowed. - * @param ip The raw client IP address - * @return true if allowed, false if rate limited. - */ - bool isAllowed(uint32_t ip); }; #endif \ No newline at end of file diff --git a/src/RAM/HashMap.h b/src/RAM/HashMap.h index 53f4055..fc2f13b 100644 --- a/src/RAM/HashMap.h +++ b/src/RAM/HashMap.h @@ -14,42 +14,6 @@ */ class HashMap { -private: - /** - * @brief A node in the hash map's linked list. - * - * Stores the key-value pair and a pointer to the next node - * to handle bucket collisions via separate chaining. - */ - struct Node - { - std::string key; /**< The unique string key identifier (e.g., Name). */ - std::string value; /**< The string data value associated with the key (e.g., Phone Number). */ - Node *next; /**< Pointer to the next node in the chain (nullptr if last). */ - - /** - * @brief Constructs a new Node object. - * @param k The string key. - * @param v The string value. - */ - Node(const std::string &k, const std::string &v); - }; - - /** * @brief Dynamic array of Node pointers representing the hash table buckets. - * - * Since it points to an array of pointers, it requires double pointer syntax (Node**). - */ - Node **table; - - int capacity; /**< The total size of the table array (number of available buckets). */ - mutable std::shared_mutex sm; /** A mutex that protects the hashmap from 2 writer threads writing at the same time */ - - /** - * @brief Converts a string key into an array index. - * @param key The string key to be hashed. - * @return An integer index between 0 and capacity - 1. - */ - int hashFunction(const std::string& key) const; public: /** @@ -88,6 +52,43 @@ class HashMap * @param callback A function to call for each key-value pair. */ void forEach(std::function callback) const; + +private: + /** + * @brief A node in the hash map's linked list. + * + * Stores the key-value pair and a pointer to the next node + * to handle bucket collisions via separate chaining. + */ + struct Node + { + std::string key; /**< The unique string key identifier (e.g., Name). */ + std::string value; /**< The string data value associated with the key (e.g., Phone Number). */ + Node *next; /**< Pointer to the next node in the chain (nullptr if last). */ + + /** + * @brief Constructs a new Node object. + * @param k The string key. + * @param v The string value. + */ + Node(const std::string &k, const std::string &v); + }; + + /** * @brief Dynamic array of Node pointers representing the hash table buckets. + * + * Since it points to an array of pointers, it requires double pointer syntax (Node**). + */ + Node **table; + + int capacity; /**< The total size of the table array (number of available buckets). */ + mutable std::shared_mutex sm; /** A mutex that protects the hashmap from 2 writer threads writing at the same time */ + + /** + * @brief Converts a string key into an array index. + * @param key The string key to be hashed. + * @return An integer index between 0 and capacity - 1. + */ + int hashFunction(const std::string &key) const; }; #endif \ No newline at end of file diff --git a/src/Server/Server.cpp b/src/Server/Server.cpp index 4bb311b..fcabeac 100644 --- a/src/Server/Server.cpp +++ b/src/Server/Server.cpp @@ -55,13 +55,13 @@ Server::~Server() std::cout << "[SERVER] Cleanup complete\n"; } -void Server::start() +std::expected Server::start() { std::cout << "[SERVER] Binding socket...\n"; if (bind(serverSocket, (sockaddr *)&serverAddr, sizeof(serverAddr)) != 0) { - throw std::runtime_error("Server Failed to Start using bind"); + return std::unexpected("Server Failed to Start using bind"); } std::cout << "[SERVER] Bind successful!\n"; @@ -69,7 +69,7 @@ void Server::start() if (listen(serverSocket, SOMAXCONN) != 0) { - throw std::runtime_error("Server Failed to Listen using listen"); + return std::unexpected("Server Failed to Listen using listen"); } std::cout << "[SERVER] Server is now accepting connections!\n"; @@ -78,6 +78,8 @@ void Server::start() // Start the event loop on its own thread now that the server is live. eventLoopThread = std::thread(&Server::runEventLoop, this); + + return {}; } void Server::acceptClients() @@ -234,6 +236,9 @@ void Server::messageHandler(SocketType clientSocket, const SessionKey &sessionKe closeConnection(clientSocket); return; } + + // Tell compiler that bytesReceived is greater than 0 so it wont do optimizations. + [[assume(bytesReceived > 0)]]; userSessionManager.update_activity(sessionKey); std::cout << "[CLIENT] Received " << bytesReceived << " bytes\n"; diff --git a/src/Server/Server.h b/src/Server/Server.h index 5605944..a900ef9 100644 --- a/src/Server/Server.h +++ b/src/Server/Server.h @@ -4,7 +4,10 @@ #include #include #include -#include +#include +#include +#include +#include #include "UserSessionBackgroundWorker.h" #include "../RAM/HashMap.h" @@ -23,38 +26,6 @@ */ class Server { -private: - SocketType serverSocket; /**< The main listening socket */ - sockaddr_in serverAddr; /**< Server address structure */ - int port; /**< Port the server listens on */ - - std::atomic running; /**< Controls whether the server is running */ - - HashMap hashMap; /** Server owns an instance of the hashmap */ - Persistence pers; /** Server owns an instance of persistance class */ - ThreadPool tpool; /** Server owns an instance of ThreadPool class */ - SnapshotScheduler ss; /** Server owns an instance of SnapshotScheduler class */ - RateLimiter rt; /** Server owns an instance of RateLimter class */ - UserSessionManager userSessionManager; /** Managing User Sessions */ - UserSessionBackgroundWorker user_session_background_worker; /** Background worker that sweeps expired sessions*/ - std::unordered_map connections; /** Client connections, keyed by socket */ - std::unique_ptr eventLoop; /** Watches all client sockets for readiness */ - std::thread eventLoopThread; /** Thread that runs runEventLoop() */ - std::mutex busyMutex; /** Guards busySockets */ - std::unordered_set busySockets; /** Sockets with a recv job already queued/running */ - - /** - * @brief Runs continuously on eventLoopThread: waits for socket readiness - * and dispatches ready clients to the thread pool. - */ - void runEventLoop(); - - /** - * @brief Fully tears down a client connection: stops watching it, removes - * its session and Connection entry, and closes the socket. - * @param sock The socket to close. - */ - void closeConnection(SocketType sock); public: /** @@ -71,8 +42,9 @@ class Server /** * @brief Starts the server, binds and begins listening. * @throws std::runtime_error if socket setup fails. + * @return on success returns void, on failure a string */ - void start(); + std::expected start(); /** * @brief Accepts incoming client connections in a loop. @@ -91,6 +63,39 @@ class Server * @param sessionKey The session tied to this client. */ void messageHandler(SocketType clientSocket, const SessionKey &sessionKey); + +private: + SocketType serverSocket; /**< The main listening socket */ + sockaddr_in serverAddr; /**< Server address structure */ + int port; /**< Port the server listens on */ + + std::atomic running; /**< Controls whether the server is running */ + + HashMap hashMap; /** Server owns an instance of the hashmap */ + Persistence pers; /** Server owns an instance of persistance class */ + ThreadPool tpool; /** Server owns an instance of ThreadPool class */ + SnapshotScheduler ss; /** Server owns an instance of SnapshotScheduler class */ + RateLimiter rt; /** Server owns an instance of RateLimter class */ + UserSessionManager userSessionManager; /** Managing User Sessions */ + UserSessionBackgroundWorker user_session_background_worker; /** Background worker that sweeps expired sessions*/ + std::flat_map connections; /** Client connections, keyed by socket */ + std::unique_ptr eventLoop; /** Watches all client sockets for readiness */ + std::thread eventLoopThread; /** Thread that runs runEventLoop() */ + std::mutex busyMutex; /** Guards busySockets */ + std::flat_set busySockets; /** Sockets with a recv job already queued/running */ + + /** + * @brief Runs continuously on eventLoopThread: waits for socket readiness + * and dispatches ready clients to the thread pool. + */ + void runEventLoop(); + + /** + * @brief Fully tears down a client connection: stops watching it, removes + * its session and Connection entry, and closes the socket. + * @param sock The socket to close. + */ + void closeConnection(SocketType sock); }; #endif \ No newline at end of file diff --git a/src/Storage/Persistence.h b/src/Storage/Persistence.h index 417b478..b15e6f7 100644 --- a/src/Storage/Persistence.h +++ b/src/Storage/Persistence.h @@ -15,11 +15,6 @@ */ class Persistence { -private: - std::ofstream aof_stream; /**< Persistent file stream handle for real-time logging. */ - const std::string aof_path = "appendonly.log"; /**< File path for the transaction log file. */ - const std::string rdb_path = "snapshot.log"; /**< File path for the point-in-time snapshot file. */ - std::mutex aof_mutex; /**< Protects aof_stream from concurrent access. */ public: /** @@ -71,6 +66,12 @@ class Persistence * @param hashmap A constant reference to the in-memory HashMap data source. */ void createSnapshot(const HashMap &hashmap); + +private: + std::ofstream aof_stream; /**< Persistent file stream handle for real-time logging. */ + const std::string aof_path = "appendonly.log"; /**< File path for the transaction log file. */ + const std::string rdb_path = "snapshot.log"; /**< File path for the point-in-time snapshot file. */ + std::mutex aof_mutex; /**< Protects aof_stream from concurrent access. */ }; #endif \ No newline at end of file diff --git a/src/Tests/test_hashmap.cpp b/src/Tests/test_hashmap.cpp index 9b5fba6..0c60f0e 100644 --- a/src/Tests/test_hashmap.cpp +++ b/src/Tests/test_hashmap.cpp @@ -55,7 +55,7 @@ TEST(HashMapTests, HandledRemoveWhenKeyIsntInHashmap) EXPECT_EQ(map.get(key), value); } -TEST(HashMapConcurrencyTests, HandlesConcurrentInsertsSafely) +TEST(HashMapTests, HandlesConcurrentInsertsSafely) { HashMap map{5}; // The map is small to make sure that there is many collitions const int num_threads = 5; diff --git a/src/Worker/SnapshotScheduler.h b/src/Worker/SnapshotScheduler.h index 5ee40ba..3afa81a 100644 --- a/src/Worker/SnapshotScheduler.h +++ b/src/Worker/SnapshotScheduler.h @@ -13,17 +13,6 @@ */ class SnapshotScheduler { -private: - std::thread schedulerThread; /**< Background thread that runs the scheduler loop. */ - std::atomic active; /**< Controls whether the scheduler is running. */ - std::function snapshotCallback; /**< The snapshot function to call every interval. */ - std::chrono::minutes interval; /**< How often to take a snapshot. */ - std::condition_variable cv; /**< Bell that wakes up sleeping workers. */ - std::mutex m; /**< Required by condition_variable for wait_for to work. */ - /** - * @brief The main loop that sleeps then triggers a snapshot. - */ - void run(); public: /** @@ -37,6 +26,18 @@ class SnapshotScheduler * @brief Stops the scheduler and joins the background thread. */ ~SnapshotScheduler(); + +private: + std::thread schedulerThread; /**< Background thread that runs the scheduler loop. */ + std::atomic active; /**< Controls whether the scheduler is running. */ + std::function snapshotCallback; /**< The snapshot function to call every interval. */ + std::chrono::minutes interval; /**< How often to take a snapshot. */ + std::condition_variable cv; /**< Bell that wakes up sleeping workers. */ + std::mutex m; /**< Required by condition_variable for wait_for to work. */ + /** + * @brief The main loop that sleeps then triggers a snapshot. + */ + void run(); }; #endif \ No newline at end of file diff --git a/src/Worker/ThreadPool.h b/src/Worker/ThreadPool.h index bd162ba..f162780 100644 --- a/src/Worker/ThreadPool.h +++ b/src/Worker/ThreadPool.h @@ -18,12 +18,6 @@ */ class ThreadPool { -private: - std::queue> jobs; /**< Queue of pending jobs to be executed. */ - std::vector threads; /**< Pool of worker threads. */ - std::mutex m; /**< Mutex protecting the job queue. */ - std::condition_variable cv; /**< Bell that wakes up sleeping workers. */ - std::atomic active; /**< Controls whether the pool is running. */ public: /** @@ -42,6 +36,13 @@ class ThreadPool * @param job A callable with no arguments or return value. */ void acceptJob(std::function job); + +private: + std::queue> jobs; /**< Queue of pending jobs to be executed. */ + std::vector threads; /**< Pool of worker threads. */ + std::mutex m; /**< Mutex protecting the job queue. */ + std::condition_variable cv; /**< Bell that wakes up sleeping workers. */ + std::atomic active; /**< Controls whether the pool is running. */ }; #endif \ No newline at end of file diff --git a/src/main.cpp b/src/main.cpp index ad05b07..23dc052 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -11,10 +11,14 @@ int main() Server server(port); - server.start(); + if (auto result = server.start(); !result) + { + std::cerr << "[MAIN] Failed to start server: " << result.error() << "\n"; + return 1; + } // Run accept loop in background thread - std::thread serverThread(&Server::acceptClients, &server); + std::jthread serverThread(&Server::acceptClients, &server); std::cout << "[MAIN] Server running. Type 'stop' to shut it down.\n"; @@ -34,9 +38,6 @@ int main() } } - // Wait for server thread to finish - serverThread.join(); - std::cout << "[MAIN] Server exited cleanly\n"; return 0;