include/ringqueue.hpp - main - Ant010ff/ffpp

include/ringqueue.hpp
545 строк20 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_RING_QUEUE_HPP
#define FFPP_RING_QUEUE_HPP

namespace FFPP::Concepts {

	template<typename TPolicy>
	concept RingQueuePolicy = 
		QueuePolicy<TPolicy>
		// && requires {
		// 	typename TPolicy::CounterT;
		// } 
		// && ConstexprInvocableStrict<TPolicy::QueueMode, QueueFlags> //queue mode flags
		// && Invocable<TPolicy::Backoff, BackoffHandlerT> //backoff handler generator
		&& Invocable<TPolicy::QueueCapacity, size_t> //fixed queue capacity, >= 2, rounded to power of 2
		&& Invocable<TPolicy::QueueMinimalCapacity, size_t> //queue minimal capacity, >= 2, rounded to power of 2
		//&& Invocable<TPolicy::OnCapacity, bool, size_t, size_t> //called on capacity change
	;//RingQueuePolicy

}//FFPP::Concepts

namespace FFPP {

	////RingQueuePolicy///////////////////////////////////////////////////////////////////////////////////////////////////////////

	struct RingQueuePolicy { 		
		using ResourcePolicyT = DefaultResourcePolicy;
		using CounterT = size_t; //uint32_t recommended when using QueueFlags::Wait in Linux environment (futex 32-bit requirement)

		//By default using scalable r/w mutex to protect dynamic ring-queue. Exclusive lock is required only when internal buffer 
		//reallocation is necessary. In all the rest cases enqueue/dequeue operations are performed concurrently under a shared 
		//lock. Fixed-capacity mode does not use mutex (QueueFlags::Grow and QueueFlags::Dynamic not specified).
		struct MutexPolicy : StripedTicketMutexPolicy {
			static size_t SharedSlots() { 
				auto const nThreads = std::thread::hardware_concurrency();
				if(nThreads < 2) return 1;
				//8 shared slots is a reasonable limit considering scalability limit of a Vyukov Bounded MPMC Queue algorithm.
				else return std::clamp(size_t(nThreads / 4), size_t(2), size_t(8)); 
			}
		};
		using MutexT = StripedTicketMutex<MutexFlags::Default, MutexPolicy>;
		using LockableT = Lockable<MutexT>;
		
		static constexpr auto Backoff() {
			//By default using periodic exponential CPU relaxation with factor 6 (loops from 2 to 64) followed by shorter exponential
			//phase with random duration for threads de-synchronization (up to 32 relaxation loops) and regular thread yield after 
			//exponential phases.
			return FFPP::JitteredBackoff<true, 6, 0, 0, 1, 5>();
		}
		
		static constexpr QueueFlags QueueMode() { return QueueFlags::Default; }
		
		static size_t QueueCapacity() { return 2 * std::thread::hardware_concurrency(); } //fixed queue capacity (if enabled)
		static size_t QueueMinimalCapacity() { return 8; } //queue minimal capacity
		static bool OnCapacity(size_t /*nCurrent*/, size_t /*nRequested*/) { return true; } //called on capacity change
	};

	////RingQueue/////////////////////////////////////////////////////////////////////////////////////////////////////////////////

	/*
		RingQueue, low-locking queue over fixed, growing (default) or growing/shrinking (dynamic) ring buffer based on the Bounded 
		MPMC queue algorithm by Dmitry Vyukov (https://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue). 
	*/
	template<std::movable TValue, Concepts::RingQueuePolicy TPolicy = RingQueuePolicy>
	class RingQueue : public TPolicy::LockableT {
	public:
		struct Policy : TPolicy {
			using CounterT = std::conditional_t<requires { typename TPolicy::CounterT; }, typename TPolicy::CounterT, size_t>;

