include/functionaltimer.hpp - main - Ant010ff/ffpp

include/functionaltimer.hpp
306 строк10 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_TIMER_HPP
#define FFPP_FUNCTIONAL_TIMER_HPP

namespace FFPP::Concepts {

	template<typename Type>
	concept TimerPolicy = 
		ResourcePolicy<typename Type::ResourcePolicyT>
		&& Lockable<typename Type::LockableT> 
		&& StripedTablePolicy<Type> 
		&& Invocable<Type::TableStripes, uint32_t>
		&& InvocableStrict<Type::OnTimerStart, bool>
		&& Invocable<Type::OnTimerStop, void>
		&& InvocableStrict<Type::OnTimerException, bool, std::exception>
		&& InvocableStrict<Type::OnWaitException, bool, std::exception>
	;//TimerPolicy

}//FFPP::Concepts

namespace FFPP {

	struct TimerPolicy : StripedTablePolicy {
		using ResourcePolicyT = DefaultResourcePolicy;
		using LockableT = Lockable<TicketMutex<>>;
		static uint32_t TableStripes() { return std::thread::hardware_concurrency(); }
		static bool OnTimerStart() noexcept { return true; }
		static void OnTimerStop() noexcept { }
		static bool OnTimerException(std::exception const&) noexcept { return true; }
		static bool OnWaitException(std::exception const&) noexcept { return false; }
	};//TimerPolicy
	
	template<Concepts::TimerPolicy TPolicy = TimerPolicy>
	class FunctionalTimer : public TPolicy::LockableT {
	public:
		using PolicyT = TPolicy;
		using ResourcePolicyT = PolicyT::ResourcePolicyT;
		using IdProviderT = ResourcePolicyT::IdProviderT;
		using Id = IdProviderT::Id;
		using HashT = IdProviderT::Hash;
		using LockableT = PolicyT::LockableT;
		template<typename TSignature, bool t_bUnique = true>
		using FunctionT = ResourcePolicyT::template FunctionT<TSignature, t_bUnique>;
		using IntervalHandlerT = FunctionT<bool(), false>;
		using ClockT = std::chrono::steady_clock;
		using DurationT = std::chrono::milliseconds;
		using FunctionalTimerT = FunctionalTimer<TPolicy>;
		template<typename Type> 
		using UniqueT = ResourcePolicyT::template UniqueT<Type>;
		template<typename Type> 
		using SharedT = ResourcePolicyT::template SharedT<Type>;
		template<typename Type> 
		using WeakT = ResourcePolicyT::template WeakT<Type>;
		using UP = UniqueT<FunctionalTimerT>;
		using SP = SharedT<FunctionalTimerT>;
		using WP = WeakT<FunctionalTimerT>;
	
		FunctionalTimer(Concepts::Duration auto drnPeriod) 
		: jtProcessor_([this, drnPeriod] (std::stop_token stop) {
			UniqueGuard ugComplete { [this] () { 
				bsComplete_.release();
			} };

			if(!PolicyT::OnTimerStart()) return;
			std::vector<SharedHandlerContextT, typename ResourcePolicyT::template AllocatorT<SharedHandlerContextT>> vHandlers;
			std::vector<Id, typename ResourcePolicyT::template AllocatorT<Id>> vComplete;
			for(auto tpStart = ClockT::now(); !stop.stop_requested();) {
				FFPP_TRY {
					vHandlers.clear();
					{
						size_t const nHandlers = tblHandlers_.Size();
						if(nHandlers == 0) {
							vHandlers.shrink_to_fit();
							vComplete.shrink_to_fit();
						}
						else {
							if(vHandlers.capacity() < nHandlers) vHandlers.reserve(nHandlers);
							if(vComplete.capacity() < nHandlers) vComplete.reserve(nHandlers);
							tblHandlers_.ForEachLocked([&vHandlers, &vComplete] (ValOrRef<Id const> idHandler, auto const& spContext) {
								if(spContext->bComplete.load(std::memory_order::relaxed)) vComplete.push_back(idHandler);
								else vHandlers.push_back(spContext);
								return true;
							});
						}
					}

					for(auto const& spContext : vHandlers) {
						if(spContext->c_drnInterval == DurationT { } || tpStart - spContext->tpInvoke >= spContext->c_drnInterval) {
							spContext->tpInvoke = tpStart;
							if(!std::invoke(spContext->c_onTime)) {
								spContext->bComplete.store(true, std::memory_order::relaxed);
							}
						}
					}

					if(!vComplete.empty()) {
						for(ValOrRef<Id const> idComplete : vComplete) tblHandlers_.Remove(idComplete);
						vComplete.clear();
					}

					if(bsLoop_.try_acquire_until(tpStart + drnPeriod)) break;
					auto const c_tpNow = ClockT::now();
					if(tpStart + 2 * drnPeriod < c_tpNow) tpStart = c_tpNow;
					else tpStart += drnPeriod;
				}
				FFPP_CATCH(std::exception, ex) {
					if(!OnTimerException(this, ex)) break;
				}
			}
			PolicyT::OnTimerStop();
		})
		{ }

		FunctionalTimer(Concepts::Duration auto const& drnPeriod, std::convertible_to<IntervalHandlerT> auto&& onTime) 
		: FunctionalTimer(drnPeriod)
		{ Submit(std::forward<decltype(onTime)>(onTime)); }

		~FunctionalTimer() { Complete(Flags::Wait); }

		FFPP_ATTR_INLINE
		static ResourcePolicyT::template SharedT<FunctionalTimerT> MakeShared(Concepts::Duration auto drnPeriod) { 
			return ResourcePolicyT::template AllocateShared<FunctionalTimerT>(drnPeriod);
		}

