async DB workers, reconnect, and session limit

src/database.cpp

@@ -1,21 +1,16 @@
 #include "database.hpp"
 #include <chrono>
 
-Database::Database(const DatabaseConfig& cfg) 
+Database::Database(const DatabaseConfig& cfg, int workers)
-    : conn(std::make_unique<pqxx::connection>(
+    : connStr(
         "host=" + cfg.host +
         " port=" + std::to_string(cfg.port) +
         " dbname=" + cfg.dbname +
         " user=" + cfg.user +
-        " password=" + cfg.password)),
+        " password=" + cfg.password) {
-      rng(std::random_device{}()) {
+    {
-    ensureTableExists();
+        pqxx::connection probe(connStr);
-}
+        pqxx::work txn(probe);
-
-void Database::ensureTableExists() {
-    pqxx::work txn(*conn);
-    
-    // Создаем таблицу, если её нет
         txn.exec(
             "CREATE TABLE IF NOT EXISTS random_data ("
             "    id SERIAL PRIMARY KEY,"
@@ -25,16 +20,77 @@ void Database::ensureTableExists() {
             "    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP"
             ")"
         );
-    
         txn.commit();
+    }
+    for (int i = 0; i < workers; ++i) {
+        threadPool.emplace_back([this, i]() { worker(i); });
+    }
 }
 
-json Database::getRandomRecord() {
+Database::~Database() {
-    std::lock_guard<std::mutex> lock(mtx);
+    stop();
+}
 
+void Database::stop() {
+    {
+        std::lock_guard<std::mutex> 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<pqxx::connection>(connStr);
+    std::mt19937 rng(std::random_device{}());
+
+    for (;;) {
+        Task task;
+        {
+            std::unique_lock<std::mutex> lock(mtx);
+            cv.wait(lock, [&] { return stopFlag || !queue.empty(); });
+            if (stopFlag && queue.empty()) return;
+            task = std::move(queue.front());
+            queue.pop();
+        }
+
+        json res;
         try {
-        // Получаем общее количество записей
+            if (task.op == Op::RandomRecord) {
-        int totalCount = getTotalCount();
+                res = getRandomRecord(*conn, rng);
+            } else {
+                res = getRandomRecords(*conn, rng, task.count, task.med, task.max);
+            }
+        } catch (const std::exception&) {
+            conn = std::make_unique<pqxx::connection>(connStr);
+            res = json{{"error", "Database error"}};
+        }
+        task.cb(std::move(res));
+    }
+}
+
+void Database::enqueue(Task task) {
+    {
+        std::lock_guard<std::mutex> lock(mtx);
+        queue.push(std::move(task));
+    }
+    cv.notify_one();
+}
+
+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{
@@ -42,22 +98,19 @@ json Database::getRandomRecord() {
         };
     }
 
-        // Генерируем случайный ID
     std::uniform_int_distribution<int> idDist(1, totalCount);
     int randomId = idDist(rng);
 
-        pqxx::work txn(*conn);
+    pqxx::work txn(conn);
 
-        // Запрашиваем случайную запись
+    auto result = txn.exec(
-        auto result = txn.exec_params(
         "SELECT id, name, value, description, created_at "
         "FROM random_data "
         "WHERE id = $1",
-            randomId
+        pqxx::params{randomId}
     );
 
     if (result.empty()) {
-            // Если не нашли, берем первую (на всякий случай)
         result = txn.exec("SELECT id, name, value, description, created_at FROM random_data LIMIT 1");
     }
 
@@ -81,21 +134,11 @@ json Database::getRandomRecord() {
 
     txn.commit();
     return response;
-
-    } catch (const std::exception& e) {
-        return json{
-            {"error", "Database error in getRandomRecord"},
-        };
-    }
 }
 
-// БЫСТРЫЙ ВАРИАНТ: Использует рекурсивный CTE
+json Database::getRandomRecordsFast(pqxx::connection& conn, const int count) {
+    pqxx::work txn(conn);
 
-json Database::getRandomRecordsFast(const int count) {
-    pqxx::work txn(*conn);
-    
-    try {
-        // Рекурсивный CTE запрос для получения случайных записей
     std::string query =
         "WITH RECURSIVE random_selection AS ("
         "    WITH bounds AS ("
@@ -174,19 +217,9 @@ json Database::getRandomRecordsFast(const int count) {
         {"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(pqxx::connection& conn, std::mt19937& rng, const int count, const int totalCount) {
-
-json Database::getRandomRecordsSlow(const int count, const int totalCount) {
-    // Генерируем уникальные случайные ID
     std::set<int> ids;
     {
         std::uniform_int_distribution<int> idDist(1, totalCount);
@@ -195,10 +228,8 @@ json Database::getRandomRecordsSlow(const int count, const int totalCount) {
         }
     }
 
-    // Запрашиваем записи по сгенерированным ID
+    pqxx::work txn(conn);
-    pqxx::work txn(*conn);
 
-    try {
     if (ids.empty()) {
         return json{
             {"data", json::array()},
@@ -207,16 +238,14 @@ json Database::getRandomRecordsSlow(const int count, const int totalCount) {
         };
     }
 
-        // Формируем запрос с IN
     std::string query = getQueryRandomIds(static_cast<int>(ids.size()));
 
-        // Подготавливаем параметры для exec_params
     pqxx::params param_list;
     for (int id : ids) {
         param_list.append(id);
     }
 
-        auto result = txn.exec_params(query, param_list);
+    auto result = txn.exec(query, param_list);
 
     json response = json::array();
     for (const auto& row : result) {
@@ -237,28 +266,17 @@ json Database::getRandomRecordsSlow(const int count, const int totalCount) {
         {"strategy", "slow"},
         {"timestamp", std::chrono::system_clock::now().time_since_epoch().count()}
     };
-        
-    } catch (const std::exception& e) {
-        txn.abort();
-        return json{
-            {"error", "Database error in slow strategy"},
-        };
-    }
 }
 
-// ГЛАВНАЯ ФУНКЦИЯ: Выбирает стратегию в зависимости от count
+json Database::getRandomRecords(pqxx::connection& conn, std::mt19937& rng, int count, int med, int max) {
-
-json Database::getRandomRecords(int count, int med, int max) {
-    std::lock_guard<std::mutex> lock(mtx);
-    // Валидация входных параметров
     if (count <= 0) {
         return json{
             {"error", "Count must be positive"},
             {"count", 0}
         };
     }
-    //  Получаем общее количество записей с блокировкой
+
-    int totalCount = getTotalCount();
+    int totalCount = getTotalCount(conn);
 
     if (totalCount == 0) {
         return json{
@@ -271,46 +289,39 @@ json Database::getRandomRecords(int count, int med, int max) {
         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::connection& conn) {
-int Database::getTotalCount() {
+    pqxx::work txn(conn);
-    pqxx::work txn(*conn);
     auto result = txn.exec("SELECT COUNT(*) FROM random_data");
     int count = result[0][0].as<int>();
     txn.commit();
     return count;
 }
 
-std::string Database::getQueryRandomIds(const int count){
+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);

src/database.hpp

@@ -5,31 +5,55 @@
 #include <random>
 #include <memory>
 #include <mutex>
+#include <condition_variable>
+#include <queue>
 #include <set>
+#include <thread>
 #include <vector>
+#include <functional>
 
 #include "config.hpp"
 #include "common.hpp"
 
 class Database {
 public:
-    explicit Database(const DatabaseConfig& config);
+    using Callback = std::function<void(json)>;
-    ~Database() = default;
 
-    json getRandomRecord();
+    explicit Database(const DatabaseConfig& config, int workers = 2);
-    json getRandomRecords(int count, int med, int max);
+    ~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<pqxx::connection> conn;
+    enum class Op { RandomRecord, RandomRecords };
-    std::mt19937 rng;
-    std::mutex mtx;
-    // Создание тестовой таблицы, если она не существует
-    void ensureTableExists();
 
-    int getTotalCount();
+    struct Task {
-    // Две стратегии получения случайных записей
+        Op op;
-    json getRandomRecordsFast(const int count);
+        int count = 0;
-    json getRandomRecordsSlow(const int count, const int totalCount);
+        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<std::thread> threadPool;
+    std::mutex mtx;
+    std::condition_variable cv;
+    std::queue<Task> queue;
+    bool stopFlag = false;
+};

src/server.cpp

@@ -3,9 +3,20 @@
 #include <iostream>
 #include <chrono>
 
-// Implementation of HttpSession
+HttpSession::HttpSession(net::ip::tcp::socket&& socket,
-HttpSession::HttpSession(net::ip::tcp::socket&& socket, std::shared_ptr<Database> db)
+                         std::shared_ptr<Database> db,
-    : stream(std::move(socket)), db(db) {}
+                         std::shared_ptr<std::atomic<std::size_t>> 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<http::string_body>&& request) {
-    http::response<http::string_body> res{http::status::ok, request.version()};
+    auto self = shared_from_this();
+
+    auto sendJson = [self](http::status status, unsigned version, json body) {
+        http::response<http::string_body> 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));
+    };
 
-    try {
     if (request.method() == http::verb::get && request.target() == "/") {
-            // Получаем случайный объект из БД
+        db->getRandomRecord([self, sendJson, version = request.version()](json body) {
-            json response = db->getRandomRecord();
+            net::post(
-            res.body() = response.dump(4);
+                self->stream.get_executor(),
-        } else if (request.method() == http::verb::get && request.target().starts_with("/cards")) {
+                [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 pos = number.find("?count=");
-            auto end = number.find("?", start);
+            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::stoi(number.substr(start, end - start));
-            count = std::max(1, std::min(count, 1000)); // валидация
+            count = std::max(1, std::min(count, 1000));
-            json response = db->getRandomRecords(count, 10, 1000);
+        } catch (const std::exception&) {
-            res.body() = response.dump(4);
+            sendJson(http::status::bad_request, request.version(),
-        } else if (request.method() == http::verb::get && request.target() == "/health") {
+                     json{{"error", "Invalid count parameter"}});
-            json response = {
+            return;
-                {"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);
         }
-    } catch (const std::exception& e) {
+        db->getRandomRecords(count, 10, 1000,
-        res.result(http::status::internal_server_error);
+            [self, sendJson, version = request.version()](json body) {
-        json error = {
+                net::post(
-            {"error", "Internal Server Error"},
+                    self->stream.get_executor(),
-        };
+                    [self, sendJson, version, body = std::move(body)]() mutable {
-        res.body() = error.dump(4);
+                        sendJson(http::status::ok, version, std::move(body));
+                    });
+            });
+        return;
     }
 
-    res.prepare_payload();
+    if (request.method() == http::verb::get && request.target() == "/health") {
-    writeResponse(std::move(res));
+        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<http::string_body>&& res) {
@@ -88,7 +120,6 @@ void HttpSession::writeResponse(http::response<http::string_body>&& res) {
             if (ec) {
                 self->handleError(ec, "write");
             }
-            // Close the connection gracefully
             self->stream.socket().shutdown(net::ip::tcp::socket::shutdown_send, ec);
         }
     );
@@ -108,8 +139,8 @@ 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<std::atomic<std::size_t>>(0)) {}
 
 void Server::run() {
     accept();
@@ -142,18 +173,22 @@ void Server::accept() {
 }
 
 void Server::onAccept(beast::error_code ec, net::ip::tcp::socket socket) {
-    if (!ec) {
+    if (ec) {
-        std::make_shared<HttpSession>(std::move(socket), db)->run();
-        accept();
-    } else {
         std::cerr << "Accept error: " << ec.message() << std::endl;
-        if(!ioc.stopped()) {
+        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<HttpSession>(std::move(socket), db, activeSessions)->run();
+    }
+    accept();
 }
-

src/server.hpp

@@ -9,6 +9,7 @@
 #include <thread>
 #include <vector>
 #include <string>
+#include <atomic>
 
 #include "common.hpp"
 #include "database.hpp"
@@ -20,7 +21,11 @@ namespace net = boost::asio;
 
 class HttpSession : public std::enable_shared_from_this<HttpSession> {
 public:
-    HttpSession(net::ip::tcp::socket&& socket, std::shared_ptr<Database> db);
+    HttpSession(net::ip::tcp::socket&& socket,
+                std::shared_ptr<Database> db,
+                std::shared_ptr<std::atomic<std::size_t>> sessionCounter);
+    ~HttpSession();
+
     void run();
 
 private:
@@ -33,6 +38,7 @@ private:
     beast::tcp_stream stream;
     beast::flat_buffer buffer;
     std::shared_ptr<Database> db;
+    std::shared_ptr<std::atomic<std::size_t>> sessionCounter;
     http::request<http::string_body> req;
 };
 
@@ -52,4 +58,6 @@ private:
     int threads;
     std::vector<std::thread> threadPool;
     net::steady_timer acceptTimer;
+    std::shared_ptr<std::atomic<std::size_t>> activeSessions;
+    static constexpr std::size_t maxSessions = 512;
 };