implemented the first version of statement_cache

source/orm/sql/connection.cpp

@@ -13,6 +13,21 @@
 
 namespace matador::sql {
 
+namespace internal {
+class connection_statement_proxy final : public statement_proxy {
+public:
+  explicit connection_statement_proxy(std::unique_ptr<statement_impl>&& stmt)
+  : statement_proxy(std::move(stmt)) {}
+
+  utils::result<size_t, utils::error> execute() override {
+    return statement_->execute();
+  }
+  utils::result<std::unique_ptr<query_result_impl>, utils::error> fetch() override {
+    return statement_->fetch();
+  }
+};
+
+}
 connection::connection(connection_info info, const std::shared_ptr<abstract_sql_logger> &sql_logger)
 : connection_info_(std::move(info))
 , logger_(sql_logger)
@@ -57,8 +72,10 @@ connection & connection::operator=(connection &&x) noexcept {
   return *this;
 }
 
-connection::~connection()
-{
+connection::~connection() {
+  if (!connection_) {
+    return;
+  }
   if (connection_->is_open()) {
     connection_->close();
   }
@@ -66,8 +83,7 @@ connection::~connection()
   connection_ = nullptr;
 }
 
-utils::result<void, utils::error> connection::open() const
-{
+utils::result<void, utils::error> connection::open() const {
   const auto res = is_open();
   if (res.is_error()) {
     return utils::failure(res.err());
@@ -79,8 +95,7 @@ utils::result<void, utils::error> connection::open() const
   return utils::ok<void>();
 }
 
-utils::result<void, utils::error> connection::close() const
-{
+utils::result<void, utils::error> connection::close() const {
   logger_->on_close();
   return connection_->close();
 }
@@ -185,27 +200,17 @@ utils::result<size_t, utils::error> connection::execute(const query_context& ctx
   return execute(ctx.sql);
 }
 
-utils::result<statement, utils::error> connection::prepare(const query_context &ctx) const
-{
-  if (ctx.command != sql_command::SQL_CREATE_TABLE && (ctx.prototype.empty() || has_unknown_columns(ctx.prototype))) {
-      if (const auto result = describe(ctx.table.name); result.is_ok()) {
-        for (auto &col: ctx.prototype) {
-          const auto rit = std::find_if(std::begin(*result), std::end(*result),
-                                        [&col](const auto &value) {
-                                          return value.name() == col.name();
-                                        });
-        if (col.type() == utils::basic_type::type_null && rit != result->end()) {
-          const_cast<object::attribute_definition&>(col).type(rit->type());
-        }
-      }
-    }
+utils::result<statement, utils::error> connection::prepare(const query_context &ctx) {
+  auto result = perform_prepare(ctx);
+
+  if (!result) {
+    return utils::failure(result.err());
   }
 
-  if (auto result = connection_->prepare(ctx); result.is_ok()) {
-    return utils::ok(statement(result.release()));
-  }
-
-  return utils::ok(statement(std::unique_ptr<statement_impl>{}));
+  return utils::ok(statement(
+    std::make_shared<internal::connection_statement_proxy>(result.release()),
+    logger_)
+  );
 }
 
 std::string connection::str( const query_context& ctx ) const
@@ -218,4 +223,27 @@ const class dialect &connection::dialect() const
   return connection_->dialect();
 }
 
+utils::result<std::unique_ptr<statement_impl>, utils::error> connection::perform_prepare(const query_context& ctx) const {
+  if (ctx.command != sql_command::SQL_CREATE_TABLE && (ctx.prototype.empty() || has_unknown_columns(ctx.prototype))) {
+    if (const auto result = describe(ctx.table.name); result.is_ok()) {
+      for (auto &col: ctx.prototype) {
+        const auto rit = std::find_if(
+          std::begin(*result),
+          std::end(*result),
+          [&col](const auto &value) { return value.name() == col.name(); }
+        );
+        if (col.type() == utils::basic_type::type_null && rit != result->end()) {
+          const_cast<object::attribute_definition&>(col).type(rit->type());
+        }
+      }
+    }
+  }
+
+  auto result = connection_->prepare(ctx);
+  if (!result) {
+    return utils::failure(result.err());
+  }
+
+  return utils::ok(result.release());
+}
 }

source/orm/sql/connection_pool.cpp

@@ -0,0 +1,155 @@
+#include "matador/sql/connection_pool.hpp"
+
+#include <chrono>
+#include <thread>
+
+namespace matador::sql {
+identifiable_connection::identifiable_connection(const size_t id, connection conn)
+: id(id)
+, conn(std::move(conn)){}
+
+connection_ptr::~connection_ptr() {
+  pool_->release(connection_);
+}
+
+connection_ptr::connection_ptr(identifiable_connection* c, connection_pool* pool)
+: connection_(c)
+, pool_(pool) {}
+
+connection_ptr::connection_ptr(connection_ptr&& x) noexcept
+: connection_(x.connection_)
+, pool_(x.pool_) {
+  x.connection_ = nullptr;
+  x.pool_ = nullptr;
+}
+
+connection_ptr& connection_ptr::operator=(connection_ptr&& x) noexcept {
+  if (this == &x) {
+    return *this;
+  }
+
+  std::swap(connection_, x.connection_);
+  std::swap(pool_, x.pool_);
+
+  return *this;
+}
+
+connection* connection_ptr::operator->() const {
+  return &connection_->conn;
+}
+
+connection& connection_ptr::operator*() const {
+  return connection_->conn;
+}
+
+std::optional<size_t> connection_ptr::id() const {
+  if (connection_) {
+    return connection_->id;
+  }
+  return std::nullopt;
+}
+
+bool connection_ptr::valid() const {
+  return connection_ != nullptr;
+}
+
+connection_pool::connection_pool(const std::string& dns, size_t count)
+: info_(connection_info::parse(dns)) {
+  connection_repo_.reserve(count);
+  while (count) {
+    connection_repo_.emplace_back(count, connection{info_});
+    auto &conn = connection_repo_.back();
+    idle_connections_.emplace(conn.id, &conn);
+    // Todo: handle result
+    const auto result = conn.conn.open();
+    if (!result) {
+      throw std::runtime_error("Failed to open connection");
+    }
+    --count;
+  }
+}
+
+connection_ptr connection_pool::acquire() {
+  std::unique_lock lock(mutex_);
+  while (idle_connections_.empty()) {
+    cv.wait(lock);
+  }
+
+  return get_next_connection();
+}
+
+connection_ptr connection_pool::try_acquire() {
+  std::unique_lock lock(mutex_);
+  if (idle_connections_.empty()) {
+    return {nullptr, this};
+  }
+
+  return get_next_connection();
+}
+
+connection_ptr connection_pool::acquire(const size_t id) {
+  using namespace std::chrono_literals;
+  pointer next_connection{nullptr};
+  auto try_count{0};
+  std::unique_lock lock(mutex_);
+
+  do {
+    if (auto it = idle_connections_.find(id); it != idle_connections_.end()) {
+      next_connection = it->second;
+      auto node = idle_connections_.extract(it);
+      inuse_connections_.insert(std::move(node));
+    } else {
+      lock.unlock();
+      std::this_thread::sleep_for(100ms);
+      lock.lock();
+    }
+  } while(try_count++ < 5);
+
+  return {next_connection, this};
+}
+
+void connection_pool::release(identifiable_connection* c) {
+  if (c == nullptr) {
+    return;
+  }
+  std::unique_lock lock(mutex_);
+  if (const auto it = inuse_connections_.find(c->id); it != inuse_connections_.end()) {
+    auto node = inuse_connections_.extract(it);
+    idle_connections_.insert(std::move(node));
+  }
+}
+
+void connection_pool::release(connection_ptr& c) {
+  release(c.connection_);
+  c.connection_ = nullptr;
+}
+
+std::size_t connection_pool::size() const {
+  return connection_repo_.size();
+}
+
+std::size_t connection_pool::idle() const {
+  std::lock_guard guard(mutex_);
+  return idle_connections_.size();
+}
+
+std::size_t connection_pool::inuse() const {
+  std::lock_guard guard(mutex_);
+  return inuse_connections_.size();
+}
+
+const connection_info& connection_pool::info() const {
+  return info_;
+}
+
+connection_ptr connection_pool::get_next_connection() {
+  pointer next_connection{nullptr};
+    for (auto & [id, conn] : idle_connections_) {
+    next_connection = conn;
+    auto node = idle_connections_.extract(id);
+    inuse_connections_.insert(std::move(node));
+    break;
+  }
+  return {next_connection, this};
+}
+}

source/orm/sql/executor.cpp

@@ -1,5 +1,4 @@
 #include "matador/sql/executor.hpp"
 
 namespace matador::sql {
-executor::~executor() = default;
 }

source/orm/sql/interface/statement_proxy.cpp

@@ -0,0 +1,17 @@
+#include "matador/sql/interface/statement_proxy.hpp"
+
+namespace matador::sql {
+statement_proxy::statement_proxy(std::unique_ptr<statement_impl>&& stmt)
+: statement_(std::move(stmt)){}
+
+void statement_proxy::bind(const size_t pos, const char* value, const size_t size) const {
+  statement_->bind(pos, value, size);
+}
+void statement_proxy::bind(const size_t pos, std::string& val, const size_t size) const {
+  statement_->bind(pos, val, size);
+}
+
+void statement_proxy::reset() const {
+  statement_->reset();
+}
+}

source/orm/sql/statement.cpp

@@ -3,30 +3,28 @@
 #include "matador/sql/field.hpp"
 
 #include <algorithm>
+#include <utility>
 
 namespace matador::sql {
 
-statement::statement(std::unique_ptr<statement_impl> impl, const std::shared_ptr<abstract_sql_logger> &logger)
-: statement_(std::move(impl))
-, logger_(logger)
-{}
+statement::statement(const std::shared_ptr<statement_proxy>& proxy, logger_ptr  logger)
+: statement_proxy_(proxy)
+, logger_(std::move(logger)){}
 
-statement &statement::bind(const size_t pos, const char *value)
-{
-  statement_->bind(pos, value, 0);
+statement &statement::bind(const size_t pos, const char *value) {
+  statement_proxy_->bind(pos, value, strlen(value));
   return *this;
 }
 
 statement &statement::bind(const size_t pos, std::string &val, const size_t size)
 {
-  statement_->bind(pos, val, size);
+  statement_proxy_->bind(pos, val, size);
   return *this;
 }
 
-utils::result<size_t, utils::error> statement::execute() const
-{
+utils::result<size_t, utils::error> statement::execute() const {
 //  logger_.info(statement_->query_.sql);
-  return statement_->execute();
+  return statement_proxy_->execute();
 }
 
 //bool is_unknown(const std::vector<object::column_definition> &columns) {
@@ -35,13 +33,13 @@ utils::result<size_t, utils::error> statement::execute() const
 //  });
 //}
 
-utils::result<query_result<record>, utils::error> statement::fetch() const
-{
+utils::result<query_result<record>, utils::error> statement::fetch() const {
 //  if (is_unknown(statement_->query_.prototype)) {
 //
 //  }
 
-  auto result = statement_->fetch();
+
+  auto result = statement_proxy_->fetch();
   if (!result.is_ok()) {
     return utils::failure(result.err());
   }
@@ -51,7 +49,7 @@ utils::result<query_result<record>, utils::error> statement::fetch() const
 
 utils::result<std::optional<record>, utils::error> statement::fetch_one() const {
     //  logger_.info(statement_->query_.sql);
-    auto result = statement_->fetch();
+    auto result = statement_proxy_->fetch();
     if (!result.is_ok()) {
         return utils::failure(result.err());
     }
@@ -67,7 +65,7 @@ utils::result<std::optional<record>, utils::error> statement::fetch_one() const
 
 void statement::reset() const
 {
-  statement_->reset();
+  statement_proxy_->reset();
 }
 
 }

source/orm/sql/statement_cache.cpp

@@ -0,0 +1,72 @@
+#include "matador/sql/statement_cache.hpp"
+#include "matador/sql/backend_provider.hpp"
+#include "matador/sql/error_code.hpp"
+#include "matador/sql/connection_pool.hpp"
+
+namespace matador::sql {
+namespace internal {
+class statement_cache_proxy final : public statement_proxy {
+public:
+  explicit statement_cache_proxy(std::unique_ptr<statement_impl>&& stmt)
+  : statement_proxy(std::move(stmt)) {}
+
+  utils::result<size_t, utils::error> execute() override {
+    return statement_->execute();
+  }
+  utils::result<std::unique_ptr<query_result_impl>, utils::error> fetch() override {
+    return statement_->fetch();
+  }
+};
+
+}
+statement_cache::statement_cache(connection_pool &pool, const size_t max_size)
+: max_size_(max_size)
+, pool_(pool)
+, dialect_(backend_provider::instance().connection_dialect(pool_.info().type)) {}
+
+utils::result<statement, utils::error> statement_cache::acquire(const query_context& ctx) {
+  std::unique_lock lock(mutex_);
+  // hash statement
+  const auto key = std::hash<std::string>{}(ctx.sql);
+  // Found in cache. Move it to of the LRU list
+  if (const auto it = cache_map_.find(key); it != cache_map_.end()) {
+    usage_list_.splice(usage_list_.begin(), usage_list_, it->second.second);
+    return utils::ok(it->second.first);
+  }
+  // Prepare a new statement
+  // acquire pool connection
+  const auto conn = pool_.acquire();
+  auto result = conn->perform_prepare(ctx);
+  if (!result) {
+    return utils::failure(utils::error{error_code::PREPARE_FAILED, std::string("Failed to prepare")});
+  }
+
+  // If cache max size reached ensure space
+  if (cache_map_.size() >= max_size_) {
+    const size_t& lru_key = usage_list_.back();
+    cache_map_.erase(lru_key);
+    usage_list_.pop_back();
+  }
+
+  usage_list_.push_front(key);
+  const auto it = cache_map_.insert({
+    key,
+    std::make_pair(statement{
+      std::make_shared<internal::statement_cache_proxy>(result.release())},
+      usage_list_.begin())
+  }).first;
+
+  return utils::ok(it->second.first);
+}
+
+size_t statement_cache::size() const {
+  return cache_map_.size();
+}
+size_t statement_cache::capacity() const {
+  return max_size_;
+}
+
+bool statement_cache::empty() const {
+  return cache_map_.empty();
+}
+}