include/functionalqueuethread.hpp - main - Ant010ff/ffpp

include/functionalqueuethread.hpp
537 строк17 KB
Ant010ff
Version 2.11.3.
08 авг 2026, 16:44
08 авг 2026, 16:44fd187d3
/*
This header is part of a Functional Flow Processing Primitives (FFPP) library, version 2.11.3.

Official repository: https://gitlab.com/ant010ff/ffpp

Licensed under the MIT License <http://opensource.org/licenses/MIT>.
SPDX-License-Identifier: MIT
Copyright (c) 2021 - 2026 Anton Nasonov <ant010fff @ gmail . com>.
*/

#ifndef FFPP_FUNCTIONAL_QUEUE_THREAD_HPP
#define FFPP_FUNCTIONAL_QUEUE_THREAD_HPP

namespace FFPP::Concepts {

	template<typename Type>
	concept ThreadPolicy = 
		FunctionalQueuePolicy<Type> 
		&& Invocable<Type::IdleBackoff, BackoffHandlerT> //idle backoff handler generator
		&& InvocableStrict<Type::OnThreadStart, bool>
		&& Invocable<Type::OnThreadStop, void>
		&& InvocableStrict<Type::OnThreadException, bool, std::exception>
		//&& InvocableStrict<Type::OnWaitException, bool, std::exception>
		//&& InvocableStrict<Type::OnFinalizeException, bool, std::exception>
	;//ThreadPolicy

}//FFPP::Concepts

namespace FFPP::Base {

