include/executable.hpp - main - Ant010ff/ffpp

include/executable.hpp
351 строка13 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_EXECUTABLE_HPP
#define FFPP_EXECUTABLE_HPP

namespace FFPP::Concepts {

	template<typename Type>
	concept ExecutablePolicy = 
		ResourcePolicy<typename Type::ResourcePolicyT> 
		&& Lockable<typename Type::LockableT> 
		&& ConstexprInvocableStrict<Type::SchedulingMethod, SchedulingFlags>
		//Enforcing compile-time use for Type::DefaultPriority() static method and result type compatibility:
		&& std::same_as<std::bool_constant<(Type::DefaultPriority(), true)>, std::true_type> 
		&& Priority<typename std::invoke_result_t<decltype(Type::DefaultPriority)>>
	;

	template<typename Type>
	concept Executable = 
		ExecutablePolicy<typename Type::PolicyT> 
        && !std::copyable<Type>
		&& ConstexprInvocableStrict<Type::IsSubmittable, bool>
		&& ConstexprInvocableStrict<Type::InvalidValue, size_t>
		&& ConstexprInvocableStrict<Type::InvalidLimit, size_t>
		&& ConstexprInvocableStrict<Type::InvalidQueueId, uint32_t>
		&& requires(Type* pExec) {			
			{ pExec->shared_from_this() };
			{ pExec->GetPriority() } -> std::same_as<uint32_t>;
			{ pExec->GetInstanceFlags() } -> std::same_as<Flags>;
			{ pExec->TestInstanceFlags(Flags(), bool()) } -> std::same_as<bool>;
			{ pExec->GetProcessingThread() } -> std::same_as<size_t>;
			{ pExec->IsProcessingThread() } -> std::same_as<bool>;
			{ pExec->Process(bool(), uint32_t()) } -> std::same_as<size_t>;
			{ pExec->Submit() } -> std::same_as<bool>;
			{ pExec->Finalize(Flags::Undefined) } -> std::same_as<bool>;
			{ pExec->IsFinalizing() } -> std::same_as<bool>;
			{ pExec->IsFinalized() } -> std::same_as<bool>;
			{ pExec->IsComplete() } -> std::same_as<bool>;
			{ pExec->Wait(Flags()) } -> std::same_as<bool>;
			{ pExec->template AddInstanceFlags<true>(Flags()) } -> std::same_as<Flags>;
			{ pExec->template ClearInstanceFlags<true>(Flags()) } -> std::same_as<Flags>;
			{ pExec->template GetExecutorId<true>() };
			{ pExec->template GetProcessorId<true>() };
			{ pExec->template GetSubmitsCounter<true>() };
			{ pExec->template GetBindingCounter<true>() };
			{ pExec->template GetStateFlags<true>() };
		}
	;

	template<typename TExecutor, typename TExecutable>
	concept Executor = 
		ResourcePolicy<typename TExecutor::ResourcePolicyT> 
		&& ResourcePolicy<typename TExecutable::ResourcePolicyT> 
		&& requires(TExecutor* pExecutor) {			
			{ pExecutor->GetInstanceId() } -> std::same_as<IdProvider::Id>;
		}
        && requires(TExecutor* pExecutor, typename TExecutable::ResourcePolicyT::template SharedT<TExecutable> const& spExecutable) {
            { pExecutor->Submit(spExecutable) } -> std::same_as<bool>;
        }
    ;

	template<typename TSharedExecutor, typename TExecutable>
	concept SharedExecutor = 
		requires { typename TSharedExecutor::element_type; }
		&& Executor<typename TSharedExecutor::element_type, TExecutable>
	;

}//FFPP::Concepts

namespace FFPP {

	////ExecutablePolicy//////////////////////////////////////////////////////////////////////////////////////////////////////////

	struct ExecutablePolicy { 
		using ResourcePolicyT = DefaultResourcePolicy;
		using LockableT = Lockable<TicketMutex<>>;
		static constexpr SchedulingFlags SchedulingMethod() { return SchedulingFlags::Default; }
		static constexpr uint32_t DefaultPriority() { return 0; }
	};

	////Executable////////////////////////////////////////////////////////////////////////////////////////////////////////////////

