++

Author: Zainullin Damir

include/ipfixprobe/outputPlugin/outputStorage/backoffScheme.hpp

@@ -9,10 +9,10 @@ namespace ipxp::output {
 class BackoffScheme {
 public:
 	explicit BackoffScheme(
-		const uint32_t shortWaitThreshold,
-		const uint32_t longWaitThreshold) noexcept
+		const std::size_t shortWaitThreshold,
+		const std::size_t longWaitThreshold) noexcept
 		: m_shortWaitThreshold(shortWaitThreshold)
-		, m_longWaitThreshold(shortWaitThreshold + longWaitThreshold)
+		, m_longWaitThreshold(std::max(shortWaitThreshold + longWaitThreshold, longWaitThreshold))
 	{
 	}
 
@@ -30,9 +30,9 @@ class BackoffScheme {
 	}
 
 private:
-	const uint32_t m_shortWaitThreshold;
-	const uint32_t m_longWaitThreshold;
-	uint32_t m_waitCounter {0};
+	const std::size_t m_shortWaitThreshold;
+	const std::size_t m_longWaitThreshold;
+	std::size_t m_waitCounter {0};
 };
 
 } // namespace ipxp::output

include/ipfixprobe/outputPlugin/outputStorage/mq2OutputStorage.hpp

@@ -0,0 +1,78 @@
+#pragma once
+
+#include "backoffScheme.hpp"
+#include "doubleBufferedValue.hpp"
+#include "mqOutputStorage.hpp"
+#include "outputStorage.hpp"
+#include "rwSpinlock.hpp"
+#include "threadUtils.hpp"
+
+#include <algorithm>
+#include <atomic>
+#include <condition_variable>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <ranges>
+#include <vector>
+
+#include <boost/container/static_vector.hpp>
+
+namespace ipxp::output {
+
+class MQ2OutputStorage : public MQOutputStorage {
+public:
+	explicit MQ2OutputStorage(const uint8_t writersCount) noexcept
+		: MQOutputStorage(writersCount)
+	{
+		std::cout << std::endl;
+	}
+
+	bool storeContainer(ContainerWrapper container, const uint8_t writerId) noexcept override
+	{
+		BackoffScheme backoff(3, std::numeric_limits<std::size_t>::max());
+		while (!m_queues[writerId].tryWrite(
+			std::move(container),
+			*m_allocationBuffer,
+			m_readerGroupsCount,
+			std::numeric_limits<std::size_t>::max())) {
+			backoff.backoff();
+		}
+		return true;
+	}
+
+	std::optional<ReferenceCounterHandler<OutputContainer>> getContainer(
+		const std::size_t readerGroupIndex,
+		const uint8_t localReaderIndex,
+		const uint8_t globalReaderIndex) noexcept override
+	{
+		const size_t tries = m_totalWritersCount / m_readerGroupSizes[readerGroupIndex] + 1;
+		BackoffScheme backoff(3, 5);
+		for (const auto _ : std::views::iota(0U, tries)) {
+			const uint8_t sequenceIndex = m_readersData[globalReaderIndex]->sequenceIndex++;
+			const uint8_t queueIndex = m_readersData[globalReaderIndex]
+										   ->queueJumpSequence[sequenceIndex % MAX_WRITERS_COUNT];
+			ContainerWrapper* container = m_queues[queueIndex].tryRead(readerGroupIndex);
+			if (container != nullptr) {
+				return std::make_optional<ReferenceCounterHandler<OutputContainer>>(
+					getReferenceCounter(*container));
+			}
+			// std::this_thread::yield();
+			backoff.backoff();
+		}
+		return std::nullopt;
+	}
+
+	bool finished(const std::size_t readerGroupIndex) noexcept override
+	{
+		return m_readerGroupSizes[readerGroupIndex] > m_totalWritersCount
+			|| (!writersPresent() && std::ranges::all_of(m_queues, [&](const Queue& queue) {
+				   return queue.finished();
+			   }));
+	}
+
+private:
+};
+
+} // namespace ipxp::output

include/ipfixprobe/outputPlugin/outputStorage/mqOutputStorage.hpp

@@ -102,7 +102,7 @@ class MQOutputStorage : public OutputStorage {
 		//  WriteLockGuard lockGuard(m_locks[writeQueueIndex]);
 
 		if (!m_queues[writerId]
-				 .tryWrite(std::move(container), *m_allocationBuffer, m_readerGroupsCount)) {
+				 .tryWrite(std::move(container), *m_allocationBuffer, m_readerGroupsCount, 3)) {
 			return false;
 		}
 		return true;
@@ -140,7 +140,7 @@ class MQOutputStorage : public OutputStorage {
 			   }));
 	}
 
-private:
+protected:
 	struct QueueIndex {
 		const uint16_t initialOffset;
 		uint8_t loops;
@@ -169,12 +169,13 @@ class MQOutputStorage : public OutputStorage {
 		bool tryWrite(
 			ContainerWrapper container,
 			AllocationBufferBase<ReferenceCounter<OutputContainer>>& origin,
-			const uint8_t readerGroupCount) noexcept
+			const uint8_t readerGroupCount,
+			const std::size_t longBackoffTries) noexcept
 		{
 			State* currentState = &m_stateBuffer.getCurrentValue();
 
 			if (currentState->written == currentState->writeBuffer.size()) {
-				BackoffScheme backoff(7, 5);
+				BackoffScheme backoff(7, longBackoffTries);
 				while (!allReadersFinished()) {
 					if (!backoff.backoff()) {
 						container.deallocate(origin);

tests/performance/outputStorage/testOutputStorage.cpp

@@ -15,6 +15,7 @@
 #include <outputStorage/lfnbOutputStorage.hpp>
 #include <outputStorage/mc2OutputStorage.hpp>
 #include <outputStorage/mcOutputStorage.hpp>
+#include <outputStorage/mq2OutputStorage.hpp>
 #include <outputStorage/mqOutputStorage.hpp>
 #include <outputStorage/ringOutputStorage.hpp>
 #include <outputStorage/serializedOutputStorage.hpp>
@@ -224,7 +225,7 @@ TEST(TestOutputStorage, Debug)
 	for (const auto testIndex : std::views::iota(0, 100)) {
 		std::cout << " Debug Loop Iteration " << testIndex << "\n";
 		// makePerformanceTest<ipxp::output::MCOutputStorage>("MCOutputStorage");
-		stressTest<ipxp::output::MC2OutputStorage>(false);
+		stressTest<ipxp::output::MQ2OutputStorage>(false);
 	}
 }