include/brokertaskloop.hpp - main - Ant010ff/ffpp

include/brokertaskloop.hpp
945 строк30 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_BROKER_TASK_LOOP_HPP
#define FFPP_FUNCTIONAL_BROKER_TASK_LOOP_HPP

namespace FFPP::dtl {

	////BrokerTaskLoop////////////////////////////////////////////////////////////////////////////////////////////////////////////

	template<typename TSharedTaskLoop>
    inline auto MakeBrokerJob(TSharedTaskLoop const& spTL) { 
		using TaskLoopT = typename TSharedTaskLoop::element_type;
		using BrokerT = typename TaskLoopT::BrokerT;
		using PolicyT = typename TaskLoopT::PolicyT;
		using ResourcePolicyT = typename PolicyT::ResourcePolicyT;
        struct Policy : PolicyT {
            static constexpr SchedulingFlags SchedulingMethod() { 
                return SchedulingFlags::Concurrent | (PolicyT::SchedulingMethod() & ~SchedulingFlags::Deferred); 
            }
        };
		static_assert(Policy::EnablePool(), "Broker's thread pool disabled by policy.");
        return MakeJob<Policy>(
			BrokerT::GetPool()
            , [wpTL = ResourcePolicyT::AsWeak(spTL)] () {
				auto spTL = wpTL.lock();
				if(!spTL || spTL->IsFinalizing()) return;
				assert(spTL->Size() == 1);
				if(spTL->Process() == 0) {
					Except<>::Throw(
						"Task loop concurrent processing failed @"
						#if FFPP_TRACK_ORIGIN
						, spTL->GetOrigin()
						#endif
					);
				}
			}
            , spTL->GetPriority()
            , spTL->GetInstanceFlags()
            #if FFPP_TRACK_ORIGIN
            , spTL->GetOrigin()
            #endif
        );
    }

	template<typename TBroker, typename TContext, typename TFailure, Concepts::FunctionalLoopPolicy TPolicy>
	class BrokerTaskLoop : public FunctionalLoop<BrokerTaskLoop<TBroker, TContext, TFailure, TPolicy>, TPolicy> {
	public:
		using PolicyT = TPolicy;
		using BrokerT = TBroker;
		using ResourcePolicyT = BrokerT::ResourcePolicyT;
		using IdProviderT = BrokerT::IdProviderT;
		using Id = BrokerT::Id;
		template<typename TSignature, bool t_bUnique = true>
		using FunctionT = ResourcePolicyT::template FunctionT<TSignature, t_bUnique>;
		using BrokerTaskLoopT = BrokerTaskLoop<TBroker, TContext, TFailure, TPolicy>;        
		using FunctionalLoopT = FunctionalLoop<BrokerTaskLoopT, TPolicy>;
		using FunctionalSequenceT = FunctionalLoopT::FunctionalSequenceT;
		friend FunctionalSequenceT;
		using ExecutableT = FunctionalSequenceT::ExecutableT;
		friend ExecutableT;
		using TestSubmitHandlerT = FunctionT<bool(BrokerTaskLoopT*), false>;
		using EventAcceptorT = EventAcceptorI<TContext, TFailure>;
		using ContextT = EventAcceptorT::ContextT;
		using FailureT = EventAcceptorT::FailureT;
		using EmptyT = EventAcceptorT::EmptyT;
		using StateT = EventAcceptorT::StateT;
		template<typename Type> 
		using SharedT = ResourcePolicyT::template SharedT<Type>;
		template<typename Type> 
		using WeakT = ResourcePolicyT::template WeakT<Type>;

		BrokerTaskLoop(
			ValOrRef<Id const>                  idTask
			, ValOrRef<Id const>                idGroup
			, uint32_t                          nPriority
			, Flags                             flgSequence
			, TestSubmitHandlerT&&              onTestSubmit
			, SharedT<EventAcceptorT> const&	spAcceptor
			#if FFPP_TRACK_ORIGIN
			, std::source_location const& 		slOrigin = std::source_location::current()
			#endif
		) 
		: FunctionalLoopT(
			nPriority
			, flgSequence
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		)
		, c_idTask_(idTask)
		, c_idGroup_(idGroup)
		, c_spAcceptor_(spAcceptor)
		, c_onTestSubmit_(std::move(onTestSubmit))
		, c_onFinalize_([
			idTask
			, idGroup
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		] (Flags flgContext) { 
			TBroker::template RemoveTask<TContext>(
				idTask
				, idGroup
				, flgContext
				#if FFPP_TRACK_ORIGIN
				, slOrigin
				#endif
			); 
		})
		{ }

