WIP ndp: update ndp plugin

Author: Pavel Siska

src/plugins/input/nfb/src/ndp.cpp

@@ -17,6 +17,9 @@
 #include <cstdio>
 #include <cstring>
 #include <iostream>
+#include <memory>
+#include <algorithm>
+#include <span>
 
 #include <ipfixprobe/pluginFactory/pluginManifest.hpp>
 #include <ipfixprobe/pluginFactory/pluginRegistrar.hpp>
@@ -47,7 +50,7 @@ static std::vector<std::string> parseDevices(const std::string& input)
 {
 	std::vector<std::string> result;
 
-	size_t colon_pos = input.find(':');
+	size_t colon_pos = input.find_last_of(':');
 	std::string suffix;
 	std::string devices;
 
@@ -69,6 +72,7 @@ static std::vector<std::string> parseDevices(const std::string& input)
 }
 
 NdpPacketReader::NdpPacketReader(const std::string& params)
+	: ndp_packet_burst(new ndp_packet[64])
 {
 	init(params.c_str());
 }
@@ -121,54 +125,57 @@ InputPlugin::Result NdpPacketReader::get(PacketBlock& packets)
 	parser_opt_t opt = {&packets, false, false, 0};
 	struct ndp_packet* ndp_packet;
 	struct timeval timestamp;
-	size_t read_pkts = 0;
 	int ret = -1;
 
 	packets.cnt = 0;
