Counters to shared memory #227

stal76 · 2024-07-11T16:46:55Z

At the current time, the operation of obtaining the counters values may require a lot of resources, the following actions are performed:

the client process opens a unix socket with a Dataplane;
a separate thread is created in the Dataplane, this thread processes the client request and returns the counter values.

In this PR, the main Dataplane counters have been moved to shared memory. To do this, the following files are created in /dev/shm/:

yanet_data plane.shm - contains general information about workers in Dataplane, metadata and some other information.
yanet_workers_node_N.shm - N is the numa node number (a separate file for each numa node), contains buffers of all workers working on this numa node.

A more detailed description of the data storage structure in these files is in the common/sdpcommon.h.

Any client process can open these files and get the values of the required counters.

ol-imorozko

First commit:

common/sdpcommon.h

common/shared_memory.h

dataplane/unittest/shared_memory.cpp

ol-imorozko · 2024-07-25T16:23:03Z

dataplane/worker.cpp

+	for (const auto& iter : counters_named)
+	{
+		metadata.counter_positions[iter.first] = metadata.start_counters / sizeof(uint64_t) + static_cast<uint64_t>(iter.second);
+	}


Since you're using both iter.first and iter.second, use structured bindings with meaningful names i.e. for (const auto& [name, offset] : counters_named)

ol-imorozko · 2024-07-25T16:24:39Z

dataplane/worker_gc.cpp

+	for (const auto& iter : counters_stats)
+	{
+		metadata.counter_positions[iter.first] = (metadata.start_stats + iter.second) / sizeof(uint64_t);
+	}


same with structured bindings, name iter.first and iter.second

ol-imorozko

second commit:

ol-imorozko · 2024-07-30T11:08:03Z

common/sdpclient.h

+#define SHARED_MEMORY_REREAD_TIMEOUT_MICROSECONDS 100
+#define SHARED_MEMORY_REREAD_MAXIMUM_ATTEMPTS 100


constexpr auto SHARED_MEMORY_REREAD_TIMEOUT_MICROSECONDS = 100; constexpr auto SHARED_MEMORY_REREAD_MAXIMUM_ATTEMPTS = 100;

ol-imorozko · 2024-07-30T11:11:03Z

common/sdpclient.h

