Calling DB server from HTTP server

Author: martin-beran

README.md

@@ -26,3 +26,20 @@ file [LICENSE.md](LICENSE.html).
     cmake --build build
 
 This will create executable `build/src/acppsrv`.
+
+## Run
+
+1. Create a testing database by script `create_db.sh`.
+1. Start the server by command `build/src/acppsrv etc/default_cfg.json`. It
+   will write log to its standard error.
+
+1. Send a request by command, for example,
+
+        curl  --data-binary @- -H 'Content-Type: application/json'-v http://localhost:8080/db
+
+    The request body in JSON format should be written to stdandard input of
+    `curl`. Request URI paths and protobuf/JSON request and response formats
+    have not been documented yet. See source code for available request types.
+
+1. Stop the server by sending it signal `SIGTERM` or `SIGINT` (by `kill` or
+   `Ctrl-C`.

create_db.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+
+if [ $# != 1 ]; then
+    echo "usage: $0 database.sqlite" >&2
+    exit 1
+fi
+
+sqlite3 "$1" <<EOF
+create table data (key primary key, value);
+EOF

src/CMakeLists.txt

@@ -2,6 +2,7 @@
 protobuf_generate_cpp(
     PROTO_CPP PROTO_HPP
     configuration.proto
+    http_hnd_db.proto
     http_hnd_stat.proto
 )
 message("Protobuf generated headers: ${PROTO_HPP}")
@@ -17,9 +18,10 @@ add_executable(
     application.cpp
     configuration.cpp
     db_server.cpp
-    http_server.cpp
+    http_hnd_db.cpp
     http_hnd_echo.cpp
     http_hnd_stat.cpp
+    http_server.cpp
     log.cpp
     sqlite3.cpp
     worker.cpp

src/application.cpp

@@ -21,12 +21,20 @@ bool application::run()
     thread_pool database_pool(tp && tp->has_database() ?
                               &tp->database() : nullptr,
                               std::string{pool_database_name});
+    // 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;
+    }
     // Set up the control pool
     boost::asio::signal_set termsig{control_pool.ctx, SIGINT, SIGTERM};
     termsig.async_wait(
-        [&main_pool, &database_pool](const boost::system::error_code& ec,
-                                     int sig)
-        {
+        [&main_pool, &database_pool, &db_srv](
+            const boost::system::error_code& ec, int sig
+        ) {
             if (ec)
                 log_msg(log_level::warning) <<
                     "Waiting for termination signal failed: " << ec.message();
@@ -47,21 +55,14 @@ bool application::run()
             }
             main_pool.stop();
             database_pool.stop();
+            db_srv.interrupt();
         });
     // Set up the main pool
-    http_server http_srv(cfg, main_pool, name, version);
+    http_server http_srv(cfg, main_pool, name, version, db_srv);
     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);

src/db_server.cpp

@@ -1,5 +1,6 @@
 #include "db_server.hpp"
 #include "configuration.hpp"
+#include "http_hnd_db.pb.h"
 #include "log.hpp"
 #include "worker.hpp"
 #include <tuple>
@@ -11,7 +12,17 @@ db_server::db_server(const proto::SQLite3* cfg, thread_pool& workers):
     cfg(cfg), workers(workers)
 {
 }
-    
+
+void db_server::interrupt()
+{
+    for (size_t tidx = 1; auto&& db_map: databases) {
+        log_msg(log_level::debug) << "Interrupting database thread " <<
+            tidx++ << '/' << workers.size();
+        for (auto&& db: db_map)
+            db.second.db.interrupt();
+    }
+}
+
 bool db_server::run()
 {
     try {
@@ -50,4 +61,16 @@ bool db_server::run()
     return true;
 }
 
+http_hnd::proto::db::Response
+db_server::run_query(http_hnd::proto::db::Request& request)
+{
+    http_hnd::proto::db::Response response;
+    DEBUG() << "Running database=\"" << request.db() <<
+        "\" query=\"" << request.query() << '"' <<
+        " thread=" << thread_pool::this_thread() << '/' << workers.size();
+    response.set_ok(true);
+    response.set_msg("ok");
+    return response;
+}
+
 } // namespace acppsrv

src/db_server.hpp

@@ -15,16 +15,30 @@ class SQLite3;
 
 } // namespace proto
 
+namespace http_hnd::proto::db {
+
+class Request;
+class Response;
+
+} // namespace http_hnd::proto::db
+
 class db_server {
 public:
     db_server(const proto::SQLite3* cfg, thread_pool& workers);
     bool run();
+    // May be called from any thread
+    template <std::invocable<http_hnd::proto::db::Response> Handler>
+    void query(http_hnd::proto::db::Request&& request, Handler&& handler); 
+    // May be called from any thread
+    void interrupt();
 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;
     };
+    http_hnd::proto::db::Response
+        run_query(http_hnd::proto::db::Request& request);
     const proto::SQLite3* cfg;
     thread_pool& workers;
     std::vector<std::map<std::string, db_def_t>> databases;

src/db_server_impl.hpp

@@ -0,0 +1,21 @@
+#pragma once
+
+#include "db_server.hpp"
+#include "http_hnd_db.pb.h"
+#include "worker.hpp"
+
+namespace acppsrv {
+
+template <std::invocable<http_hnd::proto::db::Response> Handler>
+void db_server::query(http_hnd::proto::db::Request&& request, Handler&& handler)
+{
+    boost::asio::post(workers.ctx,
+        [this, request = std::move(request),
+        handler = std::forward<Handler>(handler)]() mutable
+        {
+            auto response = run_query(request);
+            handler(std::move(response));
+        });
+}
+
+} // namespace acppsrv