		FFPP_ATTR_INLINE
		static SharedT<BrokerTaskLoopT> MakeShared(
			ValOrRef<Id const>                  idTask
			, ValOrRef<Id const>                idGroup
			, uint32_t                          nPriority
			, Flags                             flgSequence
			, TestSubmitHandlerT&&              onTestSubmit
			, SharedT<EventAcceptorT> const&	spAcceptor
			#if FFPP_TRACK_ORIGIN
			, std::source_location const& 		slOrigin = std::source_location::current()
			#endif
		) { 
			auto spTL = ResourcePolicyT::template AllocateShared<BrokerTaskLoopT>(
				idTask
				, idGroup
				, nPriority
				, flgSequence
				, std::move(onTestSubmit)
				, spAcceptor
				#if FFPP_TRACK_ORIGIN
				, slOrigin
				#endif
			);
			if constexpr(PolicyT::Asynchronous() && PolicyT::EnablePool()) {
				if(spTL->IsSequence(Flags::Concurrent)) {
					spTL->onSubmit_ = [spJob = MakeBrokerJob(spTL)] (BrokerTaskLoopT*) { 
						return spJob->Submit(); 
					};
					spTL->onProcess_ = [] (BrokerTaskLoopT* pThis) {
						return pThis->FunctionalLoopT::OnProcessAll(pThis->Size());
					};
					return spTL;
				}
			}
			assert(!spTL->IsSequence(Flags::Concurrent));
			spTL->onSubmit_ = [] (BrokerTaskLoopT* pThis) {
				//Thread bound execution serialization. It doesn't matter which thread exactly calls OnSubmit. But all requested 
				//calls (nSubmit_) must be performed (sequentially).
				if(pThis->BeginSubmit() == 0) {
					UniqueGuard sg { [pThis] () noexcept {
						pThis->nSubmit_.store(0, std::memory_order::relaxed);
					} };
					if constexpr(TBroker::PolicyT::Asynchronous()) {
						if(!pThis->c_spPool_->Submit(pThis->Self())) return false; 
					}
					else {
						do {
							if(pThis->Process() == 0) return false;
						} while(pThis->FinishSubmit() > 1 && !pThis->IsFinalizing());
					}
					sg.Abandon();
				}
				return true;
			};	
			spTL->onProcess_ = [] (BrokerTaskLoopT* pThis) {
				size_t nProcessed = pThis->FunctionalLoopT::OnProcessAll();
				if constexpr(TBroker::PolicyT::Asynchronous()) {
					if(pThis->FinishSubmit() > 1 && !pThis->IsFinalizing()) {
						if(!pThis->c_spPool_->Submit(pThis->Self())) {
							Except<>::Throw(
								"Task loop Submit failed @"
								#if FFPP_TRACK_ORIGIN
								, pThis->GetOrigin()
								#endif
							);
						}
					}
				}
				return nProcessed;
			};
			return spTL;
		}

		FFPP_ATTR_INLINE
		ValOrRef<Id const> GetTaskId() const noexcept { return c_idTask_; }

		FFPP_ATTR_INLINE
		ValOrRef<Id const> GetGroupId() const noexcept { return c_idGroup_; }

		FFPP_ATTR_INLINE
		operator ValOrRef<Id const> () const noexcept { return GetTaskId(); }

		FFPP_ATTR_INLINE
		bool Submit() { 
			if(this->IsSequence(Flags::Dongle)) return true;
			return this->OnSubmit(); 
		}

		template<typename TEmit>
		requires Concepts::OneOf<TEmit, ContextT, FailureT>
		FFPP_ATTR_INLINE
		bool Emit(TEmit const& ctx) {
			return c_spAcceptor_->Emit(ctx);
		}

		template<typename TEmit>
		requires Concepts::OneOf<TEmit, ContextT, FailureT>
		FFPP_ATTR_INLINE
		bool Emit(TEmit&& ctx) {
			return c_spAcceptor_->Emit(std::forward<TEmit>(ctx));
		}

		FFPP_ATTR_INLINE
		StateT Fetch() {
			return c_spAcceptor_->Fetch();
		}

		void Reset(bool bContinue = false) {
			c_spAcceptor_->Reset();
			auto ul = this->GetUniqueLock();
			UpdateActivityTime();
			if(bContinue) this->Continue();
		}

		FFPP_ATTR_INLINE
		bool Empty() const { return c_spAcceptor_->Empty(); }

		FFPP_ATTR_INLINE
		bool IsActive(Concepts::Duration auto const& drnTimeout) noexcept {
			if(this->IsSequence(Flags::Infinite)) return true;
			size_t const nProcess = nProcess_.load(std::memory_order::relaxed);
			auto const msTime = GetCompatibleTime();
			if(nProcess != nActivity_.exchange(nProcess, std::memory_order::relaxed)) {
				msActive_.store(msTime, std::memory_order::relaxed);
				return true;
			}
			return std::chrono::milliseconds { msTime - msActive_.load(std::memory_order::relaxed) } < drnTimeout;
		}

