diff --git a/.github/workflows/test-libbpf.yml b/.github/workflows/test-libbpf.yml index c90df60..a6fdea0 100644 --- a/.github/workflows/test-libbpf.yml +++ b/.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 diff --git a/src/53-egress-pacer/.config b/src/53-egress-pacer/.config new file mode 100644 index 0000000..ad3749c --- /dev/null +++ b/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 diff --git a/src/53-egress-pacer/.gitignore b/src/53-egress-pacer/.gitignore new file mode 100644 index 0000000..9f08618 --- /dev/null +++ b/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/ diff --git a/src/53-egress-pacer/Makefile b/src/53-egress-pacer/Makefile new file mode 100644 index 0000000..60e2e19 --- /dev/null +++ b/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 - &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: diff --git a/src/53-egress-pacer/bpf_experimental.h b/src/53-egress-pacer/bpf_experimental.h new file mode 100644 index 0000000..f1c066f --- /dev/null +++ b/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 + +#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 */ diff --git a/src/53-egress-pacer/egress_pacer.bpf.c b/src/53-egress-pacer/egress_pacer.bpf.c new file mode 100644 index 0000000..e4f52c4 --- /dev/null +++ b/src/53-egress-pacer/egress_pacer.bpf.c @@ -0,0 +1,197 @@ +// SPDX-License-Identifier: GPL-2.0 +#include "vmlinux.h" +#include +#include +#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", +}; diff --git a/src/53-egress-pacer/egress_pacer.c b/src/53-egress-pacer/egress_pacer.c new file mode 100644 index 0000000..1148b63 --- /dev/null +++ b/src/53-egress-pacer/egress_pacer.c @@ -0,0 +1,269 @@ +// SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause) +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#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, ×tamp)) + 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; +} diff --git a/src/53-egress-pacer/egress_pacer.h b/src/53-egress-pacer/egress_pacer.h new file mode 100644 index 0000000..a5fe056 --- /dev/null +++ b/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 */ diff --git a/src/53-egress-pacer/tests/test_egress_pacer.py b/src/53-egress-pacer/tests/test_egress_pacer.py new file mode 100644 index 0000000..98c0884 --- /dev/null +++ b/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()) diff --git a/src/third_party/bpftool b/src/third_party/bpftool index 3be8ac3..4222ef1 160000 --- a/src/third_party/bpftool +++ b/src/third_party/bpftool @@ -1 +1 @@ -Subproject commit 3be8ac358900c5f57028b2b6501407a14167a9bb +Subproject commit 4222ef1c8cdb0a1ea03c7a8c650ec334b782902d