Add bounded BPF egress pacer tutorial

.github/workflows/test-libbpf.yml

@@ -20,6 +20,7 @@ jobs:
         submodules: 'recursive'
     - name: install deps
       run: |
+        sudo apt-get update
         sudo apt-get install -y --no-install-recommends \
         libelf1 libelf-dev zlib1g-dev \
         make git clang llvm pkg-config build-essential dwarves \
@@ -179,6 +180,10 @@ jobs:
       run: |
           make -C src/50-tcx
 
+    - name: test 53 egress pacer
+      run: |
+          make -C src/53-egress-pacer
+
     - name: test features bpf_token
       run: |
           make -C src/features/bpf_token

src/53-egress-pacer/.config

@@ -0,0 +1,10 @@
+level=Depth
+type=Networking
+kernel_min=6.16
+kernel_min_basis=documented
+architectures=x86_64
+btf=required
+kernel_config=CONFIG_BPF=y,CONFIG_BPF_SYSCALL=y,CONFIG_BPF_JIT=y,CONFIG_DEBUG_INFO_BTF=y,CONFIG_NET_SCHED=y,CONFIG_NET_SCH_BPF=y
+hardware=network-interface
+root=required
+test_status=ci-build

src/53-egress-pacer/.gitignore

@@ -0,0 +1,15 @@
+# Build artifacts
+.output/
+*.o
+*.skel.h
+
+# Generated binary
+egress_pacer
+
+# Python cache
+__pycache__/
+
+# Editor files
+*.swp
+*~
+.vscode/

src/53-egress-pacer/Makefile