			static constexpr QueueFlags QueueMode() {
				if constexpr(requires { TPolicy::QueueMode(); }) {
					return TPolicy::QueueMode();
				}
				else {
					QueueFlags flgMode = QueueFlags::Default;
					if constexpr(requires { TPolicy::FixedCapacity(); }) {
						if constexpr(TPolicy::FixedCapacity()) flgMode = QueueFlags::Undefined;
					}
					if constexpr(requires { TPolicy::WaitOnOverflow(bool { }); }) {
						if constexpr(TPolicy::WaitOnOverflow(false)) flgMode = flgMode | QueueFlags::Wait;
						if constexpr(TPolicy::WaitOnOverflow(true)) flgMode = flgMode | QueueFlags::Wait | QueueFlags::Obligate;
					}
					return flgMode;
				}				
			}

			static bool OnCapacity(size_t nCurrent, size_t nRequested) { 
				if constexpr(requires { TPolicy::OnCapacity(size_t { }, size_t { }); }) {
					return TPolicy::OnCapacity(nCurrent, nRequested); 
				}
				else if constexpr(requires { TPolicy::OnCapacity(size_t { }); }) {
					return TPolicy::OnCapacity(nRequested); 
				}
				return true;
			}

			static constexpr auto Backoff() {
				if constexpr(requires { TPolicy::Backoff(); }) {
					return TPolicy::Backoff(); 
				}
				else {
					return FFPP::JitteredBackoff<true, 6, 0, 0, 1, 5>();
				}
			}
		};//Policy

		using ValueT = TValue;
		using PolicyT = Policy;
		using CounterT = PolicyT::CounterT;
		using ResourcePolicyT = PolicyT::ResourcePolicyT;        
		using LockableT = PolicyT::LockableT;
		using UniqueLockT = LockableT::UniqueLockT;
		using SharedLockT = LockableT::SharedLockT;
		template<typename Type> 
		using AllocatorT = ResourcePolicyT::template AllocatorT<Type>;
		
		static size_t constexpr 
			c_nInvalidLimit = ResourcePolicyT::template InvalidValue<size_t>()
			, c_nCapacityLimit = size_t(1) << (std::numeric_limits<size_t>::digits - 1)
		;

		static consteval bool IsMode(QueueFlags flg, bool bAll = false) { 
			if(bAll) return (flg & PolicyT::QueueMode()) == flg; 
			else return (flg & PolicyT::QueueMode()) != QueueFlags::Undefined;
		}

		static consteval bool IsNoexcept() { 
			return c_bNoExEnqueue_ && c_bNoExDequeue_; 
		}

		static consteval bool IsSwapSemantics() { 
			return c_bSwap_; 
		}

		RingQueue(size_t nLimit = c_nInvalidLimit) 
		: nLimit_(nLimit)
		{ 
			if constexpr(c_bDynamicCapacity_) {
				mskCapacity_ = MinimalCapacity() - 1;
			}
			else {
				if(c_nInvalidLimit == nLimit) mskCapacity_ = FixedCapacity() - 1;
				else mskCapacity_ = std::bit_ceil(std::max(nLimit, FixedCapacity())) - 1;
			}
			vBuffer_ = MakeBuffer(mskCapacity_ + 1);
		}

		RingQueue(RingQueue const&) = delete;
		RingQueue(RingQueue&&) = delete;

		RingQueue& operator = (RingQueue const&) = delete;
		RingQueue& operator = (RingQueue&&) = delete;
		