+		{
+			number_of_attempts++;
+			std::string message_error;
+			uint64_t size_mmap;


Better to init this variable: uint64_t size_mmap = 0;

ol-imorozko · 2024-07-30T11:23:45Z

common/sdpclient.h

+		while (result != eResultRead::ok)
+		{
+			number_of_attempts++;
+			std::string message_error;
+			uint64_t size_mmap;
+			result = ReadItAgainMainFileDataplane(sdp_data, size_mmap, message_error);
+			if (result == eResultRead::error)
+			{
+				YANET_LOG_ERROR("File %s. %s\n", YANET_SHARED_MEMORY_FILE_DATAPLANE, message_error.c_str());
+				sdp_data.UnmapBuffers(size_mmap);
+				return eResult::errorInitSharedMemory;
+			}
+			else if (result == eResultRead::need_reread)
+			{
+				sdp_data.UnmapBuffers(size_mmap);
+				if (number_of_attempts >= SHARED_MEMORY_REREAD_MAXIMUM_ATTEMPTS)
+				{
+					YANET_LOG_ERROR("File %s. Attempts were made to read: %d. %s\n",
+					                YANET_SHARED_MEMORY_FILE_DATAPLANE,
+					                number_of_attempts,
+					                message_error.c_str());
+					return eResult::errorInitSharedMemory;
+				}
+				YANET_LOG_WARNING("File %s. %s\n", YANET_SHARED_MEMORY_FILE_DATAPLANE, message_error.c_str());
+				std::this_thread::sleep_for(std::chrono::microseconds{SHARED_MEMORY_REREAD_TIMEOUT_MICROSECONDS});
+			}
+		}


IMHO it will be better to extract the loop body to some function like AttemptRead. Maybe even a lambda will be good enough:

int number_of_attempts = 0; eResultRead result = eResultRead::need_reread; std::string message_error; uint64_t size_mmap = 0; auto attempt_read = [&]() -> eResultRead { number_of_attempts++; result = ReadItAgainMainFileDataplane(sdp_data, size_mmap, message_error); //... std::this_thread::sleep_for(std::chrono::microseconds{/*...*/}); } return result; }; while (result != eResultRead::ok) { if (auto err = attempt_read(); err == eResult::errorInitSharedMemory) { return err; } }

ol-imorozko · 2024-07-30T11:30:44Z

common/sdpclient.h

+		for (const auto& iter : sdp_data.workers)
+		{
+			sockets_buffer[iter.second.socket] = {nullptr, 0};
+		}
+		for (const auto& iter : sdp_data.workers_gc)
+		{
+			sockets_buffer[iter.second.socket] = {nullptr, 0};
+		}


It's sad that since we have a requirement to use gcc from Ubuntu 18, we cannot just do

for (const auto& [id, worker] : sdp_data.workers) { sockets_buffer[worker.socket] = {nullptr, 0}; }

cause gcc will tell us that id is unused, which is a bug in that gcc version.

this comment is not a request to do something, just sharing my thoughts :)

ol-imorozko · 2024-07-30T11:32:07Z

common/sdpclient.h

+		for (auto& iter : sockets_buffer)
+		{
+			std::string filename = FileNameWorkerOnNumaNode(iter.first);
+			auto [buffer, size] = common::ipc::SharedMemory::OpenBuffer(filename, false);
+			if (buffer == nullptr)
+			{
+				YANET_LOG_ERROR("Error openning shared memory buffer from file: %s\n", filename.c_str());
+				return eResult::errorInitSharedMemory;
+			}
+			iter.second = {buffer, size};
+		}


Here since we use both iter.first and iter.second, we can safely use structured bindings:

for (auto& [socket, buffer_pair] : sockets_buffer) { std::string filename = FileNameWorkerOnNumaNode(socket); auto [buffer, size] = common::ipc::SharedMemory::OpenBuffer(filename, false); if (!buffer) { YANET_LOG_ERROR("Error opening shared memory buffer from file: %s\n", filename.c_str()); return eResult::errorInitSharedMemory; } buffer_pair = {buffer, size}; }

ol-imorozko · 2024-07-30T12:53:50Z

common/sdpcommon.h

+		for (const auto& iter : data_worker)
+		{
+			info[iter.first] = {tSocketId(iter.second.first), iter.second.second, nullptr};
+		}


for (const auto& [core_id, worker_data] : data_worker) { info[tCoreId(core_id)] = {tSocketId(worker_data.first), worker_data.second, nullptr}; }

ol-imorozko · 2024-07-30T12:55:32Z

common/sdpcommon.h

+				YANET_LOG_ERROR("Error munmap %d: %s", errno, strerror(errno));
+			}


do we need to exit here? If not, maybe LOG_WARNING?

ol-imorozko · 2024-07-30T12:57:05Z

controlplane/controlplane.cpp

+		uint64_t* aclCounters = common::sdp::ShiftBuffer<uint64_t*>(iter.second.buffer, start_acl_counters);
+		for (size_t i = 0; i < YANET_CONFIG_ACL_COUNTERS_SIZE; i++)
+		{
+			response[i] += aclCounters[i];
+		}


auto* aclCounters = common::sdp::ShiftBuffer<uint64_t*>(iter.second.buffer, start_acl_counters); std::transform(response.begin(), response.end(), aclCounters, response.begin(), std::plus<>{});

ol-imorozko · 2024-07-30T12:58:03Z

controlplane/telegraf.cpp

+		uint64_t* worker_bursts =
+		        common::sdp::ShiftBuffer<uint64_t*>(worker_info.buffer, sdp_data->metadata_worker.start_bursts);


auto* worker_bursts =

ol-imorozko · 2024-07-30T12:59:45Z

controlplane/telegraf.cpp

+		std::array<uint64_t, CONFIG_YADECAP_MBUFS_BURST_SIZE + 1> bursts;
+		uint64_t* worker_bursts =
+		        common::sdp::ShiftBuffer<uint64_t*>(worker_info.buffer, sdp_data->metadata_worker.start_bursts);
+		memcpy(&bursts[0], worker_bursts, sizeof(uint64_t) * (CONFIG_YADECAP_MBUFS_BURST_SIZE + 1));


std::memcpy(bursts.data(), worker_bursts, sizeof(uint64_t) * bursts.size());

ol-imorozko

Last comment:

Also use reserve to pre-allocate space for vectors to avoid multiple reallocations in bus.h functions

ol-imorozko · 2024-07-30T13:05:09Z

cli/bus.h

