I am currently learning about packet-capturing techniques on Linux and experimenting with AF_PACKET sockets and PACKET_FANOUT specifically PAKCET_FANOUT_LB.
I created a test program with 4 AF_PACKET sockets, all bound to the same network interface and belonging to the same fanout group.
Each socket is handled by its own capture thread.
The architecture is:
Network Interface
|
v
AF_PACKET fanout
PACKET_FANOUT_LB
|
+-------------+---------------------------+
| | | |
v v v v
Socket 0 Socket 1 Socket 2 Socket 3
| | | |
v v v v
Thread 0 Thread 1 Thread 2 Thread 3
| | | |
v v v v
Queue 0 Queue 1 Queue 2 Queue 3
| | | |
+-------------+-------------+-------------+
|
v
Collector thread
Each capture thread reads packets from its socket, makes a heap copy of the packet, and pushes the copied packet into its own queue. A single collector thread then consumes packets from the four queues.
For debugging, the collector prints information such as:
Pkt No , Src IP , Dest IP , Src port , Dest port , Protocol , TCP Seq No , TCP Ack No
I looked at the Linux kernel source code in net/packet/af_packet.c, specifically fanout_demux_lb(). The relevant code in the kernel source I am testing is:
static unsigned int fanout_demux_lb(struct packet_fanout *f,
struct sk_buff *skb,
unsigned int num)
{
unsigned int val = atomic_inc_return(&f->rr_cur);
return val % num;
}
Since I have 4 sockets, my understanding is that if rr_cur starts at 0, the sequence would be: 1,2,3,0,1,2,3,0...
Therefore, I expected the sockets selected by the load-balancing algorithm to follow:
Socket 1 Socket 2 Socket 3 Socket 0 Socket 1 Socket 2 Socket 3 Socket 0 ...
At the beginning of the test, the packet sequence observed by my application followed the expected pattern very well. I also captured the same traffic using Wireshark and compared the packets using the TCP sequence and acknowledgement numbers. The application log and Wireshark capture appeared to correspond correctly. I have attached screenshots showing the initial behavior.
However, after approximately one hour of continuous execution, the round-robin pattern observed by my application starts to break.
For example, after packet number 214510, the round robin sequence no longer appears to follow
I have attached screenshots showing the application log and the corresponding Wireshark capture at the point where this happens.
I repeated the same test again, and the same type of behavior occurred, so it does not appear to be a one-time occurrence.
The interesting part is that the problem does not happen immediately. The application can run for a long time while the expected pattern is observed, and then the observed sequence starts to differ.
#include "Logger.h"
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
static const int CAPTURE_THREADS = 4;
static const int START_QUEUE = 1;
static const int SNAPSHOT_LENGTH = 65536;
static const int READ_TIMEOUT_MS = 1000;
static const int KERNEL_BUFFER_SIZE = 256 * 1024 * 1024;
static const char* INTERFACE_NAME = "wlp2s0";
static uint16_t fanoutID = static_cast(getpid());
struct CapturedPacket {
std::vector data;
uint64_t packetNumber;
};
struct CollectorQueue {
std::queue packets;
std::mutex mutex;
std::condition_variable condition;
};
static std::vector captureSockets(CAPTURE_THREADS, -1);
static std::vector captureThreads;
static std::vector collectorQueues(CAPTURE_THREADS);
static std::thread collectorThread;
static std::atomic stopPacketCapture(false);
static std::atomic packetNumber(0);
bool setPromiscuousMode(int socketFd, const char* interfaceName) {
struct packet_mreq mreq{};
mreq.mr_ifindex = static_cast(if_nametoindex(interfaceName));
mreq.mr_type = PACKET_MR_PROMISC;
if (setsockopt(socketFd, SOL_PACKET, PACKET_ADD_MEMBERSHIP,
&mreq, sizeof(mreq)) < 0) {
return false;
}
return true;
}
void setKernelBuffer(int socketFd) {
int bufferSize = KERNEL_BUFFER_SIZE;
if (setsockopt(socketFd, SOL_SOCKET, SO_RCVBUFFORCE, &bufferSize, sizeof(bufferSize)) < 0) {
LOG("CAPTURE", "Failed to set kernel buffer to 256 MB");
}
}
bool setReadTimeout(int socketFd) {
struct timeval timeout{};
timeout.tv_sec = READ_TIMEOUT_MS / 1000;
timeout.tv_usec = (READ_TIMEOUT_MS % 1000) * 1000;
if (setsockopt(socketFd, SOL_SOCKET, SO_RCVTIMEO,&timeout, sizeof(timeout)) < 0) {
return false;
}
return true;
}
int createSocket(const char* interfaceName, int threadID) {
int socketFd = socket(AF_PACKET, SOCK_RAW, htons(ETH_P_ALL));
if (socketFd < 0) {
LOG("CAPTURE " + std::to_string(threadID), "Failed to create socket");
return -1;
}
int interfaceIndex = static_cast(if_nametoindex(interfaceName));
if (interfaceIndex == 0) {
LOG("CAPTURE " + std::to_string(threadID),"Failed to get interface index");
close(socketFd);
return -1;
}
struct sockaddr_ll address{};
address.sll_family = AF_PACKET;
address.sll_protocol = htons(ETH_P_ALL);
address.sll_ifindex = interfaceIndex;
if (bind(socketFd, reinterpret_cast(&address),sizeof(address)) < 0) {
LOG("CAPTURE " + std::to_string(threadID),"Failed to bind socket");
close(socketFd);
return -1;
}
if (!setPromiscuousMode(socketFd, interfaceName)) {
LOG("CAPTURE " + std::to_string(threadID),"Failed to enable promiscuous mode");
close(socketFd);
return -1;
}
setKernelBuffer(socketFd);
if (!setReadTimeout(socketFd)) {
LOG("CAPTURE " + std::to_string(threadID),"Failed to set read timeout");
close(socketFd);
return -1;
}
return socketFd;
}
bool joinFanoutGroup(int socketFd) {
uint32_t group = (static_cast(PACKET_FANOUT_LB) << 16) | fanoutID;
if (setsockopt(socketFd, SOL_PACKET, PACKET_FANOUT, &group, sizeof(group)) < 0) {
return false;
}
return true;
}
void parsePacket(const CapturedPacket& packet, int threadID) {
std::string threadName = "CAPTURE " + std::to_string(threadID);
if (packet.data.size() < sizeof(struct ethhdr)) {
LOG(threadName,"[Non IP] Pkt No : ", packet.packetNumber);
return;
}
const struct ethhdr* ethernet =
reinterpret_cast(packet.data.data());
if (ntohs(ethernet->h_proto) != ETH_P_IP) {
LOG(threadName,"[Non IP] Pkt No : ", packet.packetNumber);
return;
}
if (packet.data.size() <
sizeof(struct ethhdr) + sizeof(struct ip)) {
LOG(threadName,"[Non IP] Pkt No : ", packet.packetNumber);
return;
}
const struct ip* ipHeader = reinterpret_cast(packet.data.data() + sizeof(struct ethhdr));
char srcIP[INET_ADDRSTRLEN];
char dstIP[INET_ADDRSTRLEN];
inet_ntop(AF_INET, &ipHeader->ip_src,srcIP, sizeof(srcIP));
inet_ntop(AF_INET, &ipHeader->ip_dst,dstIP, sizeof(dstIP));
const unsigned char* transportHeader = packet.data.data() + sizeof(struct ethhdr) + (ipHeader->ip_hl * 4);
if (ipHeader->ip_p == IPPROTO_TCP) {
if (packet.data.size() < sizeof(struct ethhdr) + (ipHeader->ip_hl * 4) + sizeof(struct tcphdr)) {
return;
}
const struct tcphdr* tcp = reinterpret_cast(transportHeader);
LOG(threadName,
"[TCP] Pkt No : ", packet.packetNumber,
" Src IP : ", srcIP,
" Dst IP : ", dstIP,
" Src Port : ", ntohs(tcp->source),
" Dst Port : ", ntohs(tcp->dest),
" Seq No : ", ntohl(tcp->seq),
" Ack No : ", ntohl(tcp->ack_seq));
return;
}
if (ipHeader->ip_p == IPPROTO_UDP) {
if (packet.data.size() < sizeof(struct ethhdr) + (ipHeader->ip_hl * 4) + sizeof(struct udphdr)) {
return;
}
const struct udphdr* udp = reinterpret_cast(transportHeader);
LOG(threadName,
"[UDP] Pkt No : ", packet.packetNumber,
" Src IP : ", srcIP,
" Dst IP : ", dstIP,
" Src Port : ", ntohs(udp->source),
" Dst Port : ", ntohs(udp->dest));
return;
}
LOG(threadName,"[Non IP] Pkt No : ", packet.packetNumber);
}
void capturePacketOnBlockingMode(int socketFd, int threadID) {
uint8_t* buffer = new uint8_t[SNAPSHOT_LENGTH];
std::string threadName = "CAPTURE " + std::to_string(threadID);
LOG(threadName,"Draining the Pre-Started Packets");
while (recv(socketFd, buffer, SNAPSHOT_LENGTH,MSG_DONTWAIT) > 0) {
}
if (stopPacketCapture.load()) {
LOG(threadName,"Stop packet capture is set to true, stopping capture thread");
delete[] buffer;
return;
}
LOG(threadName,"Started Capturing Packets");
while (!stopPacketCapture.load()) {
ssize_t received = recv(socketFd, buffer, SNAPSHOT_LENGTH, 0);
if (received < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
continue;
}
if (!stopPacketCapture.load()) {
LOG(threadName,"recv failed");
}
continue;
}
if (received == 0)
continue;
CapturedPacket packet;
packet.packetNumber = ++packetNumber;
packet.data.assign(buffer,buffer + received);
CollectorQueue& queue = collectorQueues[threadID];
{
std::lock_guard lock(queue.mutex);
queue.packets.push(std::move(packet));
}
queue.condition.notify_one();
}
LOG(threadName,"Capture thread stopped");
delete[] buffer;
}
void collectPackets() {
int queueIndex = START_QUEUE;
LOG("COLLECTOR","Collector thread started");
while (!stopPacketCapture.load()) {
CollectorQueue& queue = collectorQueues[queueIndex];
std::unique_lock lock(queue.mutex);
queue.condition.wait(lock, [&queue] {
return !queue.packets.empty() ||
stopPacketCapture.load();
});
if (stopPacketCapture.load())
break;
CapturedPacket packet =
std::move(queue.packets.front());
queue.packets.pop();
lock.unlock();
parsePacket(packet, queueIndex);
queueIndex =
(queueIndex + 1) % CAPTURE_THREADS;
}
LOG("COLLECTOR","Collector thread stopped");
}
void startThreads() {
for(int i = 0; i < CAPTURE_THREADS; ++i){
if (!joinFanoutGroup(captureSockets[i])) {
LOG("CAPTURE " + std::to_string(i),"Failed to join fanout group");
stopPacketCapture = true;
return;
}
}
for (int i = 0; i < CAPTURE_THREADS; ++i) {
captureThreads.emplace_back(capturePacketOnBlockingMode,captureSockets[i],i);
}
collectorThread =std::thread(collectPackets);
}
void startAFPacketCapture(const char* interfaceName) {
for (int i = 0; i < CAPTURE_THREADS; ++i) {
captureSockets[i] = createSocket(interfaceName, i);
if (captureSockets[i] < 0) {
stopPacketCapture = true;
return;
}
}
startThreads();
return;
}
void stopThreads() {
stopPacketCapture = true;
for (int i = 0; i < CAPTURE_THREADS; ++i)
collectorQueues[i].condition.notify_all();
for (int socketFd : captureSockets) {
if (socketFd >= 0)
shutdown(socketFd, SHUT_RDWR);
}
for (std::thread& thread : captureThreads) {
if (thread.joinable())
thread.join();
}
if (collectorThread.joinable())
collectorThread.join();
for (int& socketFd : captureSockets) {
if (socketFd >= 0) {
close(socketFd);
socketFd = -1;
}
}
logger.stop();
}
int main() {
logger.start();
startAFPacketCapture(INTERFACE_NAME);
while (!stopPacketCapture.load())
std::this_thread::sleep_for(std::chrono::seconds(1));
stopThreads();
return 0;
}
I attached screenshots of wireshark capture and my code logs in here (https://unix.stackexchange.com/questions/807293/af-packet-packet-fanout-lb-why-does-the-observed-round-robin-sequence-break-a) since I can't attach here.
So , my questions are
- Does PACKET_FANOUT_LB guarantee strict round-robin assignment? , If not please give explanation.
- My actual requirement is to distribute packets from a single interface among multiple processing threads using a round-robin scheme. I would still like to use AF_PACKET. If PACKET_FANOUT_LB does not provide a guarantee that is sufficient for this requirement, is there another recommended way to achieve consistent round-robin distribution while continuing to use AF_PACKET?
Also , if you think there is some issue in my code which makes the round robin fail please correct me.