		FFPP_ATTR_INLINE
		void UpdateActivityTime() noexcept { 
			if(!this->IsSequence(Flags::Infinite)) msActive_.store(GetCompatibleTime(), std::memory_order::relaxed); 
		}

		FFPP_ATTR_INLINE
		size_t Submits() const noexcept { return nSubmits_.load(std::memory_order::relaxed); }

		template<bool t_bStrict = true>
		FFPP_ATTR_INLINE
		bool TrySubmit(size_t nPending) noexcept(t_bStrict) {
			auto nSubmits = nSubmits_.load(std::memory_order::relaxed);
			if constexpr(t_bStrict) {
				do {
					if(nSubmits + 1 != nPending) return false;
				} while(!nSubmits_.compare_exchange_weak(nSubmits, nPending, std::memory_order::relaxed, std::memory_order::relaxed));
			}
			else {
				size_t constexpr 
					c_nLimit = size_t(1) << (std::numeric_limits<size_t>::digits - 1)
					, c_mskLimit = ~(c_nLimit - 1)
				;
				do {
					if(nPending - nSubmits > c_nLimit) [[unlikely]] {
						Except<>::Throw(
							"Task loop pending/submit counters violation @"
							#if FFPP_TRACK_ORIGIN
							, this->GetOrigin()
							#endif
						);
					}
					if(((nPending - nSubmits - 1) & c_mskLimit) != 0) return false;
				} while(!nSubmits_.compare_exchange_weak(nSubmits, nSubmits + 1, std::memory_order::relaxed, std::memory_order::relaxed));
			}
			return true;
		}

	private:
		using AtomicCounterT = std::atomic<size_t>;
		
		Id const
			c_idGroup_ = IdProviderT::InvalidId(),
			c_idTask_ = IdProviderT::InvalidId();
		FunctionT<void(Flags), false> const
			c_onFinalize_;
		TestSubmitHandlerT const
			c_onTestSubmit_;
		ResourcePolicyT::template SharedT<EventAcceptorT> const
			c_spAcceptor_;
		ResourcePolicyT::template SharedT<typename TBroker::PoolT> const
			c_spPool_ = [] () { 
				using PoolT = ResourcePolicyT::template SharedT<typename TBroker::PoolT>;
				if constexpr(PolicyT::Asynchronous() && PolicyT::EnablePool()) return TBroker::GetPool();
				else return PoolT(); 
			} ();
		FunctionT<bool(BrokerTaskLoopT*)>
			onSubmit_,
			onProcess_;
		std::atomic<uint64_t>
			msActive_ = GetCompatibleTime();
		AtomicCounterT
			nSubmits_ = 0,
			nSubmit_ = 0,
			nProcess_ = 0,
			nActivity_ = 0;

		FFPP_ATTR_INLINE
		static uint64_t GetCompatibleTime() noexcept {
			using ClockT = std::chrono::system_clock;
			using DurationT = std::chrono::milliseconds;
			return std::chrono::duration_cast<DurationT>(ClockT::now().time_since_epoch()).count();
		}

		FFPP_ATTR_INLINE
		size_t BeginSubmit() noexcept { 
			return nSubmit_.fetch_add(1, std::memory_order::relaxed);
		}

		FFPP_ATTR_INLINE
		size_t FinishSubmit() noexcept { 
			auto nSubmit = nSubmit_.load(std::memory_order::relaxed);
			do {
				if(nSubmit == 0) break;
			} while(!nSubmit_.compare_exchange_weak(nSubmit, nSubmit - 1, std::memory_order::relaxed, std::memory_order::relaxed));
			return nSubmit;
		}

		FFPP_ATTR_HOT_PATH
		bool OnSubmit() { 
			if(!FunctionalLoopT::OnSubmit()) return false;
			if(c_onTestSubmit_ && !c_onTestSubmit_(this)) return true;
			return onSubmit_(this);
		}

		FFPP_ATTR_HOT_PATH
		size_t OnProcess(bool, uint32_t) {
			nProcess_.fetch_add(1, std::memory_order::relaxed);
			return onProcess_(this);
		}

		void OnFinalize(Flags flgContext) noexcept { 
			c_onFinalize_(flgContext);
			FunctionalLoopT::OnFinalize(); 
		}
	};//BrokerTaskLoop

	////BrokerTaskLoop<TBroker, Junction<>>///////////////////////////////////////////////////////////////////////////////////////