	template<typename TDerived, Concepts::ExecutablePolicy TPolicy = ExecutablePolicy>
	class Executable 
	: public std::enable_shared_from_this<TDerived>
	, public TPolicy::LockableT
	{
	public:
		using PolicyT = TPolicy;
		using ResourcePolicyT = PolicyT::ResourcePolicyT;
		using AtomicCounterT = std::atomic<size_t>;
		using AtomicIdT = std::atomic<IdProvider::Id>;
		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>;

		static_assert(
			TestModeFlags(PolicyT::SchedulingMethod(), SchedulingFlags::Once, true)
			|| !TestModeFlags(PolicyT::SchedulingMethod(), SchedulingFlags::Deferred, false) 
			|| !TestModeFlags(PolicyT::SchedulingMethod(), SchedulingFlags::Concurrent, false)
			, "Deferred and Concurrent scheduling modes are incompatible."
		);

		//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

		Executable(
			uint32_t 						nPriority 	= PolicyT::DefaultPriority()
			, Flags 						flgInstance	= Flags::Generic
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin 	= std::source_location::current()
			#endif
		) 
		: flgInstance_(flgInstance)
		, c_nPriority_(nPriority)
		#if FFPP_TRACK_ORIGIN
		, slOrigin_(slOrigin)
		#endif
		{ }

		Executable(Executable const&) = delete;

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

		//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

		static constexpr bool IsSubmittable() { return PolicyT::SchedulingMethod() != SchedulingFlags::Disabled; }

		static constexpr size_t InvalidValue() { return ResourcePolicyT::template InvalidValue<size_t>(); }
		static constexpr size_t InvalidLimit() { return ResourcePolicyT::template InvalidValue<size_t>(); }
		static constexpr uint32_t InvalidQueueId() { return ResourcePolicyT::template InvalidValue<uint32_t>(); }

        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

		FFPP_ATTR_INLINE
		static SharedT<TDerived> MakeShared(
			uint32_t 						nPriority 	= PolicyT::DefaultPriority()
			, Flags 						flgInstance	= Flags::Generic
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin 	= std::source_location::current()
			#endif
		) { 
			return ResourcePolicyT::template AllocateShared<TDerived>(
				nPriority
				, flgInstance
				#if FFPP_TRACK_ORIGIN
				, slOrigin
				#endif
			);
		}

		FFPP_ATTR_INLINE
		SharedT<TDerived> Self() { return this->shared_from_this(); }
		
		FFPP_ATTR_INLINE
		SharedT<TDerived const> Self() const { return this->shared_from_this(); }

		FFPP_ATTR_INLINE
		uint32_t GetPriority() const noexcept { return c_nPriority_; }

		FFPP_ATTR_INLINE
		Flags GetInstanceFlags() const noexcept { return flgInstance_; }

		FFPP_ATTR_INLINE
		bool TestInstanceFlags(Flags flg, bool bAll = false) const noexcept { 
			auto* pThis = static_cast<TDerived const*>(this);
			if(bAll) return (flg & pThis->GetInstanceFlags()) == flg; 
			else return (flg & pThis->GetInstanceFlags()) != Flags::Undefined;
		}

		FFPP_ATTR_INLINE
		std::source_location const& GetOrigin() const noexcept { return slOrigin_; }

		//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

		size_t GetProcessingThread() const noexcept {
			return tidProcessor_.load(std::memory_order::relaxed);
		}

		bool IsProcessingThread() const noexcept {
			return FFPP::GetThreadLid() == GetProcessingThread();
		}

		size_t Process(bool bAll = false, uint32_t idQueue = InvalidQueueId()) {
			return static_cast<TDerived*>(this)->OnProcess(bAll, idQueue); 
		}

		bool Submit() { 
			static_assert(IsSubmittable(), "Instances of this class can not be submitted for processing, scheduling disabled.");
			return static_cast<TDerived*>(this)->OnSubmit(); 
		}

		bool Finalize(Flags flgContext = Flags::Undefined) noexcept {
			bool const bFinalize = !bFinalizing_.exchange(true, std::memory_order::relaxed);
			if(bFinalize) {
				TryComplete();
				if constexpr(requires(TDerived* pThis) { pThis->OnFinalize(Flags::Undefined); }) {
					static_cast<TDerived*>(this)->OnFinalize(flgContext);
				}
				else {
					static_cast<TDerived*>(this)->OnFinalize();
				}
				bFinalized_.store(true, std::memory_order::release);
				bFinalized_.notify_all();
			}
			if(!!(flgContext & Flags::Wait)) Wait(flgContext);
			return bFinalize;
		}

		FFPP_ATTR_INLINE
		bool TryComplete() noexcept { return !bComplete_.exchange(true, std::memory_order::relaxed); }

		FFPP_ATTR_INLINE
		bool IsFinalizing() const noexcept { return bFinalizing_.load(std::memory_order::acquire); }

		FFPP_ATTR_INLINE
		bool IsFinalized() const noexcept { return bFinalized_.load(std::memory_order::acquire); }

		FFPP_ATTR_INLINE
		bool IsComplete() const noexcept { return bComplete_.load(std::memory_order::acquire); }

        bool Wait(Flags flgContext = Flags::Undefined) noexcept {
			return static_cast<TDerived*>(this)->OnWait(flgContext);
		}

		//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

		template<bool t_bInternalInvocation = false>
		Flags AddInstanceFlags(Flags flgAdd) noexcept { 
			static_assert(t_bInternalInvocation, "This method is intended for library internal use only.");
			return (flgInstance_ = flgInstance_ | flgAdd); 
		}

		template<bool t_bInternalInvocation = false>
		Flags ClearInstanceFlags(Flags flgClear) noexcept { 
			static_assert(t_bInternalInvocation, "This method is intended for library internal use only.");
			return (flgInstance_ = flgInstance_ & ~flgClear); 
		}

		template<bool t_bInternalInvocation = false>
		FFPP_ATTR_INLINE
		auto& GetExecutorId() noexcept { 
			static_assert(t_bInternalInvocation, "This method is intended for library internal use only.");
			return idExecutor_; 
		}

		template<bool t_bInternalInvocation = false>
		FFPP_ATTR_INLINE
		auto& GetProcessorId() noexcept { 
			static_assert(t_bInternalInvocation, "This method is intended for library internal use only.");
			return idProcessor_; 
		}

		template<bool t_bInternalInvocation = false>
		FFPP_ATTR_INLINE
		auto& GetSubmitsCounter() noexcept { 
			static_assert(t_bInternalInvocation, "This method is intended for library internal use only.");
			return nSubmits_; 
		}

		template<bool t_bInternalInvocation = false>
		FFPP_ATTR_INLINE
		auto& GetBindingCounter() noexcept { 
			static_assert(t_bInternalInvocation, "This method is intended for library internal use only.");
			return nBinding_; 
		}
		
		template<bool t_bInternalInvocation = false>
		FFPP_ATTR_INLINE
		auto& GetStateFlags() noexcept { 
			static_assert(t_bInternalInvocation, "This method is intended for library internal use only.");
			return flgState_; 
		}

	protected:
		friend class Factory<TDerived>;

		template<bool t_bInternalInvocation = false>
		void SetOrigin(std::source_location const& slOrigin) noexcept { 
			static_assert(t_bInternalInvocation, "This method is intended for library internal use only.");
			slOrigin_ = slOrigin; 
		}

		size_t OnProcess(bool, uint32_t) { return 0; }

		bool OnSubmit() { 
			return !bFinalizing_.load(std::memory_order::relaxed); 
		}

		void OnFinalize() noexcept { }

		bool OnWait(Flags flgContext) const noexcept { 
			bFinalized_.wait(false, std::memory_order::acquire); 
			return true;
		}

		size_t UpdateProcessingThread(size_t tid = FFPP::GetThreadLid()) noexcept {
			return tidProcessor_.exchange(tid, std::memory_order::relaxed);
		}

	protected:
		AtomicIdT
			//Implementation specific id of executor object which accepts Executable instance with its Submit() method for processing.
			idExecutor_ alignas(PolicyT::ResourcePolicyT::InterferenceSize()) = InvalidValue();
		AtomicCounterT
			//Implementation specific id of processor object invoking Process method, used by thread pool (FunctionalQueuePool).
			idProcessor_ = InvalidValue()
			//Counter of submission requests, used by thread pool.
			, nSubmits_ = 0
			//Binding counter of current instance binding to its processor object (idProcessor_), used by thread pool.
			, nBinding_ = 0
			//Implementation specific state flags, used by thread pool.
			, flgState_ = 0
		;
		std::atomic<size_t> 
			//A lightweight processing thread id. Actually, its a raw value of a pointer to (thread_local std::thread::id).
			tidProcessor_ = 0; 
		std::atomic<bool> 
			//Instance is finalized and inactive flag.
			bFinalized_ alignas(sizeof(CompatibleCounterT)) = false			
			//Instance finalization started flag.
			, bFinalizing_ = false
			//Instance completion pending flag.
			, bComplete_ = false
		;
		
		Flags 
			//Instance property flags, implies read-only access when instance is initialized and ready for use.
			flgInstance_ = Flags::Undefined; 

		uint32_t const
			//Instance processing priority.
			c_nPriority_ = PolicyT::DefaultPriority();
		std::source_location
			//Application source location of the point of instance construction.
			slOrigin_ = std::source_location::current();
	};//Executable

}//FFPP

#endif//FFPP_EXECUTABLE_HPP