Pool: bring in enhancements from Postgres

.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)

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

src/include/dbconnector/pool/acquire_mode.hpp

@@ -0,0 +1,45 @@
+#pragma once
+
+#include <cctype>
+#include <algorithm>
+#include <string>
+
+#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<char>(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<int>(mode)));
+		}
+	}
+};
+
+} // namespace pool
+} // namespace dbconnector

src/include/dbconnector/pool/connection_pool.hpp

@@ -8,6 +8,7 @@
 #include <mutex>
 #include <thread>
 
+#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,13 +23,18 @@ class ConnectionPool : public std::enable_shared_from_this<ConnectionPool<Connec
 	ConnectionPool(ConnectionPoolConfig config = ConnectionPoolConfig());
 	virtual ~ConnectionPool();
 
+	PooledConnection<ConnectionT> Acquire();
 	PooledConnection<ConnectionT> WaitAcquire();
 	PooledConnection<ConnectionT> TryAcquire();
 	PooledConnection<ConnectionT> ForceAcquire();
 
 	bool IsShutdown() const;
 	void Shutdown();
 
+	AcquireMode GetAcquireMode() const;
+	void SetAcquireMode(AcquireMode mode);
+	void SetAcquireMode(const std::string &mode_str);
+
 	uint64_t GetMaxConnections() const;
 	void SetMaxConnections(uint64_t new_max);
 	uint64_t GetWaitTimeoutMillis() const;
@@ -37,6 +43,10 @@ class ConnectionPool : public std::enable_shared_from_this<ConnectionPool<Connec
 	uint64_t GetAvailableConnections() const;
 	uint64_t GetTotalConnections() const;
 
+	uint64_t GetCacheHits() const;
+	uint64_t GetCacheMisses() const;
+	uint64_t GetTryFailures() const;
+
 	bool IsThreadLocalCacheEnabled() const;
 	void SetThreadLocalCacheEnabled(bool enabled);
 	uint64_t GetThreadLocalCacheHits() const;
@@ -47,9 +57,10 @@ class ConnectionPool : public std::enable_shared_from_this<ConnectionPool<Connec
 	uint64_t GetIdleTimeoutMillis() const;
 	void SetIdleTimeoutMillis(uint64_t new_idle_timeout_millis);
 
-	bool IsReaperRunning();
+	bool IsReaperRunning() const;
 	bool EnsureReaperRunning();
 	void ShutdownReaper();
+	uint64_t GetReaperPeriodMillis() const;
 
 protected:
 	virtual std::unique_ptr<ConnectionT> CreateNewConnection() = 0;
@@ -85,28 +96,34 @@ class ConnectionPool : public std::enable_shared_from_this<ConnectionPool<Connec
 	bool CheckConnectionNotExpiredAndHealthy(std::unique_lock<std::mutex> &lock,
 	                                         CachedConnection<ConnectionT> &cached_conn,
 	                                         std::chrono::steady_clock::time_point now);
-	uint64_t CalcReaperSleepSeconds();
+	uint64_t CalcReaperSleepSeconds() const;
 	void ReaperLoop();
 	void ShutdownReaperInternal(std::unique_lock<std::mutex> &lock);
 
 	CachedConnection<ConnectionT> TryAcquireFromThreadLocal(std::chrono::steady_clock::time_point now);
 	bool TryReturnToThreadLocal(std::unique_ptr<ConnectionT> &conn, std::chrono::steady_clock::time_point created_at,
 	                            std::chrono::steady_clock::time_point returned_at);
 	void ReturnFromThreadLocalCache(CachedConnection<ConnectionT> cached_conn);
-
 	void DecrementTotalConnections();
-
-	std::atomic<uint64_t> max_connections {0};
-	std::atomic<uint64_t> wait_timeout_millis {0};
+	bool IsPoolDisabled();
 
 	mutable std::mutex pool_lock;
 	std::condition_variable pool_cv;
 
+	std::atomic<AcquireMode> acquire_mode {AcquireMode::FORCE};
+
+	std::atomic<uint64_t> max_connections {0};
+	std::atomic<uint64_t> wait_timeout_millis {0};
+
 	std::deque<CachedConnection<ConnectionT>> available;
 	std::atomic<uint64_t> available_connections {0};
 	std::atomic<uint64_t> total_connections {0};
 	std::atomic<bool> shutdown_flag {false};
 
+	std::atomic<uint64_t> cache_hits {0};
+	std::atomic<uint64_t> cache_misses {0};
+	std::atomic<uint64_t> try_failures {0};
+
 	std::atomic<uint64_t> max_lifetime_millis {0};
 	std::atomic<uint64_t> idle_timeout_millis {0};
 

src/include/dbconnector/pool/connection_pool_config.hpp

@@ -2,16 +2,28 @@
 
 #include <cstdint>
 
+#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<uint64_t>(std::thread::hardware_concurrency());
+		uint64_t size = static_cast<uint64_t>(detected * hardware_concurrency_coef);
+		size = (std::max<uint64_t>)(min_connections, size);
+		size = (std::min<uint64_t>)(max_connections, size);
+		return size;
+	}
 };
 
 } // namespace pool