	template<
		typename 							TDerived
		, Concepts::FunctionalQueuePolicy	TPolicy = FunctionalQueuePolicy
		, std::default_initializable		TMixin = std::monostate
	>
	class FunctionalQueueThread 
	: public FunctionalQueue<TDerived, TPolicy> 
	, public TMixin
	{
	public:
		using PolicyT = TPolicy;
		using FunctionalQueueT = FunctionalQueue<TDerived, TPolicy>;
		using ResourcePolicyT = FunctionalQueueT::ResourcePolicyT;
		template<typename TSignature, bool t_bUnique = true>
		using FunctionT = ResourcePolicyT::template FunctionT<TSignature, t_bUnique>;
		using FunctionalSequenceT = FunctionalQueueT::FunctionalSequenceT;
		friend FunctionalSequenceT;
		using ExecutableT = FunctionalSequenceT::ExecutableT;
		friend ExecutableT;		

		FunctionalQueueThread(
			std::invocable<std::stop_token> auto&&	fnThread
			, size_t								nLimit
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&			slOrigin
			#endif
		) 
		: FunctionalQueueT(
			0
			, nLimit
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		)
		, jtProcessor_(std::forward<decltype(fnThread)>(fnThread))
		{ }

        FunctionalQueueThread(FunctionalQueueThread const&) = delete;
        FunctionalQueueThread(FunctionalQueueThread&&) = delete;

		FunctionalQueueThread& operator = (FunctionalQueueThread const&) = delete;
		FunctionalQueueThread& operator = (FunctionalQueueThread&&) = delete;

		~FunctionalQueueThread() { }

		static constexpr bool IsSubmittable() { return false; }

		bool IsJoinable() const noexcept { return jtProcessor_.joinable(); }

		bool IsActive() const noexcept { return !this->IsFinalized() && jtProcessor_.joinable(); }

		ThreadHandleT GetNativeHandle() {
			#if defined(_WIN32)
				#if (defined(_MSC_VER) && _MSC_VER >= 1930L)
					return jtProcessor_.native_handle(); 
				#else
					return nullptr; //unsupported on VS 2019, MinGW
				#endif
			#elif defined(__unix__) || defined(__APPLE__)
				return jtProcessor_.native_handle();
			#else
				#error Unsupported platform.
			#endif
		}

		template<std::invocable<TDerived*> TFunctional>
		bool Complete(Flags flgContext, TFunctional&& onComplete) {
			if(!!(flgContext & Flags::Wait) && this->IsProcessingThread()) {
				using FunctionalT = FunctionT<void(TDerived*)>;
				bool const bFinalize = FunctionalQueueT::Finalize(flgContext & ~Flags::Wait);                
				if(bFinalize) if(FunctionalT fnComplete { std::move(onComplete) }; fnComplete) {
					std::invoke(fnComplete, static_cast<TDerived*>(this));
				}
				return bFinalize;
			}
			else {
				return FunctionalQueueT::Complete(flgContext, std::forward<TFunctional>(onComplete));
			}
		}

		template<std::invocable<TDerived*> TFunctional>
		bool Complete(TFunctional&& onComplete) {
			return Complete(Flags::Undefined, std::forward<TFunctional>(onComplete));
		}

		bool Complete(Flags flgContext = Flags::Wait) {
			return Complete(flgContext, FunctionT<void(TDerived*)>{ });
		}
	
	protected:
		static bool OnThreadException(TDerived* pInstance, std::exception const& ex) noexcept { 
			using TDerivedPointer = TDerived*;
			if constexpr(requires { PolicyT::OnThreadException(TDerivedPointer { }, std::exception { }); }) {
				return PolicyT::OnThreadException(pInstance, ex);
			}
			else {
				return PolicyT::OnThreadException(ex);
			}
		}

		static bool OnWaitException(TDerived* pInstance, std::exception const& ex) noexcept { 
			using TDerivedPointer = TDerived*;
			if constexpr(requires { PolicyT::OnWaitException(TDerivedPointer { }, std::exception { }); }) {
				return PolicyT::OnWaitException(pInstance, ex);
			}
			else if constexpr(requires { PolicyT::OnWaitException(std::exception { }); }) {
				return PolicyT::OnWaitException(ex);
			}
			return false;
		}

		static bool OnFinalizeException(TDerived* pInstance, std::exception const& ex) noexcept { 
			using TDerivedPointer = TDerived*;
			if constexpr(requires { PolicyT::OnFinalizeException(TDerivedPointer { }, std::exception { }); }) {
				return PolicyT::OnFinalizeException(pInstance, ex);
			}
			else if constexpr(requires { PolicyT::OnFinalizeException(std::exception { }); }) {
				return PolicyT::OnFinalizeException(ex);
			}
			return false;
		}

		bool RequestStop() noexcept { return jtProcessor_.request_stop(); }

		void NotifyComplete() noexcept { bsComplete_.release(); }

		bool OnWait([[maybe_unused]] Flags flgContext) noexcept { 
			if(std::this_thread::get_id() == jtProcessor_.get_id()) return false;
			//First wait on a joinable thread uses join(), further and concurrent waits performs thread state polling (thread also 
			//could be terminated) with bsComplete_ condition variable.
			FFPP_TRY { 
				if(bJoin_.exchange(false, std::memory_order::relaxed) && jtProcessor_.joinable()) {
					jtProcessor_.join();
				}
				else {
					constexpr auto c_msPolling = std::chrono::milliseconds(50);
					while(jtProcessor_.joinable()) {
						bool bAcquired = bsComplete_.try_acquire_for(c_msPolling);
					}
				}
			}
			FFPP_CATCH(std::exception, ex) { 
				return OnWaitException(static_cast<TDerived*>(this), ex); 
			}
			return true;
		}

		bool CommitThread() noexcept { return bCommit_.exchange(true, std::memory_order::release); }

		[[nodiscard]]
		bool IsThreadCommitted() const noexcept { return bCommit_.load(std::memory_order::acquire); }

	private:
		std::atomic<bool> 
			bCommit_ { false }
			, bJoin_ { true };
		std::binary_semaphore 
			bsComplete_ { 0 };
		std::jthread 
			jtProcessor_;
	};//FunctionalQueueThread

}//FFPP::Base

namespace FFPP {

	struct ThreadPolicy : FunctionalQueuePolicy {
		using LockableT = Lockable<StripedTicketMutex<>>;

