From 539f2b7ceeeecba1d869adf78e9e8b57b1d1e417 Mon Sep 17 00:00:00 2001 From: Andrey Kutin Date: Thu, 6 Aug 2026 09:48:56 +0300 Subject: [PATCH] async DB workers, reconnect, and session limit --- src/database.cpp | 501 ++++++++++++++++++++++++----------------------- src/database.hpp | 56 ++++-- src/server.cpp | 147 ++++++++------ src/server.hpp | 18 +- 4 files changed, 400 insertions(+), 322 deletions(-) diff --git a/src/database.cpp b/src/database.cpp index 9699fab..fa3880a 100644 --- a/src/database.cpp +++ b/src/database.cpp @@ -1,192 +1,225 @@ #include "database.hpp" #include -Database::Database(const DatabaseConfig& cfg) - : conn(std::make_unique( +Database::Database(const DatabaseConfig& cfg, int workers) + : connStr( "host=" + cfg.host + " port=" + std::to_string(cfg.port) + " dbname=" + cfg.dbname + " user=" + cfg.user + - " password=" + cfg.password)), - rng(std::random_device{}()) { - ensureTableExists(); -} - -void Database::ensureTableExists() { - pqxx::work txn(*conn); - - // Создаем таблицу, если её нет - txn.exec( - "CREATE TABLE IF NOT EXISTS random_data (" - " id SERIAL PRIMARY KEY," - " name VARCHAR(100) NOT NULL," - " value INTEGER NOT NULL," - " description TEXT," - " created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP" - ")" - ); - - txn.commit(); -} - -json Database::getRandomRecord() { - std::lock_guard lock(mtx); - - try { - // Получаем общее количество записей - int totalCount = getTotalCount(); - - if (totalCount == 0) { - return json{ - {"error", "No records found in database"} - }; - } - - // Генерируем случайный ID - std::uniform_int_distribution idDist(1, totalCount); - int randomId = idDist(rng); - - pqxx::work txn(*conn); - - // Запрашиваем случайную запись - auto result = txn.exec_params( - "SELECT id, name, value, description, created_at " - "FROM random_data " - "WHERE id = $1", - randomId + " password=" + cfg.password) { + { + pqxx::connection probe(connStr); + pqxx::work txn(probe); + txn.exec( + "CREATE TABLE IF NOT EXISTS random_data (" + " id SERIAL PRIMARY KEY," + " name VARCHAR(100) NOT NULL," + " value INTEGER NOT NULL," + " description TEXT," + " created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP" + ")" ); + txn.commit(); + } + for (int i = 0; i < workers; ++i) { + threadPool.emplace_back([this, i]() { worker(i); }); + } +} - if (result.empty()) { - // Если не нашли, берем первую (на всякий случай) - result = txn.exec("SELECT id, name, value, description, created_at FROM random_data LIMIT 1"); +Database::~Database() { + stop(); +} + +void Database::stop() { + { + std::lock_guard lock(mtx); + stopFlag = true; + } + cv.notify_all(); + for (auto& t : threadPool) { + if (t.joinable()) { + t.join(); + } + } +} + +void Database::worker(int id) { + auto conn = std::make_unique(connStr); + std::mt19937 rng(std::random_device{}()); + + for (;;) { + Task task; + { + std::unique_lock lock(mtx); + cv.wait(lock, [&] { return stopFlag || !queue.empty(); }); + if (stopFlag && queue.empty()) return; + task = std::move(queue.front()); + queue.pop(); } - if (result.empty()) { - txn.abort(); - return json{ - {"error", "No records found in database"} - }; + json res; + try { + if (task.op == Op::RandomRecord) { + res = getRandomRecord(*conn, rng); + } else { + res = getRandomRecords(*conn, rng, task.count, task.med, task.max); + } + } catch (const std::exception&) { + conn = std::make_unique(connStr); + res = json{{"error", "Database error"}}; } + task.cb(std::move(res)); + } +} - const auto& row = result[0]; +void Database::enqueue(Task task) { + { + std::lock_guard lock(mtx); + queue.push(std::move(task)); + } + cv.notify_one(); +} - json response = { +void Database::getRandomRecord(Callback cb) { + enqueue(Task{Op::RandomRecord, 0, 0, 0, std::move(cb)}); +} + +void Database::getRandomRecords(int count, int med, int max, Callback cb) { + enqueue(Task{Op::RandomRecords, count, med, max, std::move(cb)}); +} + +json Database::getRandomRecord(pqxx::connection& conn, std::mt19937& rng) { + int totalCount = getTotalCount(conn); + + if (totalCount == 0) { + return json{ + {"error", "No records found in database"} + }; + } + + std::uniform_int_distribution idDist(1, totalCount); + int randomId = idDist(rng); + + pqxx::work txn(conn); + + auto result = txn.exec( + "SELECT id, name, value, description, created_at " + "FROM random_data " + "WHERE id = $1", + pqxx::params{randomId} + ); + + if (result.empty()) { + result = txn.exec("SELECT id, name, value, description, created_at FROM random_data LIMIT 1"); + } + + if (result.empty()) { + txn.abort(); + return json{ + {"error", "No records found in database"} + }; + } + + const auto& row = result[0]; + + json response = { + {"id", row[0].as()}, + {"name", row[1].as()}, + {"value", row[2].as()}, + {"description", row[3].as()}, + {"created_at", row[4].as()}, + {"timestamp", std::chrono::system_clock::now().time_since_epoch().count()} + }; + + txn.commit(); + return response; +} + +json Database::getRandomRecordsFast(pqxx::connection& conn, const int count) { + pqxx::work txn(conn); + + std::string query = + "WITH RECURSIVE random_selection AS (" + " WITH bounds AS (" + " SELECT" + " MIN(t.id) AS min_id," + " MAX(t.id) AS max_id" + " FROM random_data AS t" + " )," + " first_row AS (" + " SELECT" + " t.id," + " b.min_id," + " b.max_id," + " ARRAY[t.id] AS selected_ids," + " 1 AS depth" + " FROM random_data AS t, bounds AS b" + " WHERE" + " t.id >= b.min_id + ((b.max_id - b.min_id) * random())::INT" + " ORDER BY t.id" + " LIMIT 1" + " )," + " recursive_rows AS (" + " SELECT" + " t.id," + " r.min_id," + " r.max_id," + " r.selected_ids || t.id AS selected_ids," + " r.depth + 1 AS depth" + " FROM random_data AS t, first_row AS r" + " WHERE" + " t.id >= r.min_id + ((r.max_id - r.min_id) * random())::INT" + " AND NOT (t.id = ANY(r.selected_ids))" + " AND r.depth + 1 < " + std::to_string(count + 1) + + " ORDER BY t.id" + " LIMIT 1" + " )" + " SELECT * FROM first_row" + " UNION ALL" + " SELECT * FROM recursive_rows" + ")" + "SELECT" + " rd.id," + " rd.name," + " rd.value," + " rd.description," + " rd.created_at" + "FROM random_selection AS rs" + "JOIN random_data AS rd ON rd.id = rs.id" + "ORDER BY rs.depth"; + + auto result = txn.exec(query); + + if (result.empty()) { + return json{ + {"error", "No records found in database"}, + {"count", 0} + }; + } + + json response = json::array(); + for (const auto& row : result) { + response.push_back({ {"id", row[0].as()}, {"name", row[1].as()}, {"value", row[2].as()}, {"description", row[3].as()}, - {"created_at", row[4].as()}, - {"timestamp", std::chrono::system_clock::now().time_since_epoch().count()} - }; - - txn.commit(); - return response; - - } catch (const std::exception& e) { - return json{ - {"error", "Database error in getRandomRecord"}, - }; + {"created_at", row[4].as()} + }); } + + txn.commit(); + + return { + {"data", response}, + {"count", static_cast(response.size())}, + {"strategy", "fast_recursive_cte"}, + {"timestamp", std::chrono::system_clock::now().time_since_epoch().count()} + }; } -// БЫСТРЫЙ ВАРИАНТ: Использует рекурсивный CTE - -json Database::getRandomRecordsFast(const int count) { - pqxx::work txn(*conn); - - try { - // Рекурсивный CTE запрос для получения случайных записей - std::string query = - "WITH RECURSIVE random_selection AS (" - " WITH bounds AS (" - " SELECT" - " MIN(t.id) AS min_id," - " MAX(t.id) AS max_id" - " FROM random_data AS t" - " )," - " first_row AS (" - " SELECT" - " t.id," - " b.min_id," - " b.max_id," - " ARRAY[t.id] AS selected_ids," - " 1 AS depth" - " FROM random_data AS t, bounds AS b" - " WHERE" - " t.id >= b.min_id + ((b.max_id - b.min_id) * random())::INT" - " ORDER BY t.id" - " LIMIT 1" - " )," - " recursive_rows AS (" - " SELECT" - " t.id," - " r.min_id," - " r.max_id," - " r.selected_ids || t.id AS selected_ids," - " r.depth + 1 AS depth" - " FROM random_data AS t, first_row AS r" - " WHERE" - " t.id >= r.min_id + ((r.max_id - r.min_id) * random())::INT" - " AND NOT (t.id = ANY(r.selected_ids))" - " AND r.depth + 1 < " + std::to_string(count + 1) + - " ORDER BY t.id" - " LIMIT 1" - " )" - " SELECT * FROM first_row" - " UNION ALL" - " SELECT * FROM recursive_rows" - ")" - "SELECT" - " rd.id," - " rd.name," - " rd.value," - " rd.description," - " rd.created_at" - "FROM random_selection AS rs" - "JOIN random_data AS rd ON rd.id = rs.id" - "ORDER BY rs.depth"; - - auto result = txn.exec(query); - - if (result.empty()) { - return json{ - {"error", "No records found in database"}, - {"count", 0} - }; - } - - json response = json::array(); - for (const auto& row : result) { - response.push_back({ - {"id", row[0].as()}, - {"name", row[1].as()}, - {"value", row[2].as()}, - {"description", row[3].as()}, - {"created_at", row[4].as()} - }); - } - - txn.commit(); - - return { - {"data", response}, - {"count", static_cast(response.size())}, - {"strategy", "fast_recursive_cte"}, - {"timestamp", std::chrono::system_clock::now().time_since_epoch().count()} - }; - - } catch (const std::exception& e) { - txn.abort(); - return json{ - {"error", "Database error in fast recursive CTE strategy"}, - }; - } -} - -// МЕДЛЕННЫЙ ВАРИАНТ: Генерирует ID на клиенте, затем запрашивает по IN - -json Database::getRandomRecordsSlow(const int count, const int totalCount) { - // Генерируем уникальные случайные ID +json Database::getRandomRecordsSlow(pqxx::connection& conn, std::mt19937& rng, const int count, const int totalCount) { std::set ids; { std::uniform_int_distribution idDist(1, totalCount); @@ -194,126 +227,104 @@ json Database::getRandomRecordsSlow(const int count, const int totalCount) { ids.insert(idDist(rng)); } } - - // Запрашиваем записи по сгенерированным ID - pqxx::work txn(*conn); - - try { - if (ids.empty()) { - return json{ - {"data", json::array()}, - {"count", 0}, - {"strategy", "slow"} - }; - } - - // Формируем запрос с IN - std::string query = getQueryRandomIds(static_cast(ids.size())); - // Подготавливаем параметры для exec_params - pqxx::params param_list; - for (int id : ids) { - param_list.append(id); - } - - auto result = txn.exec_params(query, param_list); - - json response = json::array(); - for (const auto& row : result) { - response.push_back({ - {"id", row[0].as()}, - {"name", row[1].as()}, - {"value", row[2].as()}, - {"description", row[3].as()}, - {"created_at", row[4].as()} - }); - } - - txn.commit(); - - return { - {"data", response}, - {"count", static_cast(response.size())}, - {"strategy", "slow"}, - {"timestamp", std::chrono::system_clock::now().time_since_epoch().count()} - }; - - } catch (const std::exception& e) { - txn.abort(); + pqxx::work txn(conn); + + if (ids.empty()) { return json{ - {"error", "Database error in slow strategy"}, + {"data", json::array()}, + {"count", 0}, + {"strategy", "slow"} }; } + + std::string query = getQueryRandomIds(static_cast(ids.size())); + + pqxx::params param_list; + for (int id : ids) { + param_list.append(id); + } + + auto result = txn.exec(query, param_list); + + json response = json::array(); + for (const auto& row : result) { + response.push_back({ + {"id", row[0].as()}, + {"name", row[1].as()}, + {"value", row[2].as()}, + {"description", row[3].as()}, + {"created_at", row[4].as()} + }); + } + + txn.commit(); + + return { + {"data", response}, + {"count", static_cast(response.size())}, + {"strategy", "slow"}, + {"timestamp", std::chrono::system_clock::now().time_since_epoch().count()} + }; } -// ГЛАВНАЯ ФУНКЦИЯ: Выбирает стратегию в зависимости от count - -json Database::getRandomRecords(int count, int med, int max) { - std::lock_guard lock(mtx); - // Валидация входных параметров +json Database::getRandomRecords(pqxx::connection& conn, std::mt19937& rng, int count, int med, int max) { if (count <= 0) { return json{ {"error", "Count must be positive"}, {"count", 0} }; } - // Получаем общее количество записей с блокировкой - int totalCount = getTotalCount(); - + + int totalCount = getTotalCount(conn); + if (totalCount == 0) { return json{ {"error", "No records found in database"}, {"count", 0} }; } - + if (count > totalCount) { count = totalCount; } - - // Защита от слишком больших запросов + if (count > max) { count = max; } - - // Выбор стратегии: - // - Для малых выборок (<= med) используем медленный, но более точный способ - // - Для больших выборок (> med) используем быстрый способ с рекурсивным CTE запросом - + json response; if (count <= med) { - response = getRandomRecordsSlow(count, totalCount); + response = getRandomRecordsSlow(conn, rng, count, totalCount); } else { - response = getRandomRecordsFast(count); + response = getRandomRecordsFast(conn, count); } - if(response.contains("error")){ + if (response.contains("error")) { return response; } - // Добавляем общую мета-информацию + response["requested_count"] = count; response["total_available"] = totalCount; return response; } -// Вспомогательный метод для получения общего количества записей -int Database::getTotalCount() { - pqxx::work txn(*conn); +int Database::getTotalCount(pqxx::connection& conn) { + pqxx::work txn(conn); auto result = txn.exec("SELECT COUNT(*) FROM random_data"); int count = result[0][0].as(); txn.commit(); return count; } -std::string Database::getQueryRandomIds(const int count){ - std::string query = - "SELECT id, name, value, description, created_at " - "FROM random_data WHERE id IN ("; - - // Подготавливаем параметры +std::string Database::getQueryRandomIds(const int count) { + std::string query = + "SELECT id, name, value, description, created_at " + "FROM random_data WHERE id IN ("; + for (size_t i = 0; i < count; ++i) { - if (i > 0) query += ","; - query += "$" + std::to_string(i + 1); + if (i > 0) query += ","; + query += "$" + std::to_string(i + 1); } query += ") ORDER BY id"; return query; diff --git a/src/database.hpp b/src/database.hpp index d6a7863..fa254cc 100644 --- a/src/database.hpp +++ b/src/database.hpp @@ -5,31 +5,55 @@ #include #include #include +#include +#include #include +#include #include +#include #include "config.hpp" #include "common.hpp" class Database { public: - explicit Database(const DatabaseConfig& config); - ~Database() = default; - - json getRandomRecord(); - json getRandomRecords(int count, int med, int max); - + using Callback = std::function; + + explicit Database(const DatabaseConfig& config, int workers = 2); + ~Database(); + + Database(const Database&) = delete; + Database& operator=(const Database&) = delete; + + void getRandomRecord(Callback cb); + void getRandomRecords(int count, int med, int max, Callback cb); + private: - std::unique_ptr conn; - std::mt19937 rng; - std::mutex mtx; - // Создание тестовой таблицы, если она не существует - void ensureTableExists(); + enum class Op { RandomRecord, RandomRecords }; - int getTotalCount(); - // Две стратегии получения случайных записей - json getRandomRecordsFast(const int count); - json getRandomRecordsSlow(const int count, const int totalCount); + struct Task { + Op op; + int count = 0; + int med = 10; + int max = 1000; + Callback cb; + }; + + void worker(int id); + void enqueue(Task task); + void stop(); + + json getRandomRecord(pqxx::connection& conn, std::mt19937& rng); + json getRandomRecords(pqxx::connection& conn, std::mt19937& rng, int count, int med, int max); + json getRandomRecordsFast(pqxx::connection& conn, const int count); + json getRandomRecordsSlow(pqxx::connection& conn, std::mt19937& rng, const int count, const int totalCount); + int getTotalCount(pqxx::connection& conn); std::string getQueryRandomIds(const int count); -}; + std::string connStr; + std::vector threadPool; + std::mutex mtx; + std::condition_variable cv; + std::queue queue; + bool stopFlag = false; +}; diff --git a/src/server.cpp b/src/server.cpp index b42b623..3be53c1 100644 --- a/src/server.cpp +++ b/src/server.cpp @@ -3,9 +3,20 @@ #include #include -// Implementation of HttpSession -HttpSession::HttpSession(net::ip::tcp::socket&& socket, std::shared_ptr db) - : stream(std::move(socket)), db(db) {} +HttpSession::HttpSession(net::ip::tcp::socket&& socket, + std::shared_ptr db, + std::shared_ptr> sessionCounter) + : stream(std::move(socket)), db(db), sessionCounter(sessionCounter) { + if (sessionCounter) { + sessionCounter->fetch_add(1); + } +} + +HttpSession::~HttpSession() { + if (sessionCounter) { + sessionCounter->fetch_sub(1); + } +} void HttpSession::run() { readRequest(); @@ -34,49 +45,70 @@ void HttpSession::onRead(beast::error_code ec, std::size_t) { } void HttpSession::processRequest(http::request&& request) { - http::response res{http::status::ok, request.version()}; - res.set(http::field::server, "Cards Server"); - res.set(http::field::content_type, "application/json"); - - try { - if (request.method() == http::verb::get && request.target() == "/") { - // Получаем случайный объект из БД - json response = db->getRandomRecord(); - res.body() = response.dump(4); - } else if (request.method() == http::verb::get && request.target().starts_with("/cards")) { + auto self = shared_from_this(); + + auto sendJson = [self](http::status status, unsigned version, json body) { + http::response res{status, version}; + res.set(http::field::server, "Cards Server"); + res.set(http::field::content_type, "application/json"); + res.body() = body.dump(4); + res.prepare_payload(); + self->writeResponse(std::move(res)); + }; + + if (request.method() == http::verb::get && request.target() == "/") { + db->getRandomRecord([self, sendJson, version = request.version()](json body) { + net::post( + self->stream.get_executor(), + [self, sendJson, version, body = std::move(body)]() mutable { + sendJson(http::status::ok, version, std::move(body)); + }); + }); + return; + } + + if (request.method() == http::verb::get && request.target().starts_with("/cards")) { + int count = 0; + try { std::string number{request.target()}; - auto start = number.find("?count=") + 7; - auto end = number.find("?", start); + auto pos = number.find("?count="); + if (pos == std::string::npos) { + throw std::invalid_argument("missing count"); + } + size_t start = pos + 7; + size_t end = number.find('?', start); if (end == std::string::npos) { end = number.length(); } - int count = std::stoi(number.substr(start, end - start)); - count = std::max(1, std::min(count, 1000)); // валидация - json response = db->getRandomRecords(count, 10, 1000); - res.body() = response.dump(4); - } else if (request.method() == http::verb::get && request.target() == "/health") { - json response = { - {"status", "ok"}, - {"timestamp", std::chrono::system_clock::now().time_since_epoch().count()} - }; - res.body() = response.dump(4); - } else { - res.result(http::status::not_found); - json error = { - {"error", "Not Found"}, - }; - res.body() = error.dump(4); + count = std::stoi(number.substr(start, end - start)); + count = std::max(1, std::min(count, 1000)); + } catch (const std::exception&) { + sendJson(http::status::bad_request, request.version(), + json{{"error", "Invalid count parameter"}}); + return; } - } catch (const std::exception& e) { - res.result(http::status::internal_server_error); - json error = { - {"error", "Internal Server Error"}, - }; - res.body() = error.dump(4); + db->getRandomRecords(count, 10, 1000, + [self, sendJson, version = request.version()](json body) { + net::post( + self->stream.get_executor(), + [self, sendJson, version, body = std::move(body)]() mutable { + sendJson(http::status::ok, version, std::move(body)); + }); + }); + return; } - - res.prepare_payload(); - writeResponse(std::move(res)); + + if (request.method() == http::verb::get && request.target() == "/health") { + sendJson(http::status::ok, request.version(), json{ + {"status", "ok"}, + {"timestamp", std::chrono::system_clock::now().time_since_epoch().count()} + }); + return; + } + + sendJson(http::status::not_found, request.version(), json{ + {"error", "Not Found"} + }); } void HttpSession::writeResponse(http::response&& res) { @@ -88,7 +120,6 @@ void HttpSession::writeResponse(http::response&& res) { if (ec) { self->handleError(ec, "write"); } - // Close the connection gracefully self->stream.socket().shutdown(net::ip::tcp::socket::shutdown_send, ec); } ); @@ -108,12 +139,12 @@ Server::Server(const std::string& address, int port, int threads, std::shared_pt acceptor(ioc, net::ip::tcp::endpoint(net::ip::make_address(address), port), true), db(db), threads(threads), - acceptTimer(ioc) - {} + acceptTimer(ioc), + activeSessions(std::make_shared>(0)) {} void Server::run() { accept(); - + // Run the I/O service on the requested number of threads threadPool.reserve(threads); for (int i = 0; i < threads; ++i) { @@ -121,7 +152,7 @@ void Server::run() { ioc.run(); }); } - + // Wait for all threads to finish for (auto& t : threadPool) { if (t.joinable()) { @@ -142,18 +173,22 @@ void Server::accept() { } void Server::onAccept(beast::error_code ec, net::ip::tcp::socket socket) { - if (!ec) { - std::make_shared(std::move(socket), db)->run(); - accept(); - } else { + if (ec) { std::cerr << "Accept error: " << ec.message() << std::endl; - if(!ioc.stopped()) { - acceptTimer.expires_after(std::chrono::milliseconds(100)); - acceptTimer.async_wait([this](beast::error_code ec) { - if (!ioc.stopped()) { accept(); } - }); + if (!ioc.stopped()) { + acceptTimer.expires_after(std::chrono::milliseconds(100)); + acceptTimer.async_wait([this](beast::error_code ec) { + if (!ioc.stopped()) { accept(); } + }); } + return; } - -} + if (activeSessions->load() >= maxSessions) { + beast::error_code ignore; + socket.close(ignore); + } else { + std::make_shared(std::move(socket), db, activeSessions)->run(); + } + accept(); +} diff --git a/src/server.hpp b/src/server.hpp index f3615cc..84242e0 100644 --- a/src/server.hpp +++ b/src/server.hpp @@ -9,6 +9,7 @@ #include #include #include +#include #include "common.hpp" #include "database.hpp" @@ -20,19 +21,24 @@ namespace net = boost::asio; class HttpSession : public std::enable_shared_from_this { public: - HttpSession(net::ip::tcp::socket&& socket, std::shared_ptr db); + HttpSession(net::ip::tcp::socket&& socket, + std::shared_ptr db, + std::shared_ptr> sessionCounter); + ~HttpSession(); + void run(); - + private: void readRequest(); void onRead(beast::error_code ec, std::size_t bytes_transferred); void processRequest(http::request&& req); void writeResponse(http::response&& res); void handleError(beast::error_code ec, const char* what); - + beast::tcp_stream stream; beast::flat_buffer buffer; std::shared_ptr db; + std::shared_ptr> sessionCounter; http::request req; }; @@ -41,15 +47,17 @@ public: Server(const std::string& address, int port, int threads, std::shared_ptr db); void run(); void stop(); - + private: void accept(); void onAccept(beast::error_code ec, net::ip::tcp::socket socket); - + net::io_context ioc; net::ip::tcp::acceptor acceptor; std::shared_ptr db; int threads; std::vector threadPool; net::steady_timer acceptTimer; + std::shared_ptr> activeSessions; + static constexpr std::size_t maxSessions = 512; };