+	std::map<common::idp::requestType, std::string> names = {
+	        {common::idp::requestType::updateGlobalBase, "updateGlobalBase"},
+	        {common::idp::requestType::updateGlobalBaseBalancer, "updateGlobalBaseBalancer"},
+	        {common::idp::requestType::getGlobalBase, "getGlobalBase"},


Include "common/idp.h" for symbol common::idp::requestType

ol-imorozko · 2024-07-30T13:05:24Z

cli/bus.h

+	uint64_t* requests = common::sdp::ShiftBuffer<uint64_t*>(sdp_data.dataplane_data, sdp_data.metadata_bus.start_bus_requests);
+	uint64_t* durations = common::sdp::ShiftBuffer<uint64_t*>(sdp_data.dataplane_data, sdp_data.metadata_bus.start_bus_durations);


Use auto when initializing with a template cast to avoid duplicating the type name:

auto* requests = common::sdp::ShiftBuffer<uint64_t*>(sdp_data.dataplane_data, sdp_data.metadata_bus.start_bus_requests); auto* durations = common::sdp::ShiftBuffer<uint64_t*>(sdp_data.dataplane_data, sdp_data.metadata_bus.start_bus_durations);

ol-imorozko · 2024-07-30T13:08:38Z

cli/bus.h

+	std::map<common::idp::requestType, std::string> names = {
+	        {common::idp::requestType::updateGlobalBase, "updateGlobalBase"},


const std::map<common::idp::requestType, std::string_view> names = {

ol-imorozko · 2024-07-30T13:09:18Z

cli/bus.h

+	for (uint32_t index = 0; index < (uint32_t)common::idp::requestType::size; ++index)
+	{


static_cast

ol-imorozko · 2024-07-30T13:17:52Z

dataplane/bus.cpp

@@ -404,6 +404,8 @@ void cBus::clientThread(int clientSocket)

 		std::chrono::duration<double> duration = std::chrono::system_clock::now() - startTime;

+		// The duration time is measured in milliseconds
+		stats.durations[(uint32_t)type] += static_cast<uint64_t>(1000 * duration.count());


static_cast type too

ol-imorozko · 2024-07-30T13:19:06Z

cli/telegraf.h

+	for (const auto& [coreId, worker_info] : sdp_data.workers)
+	{
+		std::vector<influxdb_format::value_t> values;
+		uint64_t* buffer = common::sdp::ShiftBuffer<uint64_t*>(worker_info.buffer,
+		                                                       sdp_data.metadata_worker.start_counters);
+		for (const auto& [name, index] : sdp_data.metadata_worker.counter_positions)
+		{
+			values.emplace_back(name.data(), buffer[index]);
+		}
+		influxdb_format::print("worker", {{"coreId", coreId}}, values);
+	}
+
+	for (const auto& [coreId, worker_info] : sdp_data.workers_gc)
+	{
+		std::vector<influxdb_format::value_t> values;
+		uint64_t* buffer = common::sdp::ShiftBuffer<uint64_t*>(worker_info.buffer,
+		                                                       sdp_data.metadata_worker.start_counters);
+		for (const auto& [name, index] : sdp_data.metadata_worker_gc.counter_positions)
+		{
+			values.emplace_back(name.data(), buffer[index]);
+		}
+		influxdb_format::print("worker_gc", {{"coreId", coreId}}, values);
+	}


Code duplication. Either extract to a separate function, or use lambda:

auto process_worker_info = [](const tCoreId coreId, const auto& worker_info, const auto& counter_positions, uint64_t start_counters, std::string_view type) { std::vector<influxdb_format::value_t> values; auto* buffer = common::sdp::ShiftBuffer<uint64_t*>(worker_info.buffer, start_counters); for (const auto& [name, index] : counter_positions) { values.emplace_back(name.data(), buffer[index]); } influxdb_format::print(type.data(), {{"coreId", coreId}}, values); }; for (const auto& [coreId, worker_info] : sdp_data.workers) { process_worker_info(coreId, worker_info, sdp_data.metadata_worker.counter_positions, sdp_data.metadata_worker.start_counters, "worker"); } for (const auto& [coreId, worker_info] : sdp_data.workers_gc) { process_worker_info(coreId, worker_info, sdp_data.metadata_worker_gc.counter_positions, sdp_data.metadata_worker_gc.start_counters, "worker_gc"); }

- Work with shared memory Posix standard and System V is supported. - Two functions are implemented for each standard - to create a buffer (on the server) and to open the created buffer (on the client). - When creating a buffer, one can, if necessary, specify the socket (numa node) on which to allocate memory, and whether to use HugeMem.

…ed memory

…memory

…red memory

- Bus operation time measurements are enabled. - Added error message output when connecting via socket to a dataplane before calling an exception. - 3 commands for getting bus statistics have been added to the cli. - The command to get the main counters has been added to the cli.

I find the PR good enough to be merged despite some improvements could be still done

ol-imorozko · 2024-10-23T16:12:30Z

@GeorgyKirichenko

I find the PR good enough to be merged despite some improvements could be still done

And when do we plan to address my review notes? There are around 60 unresolved comments, most of which are relatively simple fixes. I don't understand why we are pushing code without resolving these issues, especially since they can be easily addressed.

I'm reviewing the project and noticing several areas that still need improvement, many of which I previously flagged for changes.

Things like:

or

stal76 requested a review from GeorgyKirichenko as a code owner July 11, 2024 16:46

stal76 force-pushed the counters-to-shared-memory branch 4 times, most recently from 5e663ea to 75da7ac Compare July 12, 2024 12:59

ol-imorozko requested changes Jul 25, 2024

View reviewed changes

ol-imorozko requested changes Jul 30, 2024

View reviewed changes

ol-imorozko previously requested changes Jul 30, 2024

View reviewed changes

stal76 force-pushed the counters-to-shared-memory branch from 75da7ac to 02129a3 Compare September 14, 2024 19:15

stal76 force-pushed the counters-to-shared-memory branch 3 times, most recently from ad72b55 to dddfaf5 Compare September 17, 2024 13:09

stal76 added 8 commits September 17, 2024 16:39

Added unit tests to check the work with shared memory

4bc3bf2

An interface implemented for storing data related to counters in shar…

5eba162

…ed memory

Unittests for the interface for working with counters in shared memory

c452d0d

Counters from Worker, Workers and Bus has been transferred to shared …

6251ef8

…memory

Getting the main counters via bus is replaced by getting them via sha…

da598ac

…red memory

The interface for receiving basic counters via bus has been removed

2948bd0

Added autotest for counters in shared memory

b9dd6a2

stal76 force-pushed the counters-to-shared-memory branch from dddfaf5 to b9dd6a2 Compare September 17, 2024 13:53

GeorgyKirichenko approved these changes Sep 18, 2024

View reviewed changes

GeorgyKirichenko merged commit 6d8e640 into main Sep 18, 2024
8 checks passed

GeorgyKirichenko deleted the counters-to-shared-memory branch September 18, 2024 06:29

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Counters to shared memory #227

Counters to shared memory #227

stal76 commented Jul 11, 2024

ol-imorozko left a comment

ol-imorozko Jul 25, 2024 •

edited

Loading

ol-imorozko Jul 25, 2024

ol-imorozko left a comment

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko left a comment

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko Jul 30, 2024

ol-imorozko commented Oct 23, 2024

		#define SHARED_MEMORY_REREAD_TIMEOUT_MICROSECONDS 100
		#define SHARED_MEMORY_REREAD_MAXIMUM_ATTEMPTS 100

		YANET_LOG_ERROR("Error munmap %d: %s", errno, strerror(errno));
		}

		uint64_t* worker_bursts =
		common::sdp::ShiftBuffer<uint64_t*>(worker_info.buffer, sdp_data->metadata_worker.start_bursts);

		uint64_t* requests = common::sdp::ShiftBuffer<uint64_t*>(sdp_data.dataplane_data, sdp_data.metadata_bus.start_bus_requests);
		uint64_t* durations = common::sdp::ShiftBuffer<uint64_t*>(sdp_data.dataplane_data, sdp_data.metadata_bus.start_bus_durations);

		std::map<common::idp::requestType, std::string> names = {
		{common::idp::requestType::updateGlobalBase, "updateGlobalBase"},

		for (uint32_t index = 0; index < (uint32_t)common::idp::requestType::size; ++index)
		{

Counters to shared memory #227

Counters to shared memory #227

Conversation

stal76 commented Jul 11, 2024

ol-imorozko left a comment

Choose a reason for hiding this comment

ol-imorozko Jul 25, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ol-imorozko left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ol-imorozko left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ol-imorozko commented Oct 23, 2024

ol-imorozko Jul 25, 2024 •

edited

Loading