Fail adaptive DPDK mode when RX notifications are unavailable

CMakeLists.txt

@@ -444,6 +444,19 @@ endif()
 
 if (BYCORF_BUILD_TESTS)
   if (BYCORF_KERNEL_BYPASS)
+    add_executable(bycorf_dpdk_adaptive_check tests/dpdk_adaptive_check.cpp)
+    target_link_libraries(bycorf_dpdk_adaptive_check PRIVATE bycorf::core)
+    target_include_directories(bycorf_dpdk_adaptive_check PRIVATE
+      "${BYCORF_DPDK_PREFIX}/include")
+    target_compile_options(bycorf_dpdk_adaptive_check PRIVATE
+      ${BYCORF_DPDK_COMPILE_FLAGS})
+    foreach(symbol IN ITEMS rte_eth_dev_configure rte_eth_dev_info_get
+        rte_eth_dev_rx_intr_ctl_q_get_fd rte_eth_dev_rx_intr_enable
+        rte_eth_dev_rx_intr_disable)
+      target_link_options(bycorf_dpdk_adaptive_check PRIVATE
+        "LINKER:--wrap=${symbol}")
+    endforeach()
+
     # Explicit hardware paths/allowlists are required; do not run in ctest.
     add_executable(bycorf_spdk_backpressure_check tests/spdk_backpressure_check.cpp)
     target_link_libraries(bycorf_spdk_backpressure_check PRIVATE bycorf::core)
@@ -472,6 +485,13 @@ if (BYCORF_BUILD_TESTS)
   set_tests_properties(bycorf_task_transfer_check PROPERTIES
     TIMEOUT 30 LABELS "runtime;task")
   if (BYCORF_KERNEL_BYPASS)