		static constexpr SchedulingFlags SchedulingMethod() { return SchedulingFlags::Disabled; }

		static constexpr auto IdleBackoff() {
			//By default using exponential busy-wait series before going to wait state with factors [1, 6] (CPU relaxation loops 
			//from 2 to 64).
			return FFPP::Backoff<false, 6, 0, 0, 1>();
		}

		//This enables double-queueing by default for threads and thread pool.
		static constexpr bool DoubleQueueing() { return true; }

		static bool OnThreadStart() noexcept { return true; }
		static void OnThreadStop() noexcept { }
		static bool OnThreadException(std::exception const&) noexcept { return true; }
		static bool OnWaitException(std::exception const&) noexcept { return false; }
		static bool OnFinalizeException(std::exception const&) noexcept { return false; }
	};

	template<typename TDerived, Concepts::ThreadPolicy TPolicy = ThreadPolicy, bool t_bTiming = false>
	class FunctionalQueueThread : public Base::FunctionalQueueThread<TDerived, TPolicy> {
	public:
		using PolicyT = TPolicy;
		using ResourcePolicyT = PolicyT::ResourcePolicyT;
		template<typename Type> 
		using SharedT = ResourcePolicyT::template SharedT<Type>;
		template<typename Type> 
		using WeakT = ResourcePolicyT::template WeakT<Type>;
		using SP = SharedT<TDerived>;
		using WP = WeakT<TDerived>;
		using FunctionalQueueThreadBaseT = Base::FunctionalQueueThread<TDerived, TPolicy>;
		using FunctionalQueueT = FunctionalQueueThreadBaseT::FunctionalQueueT;
		using FunctionalSequenceT = FunctionalQueueT::FunctionalSequenceT;
		friend FunctionalSequenceT;
		using ExecutableT = FunctionalSequenceT::ExecutableT;
		friend ExecutableT;

		using ClockT = std::chrono::steady_clock;

		template<std::invocable TOnThreadStart>
		FunctionalQueueThread(
			TOnThreadStart&&				onThreadStart
			, size_t 						nLimit		= FunctionalSequenceT::InvalidLimit()
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin 	= std::source_location::current()
			#endif
		) 
		: FunctionalQueueThreadBaseT(
			[this, pSelf = static_cast<TDerived*>(this), onThreadStart = std::forward<TOnThreadStart>(onThreadStart)] 
			(std::stop_token stop) mutable {
				UniqueGuard ugComplete { [this] () noexcept { this->NotifyComplete(); } };
				if(!PolicyT::OnThreadStart()) return;
				std::invoke(onThreadStart);
				this->UpdateProcessingThread();
				auto onIdleBackoff = PolicyT::IdleBackoff();
				while(!stop.stop_requested()) {
					FFPP_TRY {
						if(!onIdleBackoff()) [[unlikely]] this->WaitEnqueue();
						while(this->HasEnqueued() && !stop.stop_requested()) pSelf->OnProcessSingle();
					}
					FFPP_CATCH(std::exception, ex) {
						if(!FunctionalQueueThreadBaseT::OnThreadException(pSelf, ex)) break;
					}
				}
				PolicyT::OnThreadStop();
			}
			, nLimit
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		)
		{ }

		FunctionalQueueThread(
			size_t 							nLimit		= FunctionalQueueT::InvalidLimit()
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin	= std::source_location::current()
			#endif
		) 
		: FunctionalQueueThread(
			[] () { }
			, nLimit
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		)
		{ }

		template<std::invocable TOnThreadStart>
		FunctionalQueueThread(
			Concepts::Duration auto const&
			, TOnThreadStart&&				onThreadStart
			, size_t 						nLimit		= FunctionalSequenceT::InvalidLimit()
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin	= std::source_location::current()
			#endif
		) 
		: FunctionalQueueThread(
			std::forward<TOnThreadStart>(onThreadStart)
			, nLimit
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		)
		{ }

