The PR follows the discussion #4873, to add a Python API for bpf_prog_test_run_opts. This will allow implementations of use cases like traffic generators. Below is the sample code using the API.
// trafficgen.bpf.c
#include <uapi/linux/bpf.h>
BPF_PERCPU_ARRAY(txcnt, long, 1);
int xdp_redirect(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;
}
# trafficgen.py
from bcc import BPF
from bcc.libbcc import lib, bpf_test_run_opts
from bcc.utils import printb
import time
from scapy.all import *
import ctypes
import os
BPF_F_TEST_XDP_LIVE_FRAMES = 1 << 1
device = "eth1"
b = BPF(src_file="trafficgen.bpf.c")
func = b.load_func("xdp_redirect", BPF.XDP)
ethernet = Ether(dst="60:45:bd:41:28:71", src="00:22:48:be:14:39")
ipHeader = IPv6(src="2404:f800:8000:122::5", dst="2404:f800:8000:122::4")
udpHeader = UDP(sport=12345, dport=54321,chksum=0)
payload = "This is a test message"
packet = ethernet/ipHeader/udpHeader/payload
sendingPacket = ctypes.create_string_buffer(raw(packet), len(packet))
packet_bytes = bytes(packet)
packet_array = (ctypes.c_ubyte * len(packet_bytes)).from_buffer_copy(packet_bytes)
opts = bpf_test_run_opts(
sz=ctypes.sizeof(bpf_test_run_opts),
data_in=ctypes.cast(packet_array, ctypes.c_void_p),
data_size_in=len(packet_bytes),
repeat=1000,
flags=BPF_F_TEST_XDP_LIVE_FRAMES,
batch_size=0,
)
try:
ret = lib.bpf_prog_test_run_opts(
func.fd,
ctypes.byref(opts)
)
if ret < 0:
raise Exception("Failed to run prog test BPF program %s" % os.strerror(ctypes.get_errno()))
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
except Exception as e:
# Catch the exception and print its message
print(f"An exception occurred: {e}")
b.remove_xdp(device, 0)
The PR follows the discussion #4873, to add a Python API for bpf_prog_test_run_opts. This will allow implementations of use cases like traffic generators. Below is the sample code using the API.