Initialization of DB connections, running DB worker threads

Author: martin-beran

etc/default_cfg.json

@@ -7,10 +7,22 @@
         },
         "control": {
             "threads": 1
+        },
+        "database": {
         }
     },
     "http_server": {
         "keepalive_requests": 100,
         "port": 8080
+    },
+    "databases": {
+        "sqlite3": {
+            "db1": {
+                "file": "db1.sqlite",
+                "queries": {
+                    "trivial": "select 1"
+                }
+            }
+        }
     }
 }

src/CMakeLists.txt

@@ -7,18 +7,23 @@ protobuf_generate_cpp(
 message("Protobuf generated headers: ${PROTO_HPP}")
 message("Protobuf generated sources: ${PROTO_CPP}")
 
+add_library(proto_obj OBJECT ${PROTO_CPP})
+target_compile_options(proto_obj PRIVATE -Wno-sign-conversion)
+
 # The main ACppSrv executable
 add_executable(
     acppsrv
     acppsrv.cpp
     application.cpp
     configuration.cpp
+    db_server.cpp
     http_server.cpp
     http_hnd_echo.cpp
     http_hnd_stat.cpp
     log.cpp
+    sqlite3.cpp
     worker.cpp
-    ${PROTO_CPP}
+    $<TARGET_OBJECTS:proto_obj>
 )
 target_include_directories(acppsrv PRIVATE ${CMAKE_CURRENT_BINARY_DIR})
 target_link_libraries(

src/application.cpp

@@ -1,24 +1,32 @@
 #include "application.hpp"
 #include "configuration.hpp"
+#include "db_server.hpp"
 #include "http_server.hpp"
 #include "worker.hpp"
 #include <boost/asio/signal_set.hpp>
+#include <tuple>
+#include <utility>
 
 namespace acppsrv {
 
 bool application::run()
 {
+    // Create threads pools
     auto tp = cfg.data().has_thread_pools() ?
         &cfg.data().thread_pools() : nullptr;
-    // Create threads pools
     thread_pool control_pool(tp && tp->has_control() ? &tp->control() : nullptr,
                              std::string{pool_control_name});
     thread_pool main_pool(tp && tp->has_main() ? &tp->main() : nullptr,
                           std::string{pool_main_name});
+    thread_pool database_pool(tp && tp->has_database() ?
+                              &tp->database() : nullptr,
+                              std::string{pool_database_name});
     // Set up the control pool
     boost::asio::signal_set termsig{control_pool.ctx, SIGINT, SIGTERM};
     termsig.async_wait(
-        [&main_pool](const boost::system::error_code& ec, int sig) {
+        [&main_pool, &database_pool](const boost::system::error_code& ec,
+                                     int sig)
+        {
             if (ec)
                 log_msg(log_level::warning) <<
                     "Waiting for termination signal failed: " << ec.message();
@@ -38,16 +46,27 @@ bool application::run()
                 }
             }
             main_pool.stop();
+            database_pool.stop();
         });
     // Set up the main pool
     http_server http_srv(cfg, main_pool, name, version);
     if (!http_srv.run()) {
         log_msg(log_level::crit) << "Cannot initialize HTTP server";
         return false;
     }
+    // Set up the database pool
+    db_server db_srv(cfg.data().has_databases() ?
+                         &cfg.data().databases() : nullptr,
+                     database_pool);
+    if (!db_srv.run()) {
+        log_msg(log_level::crit) << "Cannot initialize database server";
+        return false;
+    }
     // Start processing
+    log_msg(log_level::notice) << "Initialized, starting worker threads";
     control_pool.run(false);
     main_pool.run(true);
+    database_pool.run(true);
     // We need pool's ctx to register an async op, so any async object
     // belonging to the pool must be created after the pool. But this causes
     // destruction of async objects before destruction of the pool, causing

src/application.hpp

@@ -1,17 +1,22 @@
 #pragma once
 
+#include "sqlite3.hpp"
+
+#include <map>
 #include <string_view>
 
 namespace acppsrv {
 
 class configuration;
+class thread_pool;
 
 class application {
 public:
     static constexpr std::string_view name = "ACppSrv";
     static constexpr std::string_view version = GIT_VERSION;
     static constexpr std::string_view pool_main_name{"main"};
     static constexpr std::string_view pool_control_name{"control"};
+    static constexpr std::string_view pool_database_name{"database"};
     explicit application(const configuration& cfg): cfg(cfg) {}
     application(const application&) = delete;
     application(application&&) = delete;

src/configuration.proto

@@ -46,8 +46,10 @@ message ThreadPool {
 message ThreadPools {
     // The main server client handling pool
     ThreadPool main = 1;
-    //! The administrative control pool
+    // The administrative control pool
     ThreadPool control = 2;
+    // The database handling pool
+    ThreadPool database = 3;
 }
 
 // HTTP server
@@ -62,10 +64,21 @@ message HttpServer {
     uint64 max_request_body = 8; // 0 = unlimited
 }
 
+// SQLite3 database
+message SQLite3Db {
+    string file = 1;
+    map<string, string> queries = 2;
+}
+
+message SQLite3 {
+    map<string, SQLite3Db> sqlite3 = 1;
+}
+
 /*** Top level configuration node ********************************************/
 
 message Configuration {
     Log log = 1;
     ThreadPools thread_pools = 2;
     HttpServer http_server = 3;
+    SQLite3 databases = 4;
 }

src/db_server.cpp

@@ -0,0 +1,53 @@
+#include "db_server.hpp"
+#include "configuration.hpp"
+#include "log.hpp"
+#include "worker.hpp"
+#include <tuple>
+#include <utility>
+
+namespace acppsrv {
+
+db_server::db_server(const proto::SQLite3* cfg, thread_pool& workers):
+    cfg(cfg), workers(workers)
+{
+}
+    
+bool db_server::run()
+{
+    try {
+        for (size_t tidx = 0; tidx < workers.size(); ++tidx) {
+            log_msg(log_level::notice) << "Initializing database thread " <<
+                (tidx + 1) << '/' << workers.size();
+            databases.emplace_back();
+            auto& current = databases.back();
+            auto level = tidx == 0 ? log_level::notice : log_level::debug;
+            if (cfg) {
+                for (auto&& [d_name, d_def]: cfg->sqlite3()) {
+                    log_msg(level) << "Open database=\"" << d_name <<
+                        "\" file=" << d_def.file() << '"';
+                    auto& db = current.
+                        emplace(std::piecewise_construct,
+                                std::forward_as_tuple(d_name),
+                                std::forward_as_tuple(d_def.file())).
+                        first->second;
+                    for (auto&& [q_name, q_def]: d_def.queries()) {
+                            log_msg(level) << "Preparing database=\"" <<
+                                d_name << "\" query=\"" << q_name <<
+                                "\" sql=\"" << q_def << '"';
+                            db.queries.emplace(std::piecewise_construct,
+                                               std::forward_as_tuple(q_name),
+                                               std::forward_as_tuple(db.db,
+                                                                     q_def,
+                                                                     q_name));
+                    }
+                }
+            }
+        }
+    } catch (const sqlite::error& e) {
+        log_msg(log_level::crit) << "Opening databases: " << e.what();
+        return false;
+    }
+    return true;
+}
+
+} // namespace acppsrv

src/db_server.hpp

@@ -0,0 +1,33 @@
+#pragma once
+
+#include "sqlite3.hpp"
+
+#include <map>
+#include <vector>
+
+namespace acppsrv {
+
+class thread_pool;
+
+namespace proto {
+
+class SQLite3;
+
+} // namespace proto
+
+class db_server {
+public:
+    db_server(const proto::SQLite3* cfg, thread_pool& workers);
+    bool run();
+private:
+    struct db_def_t {
+        explicit db_def_t(const std::string& file): db(file) {}
+        sqlite::connection db;
+        std::map<std::string, sqlite::query> queries;
+    };
+    const proto::SQLite3* cfg;
+    thread_pool& workers;
+    std::vector<std::map<std::string, db_def_t>> databases;
+};
+
+} // namespace acppsrv

src/log.cpp

@@ -82,38 +82,44 @@ std::optional<log_level> from_string(std::string_view str)
 
 /*** log_msg *****************************************************************/
 
-log_msg::log_msg(logger& obj, log_level level, session_type session):
+log_msg::log_msg(logger& obj, log_level level, session_type session) noexcept:
      _logger(obj), _level(level), _session(session)
 {
     namespace sc = std::chrono;
-    if (_level > log_level::off && _level <= _logger.level()) {
-        _os.emplace(std::cerr);
-        auto now = sc::system_clock::now().time_since_epoch();
-        std::array<char, 64> buf{};
-        time_t time = sc::duration_cast<sc::seconds>(now).count();
-        auto usec =
-            std::to_string(sc::duration_cast<sc::microseconds>(now).count() %
-                           1'000'000);
-        tm t{};
-        localtime_r(&time, &t);
-        assert(strftime(buf.data(), buf.size(), "%F %T.000000%z", &t) != 0);
-        static constexpr size_t time_end = 26; // after .000000
-        std::copy(usec.begin(), usec.end(),
-                  buf.data() + time_end - usec.size());
-        *this << buf.data();
-        *this << " [" << getpid() << '.' << std::this_thread::get_id();
-        for (auto&& s: session)
-            if (s)
-                *this << '.' << *s;
-        *this << ']';
-        *this << ' ' << to_string(_level) << ' ';
-    }
+    if (_level > log_level::off && _level <= _logger.level())
+        try {
+            _os.emplace(std::cerr);
+            auto now = sc::system_clock::now().time_since_epoch();
+            std::array<char, 64> buf{};
+            time_t time = sc::duration_cast<sc::seconds>(now).count();
+            auto usec =
+                std::to_string(sc::duration_cast<sc::microseconds>(now).count() %
+                               1'000'000);
+            tm t{};
+            localtime_r(&time, &t);
+            assert(strftime(buf.data(), buf.size(), "%F %T.000000%z", &t) != 0);
+            static constexpr size_t time_end = 26; // after .000000
+            std::copy(usec.begin(), usec.end(),
+                      buf.data() + time_end - usec.size());
+            *this << buf.data();
+            *this << " [" << getpid() << '.' << std::this_thread::get_id();
+            for (auto&& s: session)
+                if (s)
+                    *this << '.' << *s;
+            *this << ']';
+            *this << ' ' << to_string(_level) << ' ';
+        } catch (...) {
+            kill();
+        }
 }
 
 log_msg::~log_msg()
 {
     if (_os && _os->get_wrapped())
-        *_os << std::endl;
+        try {
+            *_os << std::endl;
+        } catch (...) {
+        }
 }
 
 /*** logger ******************************************************************/

src/log.hpp

@@ -33,19 +33,25 @@ enum class log_level: int {
 
 [[nodiscard]] std::optional<log_level> from_string(std::string_view str);
 
+// It may be called in destructor, so member functions are generally noexcept
+// and exceptions are silently ignored
 class log_msg {
 public:
     using session_type = std::array<std::optional<uint64_t>, 2>;
-    explicit log_msg(log_level level, session_type session = {});
-    log_msg(logger& obj, log_level level, session_type = {});
+    explicit log_msg(log_level level, session_type session = {}) noexcept;
+    log_msg(logger& obj, log_level level, session_type = {}) noexcept;
     log_msg(const log_msg&) = delete;
-    log_msg(log_msg&&) = default;
+    log_msg(log_msg&&) noexcept = default;
     ~log_msg();
     log_msg& operator=(const log_msg&) = delete;
     log_msg& operator=(log_msg&&) = delete;
-    template <class T> log_msg& operator<<(T&& v) & {
+    template <class T> log_msg& operator<<(T&& v) & noexcept {
         if (_os)
-            *_os << v;
+            try {
+                *_os << v;
+            } catch (...) {
+                kill();
+            }
         return *this;
     }
     template <class ...T> log_msg& operator<<(std::variant<T...>&& v) & {
@@ -60,6 +66,9 @@ class log_msg {
         return std::move(*this);
     }
 private:
+    void kill() {
+        _os.reset();
+    }
     logger& _logger;
     log_level _level;
     std::optional<std::osyncstream> _os;
@@ -92,7 +101,7 @@ class logger {
 
 /*** log_msg *****************************************************************/
 
-inline log_msg::log_msg(log_level level, session_type session):
+inline log_msg::log_msg(log_level level, session_type session) noexcept:
     log_msg(logger::global(), level, session)
 {
 }

src/sqlite3.cpp

@@ -0,0 +1,125 @@
+#include "sqlite3.hpp"
+#include "log.hpp"
+
+#include <cassert>
+#include <sqlite3.h>
+
+namespace acppsrv::sqlite {
+
+/*** connection::impl ********************************************************/
+
+class connection::impl {
+public:
+    explicit impl(connection& conn);
+    impl(const impl&) = delete;
+    impl(impl&&) = delete;
+    ~impl();
+    impl& operator=(const impl&) = delete;
+    impl& operator=(impl&&) = delete;
+    connection& conn;
+    sqlite3* db = nullptr;
+};
+
+connection::impl::impl(connection& conn): conn(conn)
+{
+    if (int status = sqlite3_open_v2(conn._file.c_str(), &db,
+                                     SQLITE_OPEN_READWRITE |
+                                     SQLITE_OPEN_URI |
+                                     SQLITE_OPEN_NOMUTEX |
+                                     SQLITE_OPEN_EXRESCODE,
+                                     nullptr);
+        status != SQLITE_OK)
+    {
+        if (!db)
+            throw error(conn._file);
+        throw error(conn, *this);
+    }
+    assert(db);
+}
+
+connection::impl::~impl()
+{
+    if (sqlite3_close_v2(db) != SQLITE_OK) {
+        assert(db);
+        log_msg(log_level::emerg) << "Cannot close database \"" << conn._file <<
+            "\" handle: " << sqlite3_errmsg(db);
+    } else
+        log_msg(log_level::debug) << "Closed database \"" << conn._file << '"';
+}
+
+/*** connection **************************************************************/
+
+connection::connection(std::string file):
+    _file(std::move(file)), _impl(std::make_unique<impl>(*this))
+{
+}
+
+connection::~connection() = default;
+
+/*** query::impl *************************************************************/
+
+class query::impl {
+public:
+    explicit impl(query& q);
+    impl(const impl&) = delete;
+    impl(impl&&) = delete;
+    ~impl();
+    impl& operator=(const impl&) = delete;
+    impl& operator=(impl&&) = delete;
+private:
+    query& q;
+    sqlite3_stmt* stmt = nullptr;
+};
+
+query::impl::impl(query& q): q(q)
+{
+    // sqlite3 allows passing size incl. terminating NUL
+    if (sqlite3_prepare_v3(q._db._impl->db, q._sql.c_str(),
+                           int(q._sql.size() + 1),
+                           SQLITE_PREPARE_PERSISTENT, &stmt,
+                           nullptr) != SQLITE_OK)
+    {
+        assert(!stmt);
+        // this->q to silence compiler warning about unused q
+        throw error(this->q._db, this->q._sql_id);
+    }
+    assert(stmt);
+}
+
+query::impl::~impl()
+{
+    sqlite3_finalize(stmt);
+}
+
+/*** query *******************************************************************/
+
+query::query(connection& db, std::string sql, std::string sql_id):
+    _db(db), _sql(std::move(sql)), _sql_id(std::move(sql_id)),
+    _impl(std::make_unique<impl>(*this))
+{
+}
+
+query::~query() = default;
+
+/*** error *******************************************************************/
+
+error::error(const std::string& file):
+    runtime_error("sqlite3 error in db \"" + file + "\"" +
+                  ": Cannot allocate database handle")
+{
+}
+
+error::error(connection& db, const std::string& sql_id):
+        runtime_error("sqlite3 error in db \"" + db._file + "\"" +
+                      (sql_id.empty() ? "" : (" (" + sql_id + ")")) +
+                      ": " + sqlite3_errmsg(db._impl->db))
+{
+}
+
+error::error(connection& db, connection::impl& impl):
+        runtime_error("sqlite3 error in db \"" + db._file + "\"" +
+                      ": " + sqlite3_errmsg(impl.db))
+{
+}
+
+} // namespace acppsrv::sqlite

src/sqlite3.hpp

@@ -0,0 +1,54 @@
+#pragma once
+
+#include <memory>
+#include <stdexcept>
+#include <string_view>
+
+namespace acppsrv::sqlite {
+
+class connection;
+class transaction;
+class query;
+
+class connection {
+public:
+    // not using string_view, because sqlite3 requires null-terminated strings
+    explicit connection(std::string file);
+    connection(const connection&) = delete;
+    connection(connection&&) = delete;
+    ~connection();
+    connection& operator=(const connection&) = delete;
+    connection& operator=(connection&&) = delete;
+private:
+    class impl;
+    std::string _file;
+    std::unique_ptr<impl> _impl;
+    friend class error;
+    friend class query;
+};
+
+class query {
+public:
+    explicit query(connection& db, std::string sql, std::string sql_id = {});
+    query(const query&) = delete;
+    query(query&&) = delete;
+    ~query();
+    query& operator=(const query&) = delete;
+    query& operator=(query&&) = delete;
+private:
+    class impl;
+    connection& _db;
+    std::string _sql;
+    std::string _sql_id;
+    std::unique_ptr<impl> _impl;
+};
+
+class error: public std::runtime_error {
+public:
+    explicit error(const std::string& file);
+    explicit error(connection& db, const std::string& sql_id = {});
+    // To be used before db._impl is initialized
+    error(connection& db, connection::impl& impl);
+};
+
+} // namespace acppsrv::sqlite

src/worker.cpp

@@ -5,6 +5,8 @@
 
 namespace acppsrv {
 
+thread_local size_t thread_pool::this_thread_idx;
+
 thread_pool::thread_pool(const proto::ThreadPool* cfg, std::string name):
     ctx(configuration::num_threads(cfg)), name(std::move(name)),
     threads(size_t(configuration::num_threads(cfg)))
@@ -28,6 +30,7 @@ void thread_pool::run(bool persistent)
     for (size_t t = 0; t < threads.size(); ++t)
         threads[t] = std::thread(
             [this, t, num_threads = threads.size()]() {
+                this_thread_idx = t;
                 log_msg(log_level::debug) << "Starting thread " << (t + 1) <<
                     '/' << num_threads << " in pool \"" << name << '"';
                 while (!ctx.stopped())

src/worker.hpp

@@ -26,9 +26,17 @@ class thread_pool {
     void run(bool persistent);
     void stop();
     void wait();
+    size_t size() {
+        return threads.size();
+    }
+    // To be called from a handler running in this thread pool
+    [[nodiscard]] static size_t this_thread() {
+        return this_thread_idx;
+    }
     boost::asio::io_context ctx;
 private:
     const std::string name;
+    static thread_local size_t this_thread_idx;
     std::vector<std::thread> threads;
     boost::asio::executor_work_guard<boost::asio::io_context::executor_type>
         work{ctx.get_executor()};