include/exclusiveringqueue.hpp - main - Ant010ff/ffpp

include/exclusiveringqueue.hpp
144 строки5 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_EXCLUSIVE_RING_QUEUE_HPP
#define FFPP_EXCLUSIVE_RING_QUEUE_HPP

namespace FFPP::Concepts {

	template<typename TPolicy>
	concept ExclusiveRingQueuePolicy = QueuePolicy<TPolicy>;

}//FFPP::Concepts

namespace FFPP {

	////ExclusiveRingQueuePolicy//////////////////////////////////////////////////////////////////////////////////////////////////

	struct ExclusiveRingQueuePolicy { 
		using ResourcePolicyT = DefaultResourcePolicy;
		using LockableT = Lockable<TicketMutex<>>;
	};

	////ExclusiveRingQueue////////////////////////////////////////////////////////////////////////////////////////////////////////

	template<std::movable TValue, Concepts::ExclusiveRingQueuePolicy TPolicy = ExclusiveRingQueuePolicy>
	class ExclusiveRingQueue : public TPolicy::LockableT {
	public:
		using ValueT = TValue;
		using PolicyT = TPolicy;
		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)
		;

		ExclusiveRingQueue(size_t nLimit = c_nInvalidLimit) 
		: nLimit_(nLimit)
		{ }
		
		ExclusiveRingQueue(ExclusiveRingQueue const&) = delete;
		ExclusiveRingQueue(ExclusiveRingQueue&&) = delete;

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

		FFPP_ATTR_HOT_PATH
		bool Enqueue(ValueT&& vEnqueue, bool bObligate = false) {
			auto ul = this->GetUniqueLock();
			size_t nCapacity = mskCapacity_ + 1, nSize = Size();
			if(!bObligate && Limit() <= nSize) return false;
			if(nCapacity == nSize) {
				size_t const nGrow = nCapacity << 1;
				vBuffer_ = MoveBuffer(vBuffer_, iFront_, iBack_, nGrow);
				nCapacity = nGrow;
				mskCapacity_ = nCapacity - 1;
			}
			vBuffer_[ToRingIndex2(iBack_++, mskCapacity_)] = std::move(vEnqueue);
			nEnqueued_.fetch_add(1, std::memory_order::relaxed);
			return true;
		}

		FFPP_ATTR_HOT_PATH
		bool Dequeue(ValueT& vDequeued) {
			auto ul = this->GetUniqueLock();
			if(iBack_ != iFront_) {
				vDequeued = std::move(vBuffer_[ToRingIndex2(iFront_++, mskCapacity_)]);
				nEnqueued_.fetch_sub(1, std::memory_order::relaxed);
				return true;
			}
			return false;
		}
		
		FFPP_ATTR_INLINE
		size_t Capacity() const { 
			auto sl = this->GetSharedLock();
			return mskCapacity_ + 1; 
		}

		FFPP_ATTR_INLINE
		size_t Size() const { return nEnqueued_.load(std::memory_order::relaxed); }

		FFPP_ATTR_INLINE
		bool Empty() const { return Size() == 0; }

		FFPP_ATTR_INLINE
		size_t Limit(size_t nLimit) { return nLimit_.exchange(nLimit, std::memory_order::relaxed); }

		FFPP_ATTR_INLINE
		size_t Limit() { return nLimit_.load(std::memory_order::relaxed); }

	private:        
		static constexpr size_t
			c_nInitialCapacity_ = 1;

		using SlotT = ValueT;
		using SlotsVectorT = std::vector<SlotT, AllocatorT<SlotT>>;

		SlotsVectorT 
			vBuffer_ = MakeBuffer(c_nInitialCapacity_);
		size_t
			mskCapacity_ = c_nInitialCapacity_ - 1,
			iFront_ = 0,
			iBack_ = 0;
		std::atomic<size_t>
			nLimit_ = c_nInvalidLimit,
			nEnqueued_ = 0;

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

		static SlotsVectorT MakeBuffer(size_t nCapacity) {
			SlotsVectorT vBuffer;
			vBuffer.resize(nCapacity);
			return vBuffer;
		}

		static SlotsVectorT MoveBuffer(SlotsVectorT& vSource, size_t iFront, size_t iBack, size_t nCapacity) {
			if(nCapacity >= c_nCapacityLimit) Except<>::Throw("Invalid capacity (", nCapacity, ")");
			if(iBack - iFront > nCapacity) Except<>::Throw("Target buffer capacity too small (", nCapacity, ")");
			size_t const mskSource = vSource.size() - 1, mskTarget = nCapacity - 1;
			SlotsVectorT vTarget; vTarget.resize(nCapacity);
			for(size_t iMove = iFront; iMove != iBack; ++iMove) {
				vTarget[ToRingIndex2(iMove, mskTarget)] = std::move(vSource[ToRingIndex2(iMove, mskSource)]);
			}
			return vTarget;
		}
	};//ExclusiveRingQueue

}//FFPP

#endif//FFPP_EXCLUSIVE_RING_QUEUE_HPP