+	constexpr size_t maxBurstSize = 64;
+	size_t burstSize = std::min(packets.size, maxBurstSize); 
+	std::span<struct ndp_packet> packetSpan(ndp_packet_burst.get(), burstSize);
+	std::span<timeval> timestampSpan(timestamps);
+
+	size_t reader_index = (m_reader_idx++) & (m_readers_count - 1);
+	NdpReader& reader = ndpReader[reader_index];
+	int received = reader.get_packets(packetSpan, timestampSpan);
+
+	if (received < 32) {
+		std::span<struct ndp_packet> packetSpan(ndp_packet_burst.get() + received, burstSize - received);
+		std::span<timeval> timestampSpan(timestamps.data() + received, burstSize - received);
+
+		size_t reader_index = (m_reader_idx++) & (m_readers_count - 1);
+		NdpReader& reader = ndpReader[reader_index];
+		received += reader.get_packets(packetSpan, timestampSpan);
+	}
+
+	for (unsigned i = 0; i < received; ++i) {
+		ndp_packet = &ndp_packet_burst[i];
+		timestamp = timestamps[i];
 
-	bool any_timeout = false;
-
-	for (unsigned r = 0; r < m_readers_count; ++r) {
-		NdpReader& reader = ndpReader[(m_reader_idx++) % m_readers_count];
-
-		for (unsigned i = 0; i < packets.size; ++i) {
-			ret = reader.get_pkt(&ndp_packet, &timestamp);
-			if (ret == 0) {
-				any_timeout = true;
-				break;
-			} else if (ret < 0) {
-				// Error occured.
-				throw PluginError(reader.error_msg);
-			}
-
-			++read_pkts;
-			parse_packet(
-				&opt,
-				m_parser_stats,
-				timestamp,
-				ndp_packet->data,
-				ndp_packet->data_length,
-				ndp_packet->data_length);
-
-			if (opt.pblock->cnt >= packets.size) {
-				break;
-			}
+		if (ndp_packet->data_length == 0) {
+			continue; // Skip empty packets
 		}
 
-		if (opt.pblock->cnt) {
+		parse_packet(
+			&opt,
+			m_parser_stats,
+			timestamp,
+			ndp_packet->data,
+			ndp_packet->data_length,
+			ndp_packet->data_length);
+
+		if (opt.pblock->cnt >= packets.size) {
 			break;
 		}
 	}
 
-	m_seen += read_pkts;
+	m_seen += received;
 	m_parsed += opt.pblock->cnt;
 
-	m_stats.receivedPackets += read_pkts;
+	m_stats.receivedPackets += received;
 	m_stats.receivedBytes += packets.bytes;
 
 	if (opt.pblock->cnt) {
 		return Result::PARSED;
-	} else if (any_timeout) {
+	} else if (received == 0) {
 		return Result::TIMEOUT;
 	} else {
 		return Result::NOT_PARSED;

src/plugins/input/nfb/src/ndp.hpp

@@ -19,6 +19,9 @@
 #include <ipfixprobe/packet.hpp>
 #include <ipfixprobe/utils.hpp>
 
+#include <memory>
+#include <span>
+
 namespace ipxp {
 
 class NdpOptParser : public OptionsParser {
@@ -86,6 +89,9 @@ class NdpPacketReader : public InputPlugin {
 	uint64_t m_reader_idx = 0;
 	RxStats m_stats = {};
 
+	std::unique_ptr<struct ndp_packet[]> ndp_packet_burst;
+	std::array<timeval, 64> timestamps;
+
 	void init_ifc(const std::string& dev);
 };
 

src/plugins/input/nfb/src/ndpReader.cpp

@@ -198,6 +198,46 @@ void NdpReader::set_sw_timestamp(struct timeval* tv)
 	tv->tv_usec = micros;
 }
 
+int NdpReader::get_packets(std::span<struct ndp_packet> packets, std::span<timeval> timestamps)
+{
+	if (blocked_packets > 128) {
+		ndp_rx_burst_put(rx_handle);
+		blocked_packets = 0;
+	}
+
+	const unsigned received = ndp_rx_burst_get(rx_handle, packets.data(), packets.size());
+	for (unsigned i = 0; i < received; i++) {
+		struct ndp_packet* ndp_packet = &packets[i];
+		if (fw_type == NdpFwType::NDP_FW_HANIC) {
+			uint64_t* fw_ts = &((NdpHeader*) (ndp_packet->header))->timestamp;
+			if (*fw_ts == 0) {
+				set_sw_timestamp((struct timeval*) &timestamps[i]);
+			} else {
+				convert_fw_ts_to_timeval(fw_ts, (struct timeval*) &timestamps[i]);
+			}
+		} else {
+			uint8_t header_id = ndp_packet_flag_header_id_get(ndp_packet);
+			if (header_id >= ndk_timestamp_offsets.size()) {
+				set_sw_timestamp((struct timeval*) &timestamps[i]);
+			} else if (ndk_timestamp_offsets[header_id] == std::numeric_limits<uint32_t>::max()) {
+				set_sw_timestamp((struct timeval*) &timestamps[i]);
+			} else {
+				uint64_t* fw_ts
+					= (uint64_t*) ((uint8_t*) ndp_packet->header + ndk_timestamp_offsets[header_id]);
+				if (*fw_ts == std::numeric_limits<uint64_t>::max()) {
+					set_sw_timestamp((struct timeval*) &timestamps[i]);
+				} else {
+					convert_fw_ts_to_timeval(fw_ts, (struct timeval*) &timestamps[i]);
+				}
+			}
+		}
+	}
+
+	blocked_packets += received;
+
+	return received;
+}
+
 int NdpReader::get_pkt(struct ndp_packet** ndp_packet_out, struct timeval* timestamp)
 {
 	if (ndp_packet_buffer_processed >= ndp_packet_buffer_packets) {

src/plugins/input/nfb/src/ndpReader.hpp

@@ -9,6 +9,7 @@
 #include <stdint.h>
 #include <sys/time.h>
 #include <unistd.h>
+#include <span>
 
 extern "C" {
 #include <nfb/ndp.h>
@@ -49,6 +50,8 @@ class NdpReader {
 	int get_pkt(struct ndp_packet** ndp_packet, struct timeval* timestamp);
 	std::string error_msg;
 
+	int get_packets(std::span<struct ndp_packet> packets, std::span<timeval> timestamps);
+
 private:
 	void set_booted_fw();
 	void convert_fw_ts_to_timeval(const uint64_t* fw_ts, struct timeval* tv);
@@ -60,6 +63,8 @@ class NdpReader {
 	uint16_t packet_bufferSize;
 	uint64_t timeout;
 
+	uint64_t blocked_packets = 0;
+
 	NdpFwType fw_type;
 	std::vector<uint32_t> ndk_timestamp_offsets;