flow/flow_learn_span/flow_learn_span_example.cpp
Description
This example is an application that receives incoming IPv4 UDP packets, and then program, aka learn, them as new flows. The flows are learned with sorted IP addresses and port, enabling a single physical port to handle packets in both directions of the flow. See SPAN port.
This application also maintains a lookup table of programmed flows, to ensure that the same flow isn't learned more than once.
This application doesn't transmit packets, and doesn't monitor traffic in a significant way. Running this application alongside of the NT tools "pktgen" and "monitoring" will help making the functionality of this application clear.
The example program uses the following NTAPI functions:
- NT_Init()
- NT_Done()
- NT_ExplainError()
- NT_ConfigOpen()
- NT_ConfigClose()
- NT_NTPL()
- NT_NetRxOpen()
- NT_NetRxClose()
- NT_NetRxGetNextPacket()
- NT_NET_GET_PKT_DESCR_PTR_DYN4()
- NT_FlowOpenAttrInit()
- NT_FlowOpenAttrSetAdapterNo()
- NT_FlowOpen_Attr()
- NT_FlowGetVersion()
- NT_FlowWrite()
- NT_FlowClose()
This example is for adapters with flowmatcher functionality.
/*
*
* Copyright 2025 Napatech A/S. All Rights Reserved.
*
* 1. Copying, modification, and distribution of this file, or executable
* versions of this file, is governed by the terms of the Napatech Software
* license agreement under which this file was made available. If you do not
* agree to the terms of the license do not install, copy, access or
* otherwise use this file.
*
* 2. Under the Napatech Software license agreement you are granted a
* limited, non-exclusive, non-assignable, copyright license to copy, modify
* and distribute this file in conjunction with Napatech SmartNIC's and
* similar hardware manufactured or supplied by Napatech A/S.
*
* 3. The full Napatech Software License Agreement is included in this
* distribution, please see "NA-0009 Software License Agreement.pdf"
*
* 4. Redistributions of source code must retain this copyright notice,
* list of conditions and the following disclaimer.
*
* THIS SOFTWARE IS PROVIDED "AS IS" WITHOUT ANY WARRANTIES, EXPRESS OR
* IMPLIED, AND NAPATECH DISCLAIMS ALL IMPLIED WARRANTIES INCLUDING ANY
* IMPLIED WARRANTY OF TITLE, MERCHANTABILITY, NONINFRINGEMENT, OR OF
* FITNESS FOR A PARTICULAR PURPOSE. TO THE EXTENT NOT PROHIBITED BY
* APPLICABLE LAW, IN NO EVENT SHALL NAPATECH BE LIABLE FOR PERSONAL INJURY,
* OR ANY INCIDENTAL, SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES WHATSOEVER,
* INCLUDING, WITHOUT LIMITATION, DAMAGES FOR LOSS OF PROFITS, CORRUPTION OR
* LOSS OF DATA, FAILURE TO TRANSMIT OR RECEIVE ANY DATA OR INFORMATION,
* BUSINESS INTERRUPTION OR ANY OTHER COMMERCIAL DAMAGES OR LOSSES, ARISING
* OUT OF OR RELATED TO YOUR USE OR INABILITY TO USE NAPATECH SOFTWARE OR
* SERVICES OR ANY THIRD PARTY SOFTWARE OR APPLICATIONS IN CONJUNCTION WITH
* THE NAPATECH SOFTWARE OR SERVICES, HOWEVER CAUSED, REGARDLESS OF THE THEORY
* OF LIABILITY (CONTRACT, TORT OR OTHERWISE) AND EVEN IF NAPATECH HAS BEEN
* ADVISED OF THE POSSIBILITY OF SUCH DAMAGES. SOME JURISDICTIONS DO NOT ALLOW
* THE EXCLUSION OR LIMITATION OF LIABILITY FOR PERSONAL INJURY, OR OF
* INCIDENTAL OR CONSEQUENTIAL DAMAGES, SO THIS LIMITATION MAY NOT APPLY TO YOU.
*
*
*/
/**
* @example flow/flow_learn_span/flow_learn_span_example.cpp
* @section flow_learn_span_example Description
*
* This example is an application that receives incoming IPv4 UDP packets,
* and then program, aka learn, them as new flows. The flows are learned with sorted
* IP addresses and port, enabling a single physical port to handle packets in
* both directions of the flow. See SPAN port.
*
* This application also maintains a lookup table of programmed flows, to ensure
* that the same flow isn't learned more than once.
*
* This application doesn't transmit packets, and doesn't monitor traffic in a
* significant way. Running this application alongside of the NT tools "pktgen"
* and "monitoring" will help making the functionality of this application clear.
*
* The example program uses the following NTAPI functions:
* - @ref NT_Init()
* - @ref NT_Done()
* - @ref NT_ExplainError()
* - @ref NT_ConfigOpen()
* - @ref NT_ConfigClose()
* - @ref NT_NTPL()
* - @ref NT_NetRxOpen()
* - @ref NT_NetRxClose()
* - @ref NT_NetRxGetNextPacket()
* - @ref NT_NET_GET_PKT_DESCR_PTR_DYN4()
* - @ref NT_FlowOpenAttrInit()
* - @ref NT_FlowOpenAttrSetAdapterNo()
* - @ref NT_FlowOpen_Attr()
* - @ref NT_FlowGetVersion()
* - @ref NT_FlowWrite()
* - @ref NT_FlowClose()
*
* This example is for adapters with flowmatcher functionality.
*/
// Include this in order to access the Napatech API
#include <nt.h>
#include <algorithm>
#include <array>
#include <atomic>
#include <cstdint>
#include <cstring>
#include <functional>
#include <iostream>
#include <memory>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>
#define STR_INNER(A) #A
#define STR(A) STR_INNER(A)
uint8_t gGfi = 0;
namespace {
////////////////////////////////////////////////////////////////////////////////
// Utility functions
////////////////////////////////////////////////////////////////////////////////
/**
* Handle errors by printing an error and exit the application.
*/
{
if (status != NT_SUCCESS) {
std::cerr << message << ": " << error_buffer << std::endl;
std::exit(EXIT_FAILURE);
}
}
/**
* Opens a config stream and programs the input NTPL.
* When the NTPL has been programmed, the config stream is closed again.
*/
{
int status;
NtConfigStream_t config_stream;
handle_error_status(status, "NT_ConfigOpen() failed");
for (const auto& ntpl : ntpls) {
NtNtplInfo_t ntpl_info;
handle_error_status(status, "NT_NTPL() failed");
}
status = NT_ConfigClose(config_stream);
handle_error_status(status, "NT_ConfigClose() failed");
}
/**
* Quick and dirty hashing function for some fields in NtFlow_t.
* The hash uniqueness in this function is not great, but since std::unordered_multimap
* is used to store the flows, the application is able to handle duplicate hashes.
*/
{
uint64_t meta = static_cast<uint64_t>((flow->ipProtocolField << 16) | (flow->keyId << 8) | flow->keySetId);
// Multiply data with some random primes.
return (ptr[0] * 45684803) ^
(ptr[1] * 198138211) ^
(ptr[2] * 994229) ^
(ptr[3] * 8349871) ^
(ptr[4] * 3294876479) ^
(meta * 663664226587);
}
////////////////////////////////////////////////////////////////////////////////
// The general thread main function used for all StreamID
////////////////////////////////////////////////////////////////////////////////
/**
* Opens a RX stream and received packets until an end-task signal is received.
*/
{
int status;
NtNetStreamRx_t net_stream;
NtNetBuf_t net_buffer;
// Open RX stream
status = NT_NetRxOpen(&net_stream, "flow_learn_span NT_NetRxOpen", NT_NET_INTERFACE_PACKET, stream_id, -1);
handle_error_status(status, "NT_NetRxOpen() failed");
// Tell main function that this is ready to receive packets, and then start the loop.
ready->fetch_add(1);
while (!end_task->load()) {
// Get next packet or continue.
status = NT_NetRxGetNextPacket(net_stream, &net_buffer, 100);
if (status == NT_STATUS_TIMEOUT || status == NT_STATUS_TRYAGAIN) continue;
handle_error_status(status, "NT_NetRxGetNextPacket() failed");
// See the handle functions.
handle(net_buffer);
}
// Close the RX stream.
status = NT_NetRxClose(net_stream);
handle_error_status(status, "NT_NetRxClose() failed");
}
////////////////////////////////////////////////////////////////////////////////
// Global variables
////////////////////////////////////////////////////////////////////////////////
#define BLACKLIST 3
#define WHITELIST 4
struct context {
uint64_t data[8];
};
struct context context_miss;
struct context context_blacklist;
struct context context_whitelist;
std::unordered_multimap<uint64_t, std::unique_ptr<NtFlow_t>> flow_map;
std::atomic<uint64_t> flow_learn_queue_read_index {0};
std::atomic<uint64_t> flow_learn_queue_write_index {0};
std::array<NtFlow_t*, 0x100000> flow_learn_queue;
constexpr uint64_t flow_learn_queue_index_mask = 0xfffff;
////////////////////////////////////////////////////////////////////////////////
// Main thread function for programming flows
////////////////////////////////////////////////////////////////////////////////
/**
* Separate task to program flows
*/
{
// Tell main function that this is ready to receive packets, and then start the loop.
ready->fetch_add(1);
while (!end_task->load()) {
// If a new flow has been added to the queue, the pop and learn it.
auto flow_raw_ptr = flow_learn_queue[flow_learn_queue_read_index];
flow_learn_queue_read_index.store((flow_learn_queue_read_index + 1) & flow_learn_queue_index_mask);
// Learn the flow and insert it into the map.
handle_error_status(status, "NT_FlowWrite() failed");
}
}
}
////////////////////////////////////////////////////////////////////////////////
// The handle functions for packets for each StreamID
////////////////////////////////////////////////////////////////////////////////
/**
* Handle used for the missed packets RX thread.
*
* Whenever a missed packet is received, the function calculates the flow learn
* data, and checks if it is a new flow, or a previously known flow. If the flow
* is new, then the flow is programmed.
*/
{
// As this is an example program, we will just use an incrementing flow ID
static uint32_t flow_id = 0;
++context_miss.data[0];
// Get the relevant pointers from the packet data.
const uint8_t* ipv4_dst = packet + dyn4->offset0 + 4;
const uint8_t* udp_dst = packet + dyn4->offset1 + 2;
// Sort the IPv4 addresses and UDP ports for the packet. This is only required
// when the option "KeySort=Sorted" is used in the KeyDef NTPL command.
// Because of endianness the comparison is actually not trivial,
// but the std algorithms library makes it a bit easier.
if (std::lexicographical_compare(ipv4_src, ipv4_src + 4, ipv4_dst, ipv4_dst + 4) ||
(std::equal(ipv4_src, ipv4_src + 4, ipv4_dst) &&
std::lexicographical_compare(udp_src, udp_src + 2, udp_dst, udp_dst + 2))) {
// Translates to: ipv4_src < ipv4_dst || (ipv4_src == ipv4_dst && udp_src < udp_dst)
std::memcpy(flow->keyData, ipv4_src, 4);
std::memcpy(flow->keyData + 4, ipv4_dst, 4);
std::memcpy(flow->keyData + 8, udp_src, 2);
std::memcpy(flow->keyData + 10, udp_dst, 2);
}
else {
std::memcpy(flow->keyData, ipv4_dst, 4);
std::memcpy(flow->keyData + 4, ipv4_src, 4);
std::memcpy(flow->keyData + 8, udp_dst, 2);
std::memcpy(flow->keyData + 10, udp_src, 2);
}
// Some packet inspection can be added here to decide whether the new flow
// should be added to the WHITELIST or the BLACKLIST.
// This example just alternates between WHITELIST and BLACKLIST
// Set other relevant fields.
flow->ipProtocolField = 0x11; // Layer 4 is UDP.
flow->keyId = 1; // Value used in the Key-test in the NTPL filter "Key(kd, KeyID=1) == WHITELIST / BLACKLIST".
flow->keySetId = key_set_id; // Value used to compare the Key-test in the NTPL filter "Key(kd, KeyID=1) == WHITELIST / BLACKLIST".
// Calculate a locally used hash. The "id" field is user defined and is used to
// store a hash in this example. For data structures other than std::unordered_multimap
// it could make sense to store the value of the pointer to NtFlow_t; this
// would enable instant and easy access to the locally stored flow data.
auto flow_raw_ptr = flow.get();
auto hash_key = flow_hash(flow_raw_ptr);
// Check if the unique key-value pair exists, and return if it does.
// The check would be a lot simpler if std::unordered_map was used instead of
// std::unordered_multimap, but when duplicate hashes would not be handled correctly.
auto range = flow_map.equal_range(hash_key);
for (auto it = range.first; it != range.second; ++it) {
flow->ipProtocolField == it->second->ipProtocolField &&
return;
}
}
flow->id = flow_id++;
flow_map.insert(std::pair<uint64_t, std::unique_ptr<NtFlow_t>>{hash_key, std::move(flow)});
// Compared to the number of packets that can be received per second on a RX stream,
// the learn rate is very slow, which can cause packet drops if the RX has to do
// that task as well. Thus it can be a good idea to implement a queue, and perform
// the learning from another thread.
flow_learn_queue[flow_learn_queue_write_index] = flow_raw_ptr;
flow_learn_queue_write_index.store((flow_learn_queue_write_index + 1) & flow_learn_queue_index_mask);
}
/**
* Handle used for the blacklist packets RX thread.
*
* In this example, packets that hit a programmed flow are only counted.
*/
{
++context_blacklist.data[0];
}
/**
* Handle used for the whitelist packets RX thread.
*
* In this example, packets that hit a programmed flow are only counted.
*/
{
++context_whitelist.data[0];
}
} // unnamed namespace
////////////////////////////////////////////////////////////////////////////////
// Main
////////////////////////////////////////////////////////////////////////////////
{
constexpr uint8_t adapter_no = 0;
if (argc == 2 && strcmp(argv[1], "gfi") == 0)
{
std::cout << "Flow information will be generated" << std::endl;
gGfi = 1;
}
else
{
std::cout << "Flow information will not be generated" << std::endl;
}
// Start NTAPI.
handle_error_status(status, "NT_Init() failed");
// Setup NTPL using a utility function.
// Delete previous filters and programmed flow before starting the new filters.
"Delete=All",
// Macros for simplifying following NTPL commands.
"DefineMacro(\"FilterCheck\", \"Port==$1 and Layer3Protocol==IPv4 and Layer4Protocol==UDP\")",
"DefineMacro(\"KeyTypeProtoSpecs\", \"(layer3header[12]/32, layer3header[16]/32, layer4header[0]/16,layer4header[2]/16)\")",
// Specify the packet field sizes in the 5-tuple. The 5-tuple contains the
// IPv4 addresses, IP protocol field, and the UDP ports. The IP protocol
// field is implicit when the "IpProtocolField=Outer" option is used.
// In this example sorting is used to receive two-way traffic on one port.
"KeyType[Name=kt] = {32, 32, 16, 16}",
"KeyDef[Name=kd; KeyType=kt; IpProtocolField=Outer; KeySort=Sorted] = KeyTypeProtoSpecs",
// Set up filters with a Key test.
"Assign[StreamId=0; Descriptor=DYN4, Offset0=Layer3Header[12], Offset1=Layer4Header[0]] = FilterCheck(0) and Key(kd, KeyID=1) == MISS",
"Assign[StreamId=1; Descriptor=DYN4, Offset0=Layer3Header[12], Offset1=Layer4Header[0]] = FilterCheck(0) and Key(kd, KeyID=1) == " STR(BLACKLIST),
"Assign[StreamId=2; Descriptor=DYN4, Offset0=Layer3Header[12], Offset1=Layer4Header[0]] = FilterCheck(0) and Key(kd, KeyID=1) == " STR(WHITELIST),
// Set up a low priority catch-all filter where packets will be dropped if
// they do not match any of the previous three filters.
"Assign[StreamId=Drop; Priority=10]=All"
});
// Start the flow stream for learning and unlearning flows.
NtFlowAttr_t flow_attr;
NT_FlowOpenAttrInit(&flow_attr);
NT_FlowOpenAttrSetAdapterNo(&flow_attr, adapter_no);
handle_error_status(status, "NT_FlowOpen_Attr() failed");
std::cout << "Detected Flow Stream API verion " << FlowStreamVersion << std::endl;
// Start RX and flow learn tasks
std::atomic<bool> end_task {false};
std::atomic<int> ready {0};
std::thread rx_task0(rx_task, &ready, &end_task, 0, handle_stream_miss);
std::thread rx_task1(rx_task, &ready, &end_task, 1, handle_stream_blacklist_hit);
std::thread rx_task2(rx_task, &ready, &end_task, 2, handle_stream_whitelist_hit);
std::thread flow_program_task0(flow_program_task, &ready, &end_task);
// Wait for RX tasks to be ready, then wait for user to stop the application.
while(ready.load() < 4) std::this_thread::yield();
std::cout << "Press enter to end application..." << std::endl;
std::cin.get();
// End application.
end_task.store(true);
if (flow_program_task0.joinable()) flow_program_task0.join();
if (rx_task2.joinable()) rx_task2.join();
if (rx_task1.joinable()) rx_task1.join();
if (rx_task0.joinable()) rx_task0.join();
// Unlearn all the programmed flows.
NtFlow_t* flow = it->second.get();
flow->op = 0;
status = NT_FlowWrite(flow_stream, flow, -1);
handle_error_status(status, "NT_FlowWrite() failed");
}
// Close the flow stream
handle_error_status(status, "NT_FlowClose() failed");
// End NTAPI.
NT_Done();
return 0;
}
| DN-0449 Napatech Software Suite Reference Documentation (v. 3.29.38.15-72ea8b4e) Copyright © 2025 Napatech A/S |