	template<typename TContext> 
	struct Junction { };
	template<typename TContext>
	struct ActualContext { using Type = TContext; };
	template<typename TContext>
	struct ActualContext<Junction<TContext>> { using Type = TContext; };
	template<typename TContext>
	using ActualContextT = ActualContext<TContext>::Type;
	
	template<typename TBroker, typename TContext, typename TFailure, Concepts::FunctionalSequencePolicy TPolicy>
	class BrokerTaskLoop<TBroker, Junction<TContext>, TFailure, TPolicy> 
	: public FunctionalLoop<BrokerTaskLoop<TBroker, Junction<TContext>, TFailure, TPolicy>, TPolicy> 
	{
	public:
		using PolicyT = TPolicy;
		using BrokerT = TBroker;
		using ResourcePolicyT = BrokerT::ResourcePolicyT;
		using IdProviderT = BrokerT::IdProviderT;
		using Id = BrokerT::Id;
		template<typename TSignature, bool t_bUnique = true>
		using FunctionT = ResourcePolicyT::template FunctionT<TSignature, t_bUnique>;
		using JunctionT = Junction<TContext>;
		using BrokerTaskLoopT = BrokerTaskLoop<TBroker, JunctionT, TFailure, TPolicy>;        
		using FunctionalLoopT = FunctionalLoop<BrokerTaskLoopT, TPolicy>;
		using FunctionalSequenceT = FunctionalLoopT::FunctionalSequenceT;
		friend FunctionalSequenceT;
		using ExecutableT = FunctionalSequenceT::ExecutableT;
		friend ExecutableT;
		using TestSubmitHandlerT = FunctionT<bool(BrokerTaskLoopT*), false>;
		using ContextT = TContext;
		using FailureT = TFailure;
		using EmptyT = FFPP::VoidT;
		using StateT = std::variant<ContextT, FailureT, EmptyT>;
		template<typename Type> 
		using SharedT = ResourcePolicyT::template SharedT<Type>;
		template<typename Type> 
		using WeakT = ResourcePolicyT::template WeakT<Type>;

		BrokerTaskLoop(
			ValOrRef<Id const>          	idTask
			, ValOrRef<Id const>          	idGroup
			, uint32_t                    	nPriority
			, Flags                       	flgSequence
			, TestSubmitHandlerT&&   		onTestSubmit
			#if FFPP_TRACK_ORIGIN
			, std::source_location const& 	slOrigin = std::source_location::current()
			#endif
		) 
		: FunctionalLoopT(
			nPriority
			, flgSequence
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		)
		, c_idTask_(idTask)
		, c_idGroup_(idGroup)
		, c_onTestSubmit_(std::move(onTestSubmit))
		, c_onFinalize_([
			idTask
			, idGroup
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		] (Flags flgContext) {
			TBroker::template RemoveTask<Junction<TContext>>(
				idTask
				, idGroup
				, flgContext
				#if FFPP_TRACK_ORIGIN
				, slOrigin
				#endif
			);
			auto spST = GetStateTable();
			if(spST) spST->Remove(idTask);
		})
		, nDeferred_([idTask] () {
			auto spST = GetStateTable();
			if(!spST) Except<>::Throw("Null state table pointer.");
			return spST->ForLockedObject(idTask, [] (auto, auto const& sdqState) { return sdqState ? sdqState->size() : 1; });
		} ())
		{ }

		FFPP_ATTR_INLINE
		static SharedT<BrokerTaskLoopT> MakeShared(
			ValOrRef<Id const>              idTask
			, ValOrRef<Id const>            idGroup
			, uint32_t                      nPriority
			, Flags                         flgSequence
			, TestSubmitHandlerT&&          onTestSubmit
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin = std::source_location::current()
			#endif
		) { 
			auto spTL = ResourcePolicyT::template AllocateShared<BrokerTaskLoopT>(
				idTask
				, idGroup
				, nPriority
				, flgSequence
				, std::move(onTestSubmit)
				#if FFPP_TRACK_ORIGIN
				, slOrigin
				#endif
			);
			if constexpr(PolicyT::Asynchronous() && PolicyT::EnablePool()) {
				if(spTL->IsSequence(Flags::Concurrent)) {
					spTL->onSubmit_ = [spJob = MakeBrokerJob(spTL)] (BrokerTaskLoopT*) { 
						return spJob->Submit(); 
					};
					spTL->onProcess_ = [] (BrokerTaskLoopT* pThis) {
						return pThis->FunctionalLoopT::OnProcessAll(pThis->Size());
					};
					return spTL;
				}
			}
			assert(!spTL->IsSequence(Flags::Concurrent));
			spTL->onSubmit_ = [] (BrokerTaskLoopT* pThis) {
				//Thread bound execution serialization. It doesn't matter which thread exactly calls OnSubmit. But all 
				//requested calls (nSubmit_) must be performed (sequentially).
				if(pThis->BeginSubmit() == 0) {
					UniqueGuard sg { [pThis] () noexcept {
						pThis->nSubmit_.store(0, std::memory_order::relaxed);
					} };
					if constexpr(TBroker::PolicyT::Asynchronous()) {
						if(!pThis->c_spPool_->Submit(pThis->Self())) return false; 
					}
					else {
						do {
							if(pThis->Process() == 0) return false;
						} while(pThis->FinishSubmit() > 1 && !pThis->IsFinalizing());
					}
					sg.Abandon();
				}
				return true;
			};	
			spTL->onProcess_ = [] (BrokerTaskLoopT* pThis) {
				size_t nProcessed = pThis->FunctionalLoopT::OnProcessAll();
				if constexpr(TBroker::PolicyT::Asynchronous()) {
					if(pThis->FinishSubmit() > 1 && !pThis->IsFinalizing()) {
						if(!pThis->c_spPool_->Submit(pThis->Self())) {
							Except<>::Throw(
								"Task loop Submit failed @"
								#if FFPP_TRACK_ORIGIN
								, pThis->GetOrigin()
								#endif
							);
						}
					}
				}
				return nProcessed;
			};
			return spTL;
		}

		FFPP_ATTR_INLINE
		ValOrRef<Id const> GetTaskId() const noexcept { return c_idTask_; }

		FFPP_ATTR_INLINE
		ValOrRef<Id const> GetGroupId() const noexcept { return c_idGroup_; }

		FFPP_ATTR_INLINE
		operator ValOrRef<Id const> () const noexcept { return GetTaskId(); }

		template<typename TEmit>
		requires Concepts::OneOf<TEmit, ContextT, FailureT>
		FFPP_ATTR_HOT_PATH
		static bool Emit(ValOrRef<Id const> id, TEmit const& ctx) {
			auto spST = GetStateTable();
			if(!spST) Except<>::Throw("Null state table pointer.");
			return spST->ForLockedObject(
				id, 
				[&ctx] (auto, auto const& sdqState) {
					if(!sdqState) Except<>::Throw("Failed to acquire state queue object.");
					sdqState->emplace_back(ctx);
					return true;
				},
				MakeStateQueue
			);
		}

		template<typename TEmit>
		requires Concepts::OneOf<TEmit, ContextT, FailureT>
		FFPP_ATTR_HOT_PATH
		static bool Emit(ValOrRef<Id const> id, TEmit&& ctx) {
			auto spST = GetStateTable();
			if(!spST) Except<>::Throw("Null state table pointer.");
			return spST->ForLockedObject(
				id, 
				[ctx = std::move(ctx)] (auto, auto const& sdqState) mutable {
					if(!sdqState) Except<>::Throw("Failed to acquire state queue object.");
					sdqState->emplace_back(std::move(ctx));
					return true;
				},
				MakeStateQueue
			);
		}

		template<typename TEmit>
		requires Concepts::OneOf<TEmit, ContextT, FailureT>
		FFPP_ATTR_HOT_PATH
		bool Emit(TEmit const& ctx) {
			return Emit(GetTaskId(), ctx);
		}

		template<typename TEmit>
		requires Concepts::OneOf<TEmit, ContextT, FailureT>
		FFPP_ATTR_HOT_PATH
		bool Emit(TEmit&& ctx) {
			return Emit(GetTaskId(), std::forward<TEmit>(ctx));
		}

		FFPP_ATTR_HOT_PATH
		StateT Fetch() {
			auto spST = GetStateTable();
			if(!spST) Except<>::Throw("Null state table pointer.");
			return spST->ForLockedObject(
				GetTaskId(), 
				[bSignal = this->IsSequence(Flags::Signal)] (auto, auto const& sdqState)->StateT {
					if(!sdqState) Except<>::Throw("Failed to acquire state queue object.");
					if(sdqState->empty()) {
						return EmptyT { };
					}
					else if(bSignal) {
						while(sdqState->size() > 1) sdqState->pop_front();
						return sdqState->front();
					}
					else {
						StateT varState = EmptyT { };
						sdqState->front().swap(varState);
						sdqState->pop_front();
						return varState;
					}
				},
				MakeStateQueue
			);
		}

		FFPP_ATTR_HOT_PATH
		bool Empty() {
			auto spST = GetStateTable();
			if(!spST) Except<>::Throw("Null state table pointer.");
			return spST->ForLockedObject(
				GetTaskId(), 
				[] (auto, auto const& sdqState) {
					return sdqState ? sdqState->empty() : true;
				}
			);
		}

		void Reset(bool bContinue = false) {
			auto spST = GetStateTable();
			if(!spST) Except<>::Throw("Null state table pointer.");
			spST->ForLockedObject(
				GetTaskId(), 
				[] (auto, auto const& sdqState) {
					if(!sdqState) Except<>::Throw("Failed to acquire state queue object.");
					sdqState->clear();
				},
				MakeStateQueue
			);
			UpdateActivityTime();
			if(bContinue) this->Continue();
		}

		FFPP_ATTR_INLINE
		bool IsActive(Concepts::Duration auto const& drnTimeout) noexcept {
			if(this->IsSequence(Flags::Infinite)) return true;
			size_t const nProcess = nProcess_.load(std::memory_order::relaxed);
			auto const msTime = GetCompatibleTime();
			if(nProcess != nActivity_.exchange(nProcess, std::memory_order::relaxed)) {
				msActive_.store(msTime, std::memory_order::relaxed);
				return true;
			}
			return std::chrono::milliseconds { msTime - msActive_.load(std::memory_order::relaxed) } < drnTimeout;
		}

		FFPP_ATTR_INLINE
		void UpdateActivityTime() noexcept { 
			if(!this->IsSequence(Flags::Infinite)) msActive_.store(GetCompatibleTime(), std::memory_order::relaxed); 
		}

		FFPP_ATTR_INLINE
		size_t Submits() const noexcept { return nSubmits_.load(std::memory_order::relaxed); }

		FFPP_ATTR_INLINE
		bool TrySubmit(size_t nPending) noexcept {
			auto nSubmits = nSubmits_.load(std::memory_order::relaxed);
			do {
				if(nSubmits + 1 != nPending) return false;
			} while(!nSubmits_.compare_exchange_weak(nSubmits, nPending, std::memory_order::relaxed, std::memory_order::relaxed));
			return true;
		}

	private:
		using StateQueueT = std::deque<StateT, typename ResourcePolicyT::template AllocatorT<StateT>>;
		using StateTableT = StripedTable<StateQueueT, false, PolicyT>;
		using SharedStateQueueT = StateTableT::SharedObjectT;
		using AtomicCounterT = std::atomic<size_t>;
		
		Id const
			c_idGroup_ = IdProviderT::InvalidId(),
			c_idTask_ = IdProviderT::InvalidId();
		FunctionT<void(Flags), false> const
			c_onFinalize_;
		ResourcePolicyT::template SharedT<typename TBroker::PoolT> const
			c_spPool_ = [] () { 
				using PoolT = ResourcePolicyT::template SharedT<typename TBroker::PoolT>;
				if constexpr(PolicyT::Asynchronous() && PolicyT::EnablePool()) return TBroker::GetPool(); 
				else return PoolT(); 
			} ();
		TestSubmitHandlerT const
			c_onTestSubmit_;
		FunctionT<bool(BrokerTaskLoopT*)>
			onSubmit_,
			onProcess_;
		std::atomic<uint64_t>
			msActive_ = GetCompatibleTime();
		AtomicCounterT
			nSubmits_ = 0,
			nSubmit_ = 0,
			nDeferred_ = 1,
			nProcess_ = 0,
			nActivity_ = 0;

		static SharedStateQueueT MakeStateQueue(ValOrRef<Id const>) { 
			return ResourcePolicyT::template AllocateShared<StateQueueT>(); 
		}

		FFPP_ATTR_INLINE
		static ResourcePolicyT::template SharedT<StateTableT> GetStateTable() {
			static auto s_spStateTable = ResourcePolicyT::template AllocateShared<StateTableT>(PolicyT::TableStripes());
			return s_spStateTable;
		}

		FFPP_ATTR_INLINE
		static uint64_t GetCompatibleTime() noexcept {
			using ClockT = std::chrono::system_clock;
			using DurationT = std::chrono::milliseconds;
			return std::chrono::duration_cast<DurationT>(ClockT::now().time_since_epoch()).count();
		}

		FFPP_ATTR_INLINE
		size_t BeginSubmit() noexcept { 
			auto nDeferred = nDeferred_.load(std::memory_order::acquire); 
			if(nDeferred == 1) {
				return nSubmit_.fetch_add(1, std::memory_order::relaxed); 
			}			
			nDeferred_.compare_exchange_strong(nDeferred, 1, std::memory_order::release, std::memory_order::acquire);
			return nSubmit_.fetch_add(nDeferred, std::memory_order::relaxed); 
		}

		FFPP_ATTR_INLINE
		size_t FinishSubmit() noexcept { 
			auto nSubmit = nSubmit_.load(std::memory_order::relaxed);
			do {
				if(nSubmit == 0) break;
			} while(!nSubmit_.compare_exchange_weak(nSubmit, nSubmit - 1, std::memory_order::relaxed, std::memory_order::relaxed));
			return nSubmit;
		}

		FFPP_ATTR_HOT_PATH
		bool OnSubmit() { 
			if(!FunctionalLoopT::OnSubmit()) return false;
			if(c_onTestSubmit_ && !c_onTestSubmit_(this)) return true;
			return onSubmit_(this);
		}

		FFPP_ATTR_HOT_PATH
		size_t OnProcess(bool, uint32_t) {
			nProcess_.fetch_add(1, std::memory_order::relaxed);
			return onProcess_(this);
		}

		void OnFinalize(Flags flgContext) noexcept { 
			c_onFinalize_(flgContext);
			FunctionalLoopT::OnFinalize(); 
		}
	};//BrokerTaskLoop<TBroker, Junction<>>