+    foreach(scenario IN ITEMS configure descriptor enable disable poll
+        runtime-enable runtime-disable mlx5-empty)
+      add_test(NAME bycorf_dpdk_adaptive_${scenario}
+        COMMAND bycorf_dpdk_adaptive_check ${scenario})
+      set_tests_properties(bycorf_dpdk_adaptive_${scenario} PROPERTIES
+        TIMEOUT 30 LABELS "runtime;net;dpdk;stub" RUN_SERIAL TRUE)
+    endforeach()
     add_test(NAME bycorf_spdk_backpressure_stub_check
       COMMAND bycorf_spdk_backpressure_stub_check)
     set_tests_properties(bycorf_spdk_backpressure_stub_check PROPERTIES

docs/architecture/networking.md

@@ -144,10 +144,16 @@ for a timer tick. RX interrupts are disabled on returning to polling.
 
 A TAP virtual device provides a readable packet descriptor instead of a
 physical interrupt eventfd. The notification code preserves that distinction:
-reading an eventfd acknowledgement from TAP would consume a packet. If the PMD
-cannot provide the required notification capability, its queue owner keeps
-polling. Virtual-device tests cover this wake protocol but do not validate
-physical NIC interrupt behavior or predict kernel-bypass throughput. The TAP
+reading an eventfd acknowledgement from TAP would consume a packet. The mlx5
+PMD instead owns acknowledgement of its RDMA event channel; an empty channel
+after a timer or mailbox wake does not indicate lost interrupt support. If the
+PMD cannot provide the required notification capability on every RX queue,
+adaptive startup fails before services start. Configuration, notification
+registration, and interrupt-arm failures never silently select polling; the
+error identifies an explicit `BYCORF_DPDK_MODE=poll` override. A notification
+failure during a wait terminates the worker loop and stops peer workers before
+teardown barriers, producing a nonzero server exit. Virtual-device tests cover
+this wake protocol but do not validate physical NIC interrupt behavior or predict kernel-bypass throughput. The TAP
 PMD additionally uses Linux realtime signals for its internal RX trigger in
 both modes; this is separate from Bycorf's worker sleep protocol.
 

docs/architecture/runtime.md

@@ -57,7 +57,9 @@ while Run and FinalizeWorker execute only on selected workers. During shutdown,
 workers first leave their loops, then quiesce I/O and reclaim coroutine frames
 on their owning native threads. Service finalizers run after all worker frames
 are reclaimed. A worker that fails initialization contributes to both teardown
-barriers so initialized peers can finish. Services outlive the server.
+barriers so initialized peers can finish. An unexpected backend wait failure
+stops peer workers before entering those barriers and yields a nonzero exit.
+Services outlive the server.
 
 With DPDK networking, runtime preparation configures the shared port before
 launch. Worker 0 initializes the native FreeBSD kernel services; remaining

docs/dpdk-prototype.md

@@ -183,7 +183,7 @@ owning process closes. No physical NIC needs rebinding for these tests.
 
 | Setting | Default | Meaning |
 |---|---|---|
-| `BYCORF_DPDK_MODE` | `poll` | `poll` keeps polling; `adaptive` can arm RX notification and sleep on io_uring |
+| `BYCORF_DPDK_MODE` | `poll` | `poll` keeps polling; `adaptive` requires RX notifications and sleeps on io_uring when idle; unsupported queues fail startup |
 | `BYCORF_DPDK_QUEUES` | worker count, capped by hardware | RX/TX pair count, between 1 and worker count |
 | `BYCORF_DPDK_RX_STEERING` | `hash` | `hash` redistributes TCP by software tuple hash; `rss` keeps it on the receiving queue's worker |
 | `BYCORF_DPDK_IP` | `198.18.0.2` | Stack's IPv4 address |
@@ -192,6 +192,12 @@ owning process closes. No physical NIC needs rebinding for these tests.
 | `BYCORF_DPDK_MEMORY_MB` | 512 for no-huge tests | EAL memory size in MiB |
 | `BYCORF_EAL_ARGS` | virtual TAP configuration | Extra arguments passed through SPDK to EAL |
 
+Adaptive mode never falls back to polling. Startup verifies notification descriptors
+and interrupt enable/disable support for every RX queue. If the PMD cannot
+provide them, the error requests an explicit `BYCORF_DPDK_MODE=poll` setting.
+Notification failures while running also stop the server instead of silently
+changing modes.
+
 An unset/empty `BYCORF_EAL_ARGS` selects no hugepages, no PCI probing, and the
 TAP test device. Explicit arguments replace this virtual default; configure
 hugepages and the dedicated device allowlist for a physical run. Include the

include/bycorf/io/dpdk_backend.h

@@ -33,6 +33,8 @@ class DpdkBackend {
   // port and bounded software queues using BYCORF_DPDK_* environment settings.
   // Rejects counts exceeding build capacity or available EAL registrations
   // before configuring the port. RSS additionally requires one pair per worker.
+  // Adaptive mode requires usable RX notifications on every queue; unsupported
+  // devices fail startup rather than silently selecting poll mode.
   static absl::Status PrepareRuntime(unsigned workers);
   // Startup-only port placement. Every segment is steered to the same member
   // of this service's worker set; non-listening control workers are excluded.

src/io/dpdk_backend.cpp

@@ -106,6 +106,7 @@ struct alignas(64) Lane {
   rte_ring* tx = nullptr;
   std::atomic<bool> parked{false};
   int ring_fd = -1;
+  int rx_notification_fd = -1;
 };
 struct Fabric {
   std::mutex mutex;
@@ -118,7 +119,7 @@ struct Fabric {
   std::uint16_t port = RTE_MAX_ETHPORTS;
   rte_mempool* pool = nullptr;
   bycorf_bsd_interface interface{};
-  bool adaptive = false, tap = false, rss_owner = false;
+  bool adaptive = false, tap = false, mlx5 = false, rss_owner = false;
   // Populated before workers start, immutable throughout their lifetime.
   absl::flat_hash_map<std::uint16_t, std::vector<unsigned>> listener_workers;
   // A source port retains its owner through TIME_WAIT and is reused only by
@@ -191,10 +192,24 @@ void StopPort() {
     fabric.pool = nullptr;
   }
 }
+int DisableRxInterrupt(unsigned queue) {
+  const int rc = rte_eth_dev_rx_intr_disable(fabric.port, queue);
+  // mlx5 uses disable to consume a one-shot CQ event. A timer/mailbox wake,
+  // or the startup capability probe, can precede that event; EAGAIN means
+  // there was nothing to acknowledge, not that interrupts are unsupported.
+  return fabric.mlx5 && rc == -EAGAIN ? 0 : rc;
+}
 absl::Status DeviceError(const char* operation, int rc) {
   return absl::FailedPreconditionError(std::string(operation) + ": " +
                                        rte_strerror(rc < 0 ? -rc : rc));
 }
+absl::Status AdaptiveError(const char* operation, int rc) {
+  const auto status = DeviceError(operation, rc);
+  return absl::FailedPreconditionError(
+      std::string(status.message()) +
+      "; adaptive mode requires RX notification support; "
+      "set BYCORF_DPDK_MODE=poll explicitly to use polling");
+}
 }  // namespace
 
 class DpdkBackend::Impl {
@@ -233,14 +248,19 @@ class DpdkBackend::Impl {
     void Complete(Worker&, int result, unsigned) override {
       armed = false;
       if (result < 0) {
+        if (result == -ECANCELED && owner->worker->stop_requested()) return;
         owner->rx_can_wait = false;
+        std::fprintf(stderr, "DPDK worker %u: %s\n", owner->id,
+                     AdaptiveError("RX notification completion", result)
+                         .ToString()
+                         .c_str());
         return;
       }
       ++owner->notifications;
-      // TAP's notification descriptor IS its packet stream. Reading an
-      // eventfd word from it would consume/corrupt a packet. Hardware eventfds
-      // are acknowledged here, then RX interrupts are disabled before polling.
-      if (!fabric.tap) {
+      // TAP exposes packets, and mlx5 exposes an RDMA event channel whose
+      // records must be consumed by the PMD's interrupt-disable callback.
+      // Only ordinary hardware eventfds accept a uint64_t acknowledgement.
+      if (!fabric.tap && !fabric.mlx5) {
         std::uint64_t value;
         (void)::read(fd, &value, sizeof(value));
       }
@@ -641,6 +661,9 @@ absl::Status DpdkBackend::PrepareRuntime(unsigned workers) {
   if (rc) return DeviceError("ethdev info", rc);
   fabric.tap =
       info.driver_name && std::string_view(info.driver_name) == "net_tap";
+  const std::string_view driver = info.driver_name ? info.driver_name : "";
+  fabric.mlx5 = driver == "mlx5_pci" || driver == "mlx5_auxiliary" ||
+                driver == "net_mlx5";
   unsigned requested = workers;
   if (const char* queues = std::getenv("BYCORF_DPDK_QUEUES")) {
     char* end = nullptr;
@@ -669,16 +692,10 @@ absl::Status DpdkBackend::PrepareRuntime(unsigned workers) {
   }
   rc =
       rte_eth_dev_configure(fabric.port, fabric.queues, fabric.queues, &config);
-  if (rc && fabric.adaptive) {
-    config.intr_conf.rxq = 0;
-    fabric.adaptive = false;
-    std::fprintf(stderr, "DPDK RX interrupts unavailable; using polling\n");
-    rc = rte_eth_dev_configure(fabric.port, fabric.queues, fabric.queues,
-                               &config);
-  }
   if (rc) {
     StopPort();
-    return DeviceError("ethdev configure", rc);
+    return fabric.adaptive ? AdaptiveError("ethdev configure RX interrupts", rc)
+                           : DeviceError("ethdev configure", rc);
   }
   std::uint16_t rx_desc = 512, tx_desc = 512;
   rc = rte_eth_dev_adjust_nb_rx_tx_desc(fabric.port, &rx_desc, &tx_desc);
@@ -734,6 +751,26 @@ absl::Status DpdkBackend::PrepareRuntime(unsigned workers) {
     StopPort();
     return DeviceError("ethdev start", rc);
   }
+  // Validate every hardware queue before any worker or service starts. Some
+  // PMDs accept intr_conf.rxq but provide no usable notification descriptor.
+  if (fabric.adaptive) {
+    for (unsigned q = 0; q < fabric.queues; ++q) {
+      const int fd = rte_eth_dev_rx_intr_ctl_q_get_fd(fabric.port, q);
+      rc = fd < 0 ? -ENOTSUP : 0;
+      if (!rc && !fabric.tap) {
+        rc = rte_eth_dev_rx_intr_enable(fabric.port, q);
+        if (!rc) rc = DisableRxInterrupt(q);
+      }
+      if (rc) {
+        const std::string operation =
+            "RX queue " + std::to_string(q) + " interrupt validation";
+        const auto status = AdaptiveError(operation.c_str(), rc);
+        StopPort();
+        return status;
+      }
+      fabric.lanes[q].rx_notification_fd = fd;
+    }
+  }
   std::memcpy(fabric.interface.mac, mac.addr_bytes, 6);
   fabric.interface.mtu = 1500;
   fabric.prepared = true;
@@ -816,18 +853,9 @@ absl::Status DpdkBackend::Init(const IoBackendOptions& options, Worker* worker,
   }
   fabric.lanes[id].ring_fd = kernel_.WakeHandle();
   impl_->notification.owner = impl_.get();
-  impl_->rx_can_wait = id >= fabric.queues;
-  if (fabric.adaptive && id < fabric.queues) {
-    impl_->notification.fd = rte_eth_dev_rx_intr_ctl_q_get_fd(fabric.port, id);
-    if (impl_->notification.fd >= 0) {
-      impl_->rx_can_wait =
-          fabric.tap || rte_eth_dev_rx_intr_disable(fabric.port, id) == 0;
-    }
-    if (!impl_->rx_can_wait)
-      std::fprintf(stderr,
-                   "DPDK worker %u RX queue lacks interrupt support; polling\n",
-                   id);
-  }
+  impl_->rx_can_wait = true;
+  if (fabric.adaptive && id < fabric.queues)
+    impl_->notification.fd = fabric.lanes[id].rx_notification_fd;
   impl_->initialized = true;
   fabric.ready.fetch_add(1, std::memory_order_release);
   return absl::OkStatus();
@@ -870,20 +898,24 @@ absl::Status DpdkBackend::Submit() {
   return kernel_.Submit();
 }
 bool DpdkBackend::Wait(int timeout_ms) {
-  if (!fabric.adaptive || !impl_->rx_can_wait ||
+  if (!fabric.adaptive ||
       fabric.ready.load(std::memory_order_acquire) != fabric.workers)
     return true;
+  if (!impl_->rx_can_wait) return false;
   const unsigned id = impl_->id;
   Lane& lane = fabric.lanes[id];
   lane.parked.store(true, std::memory_order_release);
   bool enabled = false;
   if (id < fabric.queues) {
     if (!fabric.tap) {
-      enabled = rte_eth_dev_rx_intr_enable(fabric.port, id) == 0;
+      const int rc = rte_eth_dev_rx_intr_enable(fabric.port, id);
+      enabled = rc == 0;
       if (!enabled) {
         lane.parked.store(false);
-        impl_->rx_can_wait = false;
-        return true;
+        std::fprintf(
+            stderr, "DPDK worker %u: %s\n", id,
+            AdaptiveError("enable RX interrupt", rc).ToString().c_str());
+        return false;
       }
     }
     ++impl_->arms;
@@ -892,8 +924,12 @@ bool DpdkBackend::Wait(int timeout_ms) {
                                                 &impl_->notification);
       if (!submitted.ok()) {
         lane.parked.store(false);
-        if (enabled) rte_eth_dev_rx_intr_disable(fabric.port, id);
-        return true;
+        if (enabled) DisableRxInterrupt(id);
+        std::fprintf(stderr,
+                     "DPDK worker %u: RX notification registration: %s; "
+                     "set BYCORF_DPDK_MODE=poll explicitly to use polling\n",
+                     id, submitted.ToString().c_str());
+        return false;
       }
       impl_->notification.armed = true;
     }
@@ -926,8 +962,16 @@ bool DpdkBackend::Wait(int timeout_ms) {
     }
   }
   lane.parked.store(false, std::memory_order_release);
-  if (enabled) rte_eth_dev_rx_intr_disable(fabric.port, id);
-  return ok;
+  if (enabled) {
+    const int rc = DisableRxInterrupt(id);
+    if (rc) {
+      std::fprintf(
+          stderr, "DPDK worker %u: %s\n", id,
+          AdaptiveError("disable RX interrupt", rc).ToString().c_str());
+      return false;
+    }
+  }
+  return ok && impl_->rx_can_wait;
 }
 
 absl::Status DpdkBackend::SubmitSend(const RegisteredFile& file,

src/net/server.cpp

@@ -152,6 +152,10 @@ int Server::RunWorker(unsigned index, Worker& worker) {
   }
 
   worker.Run();
+  // A backend wait failure must stop peers before entering the all-worker
+  // barriers. Preserve the error before RequestStop marks this worker too.
+  const bool failed = !worker.stop_requested();
+  if (failed) runtime_.RequestStop();
 
   // Two-phase shutdown: wait until every worker has left its event loop, so
   // no cross-core reference into another worker's coroutine frames can still
@@ -183,10 +187,7 @@ int Server::RunWorker(unsigned index, Worker& worker) {
     if ((*service)->RunsOnWorker(index)) (*service)->FinalizeWorker(worker);
   }
 
-  if (!worker.stop_requested()) [[unlikely]] {
-    return 1;
-  }
-  return 0;
+  return failed ? 1 : 0;
 }
 
 }  // namespace bycorf