Add AttachmentPool

Author: asfernandes

src/fb-cpp/AttachmentPool.cpp

@@ -0,0 +1,339 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2026 Adriano dos Santos Fernandes
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+#include "AttachmentPool.h"
+#include "Client.h"
+#include "Exception.h"
+#include <algorithm>
+#include <cassert>
+#include <exception>
+#include <string_view>
+#include <utility>
+
+using namespace fbcpp;
+using namespace fbcpp::impl;
+
+
+static std::chrono::steady_clock::time_point steadyNow()
+{
+	return std::chrono::steady_clock::now();
+}
+
+static std::unique_ptr<AttachmentPoolEntry> extractEntry(
+	std::vector<std::unique_ptr<AttachmentPoolEntry>>& entries, AttachmentPoolEntry* entry)
+{
+	const auto it = std::find_if(
+		entries.begin(), entries.end(), [entry](const auto& candidate) { return candidate.get() == entry; });
+
+	if (it == entries.end())
+		return nullptr;
+
+	auto owned = std::move(*it);
+	entries.erase(it);
+
+	return owned;
+}
+
+
+AttachmentPool::AttachmentPool(Client& client_, std::string uri_, const AttachmentPoolOptions& options)
+	: client{&client_},
+	  uri{std::move(uri_)},
+	  options{options}
+{
+	if (options.getAttachmentOptions().getCreateDatabase())
+	{
+		throw FbCppException(
+			"AttachmentPoolOptions::setAttachmentOptions must not enable AttachmentOptions::setCreateDatabase");
+	}
+
+	if (options.getMaxSize() < 1u)
+		throw FbCppException("AttachmentPoolOptions::setMaxSize must be at least 1");
+
+	if (options.getMinSize() > options.getMaxSize())
+		throw FbCppException("AttachmentPoolOptions::setMinSize must not be greater than setMaxSize");
+
+	for (std::size_t i = 0u; i < options.getMinSize(); ++i)
+	{
+		auto entry = std::make_unique<AttachmentPoolEntry>();
+		entry->attachment = std::make_unique<Attachment>(*client, uri, options.getAttachmentOptions());
+		entry->createdAt = entry->lastReturnedAt = steadyNow();
+
+		available.push_back(entry.get());
+		entries.push_back(std::move(entry));
+	}
+}
+
+AttachmentPool::~AttachmentPool() noexcept
+{
+	assert(inUse == 0u && pending == 0u);
+
+	try
+	{
+		close();
+	}
+	catch (...)
+	{
+		// swallow
+	}
+}
+
+std::size_t AttachmentPool::size()
+{
+	std::lock_guard mutexGuard{mutex};
+	return entries.size();
+}
+
+std::size_t AttachmentPool::availableCount()
+{
+	std::lock_guard mutexGuard{mutex};
+	return available.size();
+}
+
+std::size_t AttachmentPool::inUseCount()
+{
+	std::lock_guard mutexGuard{mutex};
+	return inUse;
+}
+
+PooledAttachment AttachmentPool::acquire()
+{
+	auto lease = acquireImpl(true);
+	assert(lease.has_value());
+	return std::move(*lease);
+}
+
+std::optional<PooledAttachment> AttachmentPool::tryAcquire()
+{
+	return acquireImpl(false);
+}
+
+void AttachmentPool::close()
+{
+	std::unique_lock mutexGuard{mutex};
+	closed = true;
+
+	while (!available.empty())
+	{
+		auto* entry = available.front();
+		available.pop_front();
+
+		auto owned = extractEntry(entries, entry);
+
+		mutexGuard.unlock();
+		owned.reset();  // Disconnects (network I/O) outside the lock.
+		mutexGuard.lock();
+	}
+}
+
+std::optional<PooledAttachment> AttachmentPool::acquireImpl(bool throwOnFailure)
+{
+	std::unique_lock mutexGuard{mutex};
+
+	if (closed)
+	{
+		if (throwOnFailure)
+			throw FbCppException("AttachmentPool is closed");
+
+		return std::nullopt;
+	}
+
+	const auto deadline = steadyNow() + options.getAcquireTimeout();
+
+	while (true)
+	{
+		evictLocked(mutexGuard);
+
+		if (!available.empty())
+		{
+			auto* entry = available.front();
+			available.pop_front();
+			++inUse;
+
+			if (options.getValidateOnAcquire())
+			{
+				mutexGuard.unlock();
+
+				bool valid = true;
+
+				try
+				{
+					entry->attachment->ping();
+				}
+				catch (...)
+				{
+					valid = false;
+				}
+
+				mutexGuard.lock();
+
+				if (!valid)
+				{
+					--inUse;
+					auto owned = extractEntry(entries, entry);
+
+					mutexGuard.unlock();
+					owned.reset();  // Disconnects (network I/O) outside the lock.
+					mutexGuard.lock();
+
+					continue;
+				}
+			}
+
+			return PooledAttachment{*this, *entry};
+		}
+
+		if (entries.size() + pending < options.getMaxSize())
+		{
+			++pending;
+			mutexGuard.unlock();
+
+			std::unique_ptr<AttachmentPoolEntry> newEntry;
+			std::exception_ptr failure;
+
+			try
+			{
+				newEntry = std::make_unique<AttachmentPoolEntry>();
+				newEntry->attachment = std::make_unique<Attachment>(*client, uri, options.getAttachmentOptions());
+				newEntry->createdAt = newEntry->lastReturnedAt = steadyNow();
+			}
+			catch (...)
+			{
+				failure = std::current_exception();
+			}
+
+			mutexGuard.lock();
+			--pending;
+
+			if (failure)
+			{
+				availableCondition.notify_one();  // Let another waiter try instead.
+
+				if (throwOnFailure)
+					std::rethrow_exception(failure);
+
+				return std::nullopt;
+			}
+
+			auto* entryPtr = newEntry.get();
+			entries.push_back(std::move(newEntry));
+			++inUse;
+
+			return PooledAttachment{*this, *entryPtr};
+		}
+
+		const auto remaining = deadline - steadyNow();
+
+		if (remaining <= std::chrono::steady_clock::duration::zero() ||
+			availableCondition.wait_for(mutexGuard, remaining) == std::cv_status::timeout)
+		{
+			if (throwOnFailure)
+				throw FbCppException("Timed out waiting for an available connection from AttachmentPool");
+
+			return std::nullopt;
+		}
+	}
+}
+
+void AttachmentPool::evictLocked(std::unique_lock<std::mutex>& lock)
+{
+	assert(lock.owns_lock());
+
+	const auto currentTime = steadyNow();
+	std::vector<std::unique_ptr<AttachmentPoolEntry>> toDestroy;
+
+	for (auto it = available.begin(); it != available.end();)
+	{
+		auto* entry = *it;
+		bool evict = false;
+
+		if (options.getMaxLifetime() && currentTime - entry->createdAt >= *options.getMaxLifetime())
+			evict = true;
+		else if (options.getIdleTimeout() && entries.size() > options.getMinSize() &&
+			currentTime - entry->lastReturnedAt >= *options.getIdleTimeout())
+			evict = true;
+
+		if (evict)
+		{
+			it = available.erase(it);
+
+			if (auto owned = extractEntry(entries, entry))
+				toDestroy.push_back(std::move(owned));
+		}
+		else
+			++it;
+	}
+
+	if (!toDestroy.empty())
+	{
+		lock.unlock();
+		toDestroy.clear();  // Disconnects (network I/O) outside the lock.
+		lock.lock();
+	}
+}
+
+void AttachmentPool::release(AttachmentPoolEntry& entry) noexcept
+{
+	try
+	{
+		bool resetFailed = false;
+
+		if (!closed && options.getSessionResetOnRelease() && entry.attachment && entry.attachment->isValid())
+		{
+			try
+			{
+				entry.attachment->resetSession();
+			}
+			catch (...)
+			{
+				resetFailed = true;  // Force this connection to be discarded below.
+			}
+		}
+
+		std::unique_ptr<AttachmentPoolEntry> toDestroy;
+
+		{  // scope
+			std::lock_guard mutexGuard{mutex};
+
+			assert(inUse > 0u);
+			--inUse;
+
+			const bool discard = closed || !entry.attachment || !entry.attachment->isValid() || resetFailed;
+
+			if (discard)
+				toDestroy = extractEntry(entries, &entry);
+			else
+			{
+				entry.lastReturnedAt = steadyNow();
+				available.push_back(&entry);
+			}
+		}
+
+		availableCondition.notify_one();
+		// toDestroy (if any) disconnects here, outside the lock.
+	}
+	catch (...)
+	{
+		// swallow
+	}
+}