++

Author: Zainullin Damir

include/ipfixprobe/outputPlugin/outputStorage/allocationBuffer2.hpp

@@ -37,8 +37,8 @@ class AllocationBuffer2 : public AllocationBufferBase<ElementType> {
 			// const uint64_t queueIndex = m_nextQueue++ % m_queues.size();
 			ElementType* res = m_queues[threadQueueIndex].tryPop();
 			if (res != nullptr) {
-				res->ElementType::~ElementType();
-				new (res) ElementType();
+				// res->ElementType::~ElementType();
+				//  new (res) ElementType();
 				return res;
 			}
 		}

include/ipfixprobe/outputPlugin/outputStorage/b2OutputStorage.hpp

@@ -19,31 +19,41 @@ namespace ipxp::output {
 template<typename ElementType>
 class B2OutputStorage : public BOutputStorage<ElementType> {
 public:
-	explicit B2OutputStorage(const uint8_t writersCount) noexcept
-		: BOutputStorage<ElementType>(writersCount)
+	explicit B2OutputStorage(
+		const uint8_t expectedWritersCount,
+		const uint8_t expectedReadersCount,
+		std::shared_ptr<AllocationBufferBase<ReferenceCounter<OutputContainer<ElementType>>>>
+			allocationBuffer) noexcept
+		: BOutputStorage<ElementType>(expectedWritersCount, expectedReadersCount, allocationBuffer)
 	{
 	}
 
-	bool write(ElementType* element, const uint8_t writerIndex) noexcept override
+	bool write(
+		const Reference<OutputContainer<ElementType>>& container,
+		const uint8_t writerIndex) noexcept override
 	{
 		typename BOutputStorage<ElementType>::WriterData& writerData
 			= this->m_writersData[writerIndex].get();
 		const uint16_t containersLeft = writerData.bucketAllocation.containersLeft();
 		switch (containersLeft) {
 		case 1: {
-			this->m_allocationBuffer->replace(
+			/*this->m_allocationBuffer->replace(
 				this->getNextElement(writerData.bucketAllocation),
-				element,
-				writerIndex);
+				container,
+				writerIndex);*/
+			this->getNextElement(writerData.bucketAllocation)
+				.assign(container, this->makeDeallocationCallback(writerIndex));
 		}
 			[[fallthrough]];
 		case 0:
 			break;
 		default: {
-			this->m_allocationBuffer->replace(
+			/*this->m_allocationBuffer->replace(
 				this->getNextElement(writerData.bucketAllocation),
-				element,
-				writerIndex);
+				container,
+				writerIndex);*/
+			this->getNextElement(writerData.bucketAllocation)
+				.assign(container, this->makeDeallocationCallback(writerIndex));
 			return true;
 		}
 		}
@@ -53,19 +63,19 @@ class B2OutputStorage : public BOutputStorage<ElementType> {
 		do {
 			const bool overflowed = writerData.randomShift();
 			if (overflowed) {
-				writerData.cachedLowestReaderGeneration = this->m_lowestReaderGeneration.load();
-				if (containersLeft == 0) {}
+				writerData.cachedLowestReaderGeneration
+					= this->m_lowestReaderGeneration.load(std::memory_order_acquire);
 				backoffScheme.backoff();
 			}
 
-			if (this->m_buckets[writerData.writePosition].generation
+			if (this->m_buckets[writerData.writePosition].generation.load(std::memory_order_acquire)
 					>= writerData.cachedLowestReaderGeneration
 				|| !BOutputStorage<ElementType>::BucketAllocation::isValidBucketIndex(
 					this->m_buckets[writerData.writePosition].bucketIndex)
 				|| !this->m_buckets[writerData.writePosition].lock.try_lock()) {
 				continue;
 			}
-			if (this->m_buckets[writerData.writePosition].generation
+			if (this->m_buckets[writerData.writePosition].generation.load(std::memory_order_acquire)
 					>= writerData.cachedLowestReaderGeneration
 				|| !BOutputStorage<ElementType>::BucketAllocation::isValidBucketIndex(
 					this->m_buckets[writerData.writePosition].bucketIndex)) {
@@ -77,39 +87,37 @@ class B2OutputStorage : public BOutputStorage<ElementType> {
 
 		this->m_buckets[writerData.writePosition].bucketIndex = writerData.bucketAllocation.reset(
 			this->m_buckets[writerData.writePosition].bucketIndex);
-		std::atomic_thread_fence(std::memory_order_release);
+		// std::atomic_thread_fence(std::memory_order_release);
 
 		const uint64_t highestReaderGeneration
 			= this->m_highestReaderGeneration.load(std::memory_order_acquire);
-		if (writerData.generation
+		if (writerData.generation.load(std::memory_order_acquire)
 			< highestReaderGeneration + BOutputStorage<ElementType>::WINDOW_SIZE) {
-			writerData.generation
-				= highestReaderGeneration + BOutputStorage<ElementType>::WINDOW_SIZE;
+			writerData.generation.store(
+				highestReaderGeneration + BOutputStorage<ElementType>::WINDOW_SIZE,
+				std::memory_order_release);
 			// casMax(m_highestWriterGeneration, writerData.generation);
 		}
-		this->m_buckets[writerData.writePosition].generation = writerData.generation;
+		this->m_buckets[writerData.writePosition].generation.store(
+			writerData.generation.load(std::memory_order_acquire),
+			std::memory_order_release);
 
 		this->m_buckets[writerData.writePosition].lock.unlock();
 
 		if (containersLeft == 0) {
-			this->m_allocationBuffer->replace(
-				this->getNextElement(writerData.bucketAllocation),
-				element,
-				writerIndex);
+			this->getNextElement(writerData.bucketAllocation)
+				.assign(container, this->makeDeallocationCallback(writerIndex));
 		}
 		return true;
 	}
 
-	ElementType* read(
-		const std::size_t readerGroupIndex,
-		const uint8_t localReaderIndex,
-		const uint8_t globalReaderIndex) noexcept override
+	OutputContainer<ElementType>* read(const uint8_t readerIndex) noexcept override
 	{
 		typename BOutputStorage<ElementType>::ReaderData& readerData
-			= this->m_readersData[globalReaderIndex].get();
+			= this->m_readersData[readerIndex].get();
 		// const uint64_t readPosition = readerData.readPosition;
 		if (readerData.bucketAllocation.containersLeft()) {
-			return this->getNextElement(readerData.bucketAllocation);
+			return &this->getNextElement(readerData.bucketAllocation).getData();
 		}
 
 		// uint8_t loopCounter = 0;
@@ -118,12 +126,13 @@ class B2OutputStorage : public BOutputStorage<ElementType> {
 		// 	const uint16_t initialPosition = readerData.readPosition;
 		BackoffScheme backoffScheme(0, std::numeric_limits<std::size_t>::max());
 		do {
-			readerData.shift(this->m_readerGroupSizes[readerGroupIndex], localReaderIndex);
+			const bool overflowed = readerData.shift(this->m_expectedReadersCount, readerIndex);
 
 			// auto& y = this->m_buckets[readerData.readPosition];
-			if (readerData.isOnBufferBegin(this->m_readerGroupSizes[readerGroupIndex])) {
+			// if (readerData.isOnBufferBegin(this->m_expectedReadersCount)) {
+			if (overflowed) {
 				if (!this->writersPresent()) {
-					readerData.generation++;
+					readerData.generation.fetch_add(1, std::memory_order_release);
 					updateLowestReaderGeneration();
 					return nullptr;
 				}
@@ -133,25 +142,27 @@ class B2OutputStorage : public BOutputStorage<ElementType> {
 					readerData.skipLoop = true;
 					return nullptr;
 				}
-				readerData.generation++;
+				readerData.generation.fetch_add(1, std::memory_order_release);
 				readerData.seenValidBucket = false;
 				readerData.skipLoop = false;
 				updateLowestReaderGeneration();
 			}
-			cachedGeneration = this->m_buckets[readerData.readPosition].generation;
-			std::atomic_thread_fence(std::memory_order_acquire);
+			cachedGeneration = this->m_buckets[readerData.readPosition].generation.load(
+				std::memory_order_acquire);
+			// std::atomic_thread_fence(std::memory_order_acquire);
 			cachedBucketIndex = this->m_buckets[readerData.readPosition].bucketIndex;
-			if (cachedGeneration >= readerData.generation + 2) {
+			if (cachedGeneration >= readerData.generation.load(std::memory_order_acquire)
+					+ BOutputStorage<ElementType>::WINDOW_SIZE) {
 				readerData.seenValidBucket = true;
 			}
-		} while (cachedGeneration != readerData.generation
+		} while (cachedGeneration != readerData.generation.load(std::memory_order_acquire)
 				 || !BOutputStorage<ElementType>::BucketAllocation::isValidBucketIndex(
 					 cachedBucketIndex));
 
 		readerData.seenValidBucket = true;
 		readerData.bucketAllocation.reset(this->m_buckets[readerData.readPosition].bucketIndex);
 
-		return this->getNextElement(readerData.bucketAllocation);
+		return &this->getNextElement(readerData.bucketAllocation).getData();
 	}
 
 	/*bool finished([[maybe_unused]] const std::size_t readerGroupIndex) noexcept override
@@ -166,7 +177,9 @@ class B2OutputStorage : public BOutputStorage<ElementType> {
 			= this->m_readersData
 			| std::views::transform(
 				  [](const CacheAlligned<typename BOutputStorage<ElementType>::ReaderData>&
-						 readerDataAlligned) { return readerDataAlligned->generation; })
+						 readerDataAlligned) {
+					  return readerDataAlligned->generation.load(std::memory_order_acquire);
+				  })
 			| std::ranges::to<boost::container::static_vector<
 				uint64_t,
 				OutputStorage<ElementType>::MAX_READERS_COUNT>>();
@@ -175,7 +188,7 @@ class B2OutputStorage : public BOutputStorage<ElementType> {
 		// m_highestReaderGeneration = highestReaderGeneration;
 		const uint64_t lowestReaderGeneration = *std::ranges::min_element(readerGenerations);
 		// casMin(m_lowestReaderGeneration, lowestReaderGeneration);
-		this->m_lowestReaderGeneration = lowestReaderGeneration;
+		this->m_lowestReaderGeneration.store(lowestReaderGeneration, std::memory_order_release);
 	}
 
 	std::atomic<uint64_t> m_highestWriterGeneration {0};