++

Author: Zainullin Damir

include/ipfixprobe/outputPlugin/outputStorage/backoffScheme.hpp

@@ -19,9 +19,9 @@ class BackoffScheme {
 	bool backoff() noexcept
 	{
 		if (m_waitCounter < m_shortWaitThreshold) {
-			for (const auto _ : std::views::iota(0, 10'000)) {
-				asm volatile("pause" ::: "memory");
-			}
+			asm volatile("pause" ::: "memory");
+			/*for (const auto _ : std::views::iota(0, 10'000)) {
+			}*/
 
 		} else if (m_waitCounter < m_longWaitThreshold) {
 			std::this_thread::yield();

include/ipfixprobe/outputPlugin/outputStorage/ffq2OutputStorage.hpp

@@ -31,7 +31,7 @@ class FFQ2OutputStorage : public FFQOutputStorage<ElementType> {
 			throw std::runtime_error("Container read more times than there are reader groups.");
 		}*/
 		BackoffScheme backoffScheme(70, std::numeric_limits<std::size_t>::max());
-		const uint64_t writeRank = this->m_writeRank.fetch_add(1, std::memory_order_acq_rel);
+		const uint64_t writeRank = this->m_writeRank->fetch_add(1, std::memory_order_acq_rel);
 		const uint64_t writeIndex = writeRank % OutputStorage<ElementType>::STORAGE_CAPACITY;
 		while (!this->m_cells[writeIndex].state.tryToSetWriter()) {
 			backoffScheme.backoff();
@@ -56,13 +56,13 @@ class FFQ2OutputStorage : public FFQOutputStorage<ElementType> {
 				.state.setReadingFinished();
 			this->m_readersData[readerIndex]->lastReadIndex = std::nullopt;
 		}
-		const uint64_t readRank = this->m_readRank.fetch_add(1, std::memory_order_acq_rel);
+		const uint64_t readRank = this->m_readRank->fetch_add(1, std::memory_order_acq_rel);
 		const uint64_t readIndex = readRank % OutputStorage<ElementType>::STORAGE_CAPACITY;
-		while (readRank >= this->m_writeRank.load(std::memory_order_acquire)
+		while (readRank >= this->m_writeRank->load(std::memory_order_acquire)
 			   && this->writersPresent()) {
 			backoffScheme.backoff();
 		}
-		if (readRank >= this->m_writeRank.load(std::memory_order_acquire)) {
+		if (readRank >= this->m_writeRank->load(std::memory_order_acquire)) {
 			return nullptr;
 		}
 
@@ -82,8 +82,8 @@ class FFQ2OutputStorage : public FFQOutputStorage<ElementType> {
 		return !this->writersPresent() &&
 			// m_readRanks[readerGroupIndex].get() % ALLOCATION_BUFFER_CAPACITY
 			//== m_writeRank.load() % ALLOCATION_BUFFER_CAPACITY;
-			this->m_readRank.load(std::memory_order_acquire)
-			> this->m_writeRank.load(std::memory_order_acquire);
+			this->m_readRank->load(std::memory_order_acquire)
+			> this->m_writeRank->load(std::memory_order_acquire);
 	}
 
 private:

include/ipfixprobe/outputPlugin/outputStorage/ffqOutputStorage.hpp

@@ -27,7 +27,7 @@ class FFQOutputStorage : public OutputStorage<ElementType> {
 	{
 		BackoffScheme backoffScheme(70, 1);
 		while (true) {
-			const uint64_t writeRank = this->m_writeRank.fetch_add(1, std::memory_order_acq_rel);
+			const uint64_t writeRank = this->m_writeRank->fetch_add(1, std::memory_order_acq_rel);
 			const uint64_t writeIndex = writeRank % OutputStorage<ElementType>::STORAGE_CAPACITY;
 			if (
 				/*(m_storage[writeIndex].empty()
@@ -59,7 +59,7 @@ class FFQOutputStorage : public OutputStorage<ElementType> {
 			m_cells[*m_readersData[readerIndex]->lastReadIndex].state.setReadingFinished();
 		}
 		while (!finished()) {
-			const uint64_t readRank = m_readRank.fetch_add(1, std::memory_order_acq_rel);
+			const uint64_t readRank = m_readRank->fetch_add(1, std::memory_order_acq_rel);
 			const uint64_t readIndex = readRank % OutputStorage<ElementType>::STORAGE_CAPACITY;
 			if (m_cells[readIndex].state.tryToSetReadingStarted()) {
 				// std::atomic_thread_fence(std::memory_order_acquire);
@@ -77,8 +77,10 @@ class FFQOutputStorage : public OutputStorage<ElementType> {
 	bool finished() noexcept override
 	{
 		return !this->writersPresent()
-			&& m_readRank % OutputStorage<ElementType>::STORAGE_CAPACITY
-			== m_writeRank.load() % OutputStorage<ElementType>::STORAGE_CAPACITY;
+			&& m_readRank->load(std::memory_order_acquire)
+				% OutputStorage<ElementType>::STORAGE_CAPACITY
+			== m_writeRank->load(std::memory_order_acquire)
+				% OutputStorage<ElementType>::STORAGE_CAPACITY;
 	}
 
 protected:
@@ -166,27 +168,20 @@ class FFQOutputStorage : public OutputStorage<ElementType> {
 	struct Cell {
 		constexpr static uint64_t INVALID_RANK = std::numeric_limits<uint64_t>::max();
 
-		uint64_t rank;
+		// uint64_t rank;
 		ReaderGroupState state;
-		bool gap;
+		// bool gap;
 	};
 
 	struct ReaderData {
 		std::optional<uint16_t> lastReadIndex {0};
 	};
 
-	boost::container::static_vector<Cell, OutputStorage<ElementType>::STORAGE_CAPACITY> m_cells;
-	/*std::span<Cell> d_cells {
-		m_cells.data(),
-		OutputStorage<ElementType>::ALLOCATION_BUFFER_CAPACITY};*/
-	std::atomic<uint64_t> m_writeRank {0};
-	/*std::array<
-		CacheAlligned<std::atomic<uint64_t>>,
-		OutputStorage<ElementType>::MAX_READER_GROUPS_COUNT>
-		m_readRanks;*/
-	std::atomic<uint64_t> m_readRank {0};
 	std::array<CacheAlligned<ReaderData>, OutputStorage<ElementType>::MAX_READERS_COUNT>
 		m_readersData;
+	boost::container::static_vector<Cell, OutputStorage<ElementType>::STORAGE_CAPACITY> m_cells;
+	CacheAlligned<std::atomic<uint64_t>> m_writeRank {0};
+	CacheAlligned<std::atomic<uint64_t>> m_readRank {0};
 };
 
 } // namespace ipxp::output