From 593592af93ddb3ddba9c26a15ca3d4ba5b68727f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Leo=20L=C3=A4nnenm=C3=A4ki?= Date: Mon, 20 Jul 2026 14:53:47 +0300 Subject: [PATCH 1/2] test: Add stalled client test The test fills its own receive buffer by sending pings to the gateway without reading the replies, and expects the helper to report dropped packets and to keep forwarding once the client reads again. It fails on current code: when sendmsg_x() or write() fail with ENOBUFS, write_to_vm() retries forever, so a client that stops reading from the socket blocks the host->vm forwarding queue indefinitely, delaying every flow behind it. A helper blocked this way cannot even stop: on SIGTERM the atexit handler stops the vmnet interface on the same blocked queue, so the helper never exits. The test reads the backlog before stopping the helper to avoid hanging on failure. --- testing/helper_test.py | 84 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 84 insertions(+) diff --git a/testing/helper_test.py b/testing/helper_test.py index 1baf046c..b8290e36 100644 --- a/testing/helper_test.py +++ b/testing/helper_test.py @@ -31,6 +31,7 @@ from . import helper from . import mac +from . import store from .helper import ( NET_IPV4_MASK, NET_IPV4_SUBNET, @@ -335,6 +336,68 @@ def test_partial_dhcp_range(self): retry(ping_any, h, sock, gateway_mac, external_ips) +class TestStalledClient: + """ + Test that a client that stops reading from the socket does not block + host->vm forwarding. + """ + + def test_drop_when_client_stops_reading(self): + """ + Fill our receive buffer by sending pings to the gateway without + reading the replies. The helper's writes fail with ENOBUFS since + we are not reading, and the helper must drop the packets that do + not fit and keep forwarding, instead of retrying forever and + delaying every host->vm packet until we read again. + """ + with run_helper(operation_mode="shared") as (h, sock): + gateway_mac = arp_resolve(h, sock) + gateway_ip = find_gateway_ip(h.interface) + my_mac = h.interface[VMNET_MAC_ADDRESS] + my_ip = find_my_ip(h.interface) + + # Shrink our receive buffer so a short burst of replies + # overflows it. + sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096) + + request = ( + Ether(dst=gateway_mac, src=my_mac) + / IP(src=my_ip, dst=gateway_ip) + / ICMP(type=ICMP_ECHO_REQUEST) + / (b"x" * 1000) + ) + + # Send pings without reading the replies until the helper + # reports dropped packets. The timeout is generous: the helper + # gives up after 5 milliseconds with a full buffer. + try: + message = "peer is not reading from the socket" + deadline = time.monotonic() + 10.0 + while message not in helper_log(VM_NAME): + if time.monotonic() > deadline: + raise AssertionError( + "No dropped packets warning: helper is blocked " + "writing to a stalled client" + ) + for _ in range(20): + try: + sock.send(bytes(request)) + except OSError: + # Sending can also fail with ENOBUFS when the + # helper is not reading fast enough. + break + time.sleep(0.1) + finally: + # Read the backlog: a helper blocked in write_to_vm() + # cannot stop while our receive buffer is full, hanging + # the helper stop() on failure. + sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4 * 1024 * 1024) + drain(sock) + + # Forwarding must recover when we read again. + retry(ping, h, sock, gateway_mac, gateway_ip) + + # On macOS >= 26 we can test also the --network option. if MACOS_26: @@ -567,6 +630,27 @@ def ping_any(h, sock, gateway_mac, ips): # --- Utilities --- +def helper_log(vm_name): + """ + Read the helper log for a vm. + """ + path = store.vm_path(vm_name, "vmnet-helper.log") + with open(path) as f: + return f.read() + + +def drain(sock, timeout=0.1): + """ + Read packets until the socket is empty. + """ + sock.settimeout(timeout) + while True: + try: + sock.recv(65535) + except socket.timeout: + return + + def find_gateway_ip(interface): """ Return gateway IP, supporting both modes. From f624ce8a971fc9050b78d741fcfffd94a8f9c7b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Leo=20L=C3=A4nnenm=C3=A4ki?= Date: Mon, 20 Jul 2026 14:53:47 +0300 Subject: [PATCH 2/2] helper: Drop packets when the peer stops reading from the socket write_to_vm() retried ENOBUFS forever, 50 microseconds at a time. The vm socket is a datagram socket, so ENOBUFS means the peer receive buffer is full: the client (vfkit, krunkit) is not draining the socket as fast as the host sends. Since all host->vm forwarding runs on one serial queue, one slow or stalled client blocked every host->vm packet behind it with no bound, delaying unrelated flows and latency sensitive frames such as TCP acks and HTTP/2 pings. Real clients hit this: a guest that drains slowly during large concurrent downloads can hold the queue full for seconds, and HTTP/2 clients in the guest time out their connection health pings and drop the connection with all its streams. Bound the wait to VM_MAX_RETRIES (5 ms), shared by the whole batch. When the budget is exhausted every remaining packet gets one last write() before being dropped, since a smaller packet may still fit in the peer receive buffer. Dropping is safe: ethernet does not guarantee delivery, the guest transport protocols recover from the loss, and TCP backs off, which keeps the queue short. Dropped packets are counted and logged at most once per second. Example: WARN [host->vm] dropped 11 packets (11 total): peer is not reading from the socket --- programs/helper.c | 64 +++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 56 insertions(+), 8 deletions(-) diff --git a/programs/helper.c b/programs/helper.c index 392650dd..88cfe092 100644 --- a/programs/helper.c +++ b/programs/helper.c @@ -51,6 +51,15 @@ // 13 1 | #define VM_RETRY_DELAY (50 * MICROSECOND) +// If the peer stops reading from the socket, waiting longer cannot help; give +// up and drop the packets like a congested physical network. The bound is the +// total wait per batch: 100 retries (5 ms) is far above the retry counts seen +// under load, and far below the stalls that make guest connections time out. +// All host->vm forwarding runs on one serial queue, so one unbounded wait +// delays every flow, including latency sensitive frames such as TCP acks and +// HTTP/2 pings. +#define VM_MAX_RETRIES 100 + #define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0])) static const uintptr_t SHUTDOWN_EVENT = 1; @@ -153,6 +162,11 @@ static bool use_bulk_forwarding = true; // error of creating multiple helpers using the same socket. static char *socket_lockfile; +// Total packets dropped because the peer receive buffer was full, and the +// last time we warned about it. Modified only on the host queue. +static uint64_t vm_dropped_packets; +static time_t vm_dropped_warn_time; + static const char *host_strerror(vmnet_return_t v) { switch (v) { @@ -826,6 +840,24 @@ static inline void wait_for_buffer_space(void) nanosleep(&t, NULL); } +// Count dropped packets, warning at most once per second so a stalled peer +// cannot flood the log. +static void count_dropped_packets(int dropped) +{ + struct timespec now; + + vm_dropped_packets += dropped; + + clock_gettime(CLOCK_MONOTONIC, &now); + if (now.tv_sec == vm_dropped_warn_time) { + return; + } + vm_dropped_warn_time = now.tv_sec; + + WARNF("[host->vm] dropped %d packets (%llu total): peer is not reading from the socket", + dropped, vm_dropped_packets); +} + static void write_to_vm(int count) { for (int i = 0; i < count; i++) { @@ -834,15 +866,23 @@ static void write_to_vm(int count) int sent = 0; + // The ENOBUFS wait budget is shared by the whole batch so a stalled peer + // delays forwarding by at most VM_MAX_RETRIES * VM_RETRY_DELAY. + uint64_t retries = 0; + // Fast path. if (use_bulk_forwarding) { - uint64_t retries = 0; - while (1) { ssize_t n = sendmsg_x(options.fd, &host.msgs[sent], count-sent, 0); if (n == -1) { if (errno == ENOBUFS) { + if (retries == VM_MAX_RETRIES) { + // The slow path gives every remaining packet one last + // write() before dropping it; a smaller packet may + // still fit in the peer receive buffer. + break; + } wait_for_buffer_space(); retries++; continue; @@ -865,16 +905,19 @@ static void write_to_vm(int count) size_t size = host_packets_size(sent); int dropped = 0; + int enobufs_dropped = 0; for (int i = sent; i < count; i++) { struct vmpktdesc *packet = &host.packets[i]; ssize_t len; - uint64_t retries = 0; while (1) { len = write(options.fd, packet->vm_pkt_iov[0].iov_base, packet->vm_pkt_size); if (len == -1 && errno == ENOBUFS) { + if (retries == VM_MAX_RETRIES) { + break; + } wait_for_buffer_space(); retries++; continue; @@ -883,22 +926,27 @@ static void write_to_vm(int count) } if (len < 0) { - // TODO: like socket_vmnet we drop the packet and continue. Maybe trigger shutdown? - ERRORF("[host->vm] write: %s", strerror(errno)); + if (errno == ENOBUFS) { + enobufs_dropped++; + } else { + // TODO: like socket_vmnet we drop the packet and continue. Maybe trigger shutdown? + ERRORF("[host->vm] write: %s", strerror(errno)); + } dropped++; continue; } sent++; size += packet->vm_pkt_size; - if (retries > 0) { - DEBUGF("[host->vm] write completed after %lld retries", retries); - } // Partial write should not be possible with datagram socket. assert((size_t)len == packet->vm_pkt_size); } + if (enobufs_dropped > 0) { + count_dropped_packets(enobufs_dropped); + } + DEBUGF("[host->vm] forwarded %d packets %zu bytes %d dropped", sent, size, dropped); }