perf(dpdk): copy BSD packets directly into transmit mbufs

docs/architecture/networking.md

@@ -127,7 +127,10 @@ and outgoing IPv4 TCP use BSD; Unix sockets retain the kernel path. IPv6 DPDK
 listeners and clients are unsupported.
 
 The prototype copies MTU-sized frames between DPDK and BSD and copies received
-bytes into per-connection buffers. It uses software checksums. Software queues
+bytes into per-connection buffers. On output, the host synchronously copies a
+borrowed BSD packet into host-owned storage before the stack releases its mbuf;
+DPDK uses its final packet buffer as that destination. Only host-owned packets
+cross worker queues. It uses software checksums. Software queues
 are bounded: overflow and partial device TX free unaccepted packets, leaving
 native TCP responsible for retransmission. This is a functional integration
 boundary, not a zero-copy or tuned packet allocator.

src/io/dpdk_backend.cpp

@@ -357,18 +357,9 @@ class DpdkBackend::Impl {
       rte_pktmbuf_free(packet);
     }
   }
-  static int Transmit(void* context, const void* data, std::size_t length) {
+  static int Transmit(void* context, const bycorf_bsd_packet* source) {
     auto& self = *static_cast<Impl*>(context);
-    const auto* bytes = static_cast<const unsigned char*>(data);
-    // All VNETs learn broadcast control traffic, but only worker 0 responds to
-    // ARP requests or echo requests for the shared address. ARP requests made
-    // by any worker remain necessary to resolve its outgoing neighbour cache.
-    if (self.id && length >= 42 && Read16(bytes + 12) == 0x0806 &&
-        Read16(bytes + 20) == 2)
-      return 0;
-    if (self.id && length >= 35 && Read16(bytes + 12) == 0x0800 &&
-        bytes[23] == 1 && bytes[34] == 0)
-      return 0;
+    const std::size_t length = source->length;
     if (length > kMaxFrame) {
       ++self.drops;
       return 40;
@@ -384,7 +375,18 @@ class DpdkBackend::Impl {
       ++self.drops;
       return 55;
     }
-    std::memcpy(output, data, length);
+    source->copy(source->context, output);
+    const auto* bytes = static_cast<const unsigned char*>(output);
+    // All VNETs learn broadcast control traffic, but only worker 0 responds to
+    // ARP requests or echo requests for the shared address. ARP requests made
+    // by any worker remain necessary to resolve its outgoing neighbour cache.
+    if (self.id && ((length >= 42 && Read16(bytes + 12) == 0x0806 &&
+                     Read16(bytes + 20) == 2) ||
+                    (length >= 35 && Read16(bytes + 12) == 0x0800 &&
+                     bytes[23] == 1 && bytes[34] == 0))) {
+      rte_pktmbuf_free(packet);
+      return 0;
+    }
     if (length >= 34 && Read16(bytes + 12) == 0x0800 &&
         std::memcmp(bytes + 30, &fabric.interface.address, 4) == 0) {
       // Loopback is still native BSD TCP. Defer even same-worker delivery:

src/io/freebsd/abi.h

@@ -29,9 +29,7 @@ extern "C" {
 #endif
 
 // The only boundary between the private FreeBSD namespace and the Linux host.
-// Callbacks stay alive until all worker stacks have stopped. Packet callbacks
-// must copy/retain bytes before returning: FreeBSD reclaims their mbuf
-// afterward.
+// Callbacks stay alive until all worker stacks have stopped.
 struct bycorf_bsd_host {
   void* (*allocate)(size_t size, size_t alignment);
   void (*release)(void* pointer);
@@ -42,13 +40,26 @@ struct bycorf_bsd_host {
   void (*abort_process)(void);
 };
 
+// Borrowed only until the transmit callback returns, on the owning worker.
+// copy writes exactly length bytes into a host-owned destination with at least
+// that capacity. Neither this descriptor nor context may be retained or queued;
+// FreeBSD reclaims the source mbuf after transmit, including on failure.
+struct bycorf_bsd_packet {
+  const void* context;
+  size_t length;
+  void (*copy)(const void* context, void* destination);
+};
+
 struct bycorf_bsd_interface {
   uint32_t address;
   uint32_t netmask;
   uint32_t gateway;
   unsigned char mac[6];
   uint16_t mtu;
-  int (*transmit)(void* context, const void* bytes, size_t length);
+  // Copy synchronously into host-owned storage before queueing a packet.
+  // Returns zero on success or a positive FreeBSD errno on failure. The host
+  // owns its destination storage and must release it on any rejected send.
+  int (*transmit)(void* context, const struct bycorf_bsd_packet* packet);
   void* context;
 };
 

src/io/freebsd/interface.c

@@ -32,17 +32,20 @@
 static _Thread_local if_t interface;
 static _Thread_local struct bycorf_bsd_interface config;
 
+static void copy_packet(const void* context, void* destination) {
+  const struct mbuf* m = context;
+  m_copydata(m, 0, m->m_pkthdr.len, destination);
+}
+
 static int transmit(if_t ifp, struct mbuf* m) {
   struct bycorf_bsd_interface* cfg = if_getsoftc(ifp);
-  // The first version deliberately uses copied, MTU-sized frames and software
-  // checksums. No BSD mbuf or stack-local pointer crosses to a queue owner.
-  unsigned char frame[ETHER_MAX_LEN];
-  size_t n = m->m_pkthdr.len;
+  // Let the host copy the chain directly into its final packet buffer. The
+  // borrow ends before m_freem; only host-owned bytes can cross a worker queue.
+  const struct bycorf_bsd_packet packet = {
+      .context = m, .length = m->m_pkthdr.len, .copy = copy_packet};
   int error = EMSGSIZE;
-  if (n <= sizeof(frame)) {
-    m_copydata(m, 0, n, frame);
-    error = cfg->transmit(cfg->context, frame, n);
-  }
+  if (packet.length <= ETHER_MAX_LEN)
+    error = cfg->transmit(cfg->context, &packet);
   if_inc_counter(ifp, error ? IFCOUNTER_OERRORS : IFCOUNTER_OPACKETS, 1);
   m_freem(m);
   return error;

tests/freebsd_local_connect_check.cpp

@@ -12,13 +12,16 @@
 #include <cstdlib>
 #include <cstring>
 #include <deque>
+#include <utility>
 #include <vector>
 
 #include "../src/io/freebsd/abi.h"
 
 namespace {
 std::deque<std::vector<unsigned char>> packets;
 unsigned wire_packets = 0;
+enum class Reject { kNone, kBeforeCopy, kAfterCopy };
+Reject reject_next = Reject::kNone;
 constexpr uint32_t kAddress = 0xc6130002;
 
 void Check(bool ok, const char* what, int error = 0) {
@@ -47,7 +50,7 @@ void Exchange(struct socket* source, struct socket* target) {
   std::array<unsigned char, 32769> expected{}, actual{};
   for (size_t i = 0; i < expected.size(); ++i) expected[i] = i * 37;
   size_t sent = 0, received = 0;
-  const auto deadline = Now(CLOCK_MONOTONIC) + 2000000000;
+  const auto deadline = Now(CLOCK_MONOTONIC) + 5000000000;
   while (received < actual.size() && Now(CLOCK_MONOTONIC) < deadline) {
     size_t n = 0;
     if (sent < expected.size()) {
@@ -98,12 +101,20 @@ int main() {
   interface.mac[0] = 2;
   interface.mac[5] = 2;
   interface.mtu = 1500;
-  interface.transmit = [](void*, const void* data, size_t n) {
-    auto* bytes = static_cast<const unsigned char*>(data);
+  interface.transmit = [](void*, const bycorf_bsd_packet* packet) {
+    const auto reject = std::exchange(reject_next, Reject::kNone);
+    // A host allocation failure may reject the borrow without copying it.
+    if (reject == Reject::kBeforeCopy) return 55;  // BSD ENOBUFS
+    std::vector<unsigned char> frame(packet->length);
+    packet->copy(packet->context, frame.data());
+    // A full output queue rejects an already copied packet. Neither failure
+    // may retain the BSD borrow; TCP must recover using its own send buffer.
+    if (reject == Reject::kAfterCopy) return 55;
+    const auto* bytes = frame.data();
     const uint32_t destination = htonl(kAddress);
-    if (n >= 34 && bytes[12] == 8 && bytes[13] == 0 &&
+    if (frame.size() >= 34 && bytes[12] == 8 && bytes[13] == 0 &&
         std::memcmp(bytes + 30, &destination, 4) == 0)
-      packets.emplace_back(bytes, bytes + n);
+      packets.emplace_back(std::move(frame));
     else
       ++wire_packets;
     return 0;
@@ -130,6 +141,12 @@ int main() {
   Check(bycorf_bsd_accept(listener, &accepted) == 0, "local accept");
   Exchange(client, accepted);
   Exchange(accepted, client);
+  reject_next = Reject::kBeforeCopy;
+  Exchange(client, accepted);
+  Check(reject_next == Reject::kNone, "allocation rejection exercised");
+  reject_next = Reject::kAfterCopy;
+  Exchange(accepted, client);
+  Check(reject_next == Reject::kNone, "queue rejection exercised");
   Check(bycorf_bsd_close(accepted) == 0, "close accepted");
   Pump();
   unsigned char byte;

tests/freebsd_port_check.cpp

@@ -91,15 +91,18 @@ struct Lane {
     interface.mac[5] = 2;
     interface.mtu = 1500;
     interface.context = this;
-    interface.transmit = [](void* context, const void* bytes, size_t length) {
+    interface.transmit = [](void* context, const bycorf_bsd_packet* packet) {
       auto* lane = static_cast<Lane*>(context);
+      std::array<unsigned char, 1518> frame;
+      Check(packet->length <= frame.size(), "output frame fits");
+      packet->copy(packet->context, frame.data());
       ssize_t n;
       do {
-        n = write(lane->fd, bytes, length);
+        n = write(lane->fd, frame.data(), packet->length);
       } while (n < 0 && errno == EINTR);
       // A full TAP queue may drop a frame, as a bounded NIC queue does. TCP
       // must recover; never retain an mbuf's borrowed bytes across callbacks.
-      return n == static_cast<ssize_t>(length) ? 0 : 55;  // BSD ENOBUFS
+      return n == static_cast<ssize_t>(packet->length) ? 0 : 55;  // BSD ENOBUFS
     };
     int error = bycorf_bsd_attach_interface(&interface);
     Check(error == 0, "attach interface", error);

tests/freebsd_tcp_sequence_check.cpp

@@ -90,9 +90,13 @@ struct Peer {
   uint64_t received = 0;
   bool syn_received = false;
 
-  static int Transmit(void* context, const void* data, size_t size) {
+  static int Transmit(void* context, const bycorf_bsd_packet* packet) {
     auto& peer = *static_cast<Peer*>(context);
-    const auto* p = static_cast<const uint8_t*>(data);
+    std::array<uint8_t, 1518> frame;
+    const size_t size = packet->length;
+    Check(size <= frame.size(), "output frame fits");
+    packet->copy(packet->context, frame.data());
+    const auto* p = frame.data();
     if (size < 54 || Get16(p + 12) != 0x0800 || p[23] != 6) return 0;
     const size_t ip_header = (p[14] & 15) * 4;
     const uint8_t* tcp = p + 14 + ip_header;