++

Author: Zainullin Damir

cmake/dependencies.cmake

@@ -8,6 +8,7 @@ find_package(Unwind REQUIRED)
 find_package(LZ4 REQUIRED)
 find_package(OpenSSL REQUIRED)
 find_package(Boost REQUIRED)
+find_package(NUMA REQUIRED)
 
 if (ENABLE_INPUT_PCAP)
 	pkg_check_modules(PCAP REQUIRED libpcap)
@@ -19,7 +20,6 @@ endif()
 
 if (ENABLE_INPUT_NFB)
 	find_package(NFB REQUIRED)
-	find_package(NUMA REQUIRED)
 endif()
 
 if (ENABLE_OUTPUT_UNIREC OR ENABLE_NEMEA)

include/ipfixprobe/outputPlugin/outputStorage/outputStorageReader.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "outputStorage.hpp"
+#include "threadAffinitySetter.hpp"
 
 #include <memory>
 

include/ipfixprobe/outputPlugin/outputStorage/outputStorageWriter.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "outputStorage.hpp"
+#include "threadAffinitySetter.hpp"
 
 #include <memory>
 

include/ipfixprobe/outputPlugin/outputStorage/threadAffinitySetter.hpp

@@ -0,0 +1,71 @@
+#pragma once
+
+#include "threadUtils.hpp"
+
+#include <iostream>
+#include <memory>
+#include <vector>
+
+#include <numa.h>
+
+namespace ipxp::output {
+
+class ThreadAffinitySetter {
+public:
+	static void setNumaNode(const size_t nodeIndex)
+	{
+		const uint16_t threadIndex = getThreadId();
+		const std::size_t numaIndex = nodeIndex % m_architectureInfo.cpusByNumaNode.size();
+		const std::size_t cpuIndex
+			= threadIndex % m_architectureInfo.cpusByNumaNode[numaIndex].size();
+		const int cpuToBind = m_architectureInfo.cpusByNumaNode[numaIndex][cpuIndex];
+
+		cpu_set_t cpuset;
+		CPU_ZERO(&cpuset);
+		CPU_SET(cpuToBind, &cpuset);
+		const pthread_t current_thread = pthread_self();
+		const int errCode = pthread_setaffinity_np(current_thread, sizeof(cpu_set_t), &cpuset);
+		if (errCode != 0) {
+			throw std::system_error(errCode, std::generic_category(), "Failed to pin thread");
+		}
+	}
+
+private:
+	struct ArchitectureInfo {
+		ArchitectureInfo()
+		{
+			if (numa_available() == -1) {
+				cpusByNumaNode
+					= {std::views::iota(0, static_cast<int>(std::thread::hardware_concurrency()))
+					   | std::ranges::to<std::vector>()};
+				return;
+			}
+
+			const auto deleter = [](bitmask* mask) {
+				if (mask)
+					numa_free_cpumask(mask);
+			};
+			const int maxNodes = numa_max_node();
+			for (int currentNode = 0; currentNode <= maxNodes; currentNode++) {
+				std::unique_ptr<bitmask, decltype(deleter)> mask(numa_allocate_cpumask(), deleter);
+
+				if (numa_node_to_cpus(currentNode, mask.get()) != 0) {
+					throw std::runtime_error(
+						"Failed to get CPU mask for NUMA node " + std::to_string(currentNode));
+				}
+				auto cpusOfNode = std::views::iota(0, (int) mask->size)
+					| std::views::filter([&](int cpuIndex) {
+									  return numa_bitmask_isbitset(mask.get(), cpuIndex);
+								  })
+					| std::ranges::to<std::vector<int>>();
+				cpusByNumaNode.push_back(std::move(cpusOfNode));
+			}
+		}
+
+		std::vector<std::vector<int>> cpusByNumaNode;
+	};
+
+	static inline ArchitectureInfo m_architectureInfo;
+};
+
+} // namespace ipxp::output

tests/performance/outputStorage/CMakeLists.txt

@@ -17,6 +17,7 @@ target_link_libraries(testOutputStorage PRIVATE
     GTest::gtest
     GTest::gtest_main
     ipfixprobe-core
+    numa::numa
 )
 
 add_test(NAME testOutputStorage COMMAND testOutputStorage)

tests/performance/outputStorage/testOutputStorage.cpp

@@ -18,6 +18,7 @@
 #include <outputStorage/mq2OutputStorage.hpp>
 #include <outputStorage/mqOutputStorage.hpp>
 #include <outputStorage/ringOutputStorage.hpp>
+#include <outputStorage/threadAffinitySetter.hpp>
 // #include <outputStorage/serializedOutputStorage.hpp>
 // #include <outputStorage/serializedOutputStorageBlocking.hpp>
 #include <outputStorage/outputStorageRegistrar.hpp>
@@ -67,6 +68,7 @@ void makeTest(
 		for (DummyReader& reader : readerGroup) {
 			readContainers.back().emplace_back(
 				std::async(std::launch::async, [&reader, &readersLatch]() {
+					ipxp::output::ThreadAffinitySetter::setNumaNode(0);
 					readersLatch.count_down();
 					return reader.readContainers();
 				}));
@@ -76,12 +78,14 @@ void makeTest(
 	readersLatch.wait();
 	std::vector<std::future<void>> writerFutures
 		= writers | std::views::transform([&](DummyWriter<OutputStorageType>& writer) {
-			  return std::async(std::launch::async, [&]() { writer.writeContainers(); });
+			  return std::async(std::launch::async, [&]() {
+				  ipxp::output::ThreadAffinitySetter::setNumaNode(0);
+				  writer.writeContainers();
+			  });
 		  })
 		| std::ranges::to<std::vector<std::future<void>>>();
 
 	std::ranges::for_each(writerFutures, [](std::future<void>& future) { future.get(); });
-	// writers.clear();
 
 	const std::vector<std::size_t> containersReadInGroups
 		= readContainers
@@ -164,10 +168,6 @@ void shortTestLoop(const bool immitateWork)
 template<typename OutputStorageType>
 void makePerformanceTest(std::string_view storageName)
 {
-	std::cout << "==========================================================" << std::endl;
-	std::cout << storageName << ", 32 Writers, 4 Group 8 Reader\n";
-	makeTest<OutputStorageType>(32, {8, 8, 8, 8}, false, 20'000'000);
-
 	std::cout << "==========================================================" << std::endl;
 	std::cout << storageName << ", 1 Writers, 1 Reader\n";
 	makeTest<OutputStorageType>(1, {1}, false, 80'000'000);
@@ -188,11 +188,14 @@ void makePerformanceTest(std::string_view storageName)
 	std::cout << storageName << ", 4 Writers, 4 Group 1 Reader\n";
 	makeTest<OutputStorageType>(4, {1, 1, 1, 1}, false, 80'000'000);
 
-	std::cout << std::endl;
+	std::cout << "==========================================================" << std::endl;
+	std::cout << storageName << ", 32 Writers, 4 Group 8 Reader\n";
+	makeTest<OutputStorageType>(32, {8, 8, 8, 8}, false, 20'000'000);
 }
 
 TEST(TestOutputStorage, XXX)
 {
+	ipxp::output::ThreadAffinitySetter::setNumaNode(0);
 	std::cout << "==========================================================" << std::endl;
 	std::cout << "MQ2OutputStorage, 1 Writers, 1 Reader\n";
 	makeTest<ipxp::output::MQ2OutputStorage<void*>>(1, {1}, false, 80'000'000);