		FFPP_ATTR_HOT_PATH
		bool Enqueue(ValueT&& vEnqueue, bool bObligate = false) noexcept(c_bNoExEnqueue_) {
			Slot* pSlot = nullptr;            
			CounterT iFront = 0, iBack = 0, iSequence = 0, nBackoffSkip = c_nBackoffSkip_;
			auto onBackoff = PolicyT::Backoff();
			if constexpr(!c_bDynamicCapacity_) {
				bool bBusyWait = true;
				iBack = iBack_.load(std::memory_order::relaxed);
				while(true) {
					pSlot = &vBuffer_[ToRingIndex2(iBack, mskCapacity_)];
					iSequence = pSlot->Sequence().load(std::memory_order::acquire);
					if(iSequence == iBack) [[likely]] { 
						if(iBack_.compare_exchange_weak(iBack, iBack + 1, std::memory_order::relaxed, std::memory_order::relaxed)) {
							if constexpr(c_bSwap_) pSlot->Swap(vEnqueue);
							else pSlot->Assign(std::move(vEnqueue));
							pSlot->Sequence().store(iBack + 1, std::memory_order::release);
							if constexpr(c_bWaitOnEmpty_) {
								//Notification chaining strategy:
								//1. Notify only one waiter (any).
								//2. Create a single-wakeup chain instead of thundering herd.
								iBack_.notify_one();
							}
							return true;
						}
						bBusyWait = TrySkipBackoff(onBackoff, nBackoffSkip);
					}
					else {
						if(iSequence + mskCapacity_ == iBack) { //queue full
							//Not enough space in ring buffer - return denial or loop with busy wait (QueueFlags::Wait).
							if(bObligate) {
								//In case of obligate enqueue check for Wait flag only.
								if constexpr(!c_bWaitOnOverflow_) break;
							}
							else {
								//In case of regular enqueue check for Wait and Obligate flags. The latter disables wait on 
								//regular enqueues, that is wait in obligate case only.
								if constexpr(!c_bWaitOnOverflow_ || IsMode(QueueFlags::Obligate)) break;
							}
						}
						bBusyWait = Backoff(onBackoff);
					}
					if constexpr(c_bWaitOnOverflow_) {
						//QueueFlags::Wait handling, that is busy-waiting loop series and sleep if contention was not mitigated.
						//Notification chaining strategy:
						//1. Each waiter calls notify_one() before re-spinning/re-waiting.
						//2. Create a single-wakeup chain instead of thundering herd.
						if(!bBusyWait) [[unlikely]] {
							iFront = iFront_.load(std::memory_order::relaxed);
							iBack = iBack_.load(std::memory_order::relaxed);
							if(iBack - iFront > mskCapacity_) {
								iFront_.wait(iFront, std::memory_order::acquire);
								iBack = iBack_.load(std::memory_order::relaxed);
								nBackoffSkip = c_nBackoffSkip_;
							}							
						}
						else {
							iBack = iBack_.load(std::memory_order::relaxed);
						}
					}
					else {
						iBack = iBack_.load(std::memory_order::relaxed);
					}
				}
			}
			else {
				//In a dynamic-capacity mode Enqueue/Dequeue pair requires exclusive access only when internal buffer reallocation 
				//is necessary. In case of a regular operation shared lock alows to act almost as if there is no lock at all.
				SharedLockT sl = this->GetSharedLock(); UniqueLockT ul;
				CounterT mskCapacity = mskCapacity_, nCapacity = 0, nGrow = 0;
				while(true) { //Outer loop required for shared lock upgrade logic.
					iBack = iBack_.load(std::memory_order::relaxed);
					while(true) { 
						pSlot = &vBuffer_[ToRingIndex2(iBack, mskCapacity_)];
						iSequence = pSlot->Sequence().load(std::memory_order::acquire);
						if(iSequence == iBack) [[likely]] { 
							if(iBack_.compare_exchange_weak(iBack, iBack + 1, std::memory_order::relaxed, std::memory_order::relaxed)) {
								if constexpr(c_bSwap_) pSlot->Swap(vEnqueue);
								else pSlot->Assign(std::move(vEnqueue));
								pSlot->Sequence().store(iBack + 1, std::memory_order::release);
								if constexpr(c_bWaitOnEmpty_) {
									//Notification chaining strategy:
									//1. Notify only one waiter (any).
									//2. Create a single-wakeup chain instead of thundering herd.
									iBack_.notify_one();
								}
								return true;
							}
							TrySkipBackoff(onBackoff, nBackoffSkip);
						}
						else {
							if(((iBack - iSequence) & ~(c_nCapacityLimit - 1)) == 0) { //queue full (close to), iSequence < iBack wrap-around aware
								nCapacity = mskCapacity_ + 1;
								nGrow = nCapacity << 1;
								if(nGrow >= c_nCapacityLimit) Except<>::Throw("Capacity limit reached (", nGrow, ")");
								if(nGrow < nCapacity) Except<>::Throw("Capacity overflow");
								if(!bObligate && Limit() < nGrow) return false; //Coarse limit check.
								//In case of unlimited (obligate) or normal enqueue within current limit test new capacity value 
								//with user specified policy and try to increase it if permitted [1].
								break;	
							}
							Backoff(onBackoff);
						}
						iBack = iBack_.load(std::memory_order::relaxed);
					}
					if(mskCapacity != mskCapacity_) {
						mskCapacity = mskCapacity_;
						continue;
					}
					if(sl.owns_lock()) { //Upgrade shared lock to exclusive and retry the whole operation.
						sl.unlock();
						ul = this->GetUniqueLock();
						continue;
					}
					//[1] Try to increase buffer under exclusive lock.
					if(!PolicyT::OnCapacity(nCapacity, nGrow)) return false;
					vBuffer_ = MoveBuffer(vBuffer_, iFront_.load(std::memory_order::relaxed), iBack, nGrow);
					mskCapacity_ = nGrow - 1;
				}
			}
			return false;
		}