		static constexpr Id InvalidId() { return IdProviderT::InvalidId(); }
		static Id GenerateId() noexcept { return IdProviderT::GenerateId(); }

		FunctionalTimer(FunctionalTimer const&) = delete;
		FunctionalTimer(FunctionalTimer&&) = delete;

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

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

		bool IsTimerThread() const noexcept { return std::this_thread::get_id() == jtProcessor_.get_id(); }

		template<typename TDuration, typename THandler>
		requires (Concepts::Duration<TDuration> && std::convertible_to<THandler, IntervalHandlerT>)
		Id Submit(ValOrRef<Id const> idHandler, TDuration const& drnInterval, THandler&& onTime) {
			return tblHandlers_.Insert(
				idHandler, 
				ResourcePolicyT::template AllocateShared<HandlerContext>(
					std::forward<THandler>(onTime), 
					drnInterval
				)
			);
		}

		template<typename TDuration, typename THandler>
		requires (Concepts::Duration<TDuration> && std::convertible_to<THandler, IntervalHandlerT>)
		Id Submit(TDuration const& drnInterval, THandler&& onTime) {
			return Submit(InvalidId(), drnInterval, std::forward<THandler>(onTime));
		}

		template<typename THandler>
		requires std::convertible_to<THandler, IntervalHandlerT>
		Id Submit(ValOrRef<Id const> idHandler, THandler&& onTime) {
			return Submit(idHandler, DurationT { }, std::forward<THandler>(onTime));
		}

		template<typename THandler>
		requires std::convertible_to<THandler, IntervalHandlerT>
		Id Submit(THandler&& onTime) {
			return Submit(InvalidId(), DurationT { }, std::forward<THandler>(onTime));
		}

		bool Contains(ValOrRef<Id const> idHandler) {
			return tblHandlers_.Test(idHandler);
		}

		bool Complete(std::convertible_to<Id> auto&... idHandlers) {
			bool bComplete = false;
			((bComplete |= tblHandlers_.Remove(idHandlers), idHandlers = InvalidId()), ...);
			return bComplete;
		}

		bool Complete(std::convertible_to<Id> auto const&... idHandlers) {
			bool bComplete = false;
			((bComplete |= tblHandlers_.Remove(idHandlers)), ...);
			return bComplete;
		}

		bool Finalize(Flags flgContext = Flags::Wait) { 
			bool bStop = jtProcessor_.request_stop();
			if(bStop) bsLoop_.release();
			if(!!(flgContext & Flags::Wait)) Wait(flgContext);
			return bStop;
		}

		bool Complete(Flags flgContext = Flags::Wait) { 
			return Finalize(flgContext);
		}

		bool Wait([[maybe_unused]] Flags flgContext = Flags::Undefined) 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(this, ex); 
			}
			return true;
		}

		size_t Size() const {
			return tblHandlers_.Size();
		}

		std::pair<size_t, bool> Clear(Flags flgContext = Flags::Undefined) { 
			if(!!(flgContext & Flags::Async)) {
				size_t nCleanup = 0;
				bool bException = false;
				FFPP_TRY {
					tblHandlers_.ForEachLocked([&nCleanup] (auto, auto const& spContext) {
						nCleanup++;
						spContext->bComplete.store(true, std::memory_order::relaxed);
						return true;
					});
				}
				FFPP_CATCH(std::exception, ex) {
					PolicyT::OnTableCleanupException(ex);
					bException = true;
				}
				return { nCleanup, bException };
			}
			else {
				return tblHandlers_.Clear();
			}
		}

	private:
		struct HandlerContext {
			HandlerContext(std::convertible_to<IntervalHandlerT> auto&& onTime, Concepts::Duration auto const& drnPeriod) 
			: c_onTime(std::forward<decltype(onTime)>(onTime))
			, c_drnInterval(std::chrono::duration_cast<DurationT>(drnPeriod))
			{}

			IntervalHandlerT const
				c_onTime;
			DurationT const
				c_drnInterval = { };
			ClockT::time_point
				tpInvoke = ClockT::now();
			std::atomic<bool>
				bComplete = false;
		};//HandlerContext
		template<typename Type> 
		using AllocatorT = ResourcePolicyT::template AllocatorT<Type>;
		using SharedHandlerContextT = SharedT<HandlerContext>;
		using HandlerTableT = StripedTable<HandlerContext, false, PolicyT>;

		static bool OnTimerException(FunctionalTimer<TPolicy>* pInstance, std::exception const& ex) noexcept { 
			using TInstancePointer = FunctionalTimer<TPolicy>*;
			if constexpr(requires { PolicyT::OnTimerException(TInstancePointer { }, std::exception { }); }) {
				return PolicyT::OnTimerException(pInstance, ex);
			}
			else {
				return PolicyT::OnTimerException(ex);
			}
		}

		static bool OnWaitException(FunctionalTimer<TPolicy>* pInstance, std::exception const& ex) noexcept { 
			using TInstancePointer = FunctionalTimer<TPolicy>*;
			if constexpr(requires { PolicyT::OnWaitException(TInstancePointer { }, std::exception { }); }) {
				return PolicyT::OnWaitException(pInstance, ex);
			}
			else if constexpr(requires { PolicyT::OnWaitException(std::exception { }); }) {
				return PolicyT::OnWaitException(ex);
			}
			return false;
		}

		std::atomic<bool> 
			bJoin_ = true;
		std::binary_semaphore 
			bsLoop_ { 0 },
			bsComplete_ { 0 };
		HandlerTableT
			tblHandlers_;
		std::jthread
			jtProcessor_;
	};//FunctionalTimer

}//FFPP

#endif//FFPP_FUNCTIONAL_TIMER_HPP