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); } 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.