test (0ccea64c) · Commits · jan.koester / blogi

plugins/media/checkout_pool.h

0 → 100644

+124 −0

Original line number	Diff line number	Diff line
		/*******************************************************************************
		* Copyright (c) 2026, Jan Koester jan.koester@gmx.net
		* All rights reserved.
		*
		* Redistribution and use in source and binary forms, with or without
		* modification, are permitted provided that the following conditions are met:
		* Redistributions of source code must retain the above copyright
		* notice, this list of conditions and the following disclaimer.
		* Redistributions in binary form must reproduce the above copyright
		* notice, this list of conditions and the following disclaimer in the
		* documentation and/or other materials provided with the distribution.
		* Neither the name of the <organization> nor the
		* names of its contributors may be used to endorse or promote products
		* derived from this software without specific prior written permission.
		*
		* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
		* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
		* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
		* DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY
		* DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
		* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
		* LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
		* ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
		* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
		* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
		*******************************************************************************/

		#pragma once

		#include <functional>
		#include <memory>
		#include <mutex>
		#include <vector>

		namespace blogi {

		// A checkout/return pool for objects that are unsafe to share across
		// threads (e.g. mediadb::Client / libhttppp::HttpClient — one live socket
		// plus unsynchronized parse state per instance). Needed once H2 stream
		// dispatch can be offloaded to a background thread pool: `tid`-indexed
		// per-worker-slot pooling (the old pattern) assumes exactly one thread
		// ever touches a given slot at a time, which is no longer true once
		// multiple offloaded tasks can run concurrently and don't share the same
		// tid the way the inbound epoll workers do.
		//
		// acquire() blocks only on the internal mutex (µs-scale), never while
		// holding it across the (potentially slow, real-network) factory call —
		// a miss just falls through to constructing a fresh instance outside the
		// lock.
		template <class T>
		class CheckoutPool {
		public:
		using Factory = std::function<std::unique_ptr<T>()>;

		CheckoutPool(size_t maxSize, Factory factory)
		: _maxSize(maxSize), _factory(std::move(factory)) {}

		class Handle {
		public:
		Handle() = default;
		Handle(CheckoutPool &pool, std::unique_ptr<T> obj)
		: _pool(&pool), _obj(std::move(obj)) {}
		~Handle() { if (_pool && _obj) _pool->release(std::move(_obj)); }

		Handle(Handle &&) = default;
		Handle &operator=(Handle &&) = default;
		Handle(const Handle &) = delete;
		Handle &operator=(const Handle &) = delete;

		T *get() const { return _obj.get(); }
		T *operator->() const { return _obj.get(); }
		T &operator() const { return _obj; }
		explicit operator bool() const { return _obj != nullptr; }

		// Call when the checked-out object's state may be corrupted (e.g.
		// after catching an exception from one of its calls) so a broken
		// instance never goes back into circulation for the next caller.
		void discard() { _obj.reset(); }

		// Alias for discard(), for drop-in compatibility with call sites
		// written against std::unique_ptr's reset(). Deliberately always
		// discards rather than returning to the pool — matches this
		// pool's callers, which only ever call reset() at points where
		// the pre-pooling code unconditionally tore the connection down.
		void reset() { discard(); }

		private:
		CheckoutPool *_pool = nullptr;
		std::unique_ptr<T> _obj;
		};

		// The factory call (a real blocking connect, for the clients this is
		// used for) always happens outside the lock.
		Handle acquire() {
		{
		std::lock_guard<std::mutex> lk(_mtx);
		if (!_avail.empty()) {
		auto obj = std::move(_avail.back());
		_avail.pop_back();
		return Handle(*this, std::move(obj));
		}
		}
		return Handle(*this, _factory());
		}

		private:
		friend class Handle;

		void release(std::unique_ptr<T> obj) {
		std::lock_guard<std::mutex> lk(_mtx);
		if (_avail.size() < _maxSize)
		_avail.push_back(std::move(obj));
		// else: pool already full (more concurrent checkouts than
		// maxSize happened) — just let this instance be destroyed instead
		// of growing the pool unbounded.
		}

		std::mutex _mtx;
		std::vector<std::unique_ptr<T>> _avail;
		size_t _maxSize;
		Factory _factory;
		};

		} // namespace blogi

plugins/media/media.cpp

+77 −23