	////BrokerTaskLoop<TBroker, TBroker::ChainContextT>///////////////////////////////////////////////////////////////////////////
	
	template<typename TBroker, typename TFailure, Concepts::FunctionalSequencePolicy TPolicy>
	class BrokerTaskLoop<TBroker, typename TBroker::ChainContextT, TFailure, TPolicy> 
	: public FunctionalLoop<BrokerTaskLoop<TBroker, typename TBroker::ChainContextT, TFailure, TPolicy>, TPolicy> 
	{
	public:
		using PolicyT = TPolicy;
		using BrokerT = TBroker;
		using ResourcePolicyT = BrokerT::ResourcePolicyT;
		using IdProviderT = BrokerT::IdProviderT;
		using Id = BrokerT::Id;
		using ChainContextT = TBroker::ChainContextT;
		template<typename TSignature, bool t_bUnique = true>
		using FunctionT = ResourcePolicyT::template FunctionT<TSignature, t_bUnique>;
		using BrokerTaskLoopT = BrokerTaskLoop<TBroker, ChainContextT, TFailure, TPolicy>;
		using FunctionalLoopT = FunctionalLoop<BrokerTaskLoopT, TPolicy>;
		using FunctionalSequenceT = FunctionalLoopT::FunctionalSequenceT;
		friend FunctionalSequenceT;
		using ExecutableT = FunctionalSequenceT::ExecutableT;
		friend ExecutableT;
		using TestSubmitHandlerT = FunctionT<bool(BrokerTaskLoopT*), false>;
		template<typename Type> 
		using SharedT = ResourcePolicyT::template SharedT<Type>;
		template<typename Type> 
		using WeakT = ResourcePolicyT::template WeakT<Type>;

		BrokerTaskLoop(
			ValOrRef<Id const>          	idTask
			, ValOrRef<Id const>          	idGroup
			, uint32_t                    	nPriority
			, Flags                       	flgSequence
			, TestSubmitHandlerT&&   		onTestSubmit
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin = std::source_location::current()
			#endif
		) 
		: FunctionalLoopT(
			nPriority
			, flgSequence & ~Flags::Concurrent
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		)
		, c_idTask_(idTask)
		, c_idGroup_(idGroup)
		, c_onFinalize_([
			idTask
			, idGroup
			#if FFPP_TRACK_ORIGIN
			, slOrigin
			#endif
		] (Flags flgContext) { 
			TBroker::RemoveChain(
				idTask
				, idGroup
				, flgContext
				#if FFPP_TRACK_ORIGIN
				, slOrigin
				#endif
			); 
		})
		{ 
			static_assert(
				PolicyT::Asynchronous(), 
				"Functional chaining is not available with synchronous broker, check broker's policy. "
				"Chain processing inside single thread is possible with single threaded asynchronous broker."
			);
		}

		FFPP_ATTR_INLINE
		static SharedT<BrokerTaskLoopT> MakeShared(
			ValOrRef<Id const>              idTask
			, ValOrRef<Id const>            idGroup
			, uint32_t                      nPriority
			, Flags                         flgSequence
			, TestSubmitHandlerT&&          onTestSubmit
			#if FFPP_TRACK_ORIGIN
			, std::source_location const&	slOrigin = std::source_location::current()
			#endif
		) { 
			return ResourcePolicyT::template AllocateShared<BrokerTaskLoopT>(
				idTask
				, idGroup
				, nPriority
				, flgSequence
				, std::move(onTestSubmit)
				#if FFPP_TRACK_ORIGIN
				, slOrigin
				#endif
			);
		}

		FFPP_ATTR_INLINE
		bool TryChain() {
			if(this->IsCommitted()) return true;
			AtomicCounterT::value_type tidChainer = 0, tidThis = GetThreadLid();
			if(tidChainer_.compare_exchange_strong(tidChainer, tidThis, std::memory_order::relaxed)) return true;
			return tidChainer == tidThis;
		}

		FFPP_ATTR_INLINE
		bool Busy() const noexcept { return bBusy_.load(std::memory_order::relaxed); }
		
		FFPP_ATTR_INLINE
		void Await() const noexcept { bBusy_.wait(true, std::memory_order::acquire); }
		
		FFPP_ATTR_INLINE
		size_t Submits() const noexcept { return nSubmits_.load(std::memory_order::relaxed); }
		
		FFPP_ATTR_INLINE
		size_t Forward() noexcept { return nNext_.fetch_add(1, std::memory_order::relaxed); }
		
		FFPP_ATTR_INLINE
		bool TryForward() noexcept { 
			auto nNext = nNext_.load(std::memory_order::relaxed);
			do {
				if(nNext == 0) return false;
			} while(!nNext_.compare_exchange_weak(nNext, nNext - 1, std::memory_order::relaxed, std::memory_order::relaxed));
			return true;
		}

		FFPP_ATTR_INLINE
		size_t Continue() noexcept { return nContinue_.fetch_add(1, std::memory_order::relaxed); }

		FFPP_ATTR_INLINE
		bool TryContinue() noexcept {
			auto nContinue = nContinue_.load(std::memory_order::relaxed);
			do {
				if(nContinue == 0) return false;
			} while(!nContinue_.compare_exchange_weak(nContinue, nContinue - 1, std::memory_order::relaxed, std::memory_order::relaxed));
			FunctionalLoopT::Continue();
			return true;
		}

		FFPP_ATTR_INLINE
		ValOrRef<Id const> GetTaskId() const noexcept { return c_idTask_; }

		FFPP_ATTR_INLINE
		ValOrRef<Id const> GetGroupId() const noexcept { return c_idGroup_; }

		FFPP_ATTR_INLINE
		operator ValOrRef<Id const> () const noexcept { return GetTaskId(); }

	private:
		using AtomicCounterT = std::atomic<size_t>;
		Id const
			c_idGroup_ = IdProviderT::InvalidId(),
			c_idTask_ = IdProviderT::InvalidId();
		FunctionT<void(Flags), false> const
			c_onFinalize_;
		AtomicCounterT
			tidChainer_ = 0,
			nSubmits_ = 0,
			nSubmit_ = 0,
			nProcess_ = 0,
			nNext_ = 0,
			nContinue_ = 0;
		std::atomic<bool>
			bBusy_ alignas(sizeof(CompatibleCounterT)) = false;

		FFPP_ATTR_HOT_PATH
		bool OnSubmit() { 
			if(!FunctionalSequenceT::OnSubmit()) return false; //Bypass FunctionalLoopT commit flag.
			nSubmit_.fetch_add(1, std::memory_order::relaxed);
			nSubmits_.fetch_add(1, std::memory_order::relaxed);
			if(!bBusy_.exchange(true, std::memory_order::relaxed)) {
				return this->Process() > 0;
			}
			else {
				return true;
			}
		}

		FFPP_ATTR_HOT_PATH
		size_t OnProcess(bool, uint32_t) {
			//Thread bound execution serialization. Required here because BrokerTaskLoop<TBroker::ChainContextT>::OnProcess can be 
			//called simultaneously from two threads (submitting and processing ones) during chain processing.
			if(nProcess_.fetch_add(1, std::memory_order::relaxed) > 0) return 0;
			size_t nProcessed = 0;
			bool bAgain = false;
			do if(!this->IsFinalizing()) {
				UniqueGuard sg { [this, &bAgain] () noexcept { 
					bAgain = nProcess_.fetch_sub(1, std::memory_order::relaxed) > 1; 
				} };
				nProcessed += FunctionalLoopT::OnProcessSingle(); 
				if(this->Ready()) {
					if(nSubmit_.fetch_sub(1, std::memory_order::relaxed) > 1) {
						nProcessed += FunctionalLoopT::OnProcessSingle(); 
					}
					else {
						bBusy_.store(false, std::memory_order::release);
						bBusy_.notify_all();
					}
				}
				sg.Finalize();
				if(bAgain) std::this_thread::yield();
			} while(std::exchange(bAgain, false));
			return nProcessed;
		}

		void OnFinalize(Flags flgContext) noexcept { 
			c_onFinalize_(flgContext);
			FunctionalLoopT::OnFinalize(); 
		}
	};//BrokerTaskLoop<TBroker, TBroker::ChainContextT>

}//FFPP::dtl

#endif//FFPP_FUNCTIONAL_BROKER_TASK_LOOP_HPP