diff --git a/.github/workflows/ConnectorTests.yml b/.github/workflows/ConnectorTests.yml index a87d0dc..f8b52cd 100644 --- a/.github/workflows/ConnectorTests.yml +++ b/.github/workflows/ConnectorTests.yml @@ -50,7 +50,7 @@ jobs: cd build cmake ../test cmake --build . --config Debug - test_database_connection.exe + test_database_connector.exe macos-tests-aarch64: name: macOS Tests (aarch64) diff --git a/Makefile b/Makefile index c7930d7..be6027c 100644 --- a/Makefile +++ b/Makefile @@ -31,7 +31,7 @@ build_threads: cmake --build . --config Debug test: build - build/debug/test_database_connection + build/debug/test_database_connector test_threads: build_threads - build/debug/test_database_connection + build/debug/test_database_connector diff --git a/src/include/dbconnector/pool/acquire_mode.hpp b/src/include/dbconnector/pool/acquire_mode.hpp new file mode 100644 index 0000000..04db004 --- /dev/null +++ b/src/include/dbconnector/pool/acquire_mode.hpp @@ -0,0 +1,45 @@ +#pragma once + +#include +#include +#include + +#include "dbconnector/pool/pool_exception.hpp" + +namespace dbconnector { +namespace pool { + +enum class AcquireMode { FORCE, WAIT, TRY }; + +struct AcquireModeHelpers { + static AcquireMode FromString(const std::string &mode_str) { + std::string ms(mode_str.data(), mode_str.length()); + std::transform(ms.begin(), ms.end(), ms.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + if (ms == "force") { + return AcquireMode::FORCE; + } else if (ms == "wait") { + return AcquireMode::WAIT; + } else if (ms == "try") { + return AcquireMode::TRY; + } else { + throw PoolException("Invalid unsupported acquire mode: '" + mode_str + "'"); + } + } + + std::string ToString(AcquireMode mode) { + switch (mode) { + case AcquireMode::FORCE: + return "force"; + case AcquireMode::WAIT: + return "wait"; + case AcquireMode::TRY: + return "try"; + default: + throw PoolException("Invalid unsupported acquire mode: d" + std::to_string(static_cast(mode))); + } + } +}; + +} // namespace pool +} // namespace dbconnector diff --git a/src/include/dbconnector/pool/connection_pool.hpp b/src/include/dbconnector/pool/connection_pool.hpp index ea0cf71..626368a 100644 --- a/src/include/dbconnector/pool/connection_pool.hpp +++ b/src/include/dbconnector/pool/connection_pool.hpp @@ -8,6 +8,7 @@ #include #include +#include "dbconnector/pool/acquire_mode.hpp" #include "dbconnector/pool/cached_connection.hpp" #include "dbconnector/pool/connection_pool_config.hpp" #include "dbconnector/pool/pooled_connection.hpp" @@ -22,6 +23,7 @@ class ConnectionPool : public std::enable_shared_from_this Acquire(); PooledConnection WaitAcquire(); PooledConnection TryAcquire(); PooledConnection ForceAcquire(); @@ -29,6 +31,10 @@ class ConnectionPool : public std::enable_shared_from_this CreateNewConnection() = 0; @@ -85,7 +96,7 @@ class ConnectionPool : public std::enable_shared_from_this &lock, CachedConnection &cached_conn, std::chrono::steady_clock::time_point now); - uint64_t CalcReaperSleepSeconds(); + uint64_t CalcReaperSleepSeconds() const; void ReaperLoop(); void ShutdownReaperInternal(std::unique_lock &lock); @@ -93,20 +104,26 @@ class ConnectionPool : public std::enable_shared_from_this &conn, std::chrono::steady_clock::time_point created_at, std::chrono::steady_clock::time_point returned_at); void ReturnFromThreadLocalCache(CachedConnection cached_conn); - void DecrementTotalConnections(); - - std::atomic max_connections {0}; - std::atomic wait_timeout_millis {0}; + bool IsPoolDisabled(); mutable std::mutex pool_lock; std::condition_variable pool_cv; + std::atomic acquire_mode {AcquireMode::FORCE}; + + std::atomic max_connections {0}; + std::atomic wait_timeout_millis {0}; + std::deque> available; std::atomic available_connections {0}; std::atomic total_connections {0}; std::atomic shutdown_flag {false}; + std::atomic cache_hits {0}; + std::atomic cache_misses {0}; + std::atomic try_failures {0}; + std::atomic max_lifetime_millis {0}; std::atomic idle_timeout_millis {0}; diff --git a/src/include/dbconnector/pool/connection_pool_config.hpp b/src/include/dbconnector/pool/connection_pool_config.hpp index 534c8c9..90ffa6b 100644 --- a/src/include/dbconnector/pool/connection_pool_config.hpp +++ b/src/include/dbconnector/pool/connection_pool_config.hpp @@ -2,16 +2,28 @@ #include +#include "dbconnector/pool/acquire_mode.hpp" + namespace dbconnector { namespace pool { struct ConnectionPoolConfig { - uint64_t max_connections = 4; + AcquireMode acquire_mode = AcquireMode::FORCE; + uint64_t max_connections = DefaultPoolSizeFromHardwareConcurrency(4, 32, 1.5); uint64_t wait_timeout_millis = 30000; - bool tl_cache_enabled = true; + bool tl_cache_enabled = false; uint64_t max_lifetime_millis = 0; - uint64_t idle_timeout_millis = 0; - bool start_reaper_thread = false; + uint64_t idle_timeout_millis = 60000; + bool start_reaper_thread = true; + + static uint64_t DefaultPoolSizeFromHardwareConcurrency(uint64_t min_connections, uint64_t max_connections, + float hardware_concurrency_coef) { + uint64_t detected = static_cast(std::thread::hardware_concurrency()); + uint64_t size = static_cast(detected * hardware_concurrency_coef); + size = (std::max)(min_connections, size); + size = (std::min)(max_connections, size); + return size; + } }; } // namespace pool diff --git a/src/include/dbconnector/pool/connection_pool_impl.hpp b/src/include/dbconnector/pool/connection_pool_impl.hpp index 3e95bd8..94a20ae 100644 --- a/src/include/dbconnector/pool/connection_pool_impl.hpp +++ b/src/include/dbconnector/pool/connection_pool_impl.hpp @@ -15,9 +15,9 @@ namespace pool { template ConnectionPool::ConnectionPool(ConnectionPoolConfig config) - : max_connections(config.max_connections), wait_timeout_millis(config.wait_timeout_millis), - max_lifetime_millis(config.max_lifetime_millis), idle_timeout_millis(config.idle_timeout_millis), - tl_cache_enabled(config.tl_cache_enabled) { + : acquire_mode(config.acquire_mode), max_connections(config.max_connections), + wait_timeout_millis(config.wait_timeout_millis), max_lifetime_millis(config.max_lifetime_millis), + idle_timeout_millis(config.idle_timeout_millis), tl_cache_enabled(config.tl_cache_enabled) { if (config.start_reaper_thread) { EnsureReaperRunning(); } @@ -132,7 +132,8 @@ void ConnectionPool::ReturnFromThreadLocalCache(CachedConnection lock(pool_lock); - if (expired || shutdown_flag.load(std::memory_order_relaxed)) { + if (expired || shutdown_flag.load(std::memory_order_relaxed) || + total_connections.load(std::memory_order_relaxed) > max_connections.load(std::memory_order_relaxed)) { DecrementTotalConnections(); return; } @@ -142,11 +143,37 @@ void ConnectionPool::ReturnFromThreadLocalCache(CachedConnection +PooledConnection ConnectionPool::Acquire() { + bool pool_disabled = false; + AcquireMode mode = AcquireMode::FORCE; + { + std::lock_guard lock(pool_lock); + pool_disabled = IsPoolDisabled(); + mode = GetAcquireMode(); + } + + if (pool_disabled) { + return ForceAcquire(); + } + + switch (mode) { + case AcquireMode::FORCE: + return ForceAcquire(); + case AcquireMode::WAIT: + return WaitAcquire(); + case AcquireMode::TRY: + return TryAcquire(); + default: + throw PoolException("Invalid unsupported acquire mode: " + std::to_string(static_cast(mode))); + } +} + template PooledConnection ConnectionPool::WaitAcquire() { { std::lock_guard lock(pool_lock); - if (max_connections.load(std::memory_order_relaxed) == 0) { + if (IsPoolDisabled()) { throw PoolException("Connection pool is disabled (pool_size=0). Use " "the force acquire mode to create connections without pooling."); } @@ -155,6 +182,7 @@ PooledConnection ConnectionPool::WaitAcquire() { auto now = GetNowForTimeoutPurposes(); auto tl_conn = TryAcquireFromThreadLocal(now); if (tl_conn) { + cache_hits.fetch_add(1, std::memory_order_relaxed); return PooledConnection(this->shared_from_this(), std::move(tl_conn)); } @@ -182,6 +210,7 @@ PooledConnection ConnectionPool::WaitAcquire() { } if (healthy) { + cache_hits.fetch_add(1, std::memory_order_relaxed); return PooledConnection(this->shared_from_this(), std::move(cached_conn)); } DecrementTotalConnections(); @@ -192,6 +221,7 @@ PooledConnection ConnectionPool::WaitAcquire() { lock.unlock(); try { + cache_misses.fetch_add(1, std::memory_order_relaxed); auto conn = CreateNewConnection(); return PooledConnection(this->shared_from_this(), std::move(conn), now); } catch (...) { @@ -216,7 +246,7 @@ template PooledConnection ConnectionPool::TryAcquire() { { std::lock_guard lock(pool_lock); - if (max_connections.load(std::memory_order_relaxed) == 0) { + if (IsPoolDisabled()) { throw PoolException("Connection pool is disabled (pool_size=0). Use " "the force acquire mode to create connections without pooling."); } @@ -226,6 +256,7 @@ PooledConnection ConnectionPool::TryAcquire() { auto tl_conn = TryAcquireFromThreadLocal(now); if (tl_conn) { + cache_hits.fetch_add(1, std::memory_order_relaxed); return PooledConnection(this->shared_from_this(), std::move(tl_conn)); } @@ -248,6 +279,7 @@ PooledConnection ConnectionPool::TryAcquire() { } if (healthy) { + cache_hits.fetch_add(1, std::memory_order_relaxed); return PooledConnection(this->shared_from_this(), std::move(cached_conn)); } DecrementTotalConnections(); @@ -258,6 +290,7 @@ PooledConnection ConnectionPool::TryAcquire() { lock.unlock(); try { + cache_misses.fetch_add(1, std::memory_order_relaxed); auto conn = CreateNewConnection(); return PooledConnection(this->shared_from_this(), std::move(conn), now); } catch (...) { @@ -268,6 +301,7 @@ PooledConnection ConnectionPool::TryAcquire() { } } + try_failures.fetch_add(1, std::memory_order_relaxed); return PooledConnection(); } @@ -279,6 +313,7 @@ PooledConnection ConnectionPool::ForceAcquire() { // by setting max_conn = 0. auto tl_conn = TryAcquireFromThreadLocal(now); if (tl_conn) { + cache_hits.fetch_add(1, std::memory_order_relaxed); return PooledConnection(this->shared_from_this(), std::move(tl_conn)); } @@ -288,10 +323,11 @@ PooledConnection ConnectionPool::ForceAcquire() { if (shutdown_flag.load(std::memory_order_relaxed)) { throw PoolException("Connection pool has been shut down"); } - pooling_disabled = (max_connections.load(std::memory_order_relaxed) == 0); + pooling_disabled = IsPoolDisabled(); } if (pooling_disabled) { + cache_misses.fetch_add(1, std::memory_order_relaxed); auto conn = CreateNewConnection(); return PooledConnection(nullptr, std::move(conn), now); } @@ -316,6 +352,7 @@ PooledConnection ConnectionPool::ForceAcquire() { } if (healthy) { + cache_hits.fetch_add(1, std::memory_order_relaxed); return PooledConnection(this->shared_from_this(), std::move(cached_conn)); } DecrementTotalConnections(); @@ -325,6 +362,7 @@ PooledConnection ConnectionPool::ForceAcquire() { } try { + cache_misses.fetch_add(1, std::memory_order_relaxed); auto conn = CreateNewConnection(); return PooledConnection(this->shared_from_this(), std::move(conn), now); } catch (...) { @@ -393,6 +431,23 @@ void ConnectionPool::Discard() { pool_cv.notify_one(); } +template +AcquireMode ConnectionPool::GetAcquireMode() const { + return acquire_mode.load(std::memory_order_relaxed); +} + +template +void ConnectionPool::SetAcquireMode(AcquireMode mode) { + std::lock_guard lock(pool_lock); + acquire_mode.store(mode, std::memory_order_relaxed); +} + +template +void ConnectionPool::SetAcquireMode(const std::string &mode_str) { + AcquireMode mode = AcquireModeHelpers::FromString(mode_str); + SetAcquireMode(mode); +} + template uint64_t ConnectionPool::GetMaxConnections() const { return max_connections.load(std::memory_order_relaxed); @@ -436,6 +491,21 @@ uint64_t ConnectionPool::GetTotalConnections() const { return total_connections.load(std::memory_order_relaxed); } +template +uint64_t ConnectionPool::GetCacheHits() const { + return cache_hits.load(std::memory_order_relaxed); +} + +template +uint64_t ConnectionPool::GetCacheMisses() const { + return cache_misses.load(std::memory_order_relaxed); +} + +template +uint64_t ConnectionPool::GetTryFailures() const { + return try_failures.load(std::memory_order_relaxed); +} + template void ConnectionPool::SetMaxLifetimeMillis(uint64_t new_max_lifetime_millis) { this->max_lifetime_millis.store(new_max_lifetime_millis, std::memory_order_relaxed); @@ -529,7 +599,7 @@ bool ConnectionPool::CheckConnectionNotExpiredAndHealthy(std::uniqu } template -uint64_t ConnectionPool::CalcReaperSleepSeconds() { +uint64_t ConnectionPool::CalcReaperSleepSeconds() const { uint64_t sleep_seconds = 30; uint64_t max_lifetime_seconds = max_lifetime_millis.load(std::memory_order_relaxed) / 1000; uint64_t idle_timeout_seconds = idle_timeout_millis.load(std::memory_order_relaxed) / 1000; @@ -591,7 +661,7 @@ void ConnectionPool::ReaperLoop() { } template -bool ConnectionPool::IsReaperRunning() { +bool ConnectionPool::IsReaperRunning() const { return !reaper_shutdown_flag.load(std::memory_order_relaxed); } @@ -651,6 +721,11 @@ void ConnectionPool::ShutdownReaper() { ShutdownReaperInternal(lock); } +template +uint64_t ConnectionPool::GetReaperPeriodMillis() const { + return CalcReaperSleepSeconds() * 1000; +} + template uint64_t ConnectionPool::GetThreadLocalCacheHits() const { return tl_cache_hits.load(std::memory_order_relaxed); @@ -689,5 +764,10 @@ void ConnectionPool::DecrementTotalConnections() { } } +template +bool ConnectionPool::IsPoolDisabled() { + return GetMaxConnections() == 0; +} + } // namespace pool } // namespace dbconnector diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index fa111a2..ab31b69 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -1,6 +1,6 @@ cmake_minimum_required(VERSION 3.5...3.29) -project (database_connection_test C CXX) +project (test_database_connector C CXX) if(MSVC) set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}) @@ -10,14 +10,14 @@ endif() option(ENABLE_THREAD_SANITIZER "Enable thread sanitizer." FALSE) -add_executable(test_database_connection +add_executable(${PROJECT_NAME} test_common.cpp test_defer.cpp test_make_unique.cpp test_pool.cpp ) -target_include_directories(test_database_connection PRIVATE +target_include_directories(${PROJECT_NAME} PRIVATE ../src/include catch # duckdb_capi @@ -32,22 +32,22 @@ else() endif() if(NOT MSVC) - target_compile_options(test_database_connection PRIVATE + target_compile_options(${PROJECT_NAME} PRIVATE -O0 -Wall -Werror -Wextra -fsanitize=${SANITIZER_TYPE} ) - target_link_options(test_database_connection PRIVATE + target_link_options(${PROJECT_NAME} PRIVATE -fsanitize=${SANITIZER_TYPE} ) else() - target_compile_options(test_database_connection PRIVATE + target_compile_options(${PROJECT_NAME} PRIVATE /W4 /WX ) - target_compile_definitions(test_database_connection PRIVATE + target_compile_definitions(${PROJECT_NAME} PRIVATE -D_CRT_SECURE_NO_WARNINGS ) endif() diff --git a/test/test_pool.cpp b/test/test_pool.cpp index 8b646f7..92ec936 100644 --- a/test/test_pool.cpp +++ b/test/test_pool.cpp @@ -52,12 +52,15 @@ class TestConnectionPool : public dbconnector::pool::ConnectionPool(); + REQUIRE(pool->GetAcquireMode() == dbconnector::pool::AcquireMode::FORCE); REQUIRE(pool->WaitAcquire()); REQUIRE(pool->TryAcquire()); REQUIRE(pool->ForceAcquire()); @@ -74,6 +77,34 @@ TEST_CASE("Test connection pool basic", group_name) { REQUIRE(pool->GetMaxLifetimeMillis() == 44); pool->SetIdleTimeoutMillis(45); REQUIRE(pool->GetIdleTimeoutMillis() == 45); + + REQUIRE(pool->GetCacheHits() == 2); + REQUIRE(pool->GetCacheMisses() == 1); + REQUIRE(pool->GetTryFailures() == 0); +} + +TEST_CASE("Test connection pool acquire mode", group_name) { + auto pool = std::make_shared(1); + REQUIRE(pool->GetAcquireMode() == dbconnector::pool::AcquireMode::FORCE); + REQUIRE(pool->GetMaxConnections() == 1); + auto conn1 = pool->Acquire(); + REQUIRE(conn1); + REQUIRE(pool->GetMaxConnections() == 1); + { + REQUIRE(pool->GetAvailableConnections() == 0); + auto conn2 = pool->Acquire(); + REQUIRE(conn2); + REQUIRE(pool->GetMaxConnections() == 1); + REQUIRE(pool->GetAvailableConnections() == 0); + } + pool->SetAcquireMode("try"); + REQUIRE(!pool->Acquire()); + REQUIRE(pool->GetMaxConnections() == 1); + REQUIRE(pool->GetAvailableConnections() == 0); + + REQUIRE(pool->GetCacheHits() == 0); + REQUIRE(pool->GetCacheMisses() == 2); + REQUIRE(pool->GetTryFailures() == 1); } TEST_CASE("Test pool size no thread-local", group_name) { @@ -108,6 +139,10 @@ TEST_CASE("Test pool size no thread-local", group_name) { timeout_thrown = true; } REQUIRE(timeout_thrown); + + REQUIRE(pool->GetCacheHits() == 1); + REQUIRE(pool->GetCacheMisses() == 2); + REQUIRE(pool->GetTryFailures() == 0); } TEST_CASE("Test pool size with thread-local", group_name) { @@ -144,6 +179,10 @@ TEST_CASE("Test pool size with thread-local", group_name) { REQUIRE(3 == pool->GetTotalConnections()); } REQUIRE(2 == pool->GetTotalConnections()); + + REQUIRE(pool->GetCacheHits() == 1); + REQUIRE(pool->GetCacheMisses() == 3); + REQUIRE(pool->GetTryFailures() == 0); } TEST_CASE("Test pool disabled", group_name) { @@ -168,6 +207,10 @@ TEST_CASE("Test pool disabled", group_name) { REQUIRE(pool->GetTotalConnections() == 0); } REQUIRE(pool->GetTotalConnections() == 0); + + REQUIRE(pool->GetCacheHits() == 0); + REQUIRE(pool->GetCacheMisses() == 2); + REQUIRE(pool->GetTryFailures() == 0); } TEST_CASE("Test pool disable running", group_name) { @@ -228,6 +271,10 @@ TEST_CASE("Test pool disable running", group_name) { REQUIRE(pool->GetTotalConnections() == 0); } REQUIRE(pool->GetTotalConnections() == 0); + + REQUIRE(pool->GetCacheHits() == 1); + REQUIRE(pool->GetCacheMisses() == 5); + REQUIRE(pool->GetTryFailures() == 0); } TEST_CASE("Test pool with a reaper", group_name) { @@ -238,6 +285,7 @@ TEST_CASE("Test pool with a reaper", group_name) { REQUIRE(pool->EnsureReaperRunning()); REQUIRE(pool->EnsureReaperRunning()); REQUIRE(pool->IsReaperRunning()); + REQUIRE(pool->GetReaperPeriodMillis() == 1000); { auto conn = pool->WaitAcquire(); @@ -267,6 +315,10 @@ TEST_CASE("Test pool with a reaper", group_name) { std::this_thread::sleep_for(std::chrono::milliseconds(1500)); REQUIRE(0 == pool->GetTotalConnections()); + + REQUIRE(pool->GetCacheHits() == 0); + REQUIRE(pool->GetCacheMisses() == 2); + REQUIRE(pool->GetTryFailures() == 0); } TEST_CASE("Test pool with a reaper restart", group_name) { @@ -299,4 +351,8 @@ TEST_CASE("Test pool with a reaper restart", group_name) { std::this_thread::sleep_for(std::chrono::milliseconds(1500)); REQUIRE(0 == pool->GetTotalConnections()); + + REQUIRE(pool->GetCacheHits() == 0); + REQUIRE(pool->GetCacheMisses() == 1); + REQUIRE(pool->GetTryFailures() == 0); }