src/http_hnd_db.cpp

@@ -0,0 +1,46 @@
+#include "http_hnd_db.hpp"
+#include "http_hnd_db.pb.h"
+#include "db_server.hpp"
+#include "db_server_impl.hpp"
+#include "http_server_impl.hpp"
+#include "log.hpp"
+#include <boost/asio/use_awaitable.hpp>
+
+namespace acppsrv::http_hnd {
+
+template <class Executor,
+         boost::asio::completion_token_for<void(proto::db::Response)> CT>
+auto db::call_query(Executor executor, proto::db::Request&& request, CT&& token)
+{
+    auto init =
+        [this, executor, &request]<class Handler>(Handler&& handler) mutable {
+            db_srv.query(std::move(request),
+                [executor, handler = std::forward<Handler>(handler)](
+                    proto::db::Response response
+                ) mutable {
+                    boost::asio::post(executor,
+                        [handler = std::move(handler),
+                        response = std::move(response)]() mutable {
+                            handler(std::move(response));
+                        });
+                });
+        };
+    return boost::asio::async_initiate<CT, void(proto::db::Response)>(init,
+                                                                      token);
+}
+
+boost::asio::awaitable<http_handler::http_response_type>
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-reference-coroutine-parameters)
+db::handle_async(const http_request_type& request, uint64_t sid, uint64_t req_n)
+{
+    auto [in, response] = parse<proto::db::Request>(request, sid,  req_n);
+    if (!in)
+        co_return response;
+    auto executor = co_await boost::asio::this_coro::executor;
+    proto::db::Response out = co_await call_query(executor, std::move(*in),
+                                                  boost::asio::use_awaitable);
+    serialize(request, sid, req_n, out, response);
+    co_return response;
+}
+
+} // namespace acppsrv::http_hnd

src/http_hnd_db.hpp

@@ -0,0 +1,37 @@
+#pragma once
+
+#include "http_server.hpp"
+
+namespace acppsrv {
+
+class db_server;
+
+namespace http_hnd {
+
+namespace proto::db {
+
+class Request;
+class Response;
+
+} // namespace proto::db
+
+class db: public http_handler {
+public:
+    explicit db(db_server& db_srv): db_srv(db_srv) {}
+    bool async() override {
+        return true;
+    }
+    boost::asio::awaitable<http_response_type>
+        handle_async(const http_request_type& request,
+                     uint64_t sid, uint64_t req_n) override;
+private:
+    template <class Executor,
+             boost::asio::completion_token_for<void(proto::db::Response)> CT>
+        auto call_query(Executor executor, proto::db::Request&& request,
+                        CT&& token);
+    db_server& db_srv;
+};
+
+} // namespace http_hnd
+
+} // namespace acppsrv

src/http_hnd_db.proto

@@ -0,0 +1,32 @@
+syntax = "proto3";
+
+package acppsrv.http_hnd.proto.db;
+
+enum Null {
+    NULL = 0;
+}
+
+message Value {
+    oneof val {
+        Null v_null = 1;
+        int64 v_int64 = 2;
+        double v_double = 3;
+        string v_text = 4;
+    }
+}
+
+message Row {
+    repeated Value columns = 1;
+}
+
+message Request {
+    string db = 1;
+    string query = 2;
+    repeated Value args = 3;
+}
+
+message Response {
+    bool ok = 1;
+    string msg = 2;
+    repeated Row rows = 3;
+}

src/http_server.cpp

@@ -3,6 +3,7 @@
 #include "finally.hpp"
 #include "worker.hpp"
 
+#include "http_hnd_db.hpp"
 #include "http_hnd_echo.hpp"
 #include "http_hnd_stat.hpp"
 
@@ -64,7 +65,8 @@ auto http_server::conn_limit(CompletionToken&& token)
 
 http_server::http_server(const configuration& cfg, thread_pool& workers,
                          std::string_view app_name,
-                         std::string_view app_version):
+                         std::string_view app_version,
+                         db_server& db_srv):
     header_server(std::string(app_name) + "/" + std::string(app_version)),
     port(cfg.http_port()), workers(workers), acceptor(workers.ctx)
 {
@@ -90,6 +92,7 @@ http_server::http_server(const configuration& cfg, thread_pool& workers,
             max_request_body = v;
     }
     // Initialize handlers
+    handlers["/db"] = std::make_unique<http_hnd::db>(db_srv);
     handlers["/echo"] = std::make_unique<http_hnd::echo>();
     auto p_stat = std::make_unique<http_hnd::stat>();
     stat = p_stat.get();

src/http_server.hpp

@@ -13,6 +13,7 @@
 namespace acppsrv {
 
 class configuration;
+class db_server;
 class log_msg;
 class thread_pool;
 class http_handler;
@@ -36,7 +37,8 @@ class http_server {
         boost::beast::http::response<boost::beast::http::empty_body>;
     static_assert(std::is_same_v<endpoint_type, acceptor_type::endpoint_type>);
     http_server(const configuration& cfg, thread_pool& workers,
-                std::string_view app_name, std::string_view app_version);
+                std::string_view app_name, std::string_view app_version,
+                db_server& db_srv);
     http_server(const http_server&) = delete;
     http_server(http_server&&) = delete;
     ~http_server() = default;

src/sqlite3.cpp

@@ -56,6 +56,15 @@ connection::connection(std::string file):
 
 connection::~connection() = default;
 
+void connection::interrupt()
+{
+    if (_impl->db) {
+        sqlite3_interrupt(_impl->db);
+        log_msg(log_level::debug) << "Interrupted database file \"" << _file <<
+            '"';
+    }
+}
+
 /*** query::impl *************************************************************/
 
 class query::impl {