++

Author: Zainullin Damir

include/ipfixprobe/outputPlugin/outputStorage/allocationBufferB.hpp

@@ -19,11 +19,18 @@ namespace ipxp::output {
 
 template<typename ElementType>
 class AllocationBufferB : public AllocationBufferBase<ElementType> {
-	constexpr static std::size_t BUCKET_SIZE = 64;
+	constexpr static std::size_t BUCKET_SIZE = 16;
 	constexpr static std::size_t INDEXES_IN_CACHE_LINE = 64 / sizeof(uint16_t);
-	constexpr static std::size_t WINDOW_SIZE = 4;
+	constexpr static std::size_t WINDOW_SIZE = 16;
 
 public:
+	__attribute__((noinline)) std::size_t d_test(auto& container)
+	{
+		return std::ranges::count_if(container, [](const auto& bucket) {
+			return bucket.load(std::memory_order_acquire) != std::numeric_limits<uint16_t>::max();
+		});
+	}
+
 	explicit AllocationBufferB(const std::size_t capacity, const uint8_t writersCount) noexcept
 		: m_objectPool(capacity + writersCount * BUCKET_SIZE)
 		, m_buckets(m_objectPool.size() / BUCKET_SIZE)
@@ -46,6 +53,10 @@ class AllocationBufferB : public AllocationBufferBase<ElementType> {
 			m_fullBuckets[i].store(i);
 			m_emptyBuckets[i].store(Bucket::PLACEHOLDER);
 		}
+		for (std::size_t i = m_buckets.size(); i < m_fullBuckets.size(); i++) {
+			m_fullBuckets[i].store(Bucket::PLACEHOLDER);
+			m_emptyBuckets[i].store(Bucket::PLACEHOLDER);
+		}
 		m_writersData.resize(writersCount);
 		for (uint8_t writerIndex = 0; writerIndex < writersCount; writerIndex++) {
 			m_writersData[writerIndex]->fullPushRank = writerIndex * INDEXES_IN_CACHE_LINE;
@@ -124,39 +135,54 @@ class AllocationBufferB : public AllocationBufferBase<ElementType> {
 
 	void pushBucket(auto& buckets, const std::size_t bucketIndex, std::size_t& pushRank) noexcept
 	{
+		std::size_t offset = 0;
 		while (true) {
 			uint16_t expected = buckets[pushRank].load(std::memory_order_acquire);
+			if (++offset % 100'000'000 == 0) {
+				std::cout << "d_test(push)=" << d_test(buckets) << "\n";
+			}
 			if (expected != Bucket::PLACEHOLDER) {
-				pushRank = ((pushRank / INDEXES_IN_CACHE_LINE + 1) * INDEXES_IN_CACHE_LINE)
-					% m_buckets.size();
+				const std::size_t newPushRank
+					= ((pushRank / INDEXES_IN_CACHE_LINE + 1) * INDEXES_IN_CACHE_LINE + offset)
+					% buckets.size();
+				if (newPushRank < pushRank) {
+					// offset++;
+				}
+				pushRank = newPushRank;
 				continue;
 			}
 			if (buckets[pushRank].compare_exchange_weak(
 					expected,
 					bucketIndex,
 					std::memory_order_release,
 					std::memory_order_acquire)) {
-				pushRank = (pushRank + 1) % m_buckets.size();
+				pushRank = (pushRank + 1) % buckets.size();
 				return;
 			}
 		}
 	}
 
 	uint16_t popBucket(auto& buckets, std::size_t& popRank) noexcept
 	{
+		std::size_t offset = 0;
+		popRank = (popRank - 1 + buckets.size()) % buckets.size();
 		while (true) {
+			if (++offset % 100'000'000 == 0) {
+				std::cout << "d_test(pop)=" << d_test(buckets) << "\n";
+			}
 			uint16_t expected = buckets[popRank].load(std::memory_order_acquire);
 			if (expected == Bucket::PLACEHOLDER) {
-				popRank = ((popRank / INDEXES_IN_CACHE_LINE + 1) * INDEXES_IN_CACHE_LINE)
-					% m_buckets.size();
+				popRank
+					= (popRank / INDEXES_IN_CACHE_LINE * INDEXES_IN_CACHE_LINE - 1 + buckets.size())
+					% buckets.size();
 				continue;
 			}
 			if (buckets[popRank].compare_exchange_weak(
 					expected,
 					Bucket::PLACEHOLDER,
 					std::memory_order_release,
 					std::memory_order_acquire)) {
-				popRank = (popRank + 1) % m_buckets.size();
+				// popRank = (popRank - 1 + buckets.size()) % buckets.size();
 				return expected;
 			}
 		}