		FunctionalQueueThread(
			Concepts::Duration auto const&
			, size_t 						nLimit		= FunctionalQueueT::InvalidLimit()
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin	= std::source_location::current()
			#endif
		) 
		: FunctionalQueueThread(
			[] () { }
			, nLimit
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		)
		{ }

		~FunctionalQueueThread() { }

		template<std::invocable TOnThreadStart>
		FFPP_ATTR_INLINE
		static SharedT<TDerived> MakeShared(
			TOnThreadStart&&				onThreadStart
			, size_t 						nLimit		= FunctionalSequenceT::InvalidLimit()
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin 	= std::source_location::current()
			#endif
		) { 
			return ResourcePolicyT::template AllocateShared<TDerived>(
				std::forward<TOnThreadStart>(onThreadStart)
				, nLimit
				#if FFPP_TRACK_ORIGIN
				, slOrigin
				#endif
			);
		}

		FFPP_ATTR_INLINE
		static SharedT<TDerived> MakeShared(
			size_t 							nLimit		= FunctionalSequenceT::InvalidLimit()
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin 	= std::source_location::current()
			#endif
		) { 
			return MakeShared(
				[] () { }
				, nLimit
				#if FFPP_TRACK_ORIGIN
				, slOrigin
				#endif
			);
		}

	protected:
		void OnFinalize() noexcept {
			FunctionalQueueT::OnFinalize();
			if(this->RequestStop()) this->OnEnqueued(FunctionalQueueT::InvalidQueueId());
		}
	};//FunctionalQueueThread

	namespace dtl {

		class TimingThreadMixin {
		protected:
			std::binary_semaphore bsLoop_ { 0 };
			std::atomic_bool bNotify_ { false };

			FFPP_ATTR_INLINE
			void NotifyLoop() {
				if(!bNotify_.exchange(true, std::memory_order::relaxed)) {
					bsLoop_.release();				
				}
			}

			template<typename TTimePoint, typename TBackoff>
			FFPP_ATTR_INLINE
			bool WaitForNotification(TTimePoint tpUntil, TBackoff& onIdleBackoff) {
				if(onIdleBackoff()) [[likely]] return false;
				bool const bTimeout = !bsLoop_.try_acquire_until(tpUntil);
				if(!bTimeout) bNotify_.store(false, std::memory_order::relaxed);
				return bTimeout;
			}
		};//TimingThreadMixin

	}//dtl

