helper: Drop packets when the peer stops reading from the socket

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

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.