@@ -0,0 +1,92 @@
+# SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause)
+OUTPUT := .output
+CLANG ?= clang
+LIBBPF_SRC := $(abspath ../third_party/libbpf/src)
+BPFTOOL_SRC := $(abspath ../third_party/bpftool/src)
+LIBBPF_OBJ := $(abspath $(OUTPUT)/libbpf.a)
+BPFTOOL_OUTPUT ?= $(abspath $(OUTPUT)/bpftool)
+BPFTOOL ?= $(BPFTOOL_OUTPUT)/bootstrap/bpftool
+ARCH ?= $(shell uname -m | sed 's/x86_64/x86/' \
+			 | sed 's/arm.*/arm/' \
+			 | sed 's/aarch64/arm64/' \
+			 | sed 's/ppc64le/powerpc/' \
+			 | sed 's/mips.*/mips/' \
+			 | sed 's/riscv64/riscv/' \
+			 | sed 's/loongarch64/loongarch/')
+VMLINUX := ../third_party/vmlinux/$(ARCH)/vmlinux.h
+INCLUDES := -I$(OUTPUT) -I../third_party/libbpf/include/uapi -I$(dir $(VMLINUX)) -I.
+CFLAGS := -g -Wall -Wextra
+ALL_LDFLAGS := $(LDFLAGS) $(EXTRA_LDFLAGS)
+
+APPS = egress_pacer
+
+CLANG_BPF_SYS_INCLUDES ?= $(shell $(CLANG) -v -E - </dev/null 2>&1 \
+	| sed -n '/<...> search starts here:/,/End of search list./{ s| \(/.*\)|-idirafter \1|p }')
+
+ifeq ($(V),1)
+	Q =
+	msg =
+else
+	Q = @
+	msg = @printf '  %-8s %s%s\n' "$(1)" \
+		      "$(patsubst $(abspath $(OUTPUT))/%,%,$(2))" \
+		      "$(if $(3), $(3))";
+	MAKEFLAGS += --no-print-directory
+endif
+
+define allow-override
+  $(if $(or $(findstring environment,$(origin $(1))),\
+            $(findstring command line,$(origin $(1)))),,\
+    $(eval $(1) = $(2)))
+endef
+
+$(call allow-override,CC,$(CROSS_COMPILE)cc)
+$(call allow-override,LD,$(CROSS_COMPILE)ld)
+
+.PHONY: all clean test
+all: $(APPS)
+
+clean:
+	$(call msg,CLEAN)
+	$(Q)rm -rf $(OUTPUT) $(APPS)
+
+test: all
+	python3 tests/test_egress_pacer.py ./egress_pacer
+
+$(OUTPUT) $(OUTPUT)/libbpf $(BPFTOOL_OUTPUT):
+	$(call msg,MKDIR,$@)
+	$(Q)mkdir -p $@
+
+$(LIBBPF_OBJ): $(wildcard $(LIBBPF_SRC)/*.[ch] $(LIBBPF_SRC)/Makefile) | $(OUTPUT)/libbpf
+	$(call msg,LIB,$@)
+	$(Q)$(MAKE) -C $(LIBBPF_SRC) BUILD_STATIC_ONLY=1 \
+		    OBJDIR=$(dir $@)/libbpf DESTDIR=$(dir $@) \
+		    INCLUDEDIR= LIBDIR= UAPIDIR= install
+
+$(BPFTOOL): | $(BPFTOOL_OUTPUT)
+	$(call msg,BPFTOOL,$@)
+	$(Q)$(MAKE) ARCH= CROSS_COMPILE= OUTPUT=$(BPFTOOL_OUTPUT)/ -C $(BPFTOOL_SRC) bootstrap
+
+$(OUTPUT)/%.bpf.o: %.bpf.c $(LIBBPF_OBJ) $(wildcard %.h) $(VMLINUX) | $(OUTPUT) $(BPFTOOL)
+	$(call msg,BPF,$@)
+	$(Q)$(CLANG) -g -O2 -target bpf -D__TARGET_ARCH_$(ARCH) \
+		     $(INCLUDES) $(CLANG_BPF_SYS_INCLUDES) \
+		     -c $(filter %.c,$^) -o $(patsubst %.bpf.o,%.tmp.bpf.o,$@)
+	$(Q)$(BPFTOOL) gen object $@ $(patsubst %.bpf.o,%.tmp.bpf.o,$@)
+
+$(OUTPUT)/%.skel.h: $(OUTPUT)/%.bpf.o | $(OUTPUT) $(BPFTOOL)
+	$(call msg,GEN-SKEL,$@)
+	$(Q)$(BPFTOOL) gen skeleton $< > $@
+
+$(patsubst %,$(OUTPUT)/%.o,$(APPS)): %.o: %.skel.h
+
+$(OUTPUT)/%.o: %.c $(wildcard %.h) | $(OUTPUT)
+	$(call msg,CC,$@)
+	$(Q)$(CC) $(CFLAGS) $(INCLUDES) -c $(filter %.c,$^) -o $@
+
+$(APPS): %: $(OUTPUT)/%.o $(LIBBPF_OBJ) | $(OUTPUT)
+	$(call msg,BINARY,$@)
+	$(Q)$(CC) $(CFLAGS) $^ $(ALL_LDFLAGS) -lelf -lz -o $@
+
+.DELETE_ON_ERROR:
+.SECONDARY:

src/53-egress-pacer/bpf_experimental.h

@@ -0,0 +1,32 @@
+/* SPDX-License-Identifier: GPL-2.0 */
+#ifndef __EGRESS_PACER_BPF_EXPERIMENTAL_H
+#define __EGRESS_PACER_BPF_EXPERIMENTAL_H
+
+#include <bpf/bpf_core_read.h>
+
+#define __contains(name, node) \
+	__attribute__((btf_decl_tag("contains:" #name ":" #node)))
+
+extern void *bpf_obj_new_impl(__u64 local_type_id, void *meta) __ksym;
+#define bpf_obj_new(type) \
+	((type *)bpf_obj_new_impl(bpf_core_type_id_local(type), NULL))
+
+extern void bpf_obj_drop_impl(void *kptr, void *meta) __ksym;
+#define bpf_obj_drop(kptr) bpf_obj_drop_impl(kptr, NULL)
+
+extern int bpf_list_push_front_impl(struct bpf_list_head *head,
+				    struct bpf_list_node *node,
+				    void *meta, __u64 off) __ksym;
+#define bpf_list_push_front(head, node) \
+	bpf_list_push_front_impl(head, node, NULL, 0)
+
+extern int bpf_list_push_back_impl(struct bpf_list_head *head,
+				   struct bpf_list_node *node,
+				   void *meta, __u64 off) __ksym;
+#define bpf_list_push_back(head, node) \
+	bpf_list_push_back_impl(head, node, NULL, 0)
+
+extern struct bpf_list_node *
+bpf_list_pop_front(struct bpf_list_head *head) __ksym;
+
+#endif /* __EGRESS_PACER_BPF_EXPERIMENTAL_H */

src/53-egress-pacer/egress_pacer.bpf.c

@@ -0,0 +1,197 @@
+// SPDX-License-Identifier: GPL-2.0
+#include "vmlinux.h"
+#include <bpf/bpf_helpers.h>
+#include <bpf/bpf_tracing.h>
+#include "bpf_experimental.h"
+#include "egress_pacer.h"
+
+char LICENSE[] SEC("license") = "GPL";
+
+#define NET_XMIT_SUCCESS 0x00
+#define NET_XMIT_DROP 0x01
+#define private(name) SEC(".data." #name) __hidden __attribute__((aligned(8)))
+
+struct bpf_sk_buff_ptr {
+	struct sk_buff *skb;
+};
+
+extern void bpf_qdisc_skb_drop(struct sk_buff *skb,
+				       struct bpf_sk_buff_ptr *to_free) __ksym;
+extern void bpf_qdisc_watchdog_schedule(struct Qdisc *sch, __u64 expire,
+					__u64 delta_ns) __ksym;
+extern void bpf_qdisc_bstats_update(struct Qdisc *sch,
+				    const struct sk_buff *skb) __ksym;
+extern void bpf_kfree_skb(struct sk_buff *skb) __ksym;
+
+const volatile __u64 rate_kbps = 1024;
+const volatile __u32 queue_limit = 256;
+
+struct pacer_stats stats;
+
+struct packet_node {
+	__u64 eligible_ns;
+	__u32 packet_len;
+	struct sk_buff __kptr *skb;
+	struct bpf_list_node node;
+};
+
+private(A) struct bpf_spin_lock queue_lock;
+private(A) struct bpf_list_head packet_queue __contains(packet_node, node);
+private(A) __u64 next_departure_ns;
+
+static struct qdisc_skb_cb *qdisc_skb_cb(const struct sk_buff *skb)
+{
+	return (struct qdisc_skb_cb *)skb->cb;
+}
+
+static __u32 qdisc_packet_len(const struct sk_buff *skb)
+{
+	return qdisc_skb_cb(skb)->pkt_len;
+}
+
+SEC("struct_ops/egress_pacer_enqueue")
+int BPF_PROG(egress_pacer_enqueue, struct sk_buff *skb, struct Qdisc *sch,
+	     struct bpf_sk_buff_ptr *to_free)
+{
+	struct packet_node *packet;
+	__u64 eligible_ns, interval_ns, now;
+	__u32 packet_len;
+
+	packet_len = qdisc_packet_len(skb);
+	if (sch->q.qlen >= sch->limit)
+		goto drop;
+
+	packet = bpf_obj_new(typeof(*packet));
+	if (!packet)
+		goto drop;
+
+	now = bpf_ktime_get_ns();
+	interval_ns = (__u64)packet_len * 8 * 1000000ULL / rate_kbps;
+	if (!interval_ns)
+		interval_ns = 1;
+	packet->packet_len = packet_len;
+	skb = bpf_kptr_xchg(&packet->skb, skb);
+	if (skb)
+		bpf_qdisc_skb_drop(skb, to_free);
+
+	bpf_spin_lock(&queue_lock);
+	eligible_ns = next_departure_ns > now ? next_departure_ns : now;
+	next_departure_ns = eligible_ns + interval_ns;
+	packet->eligible_ns = eligible_ns;
+	bpf_list_push_back(&packet_queue, &packet->node);
+	sch->q.qlen++;
+	sch->qstats.backlog += packet_len;
+	__sync_fetch_and_add(&stats.enqueued, 1);
+	if (sch->q.qlen > stats.max_qlen)
+		stats.max_qlen = sch->q.qlen;
+	bpf_spin_unlock(&queue_lock);
+	return NET_XMIT_SUCCESS;
+
+drop:
+	bpf_qdisc_skb_drop(skb, to_free);
+	__sync_fetch_and_add(&stats.policy_dropped, 1);
+	return NET_XMIT_DROP;
+}
+
+SEC("struct_ops/egress_pacer_dequeue")
+struct sk_buff *BPF_PROG(egress_pacer_dequeue, struct Qdisc *sch)
+{
+	struct bpf_list_node *node;
+	struct packet_node *packet;
+	struct sk_buff *skb = NULL;
+	__u64 expire, now;
+	__u32 packet_len;
+
+	bpf_spin_lock(&queue_lock);
+	node = bpf_list_pop_front(&packet_queue);
+	bpf_spin_unlock(&queue_lock);
+	if (!node) {
+		return NULL;
+	}
+
+	packet = container_of(node, struct packet_node, node);
+	now = bpf_ktime_get_ns();
+	if (now < packet->eligible_ns) {
+		expire = packet->eligible_ns;
+		bpf_spin_lock(&queue_lock);
+		bpf_list_push_front(&packet_queue, &packet->node);
+		bpf_spin_unlock(&queue_lock);
+		bpf_qdisc_watchdog_schedule(sch, expire, 0);
+		return NULL;
+	}
+
+	packet_len = packet->packet_len;
+	skb = bpf_kptr_xchg(&packet->skb, skb);
+	bpf_spin_lock(&queue_lock);
+	sch->q.qlen--;
+	sch->qstats.backlog -= packet_len;
+	bpf_spin_unlock(&queue_lock);
+	bpf_obj_drop(packet);
+
+	if (!skb)
+		return NULL;
+
+	bpf_qdisc_bstats_update(sch, skb);
+	__sync_fetch_and_add(&stats.dequeued, 1);
+	__sync_fetch_and_add(&stats.bytes_dequeued, packet_len);
+	return skb;
+}
+
+SEC("struct_ops/egress_pacer_init")
+int BPF_PROG(egress_pacer_init, struct Qdisc *sch, struct nlattr *opt,
+	     struct netlink_ext_ack *extack)
+{
+	(void)opt;
+	(void)extack;
+	sch->limit = queue_limit;
+	return 0;
+}
+
+SEC("struct_ops/egress_pacer_reset")
+void BPF_PROG(egress_pacer_reset, struct Qdisc *sch)
+{
+	int queued = sch->q.qlen;
+	int i;
+
+	bpf_for(i, 0, queued) {
+		struct bpf_list_node *node;
+		struct packet_node *packet;
+		struct sk_buff *skb = NULL;
+
+		bpf_spin_lock(&queue_lock);
+		node = bpf_list_pop_front(&packet_queue);
+		bpf_spin_unlock(&queue_lock);
+		if (!node)
+			break;
+
+		packet = container_of(node, struct packet_node, node);
+		skb = bpf_kptr_xchg(&packet->skb, skb);
+		if (skb) {
+			bpf_kfree_skb(skb);
+			__sync_fetch_and_add(&stats.cleanup_dropped, 1);
+		}
+		bpf_obj_drop(packet);
+	}
+
+	bpf_spin_lock(&queue_lock);
+	next_departure_ns = 0;
+	sch->q.qlen = 0;
+	sch->qstats.backlog = 0;
+	bpf_spin_unlock(&queue_lock);
+}
+
+SEC("struct_ops/egress_pacer_destroy")
+void BPF_PROG(egress_pacer_destroy, struct Qdisc *sch)
+{
+	(void)sch;
+}
+
+SEC(".struct_ops")
+struct Qdisc_ops pacer = {
+	.enqueue = (void *)egress_pacer_enqueue,
+	.dequeue = (void *)egress_pacer_dequeue,
+	.init = (void *)egress_pacer_init,
+	.reset = (void *)egress_pacer_reset,
+	.destroy = (void *)egress_pacer_destroy,
+	.id = "bpf_pacer",
+};

src/53-egress-pacer/egress_pacer.c

@@ -0,0 +1,269 @@
+// SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause)
+#include <errno.h>
+#include <getopt.h>
+#include <linux/pkt_sched.h>
+#include <net/if.h>
+#include <signal.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#include <unistd.h>
+#include <bpf/libbpf.h>
+#include "egress_pacer.h"
+#include "egress_pacer.skel.h"
+
+static volatile sig_atomic_t exiting;
+
+static struct env {
+	const char *interface;
+	unsigned long long rate_kbps;
+	unsigned int queue_limit;
+	unsigned int duration;
+	bool verbose;
+} env = {
+	.rate_kbps = 1024,
+	.queue_limit = 256,
+	.duration = 10,
+};
+
+static void handle_signal(int signal)
+{
+	(void)signal;
+	exiting = 1;
+}
+
+static int libbpf_print_fn(enum libbpf_print_level level, const char *format,
+			   va_list args)
+{
+	if (level == LIBBPF_DEBUG && !env.verbose)
+		return 0;
+	return vfprintf(stderr, format, args);
+}
+
+static void usage(const char *program)
+{
+	fprintf(stderr,
+		"Usage: %s --interface IFACE [--rate-kbps KBPS] "
+		"[--queue-limit PACKETS] [--duration SEC] [--verbose]\n\n"
+		"Temporarily pace one interface with a bounded BPF qdisc.\n\n"
+		"Options:\n"
+		"  -i, --interface IFACE       interface to control (required)\n"
+		"  -r, --rate-kbps KBPS       egress rate, 8-100000000 "
+		"(default: 1024)\n"
+		"  -q, --queue-limit PACKETS  queue capacity, 1-65535 "
+		"(default: 256)\n"
+		"  -d, --duration SEC         control window, 1-86400 "
+		"(default: 10)\n"
+		"  -v, --verbose              print libbpf diagnostics\n"
+		"  -h, --help                 show this help\n",
+		program);
+}
+
+static int parse_u64(const char *value, unsigned long long maximum,
+		     unsigned long long *result)
+{
+	char *end = NULL;
+	unsigned long long parsed;
+
+	errno = 0;
+	parsed = strtoull(value, &end, 10);
+	if (errno || end == value || *end || parsed > maximum)
+		return -EINVAL;
+	*result = parsed;
+	return 0;
+}
+
+static int parse_args(int argc, char **argv)
+{
+	static const struct option options[] = {
+		{ "interface", required_argument, NULL, 'i' },
+		{ "rate-kbps", required_argument, NULL, 'r' },
+		{ "queue-limit", required_argument, NULL, 'q' },
+		{ "duration", required_argument, NULL, 'd' },
+		{ "verbose", no_argument, NULL, 'v' },
+		{ "help", no_argument, NULL, 'h' },
+		{},
+	};
+	unsigned long long parsed;
+	int option;
+
+	while ((option = getopt_long(argc, argv, "i:r:q:d:vh", options, NULL)) != -1) {
+		switch (option) {
+		case 'i':
+			env.interface = optarg;
+			break;
+		case 'r':
+			if (parse_u64(optarg, 100000000, &parsed) || parsed < 8) {
+				fprintf(stderr, "invalid rate in Kbit/s: %s\n", optarg);
+				return -EINVAL;
+			}
+			env.rate_kbps = parsed;
+			break;
+		case 'q':
+			if (parse_u64(optarg, 65535, &parsed) || parsed == 0) {
+				fprintf(stderr, "invalid queue limit: %s\n", optarg);
+				return -EINVAL;
+			}
+			env.queue_limit = parsed;
+			break;
+		case 'd':
+			if (parse_u64(optarg, 86400, &parsed) || parsed == 0) {
+				fprintf(stderr, "invalid duration in seconds: %s\n", optarg);
+				return -EINVAL;
+			}
+			env.duration = parsed;
+			break;
+		case 'v':
+			env.verbose = true;
+			break;
+		case 'h':
+			usage(argv[0]);
+			exit(0);
+		default:
+			return -EINVAL;
+		}
+	}
+
+	if (!env.interface) {
+		fprintf(stderr, "--interface is required\n");
+		return -EINVAL;
+	}
+	if (optind != argc)
+		return -EINVAL;
+	return 0;
+}
+
+static long long monotonic_milliseconds(void)
+{
+	struct timespec timestamp;
+
+	if (clock_gettime(CLOCK_MONOTONIC, &timestamp))
+		return -errno;
+	return timestamp.tv_sec * 1000LL + timestamp.tv_nsec / 1000000;
+}
+
+static int wait_for_duration(void)
+{
+	struct timespec interval = { .tv_nsec = 100000000 };
+	long long deadline, now;
+
+	now = monotonic_milliseconds();
+	if (now < 0)
+		return (int)now;
+	deadline = now + env.duration * 1000LL;
+
+	while (!exiting) {
+		now = monotonic_milliseconds();
+		if (now < 0)
+			return (int)now;
+		if (now >= deadline)
+			break;
+		if (nanosleep(&interval, NULL) && errno != EINTR)
+			return -errno;
+	}
+
+	return 0;
+}
+
+int main(int argc, char **argv)
+{
+	struct egress_pacer_bpf *skel = NULL;
+	struct pacer_stats final_stats = {};
+	struct bpf_tc_hook hook = {
+		.sz = sizeof(hook),
+		.attach_point = BPF_TC_QDISC,
+		.parent = TC_H_ROOT,
+		.handle = TC_H_MAKE(1 << 16, 0),
+		.qdisc = "bpf_pacer",
+	};
+	bool qdisc_created = false;
+	unsigned int ifindex;
+	int cleanup_err;
+	int err;
+
+	err = parse_args(argc, argv);
+	if (err) {
+		usage(argv[0]);
+		return 1;
+	}
+
+	ifindex = if_nametoindex(env.interface);
+	if (!ifindex) {
+		fprintf(stderr, "interface does not exist: %s\n", env.interface);
+		return 1;
+	}
+	hook.ifindex = ifindex;
+
+	libbpf_set_print(libbpf_print_fn);
+	signal(SIGINT, handle_signal);
+	signal(SIGTERM, handle_signal);
+
+	skel = egress_pacer_bpf__open();
+	if (!skel) {
+		fprintf(stderr, "failed to open BPF skeleton\n");
+		return 1;
+	}
+	skel->rodata->rate_kbps = env.rate_kbps;
+	skel->rodata->queue_limit = env.queue_limit;
+
+	err = egress_pacer_bpf__load(skel);
+	if (err) {
+		fprintf(stderr,
+			"failed to load BPF qdisc: %s\n"
+			"This tool requires Linux 6.16+, CONFIG_NET_SCH_BPF, "
+			"BTF, and BPF JIT.\n",
+			strerror(-err));
+		goto cleanup;
+	}
+
+	err = egress_pacer_bpf__attach(skel);
+	if (err) {
+		fprintf(stderr, "failed to register bpf_pacer qdisc: %s\n",
+			strerror(-err));
+		goto cleanup;
+	}
+
+	err = bpf_tc_hook_create(&hook);
+	if (err) {
+		if (err == -EEXIST)
+			fprintf(stderr,
+				"refusing to replace the existing root qdisc on %s\n",
+				env.interface);
+		else
+			fprintf(stderr, "failed to attach bpf_pacer to %s: %s\n",
+				env.interface, strerror(-err));
+		goto cleanup;
+	}
+	qdisc_created = true;
+
+	printf("READY interface=%s rate_kbps=%llu queue_limit=%u duration=%u\n",
+	       env.interface, env.rate_kbps, env.queue_limit, env.duration);
+	fflush(stdout);
+
+	err = wait_for_duration();
+
+cleanup:
+	if (qdisc_created) {
+		cleanup_err = bpf_tc_hook_destroy(&hook);
+		if (cleanup_err) {
+			fprintf(stderr, "failed to remove bpf_pacer from %s: %s\n",
+				env.interface, strerror(-cleanup_err));
+			if (!err)
+				err = cleanup_err;
+		}
+	}
+	if (skel && skel->bss)
+		final_stats = skel->bss->stats;
+	if (qdisc_created) {
+		printf("SUMMARY enqueued=%llu dequeued=%llu policy_dropped=%llu "
+		       "cleanup_dropped=%llu bytes_dequeued=%llu max_qlen=%llu\n",
+		       final_stats.enqueued, final_stats.dequeued,
+		       final_stats.policy_dropped, final_stats.cleanup_dropped,
+		       final_stats.bytes_dequeued, final_stats.max_qlen);
+	}
+	egress_pacer_bpf__destroy(skel);
+	return err != 0;
+}

src/53-egress-pacer/egress_pacer.h

@@ -0,0 +1,14 @@
+/* SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause) */
+#ifndef __EGRESS_PACER_H
+#define __EGRESS_PACER_H
+
+struct pacer_stats {
+	unsigned long long enqueued;
+	unsigned long long dequeued;
+	unsigned long long policy_dropped;
+	unsigned long long cleanup_dropped;
+	unsigned long long bytes_dequeued;
+	unsigned long long max_qlen;
+};
+
+#endif /* __EGRESS_PACER_H */

src/53-egress-pacer/tests/test_egress_pacer.py

@@ -0,0 +1,294 @@
+#!/usr/bin/env python3
+"""Deterministic KVM integration checks for the BPF egress pacer."""
+
+from __future__ import annotations
+
+import errno
+import os
+import re
+import select
+import signal
+import socket
+import struct
+import subprocess
+import sys
+import threading
+import time
+
+
+TX_INTERFACE = "epac_tx"
+RX_INTERFACE = "epac_rx"
+ETHERTYPE = 0x88B5
+MAGIC = b"EPAC"
+FRAME_SIZE = 1024
+ATTEMPTS = 40
+RATE_KBPS = 64
+QUEUE_LIMIT = 8
+
+SUMMARY = re.compile(
+    r"SUMMARY enqueued=(\d+) dequeued=(\d+) policy_dropped=(\d+) "
+    r"cleanup_dropped=(\d+) bytes_dequeued=(\d+) max_qlen=(\d+)"
+)
+
+
+def run(*command: str, check: bool = True) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        command,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        check=check,
+    )
+
+
+def delete_test_link() -> None:
+    run("ip", "link", "del", TX_INTERFACE, check=False)
+
+
+def create_test_link() -> None:
+    delete_test_link()
+    run("ip", "link", "add", TX_INTERFACE, "type", "veth", "peer", "name", RX_INTERFACE)
+    run("ip", "link", "set", "dev", TX_INTERFACE, "addrgenmode", "none")
+    run("ip", "link", "set", "dev", RX_INTERFACE, "addrgenmode", "none")
+    run("ip", "link", "set", "dev", TX_INTERFACE, "up")
+    run("ip", "link", "set", "dev", RX_INTERFACE, "up")
+
+
+def wait_until_ready(process: subprocess.Popen[str]) -> list[str]:
+    lines: list[str] = []
+    deadline = time.monotonic() + 8
+    assert process.stdout is not None
+
+    while time.monotonic() < deadline:
+        readable, _, _ = select.select([process.stdout], [], [], 0.25)
+        if readable:
+            line = process.stdout.readline()
+            if line:
+                lines.append(line)
+                if line.startswith("READY "):
+                    return lines
+        if process.poll() is not None:
+            break
+
+    process.kill()
+    remainder, _ = process.communicate()
+    raise AssertionError("pacer did not become ready:\n" + "".join(lines) + remainder)
+
+
+def receive_frames(
+    stop: threading.Event,
+    ready: threading.Event,
+    times: list[float],
+    errors: list[Exception],
+) -> None:
+    receiver: socket.socket | None = None
+    try:
+        receiver = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETHERTYPE))
+        receiver.bind((RX_INTERFACE, 0))
+        receiver.settimeout(0.1)
+        ready.set()
+        while not stop.is_set():
+            try:
+                frame = receiver.recv(2048)
+            except TimeoutError:
+                continue
+            if len(frame) >= 18 and frame[12:14] == struct.pack("!H", ETHERTYPE):
+                if frame[14:18] == MAGIC:
+                    times.append(time.monotonic())
+    except Exception as error:
+        errors.append(error)
+        ready.set()
+    finally:
+        if receiver is not None:
+            receiver.close()
+
+
+def send_burst() -> int:
+    sender = socket.socket(socket.AF_PACKET, socket.SOCK_RAW)
+    sender.bind((TX_INTERFACE, 0))
+    send_errors = 0
+    try:
+        for sequence in range(ATTEMPTS):
+            header = b"\xff" * 6 + b"\x02\x00\x00\x00\x00\x01" + struct.pack("!H", ETHERTYPE)
+            payload = MAGIC + struct.pack("!I", sequence)
+            frame = header + payload + b"x" * (FRAME_SIZE - len(header) - len(payload))
+            try:
+                sender.send(frame)
+            except OSError as error:
+                if error.errno != errno.ENOBUFS:
+                    raise
+                send_errors += 1
+    finally:
+        sender.close()
+    return send_errors
+
+
+def parse_summary(output: str) -> dict[str, int]:
+    match = SUMMARY.search(output)
+    if not match:
+        raise AssertionError("missing summary:\n" + output)
+    names = (
+        "enqueued",
+        "dequeued",
+        "policy_dropped",
+        "cleanup_dropped",
+        "bytes_dequeued",
+        "max_qlen",
+    )
+    return dict(zip(names, map(int, match.groups())))
+
+
+def main() -> int:
+    binary = os.path.abspath(sys.argv[1] if len(sys.argv) > 1 else "./egress_pacer")
+    stop = threading.Event()
+    receiver: threading.Thread | None = None
+    process: subprocess.Popen[str] | None = None
+
+    invalid = run(binary, "--interface", "definitely_missing", "--duration", "1", check=False)
+    assert invalid.returncode != 0
+    assert "interface does not exist" in invalid.stdout
+
+    create_test_link()
+    try:
+        run("tc", "qdisc", "add", "dev", TX_INTERFACE, "root", "handle", "1:",
+            "pfifo", "limit", "16")
+        conflict = run(
+            binary,
+            "--interface", TX_INTERFACE,
+            "--rate-kbps", str(RATE_KBPS),
+            "--queue-limit", str(QUEUE_LIMIT),
+            "--duration", "1",
+            check=False,
+        )
+        assert conflict.returncode != 0, conflict.stdout
+        assert "refusing to replace" in conflict.stdout, conflict.stdout
+        run("tc", "qdisc", "del", "dev", TX_INTERFACE, "root")
+
+        receiver_ready = threading.Event()
+        receive_times: list[float] = []
+        receiver_errors: list[Exception] = []
+        receiver = threading.Thread(
+            target=receive_frames,
+            args=(stop, receiver_ready, receive_times, receiver_errors),
+        )
+        receiver.start()
+        assert receiver_ready.wait(timeout=2)
+        assert not receiver_errors, receiver_errors
+
+        process = subprocess.Popen(
+            [
+                binary,
+                "--interface", TX_INTERFACE,
+                "--rate-kbps", str(RATE_KBPS),
+                "--queue-limit", str(QUEUE_LIMIT),
+                "--duration", "2",
+            ],
+            text=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            bufsize=1,
+        )
+        lines = wait_until_ready(process)
+        send_errors = send_burst()
+        remainder, _ = process.communicate(timeout=8)
+        output = "".join(lines) + remainder
+        stop.set()
+        receiver.join(timeout=2)
+        assert not receiver.is_alive()
+        assert not receiver_errors, receiver_errors
+        assert process.returncode == 0, output
+
+        stats = parse_summary(output)
+        assert stats["enqueued"] + stats["policy_dropped"] == ATTEMPTS, stats
+        assert stats["policy_dropped"] > 0, stats
+        assert stats["dequeued"] == stats["enqueued"], stats
+        assert stats["cleanup_dropped"] == 0, stats
+        assert stats["max_qlen"] <= QUEUE_LIMIT, stats
+        assert stats["max_qlen"] >= 2, stats
+        assert len(receive_times) == stats["dequeued"], (len(receive_times), stats)
+        assert len(receive_times) >= 5, receive_times
+
+        span = receive_times[-1] - receive_times[0]
+        assert span >= 0.4, span
+        observed_kbps = (len(receive_times) - 1) * FRAME_SIZE * 8 / span / 1000
+        assert observed_kbps <= RATE_KBPS * 1.5, observed_kbps
+
+        qdisc_after = run("tc", "qdisc", "show", "dev", TX_INTERFACE).stdout
+        assert "bpf_pacer" not in qdisc_after, qdisc_after
+
+        process = subprocess.Popen(
+            [
+                binary,
+                "--interface", TX_INTERFACE,
+                "--rate-kbps", str(RATE_KBPS),
+                "--queue-limit", str(QUEUE_LIMIT),
+                "--duration", "30",
+            ],
+            text=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            bufsize=1,
+        )
+        signal_lines = wait_until_ready(process)
+        process.terminate()
+        signal_remainder, _ = process.communicate(timeout=4)
+        signal_output = "".join(signal_lines) + signal_remainder
+        assert process.returncode == 0, signal_output
+        parse_summary(signal_output)
+        qdisc_after_signal = run("tc", "qdisc", "show", "dev", TX_INTERFACE).stdout
+        assert "bpf_pacer" not in qdisc_after_signal, qdisc_after_signal
+
+        process = subprocess.Popen(
+            [
+                binary,
+                "--interface", TX_INTERFACE,
+                "--rate-kbps", str(RATE_KBPS),
+                "--queue-limit", str(QUEUE_LIMIT),
+                "--duration", "30",
+            ],
+            text=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            bufsize=1,
+        )
+        wait_until_ready(process)
+        process.kill()
+        process.communicate(timeout=4)
+        assert process.returncode == -signal.SIGKILL, process.returncode
+        qdisc_after_kill = run("tc", "qdisc", "show", "dev", TX_INTERFACE).stdout
+        assert "bpf_pacer" in qdisc_after_kill, qdisc_after_kill
+        run("tc", "qdisc", "del", "dev", TX_INTERFACE, "root")
+        qdisc_after_recovery = run("tc", "qdisc", "show", "dev", TX_INTERFACE).stdout
+        assert "bpf_pacer" not in qdisc_after_recovery, qdisc_after_recovery
+
+        ready_line = next(line for line in output.splitlines() if line.startswith("READY "))
+        summary_line = next(line for line in output.splitlines() if line.startswith("SUMMARY "))
+        print(ready_line)
+        print(summary_line)
+        print(
+            f"TEST-SUMMARY attempts={ATTEMPTS} received={len(receive_times)} "
+            f"send_errors={send_errors} span_ms={span * 1000:.0f} "
+            f"observed_kbps={observed_kbps:.1f}"
+        )
+        print(
+            "PASS: conflict refusal, bounded drops, pacing, accounting, "
+            "normal/signal cleanup, and SIGKILL recovery succeeded"
+        )
+    finally:
+        if process is not None and process.poll() is None:
+            process.terminate()
+            try:
+                process.communicate(timeout=2)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.communicate()
+        stop.set()
+        if receiver is not None:
+            receiver.join(timeout=2)
+        delete_test_link()
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())