		FFPP_ATTR_HOT_PATH
		bool Dequeue(ValueT& vDequeued) noexcept(c_bNoExDequeue_) {
			SharedLockT sl;
			//Acquire shared lock only if dynamic capacity is enabled.
			if constexpr(c_bDynamicCapacity_) sl = this->GetSharedLock();

			auto onBackoff = PolicyT::Backoff();
			Slot* pSlot = nullptr;
			CounterT 
				iFront = iFront_.load(std::memory_order::relaxed), 
				iBack = 0, 
				iNext = 0, 
				iSequence = 0, 
				nBackoffSkip = c_nBackoffSkip_;
			while(true) {
				iNext = iFront + 1;
				pSlot = &vBuffer_[ToRingIndex2(iFront, mskCapacity_)];
				iSequence = pSlot->Sequence().load(std::memory_order::acquire);
				if(iSequence == iNext) { 
					if(iFront_.compare_exchange_weak(iFront, iNext, std::memory_order::relaxed, std::memory_order::relaxed)) {
						if constexpr(c_bSwap_) pSlot->Swap(vDequeued);
						else vDequeued = pSlot->Move();
						pSlot->Sequence().store(iFront + mskCapacity_ + 1, std::memory_order::release);
						if constexpr(c_bShrinkableCapacity_) {							
							CounterT const
								mskCapacity = mskCapacity_
								, nCapacity = mskCapacity + 1
								, nMinimal = MinimalCapacity()
								, nShrink = nCapacity >> 1
							;
							if(nShrink > nMinimal) {
						 		//Queue size must be 4 times lower than its capacity to shrink it to the half on the current.
								CounterT const nLower = nCapacity >> 2;
								UniqueLockT ul;	
								while(true) {
									iBack = iBack_.load(std::memory_order::relaxed);
									iFront = iFront_.load(std::memory_order::relaxed);
									if(iBack - iFront < nLower && mskCapacity == mskCapacity_) {
										if(sl.owns_lock()) { //Upgrade shared lock to exclusive and retry the whole operation.
											sl.unlock();
											ul = this->GetUniqueLock();
											continue;
										}
										if(PolicyT::OnCapacity(nCapacity, nShrink)) {
											mskCapacity_ = nShrink - 1;
											vBuffer_ = MoveBuffer(vBuffer_, iFront, iBack, nShrink);
										}
									}
									break;
								}
							}
						}
						if constexpr(c_bWaitOnOverflow_) {
							//Notification chaining strategy:
							//1. Notify only one waiter (any).
							//2. Create a single-wakeup chain instead of thundering herd (see Enqueue).
							iFront_.notify_one();
						}
						return true;
					}
					TrySkipBackoff(onBackoff, nBackoffSkip);
				}
				else {
					if(iSequence == iFront) [[unlikely]] {
						if(iBack_.load(std::memory_order::relaxed) == iFront) { //queue empty
							if constexpr(c_bWaitOnEmpty_) {
								if(!Backoff(onBackoff)) [[unlikely]] { //Run back-off first.
									iBack = iBack_.load(std::memory_order::relaxed);
									iFront = iFront_.load(std::memory_order::relaxed);
									if(iBack == iFront) { //Sleep after busy-wait loops series.
										if constexpr(c_bDynamicCapacity_) sl.unlock();
										iBack_.wait(iBack, std::memory_order::acquire); 
										nBackoffSkip = c_nBackoffSkip_;
										if constexpr(c_bDynamicCapacity_) sl = this->GetSharedLock();
									}
								}
							}
							else {
								break; //Do not wait on empty (default) - break and return false.
							}
						}
						else {
							Backoff(onBackoff);
						}
					}
					else {
						Backoff(onBackoff);
					}
				}
				iFront = iFront_.load(std::memory_order::relaxed);
			}
			return false;
		}