Original line number	Diff line number	Diff line
		@@ -32,6 +32,7 @@
		#include <set>
		#include <mutex>
		#include <shared_mutex>
		#include <condition_variable>
		#include <atomic>
		#include <memory>
		#include <algorithm>
		@@ -50,6 +51,7 @@
		#include <i18n.h>

		#include "icon.webp.h"
		#include "checkout_pool.h"

		namespace blogi {

		@@ -829,8 +831,23 @@ namespace blogi {

		_mdb.resize(Args->maxthreads);
		_mdbHttpUrl = std::make_unique<libhttppp::HttpUrl>(_mdbUrl);
		_mdbClientPool.resize(Args->maxthreads);
		_loadAllowedSizes(0);

		// Thread-safe pools for the /media/getimage/ route, which alone
		// may run on the H2 offload pool (see Blogi::shouldOffloadH2Dispatch)
		// instead of the single epoll worker that would otherwise own
		// `tid` for the whole call — _mdb/_mdbClientPool above stay
		// tid-indexed since every OTHER route here only ever runs on
		// the synchronous dispatch path.
		_mdbGetImagePool = std::make_unique<CheckoutPool<mediadb::Client>>(
		static_cast<size_t>(Args->maxthreads),
		[url = _mdbUrl]() { return std::make_unique<mediadb::Client>(url, false, false); });
		_mdbStreamPool = std::make_unique<CheckoutPool<libhttppp::HttpClient>>(
		static_cast<size_t>(Args->maxthreads),
		[this]() { return std::make_unique<libhttppp::HttpClient>(*_mdbHttpUrl, 0); });

		for (int i = 0; i < Args->maxthreads; ++i)
		_dbAvailableIdx.push_back(i);
		}

		bool Controller(const int tid,libhttppp::HttpRequest &req,libhtmlpp::HtmlElement *page,const std::string &sessionid){
		@@ -877,20 +894,36 @@ namespace blogi {
		}

		std::string _sessionToken = _getSessionToken(tid, sessionid);
		_ensureMdbReady(tid, _sessionToken);

		// Uses its own checkout pool rather than _mdb[tid]/
		// _ensureMdbReady: this route alone may run on the H2
		// offload pool (see Blogi::shouldOffloadH2Dispatch), where
		// tid no longer identifies an exclusively-owned slot.
		auto mdbHandle = _mdbGetImagePool->acquire();
		if (!_sessionToken.empty()) {
		try {
		mdbHandle->set_token(_sessionToken);
		} catch(...) {
		mdbHandle.discard();
		mdbHandle = _mdbGetImagePool->acquire();
		try { mdbHandle->set_token(_sessionToken); } catch(...) {}
		}
		}

		// Look up media info from mediadb
		mediadb::MediaInfo mi;
		try {
		mi = _mdb[tid]->get_media(mediaId);
		mi = mdbHandle->get_media(mediaId);
		} catch(std::exception &e) {
		std::cerr << "media: get_media failed for " << mediaId << ": " << e.what() << std::endl;
		mdbHandle.discard();
		libhttppp::HttpResponse curres;
		curres.setState(HTTP404);
		curres.send(req,"",0);
		return true;
		} catch(...) {
		std::cerr << "media: get_media failed for " << mediaId << ": unknown error" << std::endl;
		mdbHandle.discard();
		libhttppp::HttpResponse curres;
		curres.setState(HTTP404);
		curres.send(req,"",0);
		@@ -961,14 +994,18 @@ namespace blogi {
		dataReq.setHeaderData("Range")->push_back(upstreamRange);
		}

		auto streamClient = _acquireMdbClient(tid);
		auto streamClient = _mdbStreamPool->acquire();
		std::unique_ptr<libhttppp::HttpResponse> dataResPtr;
		try {
		dataResPtr = std::make_unique<libhttppp::HttpResponse>(streamClient->GetStream(dataReq));
		} catch(...) {
		// Pooled connection may be stale — retry with a fresh one
		// Pooled connection may be stale — retry with a fresh one.
		// Assigning a new Handle disposes of the bad connection
		// (via unique_ptr's own move-assignment) rather than
		// returning it to the pool for reuse.
		std::cerr << "media: upstream GetStream failed, retrying with fresh connection" << std::endl;
		streamClient = std::make_unique<libhttppp::HttpClient>(*_mdbHttpUrl, true);
		streamClient = CheckoutPool<libhttppp::HttpClient>::Handle(
		_mdbStreamPool, std::make_unique<libhttppp::HttpClient>(_mdbHttpUrl, true));
		try {
		dataResPtr = std::make_unique<libhttppp::HttpResponse>(streamClient->GetStream(dataReq));
		} catch(std::exception &e2) {
		@@ -1613,14 +1650,16 @@ namespace blogi {
		private:
		// ---- mediadb client management ----

		std::string _getSessionToken(const int tid, const std::string &sessionid){
		std::string _getSessionToken(const int /tid/, const std::string &sessionid){
		if(sessionid.empty())
		return "";
		std::string authid;
		int idx = _acquireDbIndex();
		try {
		Session sess(*Args->database[tid]);
		Session sess(*Args->database[idx]);
		sess.getSessionData(sessionid, "authid", authid);
		} catch(...) {}
		_releaseDbIndex(idx);
		return authid;
		}

		@@ -1779,14 +1818,14 @@ namespace blogi {
		// ---- streaming state per-request ----

		struct StreamState {
		std::unique_ptr<libhttppp::HttpClient> client;
		CheckoutPool<libhttppp::HttpClient>::Handle client;
		size_t sendSize = 0;
		size_t produced = 0;
		int tid = 0;
		};

		void _storeStreamState(libhttppp::HttpRequest &req,
		std::unique_ptr<libhttppp::HttpClient> client,
		CheckoutPool<libhttppp::HttpClient>::Handle client,
		size_t sendSize, int tid){
		std::unique_lock lock(_streamMutex);
		auto &ss = _streams[reinterpret_cast<uintptr_t>(&req)];
		@@ -1808,18 +1847,6 @@ namespace blogi {
		return ptr;
		}

		// ---- HttpClient pool for mediadb streaming ----

		std::unique_ptr<libhttppp::HttpClient> _acquireMdbClient(int tid){
		auto &pool = _mdbClientPool[tid];
		if(!pool.empty()){
		auto client = std::move(pool.back());
		pool.pop_back();
		return client;
		}
		return std::make_unique<libhttppp::HttpClient>(*_mdbHttpUrl, 0);
		}

		// ---- image dimension parsing ----

		static void _parseImageDimensions(const std::vector<std::uint8_t> &data,
		@@ -1893,10 +1920,37 @@ namespace blogi {
		std::vector<std::unique_ptr<mediadb::Client>> _mdb;
		std::unique_ptr<libhttppp::HttpUrl> _mdbHttpUrl;
		std::mutex _mdbSetupMutex;
		std::vector<std::vector<std::unique_ptr<libhttppp::HttpClient>>> _mdbClientPool;
		std::mutex _streamMutex;
		std::map<uintptr_t, StreamState> _streams;

		// ---- thread-safe resources for the offloadable getimage route ----

		std::unique_ptr<CheckoutPool<mediadb::Client>> _mdbGetImagePool;
		std::unique_ptr<CheckoutPool<libhttppp::HttpClient>> _mdbStreamPool;

		// Args->database is sized to Args->maxthreads and, like _mdb above,
		// was only ever safe to index by tid because exactly one thread
		// used a given tid at a time. _getSessionToken is called from both
		// offloadable and non-offloadable routes, so it checks an index
		// out of this pool instead — blocking briefly (never racing) if
		// every slot is momentarily in use.
		std::mutex _dbCheckoutMutex;
		std::condition_variable _dbCheckoutCv;
		std::vector<int> _dbAvailableIdx;

		int _acquireDbIndex() {
		std::unique_lock<std::mutex> lk(_dbCheckoutMutex);
		_dbCheckoutCv.wait(lk, [this]{ return !_dbAvailableIdx.empty(); });
		int idx = _dbAvailableIdx.back();
		_dbAvailableIdx.pop_back();
		return idx;
		}
		void _releaseDbIndex(int idx) {
		std::lock_guard<std::mutex> lk(_dbCheckoutMutex);
		_dbAvailableIdx.push_back(idx);
		_dbCheckoutCv.notify_one();
		}

		// ---- allowed preview sizes ----

		struct PreviewSize {

src/blogi.cpp

+13 −1

Original line number	Diff line number	Diff line
		@@ -297,7 +297,8 @@ static bool handleDeployFallback(libhttppp::HttpRequest &curreq, const int tid,
		return false;
		}

		blogi::Blogi::Blogi(std::vector<netplus::socket *> serversocket, bool debug) : HttpEvent(serversocket), _debug(debug)
		blogi::Blogi::Blogi(std::vector<netplus::socket *> serversocket, bool debug)
		: HttpEvent(serversocket, 1000, static_cast<size_t>(sysconf(_SC_NPROCESSORS_ONLN))), _debug(debug)
		{
		Blogi::ActiveInstance = this;

		@@ -671,6 +672,17 @@ blogi::DomainContext *blogi::Blogi::resolveDomain(libhttppp::HttpRequest &curreq
		return nullptr;
		}

		bool blogi::Blogi::shouldOffloadH2Dispatch(libhttppp::HttpRequest &tempreq, uint32_t /streamId/) const {
		// resolveDomain locks _domainsMutex but doesn't otherwise touch any
		// state this override can observe — same rationale as any other
		// logically-const accessor that happens to need a lock.
		DomainContext ctx = const_cast<Blogi>(this)->resolveDomain(tempreq);
		if (!ctx) return false;
		const std::string prefix = ctx->buildurl("media/getimage/");
		const std::string &url = tempreq.getRequestURL();
		return url.compare(0, prefix.length(), prefix) == 0;
		}

		void blogi::Blogi::api(libhttppp::HttpRequest &curreq, const int tid, DomainContext &ctx)
		{
		auto &PlgArgs = ctx.plgArgs;

src/blogi.h

+6 −0

Original line number	Diff line number	Diff line
		@@ -74,6 +74,12 @@ namespace blogi {
		void RequestEvent(libhttppp::HttpRequest &curreq,const int tid,ULONG_PTR args);
		void ResponseEvent(libhttppp::HttpRequest &curreq,const int tid,ULONG_PTR args);

		// Only the media plugin's getimage route makes a slow, blocking
		// backend round-trip (mediadb) inside RequestEvent — opting just
		// it into the H2 offload pool keeps every other route (backed by
		// local DB reads) on the original synchronous dispatch path.
		bool shouldOffloadH2Dispatch(libhttppp::HttpRequest &tempreq, uint32_t streamId) const override;

		DomainContext *resolveDomain(libhttppp::HttpRequest &curreq);

		// Reload SSL certificates at runtime (called after cert renewal)