From 945e9b4475007d99e6471d816d0843a286e18177 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Sat, 10 Oct 2026 13:38:06 +0800 Subject: [PATCH] fix(dpdk): require RX notifications in adaptive mode --- CMakeLists.txt | 20 +++++ docs/architecture/networking.md | 14 ++- docs/architecture/runtime.md | 4 +- docs/dpdk-prototype.md | 8 +- include/bycorf/io/dpdk_backend.h | 2 + src/io/dpdk_backend.cpp | 110 ++++++++++++++++------- src/net/server.cpp | 9 +- tests/dpdk_adaptive_check.cpp | 147 +++++++++++++++++++++++++++++++ 8 files changed, 271 insertions(+), 43 deletions(-) create mode 100644 tests/dpdk_adaptive_check.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 2970445..7364d9d 100644 --- a/CMakeLists.txt +++ b/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 diff --git a/docs/architecture/networking.md b/docs/architecture/networking.md index f90838e..4fca28c 100644 --- a/docs/architecture/networking.md +++ b/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. diff --git a/docs/architecture/runtime.md b/docs/architecture/runtime.md index 94e8070..a6b5062 100644 --- a/docs/architecture/runtime.md +++ b/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 diff --git a/docs/dpdk-prototype.md b/docs/dpdk-prototype.md index c38c055..c22a529 100644 --- a/docs/dpdk-prototype.md +++ b/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 diff --git a/include/bycorf/io/dpdk_backend.h b/include/bycorf/io/dpdk_backend.h index 48a6bb0..8b14a93 100644 --- a/include/bycorf/io/dpdk_backend.h +++ b/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. diff --git a/src/io/dpdk_backend.cpp b/src/io/dpdk_backend.cpp index a67e22e..75c7f31 100644 --- a/src/io/dpdk_backend.cpp +++ b/src/io/dpdk_backend.cpp @@ -106,6 +106,7 @@ struct alignas(64) Lane { rte_ring* tx = nullptr; std::atomic 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> 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, diff --git a/src/net/server.cpp b/src/net/server.cpp index ab9c5d2..56100c4 100644 --- a/src/net/server.cpp +++ b/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 diff --git a/tests/dpdk_adaptive_check.cpp b/tests/dpdk_adaptive_check.cpp new file mode 100644 index 0000000..79e1226 --- /dev/null +++ b/tests/dpdk_adaptive_check.cpp @@ -0,0 +1,147 @@ +/* + * Copyright (C) 2026 EloqData Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "bycorf/net/server.h" + +namespace { +std::string_view scenario; +int notification_fd = -1; +std::atomic enables{0}; +std::atomic disables{0}; +std::atomic configured{0}; +std::atomic injected{false}; + +class IdleService final : public bycorf::Service { + public: + void Prepare(unsigned) override {} + bycorf::Task Run(bycorf::Worker&, + bycorf::ServiceContext) override { + co_return absl::OkStatus(); + } + void Stop() noexcept override {} +}; +} // namespace + +// Keep fault injection outside production code. The ring PMD owns no physical +// NIC, and eventfd emulates the notification descriptor for runtime failures. +extern "C" int __real_rte_eth_dev_info_get(uint16_t, rte_eth_dev_info*); +extern "C" int __wrap_rte_eth_dev_info_get(uint16_t port, + rte_eth_dev_info* info) { + const int rc = __real_rte_eth_dev_info_get(port, info); + if (!rc && scenario == "mlx5-empty") info->driver_name = "mlx5_pci"; + return rc; +} +extern "C" int __real_rte_eth_dev_configure(uint16_t, uint16_t, uint16_t, + const rte_eth_conf*); +extern "C" int __wrap_rte_eth_dev_configure(uint16_t port, uint16_t rx, + uint16_t tx, + const rte_eth_conf* conf) { + ++configured; + if (scenario == "configure" && conf->intr_conf.rxq) { + injected = true; + return -ENOTSUP; + } + rte_eth_conf ring_conf = *conf; + ring_conf.intr_conf.rxq = 0; + return __real_rte_eth_dev_configure(port, rx, tx, &ring_conf); +} +extern "C" int __wrap_rte_eth_dev_rx_intr_ctl_q_get_fd(uint16_t, uint16_t) { + if (scenario == "descriptor") { + injected = true; + return -1; + } + return notification_fd; +} +extern "C" int __wrap_rte_eth_dev_rx_intr_enable(uint16_t, uint16_t) { + const unsigned count = ++enables; + if (scenario == "enable" || (scenario == "runtime-enable" && count > 1)) { + injected = true; + return -ENOTSUP; + } + return 0; +} +extern "C" int __wrap_rte_eth_dev_rx_intr_disable(uint16_t, uint16_t) { + const unsigned count = ++disables; + if (scenario == "mlx5-empty") { + injected = true; + return -EAGAIN; + } + if (scenario == "disable" || (scenario == "runtime-disable" && count > 1)) { + injected = true; + return -EIO; + } + return 0; +} + +int main(int argc, char** argv) { + if (argc != 2) return 2; + scenario = argv[1]; + const bool poll = scenario == "poll"; + const bool mlx5 = scenario == "mlx5-empty"; + const bool runtime = scenario.starts_with("runtime-"); + // Ignore inherited hardware configuration; each process gets its own EAL. + setenv("BYCORF_EAL_ARGS", "--no-pci --no-huge --vdev=net_ring0", 1); + setenv("BYCORF_DPDK_MODE", poll ? "poll" : "adaptive", 1); + setenv("BYCORF_DPDK_QUEUES", "1", 1); + setenv("BYCORF_DPDK_RX_STEERING", "hash", 1); + setenv("BYCORF_DPDK_IP", "198.18.0.2", 1); + setenv("BYCORF_DPDK_NETMASK", "255.255.255.0", 1); + unsetenv("BYCORF_DPDK_GATEWAY"); + setenv("BYCORF_DPDK_MEMORY_MB", "512", 1); + notification_fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); + if (notification_fd < 0 || + !bycorf::ConfigureIoBackends({.dpdk_network = true}).ok()) + return 2; + IdleService service; + bycorf::Server server; + server.AddService(&service); + const auto status = server.Start({.thread_count_ = 2, .pin_workers_ = false}); + if (!poll && !runtime && !mlx5) { + if (status.ok() || !injected || configured != 1 || + status.message().find("BYCORF_DPDK_MODE=poll") == + std::string_view::npos) { + std::cerr << "expected actionable startup rejection: " << status << '\n'; + return 1; + } + std::cout << status << '\n'; + } else { + if (!status.ok()) { + std::cerr << status << '\n'; + return 1; + } + if (mlx5) std::this_thread::sleep_for(std::chrono::milliseconds(100)); + if (poll || mlx5) server.RequestStop(); + // A CTest timeout catches peers stranded at the all-worker barriers. + server.WaitUntilStopped(); + if (poll && (server.exit_code() != 0 || enables != 0)) return 1; + if (mlx5 && (server.exit_code() != 0 || enables < 2 || !injected)) return 1; + if (runtime && (server.exit_code() == 0 || !injected)) return 1; + } + close(notification_fd); + std::cout << "PASS " << scenario << '\n'; +}