		FFPP_ATTR_INLINE
		size_t Size() const noexcept { 
			return iBack_.load(std::memory_order::relaxed) - iFront_.load(std::memory_order::relaxed); 
		}

		FFPP_ATTR_INLINE
		bool Empty() const noexcept { 
			return iBack_.load(std::memory_order::relaxed) == iFront_.load(std::memory_order::relaxed); 
		}

		size_t Limit(size_t nLimit) noexcept { 
			if constexpr(c_bDynamicCapacity_) {
				return nLimit_.exchange(nLimit, std::memory_order::relaxed); 
			}
			else {
				return mskCapacity_ + 1; 
			}
		}

		FFPP_ATTR_INLINE
		size_t Limit() const noexcept { 
			if constexpr(!c_bDynamicCapacity_) return mskCapacity_ + 1; 
			else return nLimit_.load(std::memory_order::relaxed); 
		}

	private:        
		static constexpr bool 
			c_bDynamicCapacity_ = IsMode(QueueFlags::Dynamic, false)
			, c_bGrowableCapacity_ = IsMode(QueueFlags::Grow, false)
			, c_bShrinkableCapacity_ = IsMode(QueueFlags::Shrink, false)
			, c_bWaitOnOverflow_ = IsMode(QueueFlags::Wait | QueueFlags::Overflow, true) && !c_bDynamicCapacity_
			, c_bWaitOnEmpty_ = IsMode(QueueFlags::Wait | QueueFlags::Empty, true)
			, c_bNoExMoveConstruct_ = std::is_nothrow_move_constructible_v<ValueT>
			, c_bNoExMoveAssign_ = std::is_nothrow_move_assignable_v<ValueT>
			, c_bNoExMovable_ = c_bNoExMoveConstruct_ && c_bNoExMoveAssign_
			, c_bNoExSwappable_ = Concepts::NoexceptMemberSwappable<ValueT>
			, c_bSwappable_ = Concepts::MemberSwappable<ValueT> && std::default_initializable<ValueT>
			//Preferring noexcept value move over swap if the latter is not noexcept. If both are noexcept or not using swap when 
			//available and explicitly enabled in the queue policy.
			, c_bSwap_ = IsMode(QueueFlags::Swap, false) && c_bSwappable_ && (c_bNoExSwappable_ || !c_bNoExMovable_)
			, c_bNoExEnqueue_ = !c_bDynamicCapacity_ && (c_bSwap_ ? c_bNoExSwappable_ : c_bNoExMovable_)
			, c_bNoExDequeue_ = c_bNoExEnqueue_
		;
		static constexpr size_t
			c_nAlignment_ = PolicyT::ResourcePolicyT::InterferenceSize()
			, c_nBackoffSkip_ = 1
		;

		using AtomicIndexT = std::atomic<CounterT>;
		
		class alignas(c_nAlignment_) Slot {
		public:
			Slot() { }
			
			Slot(Slot&& that) noexcept(c_bNoExMovable_) { 
				Assign(std::move(that)); 
			}

			FFPP_ATTR_INLINE
			void operator = (Slot&& that) noexcept(c_bNoExMovable_) {
				Assign(std::move(that));
			}

			FFPP_ATTR_INLINE
			void Assign(Slot&& that) noexcept(c_bNoExMovable_) {
				vPayload_ = std::move(that.vPayload_);
				iSequence_.store(that.iSequence_.load(std::memory_order::relaxed), std::memory_order::relaxed);
			}

