++

Author: Zainullin Damir

include/ipfixprobe/outputPlugin/outputStorage/allocationBufferS.hpp

@@ -19,7 +19,7 @@ template<typename ElementType>
 class AllocationBufferS : public AllocationBufferBase<ElementType> {
 public:
 	explicit AllocationBufferS(const std::size_t capacity, const uint8_t writersCount) noexcept
-		: m_objectPool(capacity + writersCount)
+		: m_objectPool(capacity + writersCount * 4)
 	{
 		static_assert(
 			std::atomic<HelpState>::is_always_lock_free,
@@ -73,10 +73,21 @@ class AllocationBufferS : public AllocationBufferBase<ElementType> {
 private:
 	void handleStealRequest(const uint8_t writerIndex) noexcept
 	{
-		if (!m_helpStates[writerIndex]->load(std::memory_order_acquire).stealingRequested) {
-			return;
-		}
-		m_helpStates[writerIndex]->store(HelpState {true, true}, std::memory_order_release);
+		HelpState expected;
+		HelpState desired;
+		do {
+			expected = m_helpStates[writerIndex]->load(std::memory_order_acquire);
+			if (!expected.stealingRequested) {
+				return;
+			}
+			desired = expected;
+			desired.stealingAllowed = true;
+		} while (!m_helpStates[writerIndex]->compare_exchange_weak(
+			expected,
+			desired,
+			std::memory_order_release,
+			std::memory_order_acquire));
+
 		while (m_helpStates[writerIndex]->load(std::memory_order_acquire).stealingRequested) {
 			BackoffScheme(1, 0).backoff();
 		}
@@ -95,8 +106,9 @@ class AllocationBufferS : public AllocationBufferBase<ElementType> {
 			if (!setStealRequest(stealVictimIndex)) {
 				continue;
 			}
-			while (!isStealAllowed(stealVictimIndex)) {
-				BackoffScheme(1, 0).backoff();
+			if (!waitForStealApproval(stealVictimIndex)) {
+				clearStealRequest(stealVictimIndex);
+				continue;
 			}
 			WriterData& stealVictimData = m_writersData[stealVictimIndex].get();
 			for (std::size_t i = 0; i < stealVictimData.storage.size() / 2; i++) {
@@ -108,6 +120,17 @@ class AllocationBufferS : public AllocationBufferBase<ElementType> {
 		}
 	}
 
+	bool waitForStealApproval(const uint8_t victimIndex) noexcept
+	{
+		BackoffScheme backoff(200, 5);
+		while (!isStealAllowed(victimIndex)) {
+			if (!backoff.backoff()) {
+				return false;
+			}
+		}
+		return true;
+	}
+
 	bool setStealRequest(const uint8_t victimIndex) noexcept
 	{
 		HelpState expected;

include/ipfixprobe/outputPlugin/outputStorage/outputStorageRegistrar.hpp

@@ -24,7 +24,8 @@ class OutputStorageRegistrar {
 		m_allocationBuffer
 			= std::make_shared<AllocationBufferS<ReferenceCounter<OutputContainer<ElementType>>>>(
 				OutputStorage<ElementType>::STORAGE_CAPACITY
-					* OutputStorage<ElementType>::MAX_READER_GROUPS_COUNT,
+				//* OutputStorage<ElementType>::MAX_READER_GROUPS_COUNT
+				,
 				writersCount);
 	}