Setting BPF_F_TEST_XDP_LIVE_FRAMES flag with running BPF_PROG_TEST_RUN

[tariromukute]

Hi. I am trying to implement traffic generation using bcc. I would like to run BPF_PROG_TEST_RUN in live mode but I don't know how I can set the flags, and I can't find the signature of the method.

I have the following test code that get's invoked, but the packets are not being sent out.

from bcc import BPF, libbcc
from bcc.utils import printb
import time
from scapy.all import Ether, IP, raw, TCP, UDP, ICMP
from scapy.contrib.gtp import (
        GTP_U_Header,
        GTPPDUSessionContainer)
import ctypes

device = "eth1"

b = BPF(src_file="gtpu.bpf.c")
func = b.load_func("xdp_redirect_update_gtpu", BPF.XDP)

binary_data = bytes.fromhex(hex_data)

ethernet = Ether(dst="60:45:bd:44:07:d9", src="60:45:bd:42:8b:5c")
outerIp = IP(src="10.0.2.4", dst="10.0.2.5")
outerUdp = UDP(sport=2152, dport=2152,chksum=0)
innerIp = IP(src="12.1.1.4", dst="10.50.100.1")
icmpPkt = ICMP()

gtpHeader = GTP_U_Header(teid=0, next_ex=133)/GTPPDUSessionContainer(type=1, QFI=9)

sendingPacket = ethernet/outerIp/outerUdp/gtpHeader/innerIp/icmpPkt
size = len(sendingPacket)
print(f"Send GTPU packet \n{sendingPacket.show(dump=True)}")
sendingPacket = ctypes.create_string_buffer(raw(sendingPacket), len(sendingPacket))

SKB_OUT_SIZE = 1514  # mtu 1500 + 14 ethernet size

packet_output_size = ctypes.c_uint32()
test_retval = ctypes.c_uint32()
duration = ctypes.c_uint32()
repeat = 1000

libbcc.lib.bpf_prog_test_run(
    func.fd,
    repeat,
    ctypes.byref(sendingPacket),
    len(sendingPacket),
    ctypes.byref(packet_output),
    ctypes.byref(packet_output_size),
    ctypes.byref(test_retval),
    ctypes.byref(duration),

    )

txcnt = b.get_table("txcnt")
prev = 0
print("Printing generated packets, hit CTRL+C to stop")
while 1:
    try:
        val = txcnt.sum(0).value
        if val:
            delta = val - prev
            prev = val
            print("{} pkt/s".format(delta))
        time.sleep(1)
    except KeyboardInterrupt:
        print("Removing filter from device")
        break

b.remove_xdp(device, 0)

The XDP program is as follows

#include "utils.h"

BPF_PERCPU_ARRAY(txcnt, long, 1);

int xdp_redirect_update_gtpu(struct xdp_md *ctx)
{
	int action = XDP_ABORTED;
	__u32 key = 0;

	action = bpf_redirect(3, 0);
	long *value = txcnt.lookup(&key);
    if (value)
        *value += 1;
out:
	return action;
}

The count goes up to the 1000 packets specified, indicating the program is being invoked. How do I set the flag to be BPF_F_TEST_XDP_LIVE_FRAMES. I can't find the signature of the method libbcc.lib.bpf_prog_test_run, how can I go about it. It's not in the generated .local/lib/python3.10/site-packages/bcc/libbcc.py

[yonghong-song]

I don't think we have python API for bpf_prog_test_run. If possible, could you add this one to source/python/bcc/libbcc.py?

[tariromukute]

Thanks @yonghong-song. I have added it in the PR #4877. Tested it on my end, also add the test program on the PR.