	template<typename TDerived, Concepts::ThreadPolicy TPolicy>
	class FunctionalQueueThread<TDerived, TPolicy, true> 
	: public Base::FunctionalQueueThread<TDerived, TPolicy, dtl::TimingThreadMixin> 
	{
	public:
		using PolicyT = TPolicy;
		using ResourcePolicyT = PolicyT::ResourcePolicyT;
		template<typename Type> 
		using SharedT = ResourcePolicyT::template SharedT<Type>;
		template<typename Type> 
		using WeakT = ResourcePolicyT::template WeakT<Type>;
		using SP = SharedT<TDerived>;
		using WP = WeakT<TDerived>;
		using FunctionalQueueThreadBaseT = Base::FunctionalQueueThread<TDerived, TPolicy, dtl::TimingThreadMixin>;
		using FunctionalQueueT = FunctionalQueueThreadBaseT::FunctionalQueueT;
		using FunctionalSequenceT = FunctionalQueueT::FunctionalSequenceT;
		friend FunctionalSequenceT;
		using ExecutableT = FunctionalSequenceT::ExecutableT;
		friend ExecutableT;

		using ClockT = std::chrono::steady_clock;

		template<std::invocable TOnThreadStart>
		FunctionalQueueThread(
			Concepts::Duration auto const&	drnPeriod
			, TOnThreadStart&&				onThreadStart
			, size_t                        nLimit		= FunctionalSequenceT::InvalidLimit()
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin 	= std::source_location::current()
			#endif
		) 
		: FunctionalQueueThreadBaseT(
			[this, pSelf = static_cast<TDerived*>(this), drnPeriod, onThreadStart = std::forward<TOnThreadStart>(onThreadStart)] 
			(std::stop_token stop) mutable {
				UniqueGuard ugComplete { [this] () noexcept { this->NotifyComplete(); } };
				if(!PolicyT::OnThreadStart()) return;
				std::invoke(onThreadStart);
				this->UpdateProcessingThread();
				auto onIdleBackoff = PolicyT::IdleBackoff();
				for(auto tpStart = ClockT::now(); !stop.stop_requested();) {
					FFPP_TRY {
						bool const bTimeout = this->WaitForNotification(tpStart + drnPeriod, onIdleBackoff);
						if(bTimeout) {
							tpStart += drnPeriod;
							pSelf->OnThreadTiming(tpStart);
						}
						while(this->HasEnqueued() && !stop.stop_requested()) {
							if(ClockT::now() - tpStart > drnPeriod) {
								tpStart += drnPeriod;
								pSelf->OnThreadTiming(tpStart);
							}
							pSelf->OnProcessSingle();
						}
					}
					FFPP_CATCH(std::exception, ex) {
						if(!FunctionalQueueThreadBaseT::OnThreadException(pSelf, ex)) break;
					}
				}
				PolicyT::OnThreadStop();
			}
			, nLimit
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		)
		{ }

		FunctionalQueueThread(
			Concepts::Duration auto const&	drnPeriod	= std::chrono::milliseconds(1000)
			, size_t                        nLimit		= FunctionalQueueT::InvalidLimit()
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin 	= std::source_location::current()
			#endif
		) 
		: FunctionalQueueThread(
			drnPeriod
			, [] () { }
			, nLimit
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		)
		{ }

		~FunctionalQueueThread() { }

		template<std::invocable TOnThreadStart>
		FFPP_ATTR_INLINE
		static SharedT<TDerived> MakeShared(
			Concepts::Duration auto const&	drnPeriod 	
			, TOnThreadStart&&				onThreadStart
			, size_t 						nLimit 		= FunctionalSequenceT::InvalidLimit()	
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin 	= std::source_location::current()
			#endif
		) { 
			return ResourcePolicyT::template AllocateShared<TDerived>(
				drnPeriod
				, std::forward<TOnThreadStart>(onThreadStart)
				, nLimit
				#if FFPP_TRACK_ORIGIN
				, slOrigin
				#endif
			);
		}

		FFPP_ATTR_INLINE
		static SharedT<TDerived> MakeShared(
			Concepts::Duration auto const&	drnPeriod	= std::chrono::milliseconds(1000)
			, size_t 						nLimit		= FunctionalSequenceT::InvalidLimit()
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin 	= std::source_location::current()
			#endif
		) { 
			return MakeShared(
				drnPeriod
				, [] () { }
				, nLimit
				#if FFPP_TRACK_ORIGIN
				, slOrigin
				#endif
			);
		}

	protected:
		void OnThreadTiming(ClockT::time_point const&) { }

		template<bool t_bNotify = false> //Specialization with timer uses semaphore to wait for task enqueue.
		std::pair<bool, size_t> OnEnqueued(uint32_t idQueue) {
			auto const [bEnqueued, nEnqueued] = FunctionalQueueT::template OnEnqueued<t_bNotify>(idQueue);
			if(nEnqueued == 0) this->NotifyLoop();
			return { bEnqueued, nEnqueued };
		}

		void OnFinalize() noexcept {
			FunctionalQueueT::OnFinalize();
			FFPP_TRY {
				if(this->RequestStop()) this->NotifyLoop();
			}
			FFPP_CATCH(std::exception, ex) {
				FunctionalQueueThreadBaseT::OnFinalizeException(static_cast<TDerived*>(this), ex);
			}
		}
	};//FunctionalQueueThread
	
}//FFPP

#endif//FFPP_FUNCTIONAL_QUEUE_THREAD_HPP