			FFPP_ATTR_INLINE
			void Assign(ValueT&& that) noexcept(c_bNoExMovable_) {
				vPayload_ = std::move(that);
			}
			
			FFPP_ATTR_INLINE
			void Swap(ValueT& that) noexcept(c_bNoExSwappable_) {
				vPayload_.swap(that);
			}

			FFPP_ATTR_INLINE
			ValueT Move() noexcept(c_bNoExMoveConstruct_) { 
				return std::move(vPayload_); 
			}

			FFPP_ATTR_INLINE
			AtomicIndexT& Sequence() noexcept { 
				return iSequence_; 
			}
		
		private:
			alignas(c_nAlignment_) AtomicIndexT 
				iSequence_ = 0;
			ValueT 
				vPayload_;
		};
		using SlotsVectorT = std::vector<Slot, AllocatorT<Slot>>;

		alignas(c_nAlignment_) SlotsVectorT
			vBuffer_;
		CounterT
			mskCapacity_ = 0;
		AtomicIndexT 
			nLimit_ = c_nInvalidLimit;
		alignas(c_nAlignment_) AtomicIndexT 
			iFront_ = 0;
		alignas(c_nAlignment_) AtomicIndexT 
			iBack_ = 0;

		static size_t MinimalCapacity() { 
			return std::max(size_t(2), std::bit_ceil(size_t(PolicyT::QueueMinimalCapacity()))); 
		}

		static size_t FixedCapacity() { 
			return size_t(std::bit_ceil(std::max(PolicyT::QueueCapacity(), MinimalCapacity()))); 
		}

		FFPP_ATTR_INLINE
		static CounterT ToRingIndex2(CounterT iAbsolute, CounterT mskCapacity) noexcept { 
			return iAbsolute & mskCapacity; 
		}

		template<typename TBackoff>
		FFPP_ATTR_INLINE
		static bool Backoff(TBackoff& onBackoff) noexcept {
			return onBackoff();
		}

		template<typename TBackoff>
		FFPP_ATTR_INLINE
		static bool TrySkipBackoff(TBackoff& onBackoff, CounterT& nSkip) noexcept {
			if(nSkip == 0) return Backoff(onBackoff);
			else nSkip--;
			return true;
		}

		static SlotsVectorT MakeBuffer(CounterT nCapacity) {
			if(nCapacity >= c_nCapacityLimit) Except<>::Throw("Invalid capacity (", nCapacity, ")");
			SlotsVectorT vBuffer;
			vBuffer.resize(nCapacity);
			for(CounterT i = 0; i < nCapacity; ++i) vBuffer[i].Sequence().store(i, std::memory_order::relaxed);
			return vBuffer;
		}

		static SlotsVectorT MoveBuffer(SlotsVectorT& vSource, CounterT iFront, CounterT iBack, CounterT nCapacity) {
			if(nCapacity >= c_nCapacityLimit) Except<>::Throw("Invalid capacity (", nCapacity, ")");
			if(iBack - iFront > nCapacity) Except<>::Throw("Target buffer capacity too small (", nCapacity, ")");
			CounterT const mskSource = vSource.size() - 1, mskTarget = nCapacity - 1;
			SlotsVectorT vTarget; vTarget.resize(nCapacity);
			for(CounterT iMove = iFront; iMove != iBack; ++iMove) {
				vTarget[ToRingIndex2(iMove, mskTarget)] = std::move(vSource[ToRingIndex2(iMove, mskSource)]);
			}
			bool bFront = false;
			for(CounterT iTarget = iBack - nCapacity; iTarget != iBack; ++iTarget) {
				auto& nSequence = vTarget[ToRingIndex2(iTarget, mskTarget)].Sequence();
				CounterT nValue = iTarget;
				if(iTarget == iFront) bFront = true;
				if(bFront) nValue += 1;
				else nValue += nCapacity;
				nSequence.store(nValue, std::memory_order::relaxed);
			}
			return vTarget;
		}
	};//RingQueue

}//FFPP

#endif//FFPP_RING_QUEUE_HPP